feat(activitywatch): add ActivityWatch client with native fallback

catser/activitywatch.py

@@ -0,0 +1,178 @@
+"""
+ActivityWatch Client and Window Tracker for Catser.
+
+Integrates with the ActivityWatch backend (https://activitywatch.net/) running
+locally at http://localhost:5600/api/0.
+
+Features:
+- Automatic discovery of 'aw-watcher-window_*' buckets.
+- Polling of active window events: executable name, window title, and timestamp.
+- Evaluation against distraction/drift content rules (Shorts, Reels, TikTok, etc.).
+- Seamless fallback to native Win32 window APIs when ActivityWatch is offline,
+  ensuring Catser remains fully functional under all conditions.
+"""
+
+import time
+import json
+import logging
+import urllib.request
+import urllib.error
+from dataclasses import dataclass
+from typing import Optional, Dict, Any, List
+
+from .config import Config
+from .window_manager import WindowManager, WindowInfo
+
+logger = logging.getLogger("catser.aw")
+
+
+@dataclass
+class DistractionEvent:
+    """Represents a detected distraction target window."""
+    app: str
+    title: str
+    matched_rule: str
+    window_info: WindowInfo
+    timestamp: float = 0.0
+
+    def __post_init__(self):
+        if self.timestamp == 0.0:
+            self.timestamp = time.time()
+
+
+class ActivityWatchClient:
+    """
+    Client interface for ActivityWatch API with automatic native Win32 fallback.
+    """
+
+    def __init__(self, config: Config):
+        self.config = config
+        self.base_url = config.aw_url.rstrip("/")
+        self.api_url = f"{self.base_url}/api/0"
+        self.bucket_id: Optional[str] = None
+        self.is_connected: bool = False
+        self.last_connect_attempt: float = 0.0
+        self.connect_retry_delay: float = 5.0  # Retry connecting every 5s
+
+    def check_connection(self) -> bool:
+        """Pings the ActivityWatch /api/0/info endpoint."""
+        now = time.time()
+        if now - self.last_connect_attempt < self.connect_retry_delay and not self.is_connected:
+            return False
+
+        self.last_connect_attempt = now
+        try:
+            req = urllib.request.Request(f"{self.api_url}/info")
+            with urllib.request.urlopen(req, timeout=1.5) as resp:
+                if resp.status == 200:
+                    data = json.loads(resp.read().decode("utf-8"))
+                    logger.info(f"Connected to ActivityWatch server v{data.get('version', 'unknown')}")
+                    self.is_connected = True
+                    self._discover_window_bucket()
+                    return True
+        except Exception:
+            if self.is_connected:
+                logger.warning("ActivityWatch server disconnected. Falling back to native tracking.")
+            self.is_connected = False
+            return False
+
+        return False
+
+    def _discover_window_bucket(self) -> Optional[str]:
+        """Discovers the active 'aw-watcher-window_*' bucket."""
+        try:
+            req = urllib.request.Request(f"{self.api_url}/buckets")
+            with urllib.request.urlopen(req, timeout=2.0) as resp:
+                if resp.status == 200:
+                    buckets = json.loads(resp.read().decode("utf-8"))
+                    for b_id in buckets.keys():
+                        if b_id.startswith("aw-watcher-window"):
+                            self.bucket_id = b_id
+                            logger.info(f"Using ActivityWatch window bucket: {self.bucket_id}")
+                            return self.bucket_id
+        except Exception as e:
+            logger.warning(f"Error querying ActivityWatch buckets: {e}")
+        return None
+
+    def get_current_window_event(self) -> Optional[Dict[str, Any]]:
+        """
+        Fetches the latest window event from ActivityWatch.
+        Returns dict with 'app' and 'title', or None.
+        """
+        if not self.is_connected or not self.bucket_id:
+            if not self.check_connection() or not self.bucket_id:
+                return None
+
+        try:
+            url = f"{self.api_url}/buckets/{self.bucket_id}/events?limit=1"
+            req = urllib.request.Request(url)
+            with urllib.request.urlopen(req, timeout=1.5) as resp:
+                if resp.status == 200:
+                    events = json.loads(resp.read().decode("utf-8"))
+                    if events and isinstance(events, list) and len(events) > 0:
+                        ev_data = events[0].get("data", {})
+                        return {
+                            "app": ev_data.get("app", ""),
+                            "title": ev_data.get("title", ""),
+                            "timestamp": events[0].get("timestamp"),
+                        }
+        except Exception as e:
+            logger.debug(f"ActivityWatch query error: {e}")
+            self.is_connected = False
+
+        return None
+
+    def check_for_distraction(self) -> Optional[DistractionEvent]:
+        """
+        Checks current active window against distraction rules.
+        Uses ActivityWatch if available, otherwise falls back to native Win32.
+        """
+        app_name = ""
+        window_title = ""
+        active_window_info: Optional[WindowInfo] = None
+
+        # 1. Try ActivityWatch
+        aw_event = self.get_current_window_event()
+        if aw_event:
+            app_name = (aw_event.get("app") or "").lower()
+            window_title = aw_event.get("title") or ""
+
+            # Locate corresponding Win32 HWND for screen coordinates and close button
+            active_window_info = WindowManager.find_window_by_match(
+                title_predicate=lambda t: window_title.lower() in t.lower() or t.lower() in window_title.lower(),
+                process_predicate=lambda p: app_name in p.lower(),
+            )
+
+        # 2. Fallback to native Win32 if ActivityWatch is unavailable or HWND not found
+        if not active_window_info and self.config.auto_fallback_to_win32:
+            active_window_info = WindowManager.get_foreground_window_info()
+            if active_window_info:
+                app_name = active_window_info.process_name.lower()
+                window_title = active_window_info.title
+
+        if not active_window_info or not window_title:
+            return None
+
+        title_lower = window_title.lower()
+
+        # Check for title keyword matches (e.g. "shorts", "reels", "tiktok")
+        for kw in self.config.distraction_keywords:
+            if kw.lower() in title_lower:
+                return DistractionEvent(
+                    app=app_name,
+                    title=window_title,
+                    matched_rule=f"keyword '{kw}'",
+                    window_info=active_window_info,
+                )
+
+        # Check for application process matches
+        for target_app in self.config.distraction_apps:
+            if target_app.lower() in app_name:
+                return DistractionEvent(
+                    app=app_name,
+                    title=window_title,
+                    matched_rule=f"app '{target_app}'",
+                    window_info=active_window_info,
+                )
+
+        return None