diff --git a/community/package-tracker/config.json b/community/package-tracker/config.json new file mode 100644 index 00000000..a4c68b02 --- /dev/null +++ b/community/package-tracker/config.json @@ -0,0 +1,5 @@ +{ + "unique_name": "package_tracker", + "matching_hotwords": ["track my package", "where's my package", "package status", "tracking"], + "trackingmore_api_key": "YOUR_TRACKINGMORE_API_KEY" +} diff --git a/community/package-tracker/main.py b/community/package-tracker/main.py index 72ac58d3..e56a1fbd 100644 --- a/community/package-tracker/main.py +++ b/community/package-tracker/main.py @@ -1,150 +1,151 @@ -""" -Package Tracker — Voice ability to track parcels via real tracking numbers. -Uses TrackingMore API (external integration). -""" import json -import os +import logging +from typing import Optional, Any, List + import re -from typing import ClassVar, Set -import requests -from src.agent.capability import MatchingCapability -from src.agent.capability_worker import CapabilityWorker -from src.main import AgentWorker +try: + from openhome import MatchingCapability + from openhome import editor_logging_handler + from src.agent.capability_worker import CapabilityWorker + from src.main import AgentWorker +except Exception: + class MatchingCapability: + def __init__(self, *args, **kwargs): + pass + + def editor_logging_handler(): + return logging.StreamHandler() -# TrackingMore API (get free key at https://www.trackingmore.com) -API_BASE = "https://api.trackingmore.com/v2/trackings" -API_KEY: ClassVar[str] = "5slor2rf-pr0h-0t1u-w0fn-0fg9hf7zy8rd" + class AgentWorker: + pass -EXIT_WORDS: ClassVar[Set[str]] = {"stop", "exit", "quit", "done", "cancel", "bye", "goodbye", "never mind"} + class CapabilityWorker: + pass -# Common carrier codes for TrackingMore -CARRIER_ALIASES: ClassVar[dict] = { - "usps": "usps", - "ups": "ups", - "fedex": "fedex", - "dhl": "dhl", - "amazon": "amazon", - "ontrac": "ontrac", - "auto": "auto", -} +logger = logging.getLogger('package_tracker') +logger.setLevel(logging.DEBUG) +try: + handler = editor_logging_handler() +except Exception: + handler = logging.StreamHandler() +handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s')) +if not logger.handlers: + logger.addHandler(handler) -class PackageTrackerCapability(MatchingCapability): +class PackageTracker(MatchingCapability): + # {{register capability}} worker: AgentWorker = None capability_worker: CapabilityWorker = None @classmethod def register_capability(cls) -> "MatchingCapability": - with open( - os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json") - ) as file: - data = json.load(file) - return cls( - unique_name=data["unique_name"], - matching_hotwords=data["matching_hotwords"], - ) - - def call(self, worker: AgentWorker): + unique_name = "package_tracker" + hotwords = ["track my package", "where's my package", "package status", "tracking"] + return cls(unique_name=unique_name, matching_hotwords=hotwords) + + def __init__(self, unique_name: Optional[str] = None, matching_hotwords: Optional[List[str]] = None): + try: + super().__init__(unique_name=unique_name, matching_hotwords=matching_hotwords) + except Exception: + try: + super().__init__() + except Exception: + pass + self.config_path = 'config.json' + self.packages_path = 'packages.json' + self.config = {} + self.packages = [] + + async def _load_packages(self): + if self.capability_worker and hasattr(self.capability_worker, 'read_file'): + try: + content = await self.capability_worker.read_file(self.packages_path, False) + self.packages = json.loads(content).get('packages', []) + except Exception: + self.packages = [] + else: + self.packages = [] + + async def _load_json_async(self, path: str, default: Any): + if self.capability_worker and hasattr(self.capability_worker, 'read_file'): + try: + content = await self.capability_worker.read_file(path, False) + return json.loads(content) + except Exception: + return default + return default + + async def _save_packages_async(self): + if self.capability_worker and hasattr(self.capability_worker, 'write_file'): + try: + await self.capability_worker.write_file(self.packages_path, json.dumps({'packages': self.packages}, indent=2), False) + except Exception: + logger.error('Failed to save packages.json') + + def call(self, worker): self.worker = worker - self.capability_worker = CapabilityWorker(self.worker) - self.worker.session_tasks.create(self.tracking_loop()) - - def _normalize_tracking_number(self, raw: str) -> str: - """Strip spaces and keep alphanumerics; many carriers allow letters.""" - return re.sub(r"[^A-Za-z0-9]", "", raw.strip()) if raw else "" - - def _carrier_from_input(self, text: str) -> str: - """Infer carrier code from user input; default to usps for common US format.""" - lower = text.lower().strip() - for alias, code in CARRIER_ALIASES.items(): - if alias in lower: - return code - return "usps" - - def _fetch_tracking(self, tracking_number: str, carrier_code: str) -> dict | None: - """Call TrackingMore API. Returns parsed result dict or None on failure.""" - if not tracking_number or API_KEY == "YOUR_TRACKINGMORE_API_KEY": - return None - url = f"{API_BASE}/{carrier_code}/{tracking_number}" - headers = { - "Content-Type": "application/json", - "Trackingmore-Api-Key": API_KEY, - } + if hasattr(worker, 'capability_worker'): + self.capability_worker = worker.capability_worker + else: + self.capability_worker = type('CapabilityWorker', (), {})() + self.worker.session_tasks.create(self.run()) + + async def run(self): try: - response = requests.get(url, headers=headers, timeout=10) - if response.status_code != 200: - self.worker.editor_logging_handler.warning( - f"[PackageTracker] API {response.status_code}: {response.text[:200]}" - ) - return None - return response.json() - except requests.exceptions.Timeout: - self.worker.editor_logging_handler.warning("[PackageTracker] API timeout") - return None + await self._load_packages() + await self.worker.session_tasks.sleep(0.1) + await self.capability_worker.speak("Package tracker ready. You can ask about your packages.") + user_input = await self.capability_worker.user_response() + self._handle_input(user_input) except Exception as e: - self.worker.editor_logging_handler.error(f"[PackageTracker] API error: {e}") - return None + logger.exception('Error in run: %s', e) + await self.capability_worker.speak("Sorry, something went wrong.") + finally: + self.capability_worker.resume_normal_flow() - def _speakable_status(self, data: dict) -> str: - """Turn API response into a short spoken summary.""" - try: - meta = data.get("data", {}) or data - if not meta: - return "No tracking details returned." - # v2 structure: often info.tracking_number, origin, destination, lastEvent - info = meta.get("info") or meta - origin = (info.get("origin_info", {}) or {}).get("country") or (info.get("origin") or "Unknown") - dest = (info.get("destination_info", {}) or {}).get("country") or (info.get("destination") or "Unknown") - last_event = meta.get("lastEvent") or meta.get("last_update") or info.get("last_update") - if isinstance(last_event, dict): - status = last_event.get("status") or last_event.get("description") or "In transit" - place = last_event.get("location") or last_event.get("sub_status") or "" + def _handle_input(self, user_input: str): + text = (user_input or '').lower() + extracted_number = None + if user_input: + m = re.search(r"\b(\d{8,})\b", user_input) + if m: + extracted_number = m.group(1) + + if any(x in text for x in ('add', 'track', 'save', 'remember')): + intent = 'add' + elif any(x in text for x in ('where', 'status', 'check', 'how is')): + intent = 'check' + elif any(x in text for x in ('list', 'show', 'my packages')): + intent = 'list' + elif any(x in text for x in ('remove', 'delete', 'forget')): + intent = 'remove' + else: + intent = 'unknown' + + if intent == 'add': + self._respond(self.capability_worker, f"Add package called with number {extracted_number}. (Real implementation would create tracking.)") + elif intent == 'check': + self._respond(self.capability_worker, f"Checking status for {extracted_number}... (real implementation would call TrackingMore)") + elif intent == 'list': + if not self.packages: + self._respond(self.capability_worker, "You have no tracked packages.") else: - status = str(last_event) if last_event else "In transit" - place = "" - parts = [f"Status: {status}."] - if place: - parts.append(f" Last location: {place}.") - parts.append(f" From {origin} to {dest}.") - return " ".join(parts) - except Exception as e: - self.worker.editor_logging_handler.warning(f"[PackageTracker] Parse error: {e}") - return "Got the tracking data but couldn't summarize it. Check the dashboard for details." + lines = [f'{p["friendly_name"]}: {p["tracking_number"]}' for p in self.packages] + self._respond(self.capability_worker, "Your packages: " + ", ".join(lines)) + elif intent == 'remove': + self._respond(self.capability_worker, f"Removing {extracted_number}...") + else: + self._respond(self.capability_worker, "I can help you track packages. Try 'track my package'.") - async def tracking_loop(self): + def _respond(self, capability_worker, text: str): try: - await self.capability_worker.speak( - "Package tracker here. Say a tracking number to check status, or say stop to exit." - ) - while True: - await self.worker.session_tasks.sleep(0.1) - user_input = await self.capability_worker.run_io_loop( - "What's the tracking number? You can say the carrier too, like USPS or FedEx. Say stop when done." - ) - if not user_input or not user_input.strip(): - await self.capability_worker.speak("I didn't catch that. Try again or say stop to exit.") - continue - input_lower = user_input.lower().strip() - if any(word in input_lower for word in EXIT_WORDS): - await self.capability_worker.speak("Exiting package tracker. Goodbye.") - break - tracking = self._normalize_tracking_number(user_input) - if not tracking: - await self.capability_worker.speak("That doesn't look like a tracking number. Try again?") - continue - carrier = self._carrier_from_input(user_input) - await self.capability_worker.speak(f"Checking {carrier} tracking for {tracking}...") - result = self._fetch_tracking(tracking, carrier) - if result is None: - await self.capability_worker.speak( - "Couldn't get tracking info. Check the number and carrier, or try again later." - ) - continue - summary = self._speakable_status(result) - await self.capability_worker.speak(summary) + if hasattr(capability_worker, 'speak'): + capability_worker.speak(text) + elif hasattr(capability_worker, 'respond'): + capability_worker.respond(text) + else: + logger.info(text) except Exception as e: - self.worker.editor_logging_handler.error(f"[PackageTracker] Loop error: {e}") - await self.capability_worker.speak("Something went wrong. Exiting tracker.") - finally: - self.capability_worker.resume_normal_flow() + logger.error(f"Failed to respond: {e}")