adapters.activitywatch¶

def normalize_app(app_name: str) -> AppInfo:
    """Map an AW application name to a reverse-domain ID, flags, and category.

    Performs a case-insensitive lookup in :data:`KNOWN_APPS`.  Unknown
    applications fall back to ``"unknown.<sanitized_name>"`` with all
    flags set to ``False`` and category ``"other"``.

    Args:
        app_name: Application name as reported by ActivityWatch
            (e.g. ``"Firefox"``).

    Returns:
        A ``(app_id, is_browser, is_editor, is_terminal, app_category)``
        tuple.
    """
    key = app_name.strip().lower()
    if key in KNOWN_APPS:
        return KNOWN_APPS[key]
    sanitized = key.replace(" ", "_").replace("/", "_")
    return (f"unknown.{sanitized}", False, False, False, "other")

def parse_aw_export(path: Path, *, title_salt: str) -> list[AWEvent]:
    """Parse an ActivityWatch JSON export file into normalized events.

    Filters for buckets of type ``currentwindow`` (i.e.
    ``aw-watcher-window`` data).  Each event's application name is
    normalized and its window title is replaced with a salted hash.

    Args:
        path: Path to the AW export JSON file.
        title_salt: Salt used for hashing window titles.

    Returns:
        Sorted (by timestamp) list of :class:`AWEvent` instances.

    Raises:
        FileNotFoundError: If *path* does not exist.
        KeyError: If the JSON structure is missing expected keys.
    """
    raw = json.loads(path.read_text(encoding="utf-8"))

    buckets: dict[str, Any] = raw.get("buckets", raw)

    events: list[AWEvent] = []
    for bucket_id, bucket in buckets.items():
        bucket_type = bucket.get("type", "")
        if bucket_type != _CURRENTWINDOW_TYPE:
            logger.debug("Skipping bucket %s (type=%s)", bucket_id, bucket_type)
            continue

        logger.info(
            "Processing bucket %s (%d events)",
            bucket_id,
            len(bucket.get("events", [])),
        )
        for raw_event in bucket.get("events", []):
            events.append(_raw_event_to_aw_event(raw_event, title_salt=title_salt))

    events.sort(key=lambda e: e.timestamp)
    return events

def parse_aw_input_export(path: Path) -> list[AWInputEvent]:
    """Parse ``aw-watcher-input`` events from an AW JSON export.

    Filters for buckets of type ``os.hid.input``.  These events carry
    only aggregate counts (key presses, mouse clicks, movement, scroll)
    and contain no sensitive payload.

    Args:
        path: Path to the AW export JSON file.

    Returns:
        Sorted (by timestamp) list of :class:`AWInputEvent` instances.
        Empty if no ``os.hid.input`` bucket exists in the export.
    """
    raw = json.loads(path.read_text(encoding="utf-8"))
    buckets: dict[str, Any] = raw.get("buckets", raw)

    events: list[AWInputEvent] = []
    for bucket_id, bucket in buckets.items():
        bucket_type = bucket.get("type", "")
        if bucket_type != _INPUT_TYPE:
            continue

        logger.info(
            "Processing input bucket %s (%d events)",
            bucket_id,
            len(bucket.get("events", [])),
        )
        for raw_event in bucket.get("events", []):
            events.append(_raw_to_input_event(raw_event))

    events.sort(key=lambda e: e.timestamp)
    return events

def list_aw_buckets(
    host: str,
    *,
    timeout: int = DEFAULT_AW_TIMEOUT_SECONDS,
) -> dict[str, dict]:
    """List all buckets from a running AW server.

    Args:
        host: Base URL of the AW server (e.g. ``"http://localhost:5600"``).
        timeout: Seconds to wait for a response.

    Returns:
        Dict mapping bucket IDs to their metadata.
    """
    url = f"{host.rstrip('/')}/api/0/buckets/"
    return _api_get(url, timeout=timeout)

