Package tracker

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"
+}

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}")