From ddf99b8664a995edded6a8788aad63543c9cabad Mon Sep 17 00:00:00 2001 From: max Date: Tue, 8 Sep 2026 23:04:44 +0200 Subject: [PATCH] feat(activitywatch): add ActivityWatch client with native fallback --- catser/activitywatch.py | 178 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 178 insertions(+) create mode 100644 catser/activitywatch.py diff --git a/catser/activitywatch.py b/catser/activitywatch.py new file mode 100644 index 0000000..c7165c9 --- /dev/null +++ b/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