def find_window_bucket_id(
    host: str,
    *,
    timeout: int = DEFAULT_AW_TIMEOUT_SECONDS,
) -> str:
    """Auto-discover the ``aw-watcher-window`` bucket on *host*.

    Args:
        host: Base URL of the AW server.
        timeout: Seconds to wait for a response.

    Returns:
        The bucket ID whose ``type`` is ``currentwindow``.

    Raises:
        ValueError: If no ``currentwindow`` bucket exists on the server.
    """
    buckets = list_aw_buckets(host, timeout=timeout)
    for bucket_id, meta in buckets.items():
        if meta.get("type") == _CURRENTWINDOW_TYPE:
            return bucket_id
    raise ValueError(
        f"No bucket with type={_CURRENTWINDOW_TYPE!r} found on {host}. "
        f"Available: {list(buckets.keys())}"
    )

def find_input_bucket_id(
    host: str,
    *,
    timeout: int = DEFAULT_AW_TIMEOUT_SECONDS,
) -> str | None:
    """Auto-discover the ``aw-watcher-input`` bucket on *host*.

    Unlike :func:`find_window_bucket_id`, this returns ``None`` when no
    input bucket exists because ``aw-watcher-input`` is an optional
    watcher that many users don't run.

    Args:
        host: Base URL of the AW server.
        timeout: Seconds to wait for a response.

    Returns:
        The bucket ID whose ``type`` is ``os.hid.input``, or ``None``.
    """
    buckets = list_aw_buckets(host, timeout=timeout)
    for bucket_id, meta in buckets.items():
        if meta.get("type") == _INPUT_TYPE:
            return bucket_id
    return None

def fetch_aw_events(
    host: str,
    bucket_id: str,
    start: datetime,
    end: datetime,
    *,
    title_salt: str,
    timeout: int = DEFAULT_AW_TIMEOUT_SECONDS,
) -> list[AWEvent]:
    """Fetch events from the AW REST API for a time range.

    Args:
        host: Base URL of the AW server (e.g. ``"http://localhost:5600"``).
        bucket_id: Bucket to query (e.g. ``"aw-watcher-window_myhostname"``).
        start: Inclusive start of the query window (UTC).
        end: Exclusive end of the query window (UTC).
        title_salt: Salt used for hashing window titles.
        timeout: Seconds to wait for a response.

    Returns:
        Sorted list of :class:`AWEvent` instances.
    """
    base = host.rstrip("/")
    start_iso = start.isoformat() + "Z" if start.tzinfo is None else start.isoformat()
    end_iso = end.isoformat() + "Z" if end.tzinfo is None else end.isoformat()

    qs = urllib.parse.urlencode(
        {"start": start_iso, "end": end_iso},
        safe=":",
        quote_via=urllib.parse.quote,
    )
    url = f"{base}/api/0/buckets/{bucket_id}/events?{qs}"
    raw_events: list[dict] = _api_get(url, timeout=timeout)

    events = [_raw_event_to_aw_event(e, title_salt=title_salt) for e in raw_events]
    events.sort(key=lambda e: e.timestamp)
    return events

def fetch_aw_input_events(
    host: str,
    bucket_id: str,
    start: datetime,
    end: datetime,
    *,
    timeout: int = DEFAULT_AW_TIMEOUT_SECONDS,
) -> list[AWInputEvent]:
    """Fetch input events from the AW REST API for a time range.

    Args:
        host: Base URL of the AW server.
        bucket_id: Input bucket to query (e.g.
            ``"aw-watcher-input_myhostname"``).
        start: Inclusive start of the query window (UTC).
        end: Exclusive end of the query window (UTC).
        timeout: Seconds to wait for a response.

    Returns:
        Sorted list of :class:`AWInputEvent` instances.
    """
    base = host.rstrip("/")
    start_iso = start.isoformat() + "Z" if start.tzinfo is None else start.isoformat()
    end_iso = end.isoformat() + "Z" if end.tzinfo is None else end.isoformat()

    qs = urllib.parse.urlencode(
        {"start": start_iso, "end": end_iso},
        safe=":",
        quote_via=urllib.parse.quote,
    )
    url = f"{base}/api/0/buckets/{bucket_id}/events?{qs}"
    raw_events: list[dict] = _api_get(url, timeout=timeout)

    events = [_raw_to_input_event(e) for e in raw_events]
    events.sort(key=lambda e: e.timestamp)
    return events