diff --git a/focus_project/focus_project/sources/browser.py b/focus_project/focus_project/sources/browser.py new file mode 100644 index 0000000..c1b9e22 --- /dev/null +++ b/focus_project/focus_project/sources/browser.py @@ -0,0 +1,188 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Browser evidence: active window/URL time on the project's sites. + +Reads the kuhytrack activity server (read-only GETs, the same data the PC +watcher and the phone bridge write): ``currentwindow`` buckets give +``{app, title[, url]}`` spans, ``afkstatus`` buckets of the same host trim +them to time the user was present. A host without an AFK bucket (the phone) +is taken as present. + +Matching is case-insensitive. A ``site_patterns`` entry such as +``leetcode.com`` matches as-is in the url/title/app, or, when the window is +a browser, by its stem (``leetcode``) in the title: Chrome-family titles +carry the page name, not the host. Stems under four characters are ignored. + +Failure modes (named): no tracker config -> inactive, 0 minutes; tracker +unreachable or broken -> raises, so the registry logs it and counts 0; an +AFK bucket with no events that day -> 0 for that host (fail closed). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, date, datetime +import json +import logging +import os +from pathlib import Path +from typing import TYPE_CHECKING, Any +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +from focus_project.sources import Interval, day_bounds + +if TYPE_CHECKING: + from focus_project._model import Project + +_LOG = logging.getLogger(__name__) +# Long window events start before midnight; ask a little earlier. +LOOKBACK_S = 3600 +TIMEOUT_S = 5 +MIN_STEM = 4 +BROWSERS = ( + "helium", + "librewolf", + "firefox", + "chrom", + "brave", + "vivaldi", + "opera", + "msedge", + "zen", +) +_ENV_DEFAULT = "~/.config/kuhytrack/env" +_urlopen = urlopen + + +@dataclass(frozen=True) +class Config: + """Where the tracker listens and the bearer token (may be empty).""" + + url: str + token: str + + +def _config() -> Config | None: + """The tracker's own ``KT_URL``/``KT_TOKEN`` env file, or None if absent.""" + path = Path(os.environ.get("KUHYTRACK_ENV", _ENV_DEFAULT)).expanduser() + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError: + return None + values: dict[str, str] = {} + for line in lines: + key, sep, value = line.strip().partition("=") + if sep and not key.startswith("#"): + values[key.strip()] = value.strip().strip("'\"") + url = values.get("KT_URL", "") + return Config(url.rstrip("/"), values.get("KT_TOKEN", "")) if url else None + + +def _get_json(cfg: Config, path: str, params: dict[str, str] | None = None) -> Any: # noqa: ANN401 + """One authenticated GET returning parsed JSON.""" + url = cfg.url + path + (f"?{urlencode(params)}" if params else "") + request = Request(url) # noqa: S310 (operator-configured http URL) + if cfg.token: + request.add_header("Authorization", f"Bearer {cfg.token}") + with _urlopen(request, timeout=TIMEOUT_S) as response: + return json.load(response) + + +def _merge(spans: list[Interval]) -> list[Interval]: + """Sorted, non-overlapping union of ``spans``.""" + out: list[Interval] = [] + for start, end in sorted(spans): + if out and start <= out[-1][1]: + out[-1] = (out[-1][0], max(out[-1][1], end)) + else: + out.append((start, end)) + return out + + +def _intersect(left: list[Interval], right: list[Interval]) -> list[Interval]: + """Overlap of two merged span lists.""" + out: list[Interval] = [] + for a0, a1 in left: + for b0, b1 in right: + lo, hi = max(a0, b0), min(a1, b1) + if hi > lo: + out.append((lo, hi)) + return out + + +def _needles(patterns: tuple[str, ...]) -> tuple[list[str], list[str]]: + """``(full, stems)``: lowercase host patterns and their page-title stems.""" + full: list[str] = [] + stems: list[str] = [] + for raw in patterns: + pattern = raw.strip().lower().removeprefix("www.") + if not pattern: + continue + full.append(pattern) + stem = pattern.split(".", 1)[0] + if "." in pattern and len(stem) >= MIN_STEM: + stems.append(stem) + return full, stems + + +def _matches(data: dict[str, Any], full: list[str], stems: list[str]) -> bool: + fields = { + k: str(data.get(k, "")).lower() for k in ("url", "title", "app", "package") + } + text = " ".join(fields.values()).replace("www.", "") + if any(needle in text for needle in full): + return True + if any(b in fields["app"] for b in BROWSERS): + return any(stem in fields["title"] for stem in stems) + return False + + +def _epoch(stamp: str) -> float: + return datetime.fromisoformat(stamp).timestamp() + + +def _spans(events: list[dict[str, Any]]) -> list[Interval]: + out: list[Interval] = [] + for event in events: + start = _epoch(event["timestamp"]) + out.append((start, start + float(event.get("duration") or 0))) + return out + + +def intervals(day: date, project: Project) -> list[Interval]: + """Present time on the project's sites; see the module docstring.""" + full, stems = _needles(project.site_patterns) + if not full: + return [] + cfg = _config() + if cfg is None: + _LOG.info("browser source: no kuhytrack config (inactive)") + return [] + lo, hi = day_bounds(day) + stamp = "%Y-%m-%dT%H:%M:%S.000Z" + params = { + "start": datetime.fromtimestamp(lo - LOOKBACK_S, UTC).strftime(stamp), + "end": datetime.fromtimestamp(hi, UTC).strftime(stamp), + } + + def events(bucket: dict[str, Any]) -> list[dict[str, Any]]: + return list(_get_json(cfg, f"/api/0/buckets/{bucket['id']}/events", params)) + + buckets = list(_get_json(cfg, "/api/0/buckets").values()) + present: dict[str, list[Interval]] = {} + for bucket in buckets: + if bucket["type"] == "afkstatus": + live = [e for e in events(bucket) if e["data"].get("status") == "not-afk"] + present.setdefault(bucket["hostname"], []).extend(_spans(live)) + on_site: dict[str, list[Interval]] = {} + for bucket in buckets: + if bucket["type"] == "currentwindow": + hits = [e for e in events(bucket) if _matches(e["data"], full, stems)] + on_site.setdefault(bucket["hostname"], []).extend(_spans(hits)) + out: list[Interval] = [] + for host, spans in on_site.items(): + merged = _merge(spans) + out.extend( + _intersect(merged, _merge(present[host])) if host in present else merged + ) + return out diff --git a/focus_project/focus_project/tests/test_source_browser.py b/focus_project/focus_project/tests/test_source_browser.py new file mode 100644 index 0000000..0c92e93 --- /dev/null +++ b/focus_project/focus_project/tests/test_source_browser.py @@ -0,0 +1,208 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""The browser adapter: window/URL time on a project's sites, AFK-trimmed.""" + +from __future__ import annotations + +from datetime import UTC, date, datetime +from typing import Any +from urllib.error import URLError + +import pytest + +from focus_project import sources +from focus_project._model import Project +from focus_project.sources import browser + +DAY = date(2026, 11, 2) +LO, HI = sources.day_bounds(DAY) +CFG = browser.Config("http://tracker.test", "tok") + + +def _project(*patterns: str) -> Project: + return Project( + id="p", + name="P", + start=DAY, + end=DAY, + replaces=("leetcode",), + daily_minutes=120, + unlock_minutes=30, + site_patterns=patterns, + ) + + +def _iso(offset: float) -> str: + stamp = datetime.fromtimestamp(LO + offset, UTC) + return stamp.strftime("%Y-%m-%dT%H:%M:%S.000Z") + + +def _ev(offset: float, dur: float, **data: str) -> dict[str, Any]: + return {"timestamp": _iso(offset), "duration": dur, "data": data} + + +def _bucket(bid: str, kind: str, host: str) -> dict[str, Any]: + return {"id": bid, "type": kind, "hostname": host} + + +def _fake_tracker( + monkeypatch: pytest.MonkeyPatch, + buckets: list[dict[str, Any]], + events: dict[str, list[dict[str, Any]]], +) -> list[tuple[str, dict[str, str] | None]]: + calls: list[tuple[str, dict[str, str] | None]] = [] + + def fake( + _cfg: browser.Config, path: str, params: dict[str, str] | None = None + ) -> Any: # noqa: ANN401 + calls.append((path, params)) + if path == "/api/0/buckets": + return {b["id"]: b for b in buckets} + bid = path.split("/")[4] + return events.get(bid, []) + + monkeypatch.setattr(browser, "_get_json", fake) + monkeypatch.setattr(browser, "_config", lambda: CFG) + return calls + + +WIN = _bucket("win_pc", "currentwindow", "pc") +AFK = _bucket("afk_pc", "afkstatus", "pc") +ACTIVE_ALL = [_ev(0, 86000, status="not-afk")] + + +def test_title_stem_in_browser_counts(monkeypatch: pytest.MonkeyPatch) -> None: + _fake_tracker( + monkeypatch, + [WIN, AFK], + { + "win_pc": [ + _ev(100, 600, app="Helium", title="Two Sum - LeetCode - Helium") + ], + "afk_pc": ACTIVE_ALL, + }, + ) + assert browser.intervals(DAY, _project("leetcode.com")) == [(LO + 100, LO + 700)] + + +def test_stem_ignored_outside_browsers(monkeypatch: pytest.MonkeyPatch) -> None: + _fake_tracker( + monkeypatch, + [WIN, AFK], + { + "win_pc": [_ev(0, 600, app="Terminator", title="notes on leetcode")], + "afk_pc": ACTIVE_ALL, + }, + ) + assert browser.intervals(DAY, _project("leetcode.com")) == [] + + +def test_full_pattern_matches_any_app_and_url(monkeypatch: pytest.MonkeyPatch) -> None: + _fake_tracker( + monkeypatch, + [WIN], + { + "win_pc": [ + _ev(0, 60, app="x", title="t", url="https://HackerRank.com/a"), + _ev(100, 60, app="mpv", title="www.hackerrank.com talk"), + ] + }, + ) + got = browser.intervals(DAY, _project("hackerrank.com")) + assert got == [(LO, LO + 60), (LO + 100, LO + 160)] + + +def test_afk_time_is_cut(monkeypatch: pytest.MonkeyPatch) -> None: + _fake_tracker( + monkeypatch, + [WIN, AFK], + { + "win_pc": [_ev(0, 1000, app="Helium", title="LeetCode")], + "afk_pc": [ + _ev(100, 200, status="not-afk"), + _ev(500, 100, status="not-afk"), + ], + }, + ) + assert browser.intervals(DAY, _project("leetcode.com")) == [ + (LO + 100, LO + 300), + (LO + 500, LO + 600), + ] + + +def test_afk_bucket_without_events_fails_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _fake_tracker( + monkeypatch, + [WIN, AFK], + {"win_pc": [_ev(0, 1000, app="Helium", title="LeetCode")], "afk_pc": []}, + ) + assert browser.intervals(DAY, _project("leetcode.com")) == [] + + +def test_phone_bucket_has_no_afk_filter(monkeypatch: pytest.MonkeyPatch) -> None: + phone = _bucket("win_ph", "currentwindow", "pixel6a") + _fake_tracker( + monkeypatch, + [phone], + {"win_ph": [_ev(10, 90, app="Chrome", title="HackerRank")]}, + ) + assert browser.intervals(DAY, _project("hackerrank.com")) == [(LO + 10, LO + 100)] + + +def test_hosts_do_not_borrow_each_others_afk(monkeypatch: pytest.MonkeyPatch) -> None: + other = _bucket("afk_other", "afkstatus", "other") + _fake_tracker( + monkeypatch, + [WIN, AFK, other], + { + "win_pc": [_ev(0, 100, app="Helium", title="LeetCode")], + "afk_pc": ACTIVE_ALL, + "afk_other": [], + }, + ) + assert browser.intervals(DAY, _project("leetcode.com")) == [(LO, LO + 100)] + + +def test_no_patterns_no_calls(monkeypatch: pytest.MonkeyPatch) -> None: + calls = _fake_tracker(monkeypatch, [WIN], {}) + assert browser.intervals(DAY, _project()) == [] + assert calls == [] + + +def test_short_stem_needs_the_full_pattern(monkeypatch: pytest.MonkeyPatch) -> None: + _fake_tracker( + monkeypatch, + [WIN], + {"win_pc": [_ev(0, 60, app="Helium", title="go to the shop")]}, + ) + assert browser.intervals(DAY, _project("go.dev")) == [] + + +def test_query_is_bounded_to_the_day(monkeypatch: pytest.MonkeyPatch) -> None: + calls = _fake_tracker(monkeypatch, [WIN], {}) + browser.intervals(DAY, _project("leetcode.com")) + params = next(p for path, p in calls if path.endswith("/events")) + assert params is not None + assert params["start"] < params["end"] + + +def test_no_config_is_inactive(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(browser, "_config", lambda: None) + assert browser.intervals(DAY, _project("leetcode.com")) == [] + + +def test_unreachable_tracker_raises_for_the_registry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + reason = "down" + + def boom(*_a: object, **_k: object) -> None: + raise URLError(reason) + + monkeypatch.setattr(browser, "_get_json", boom) + monkeypatch.setattr(browser, "_config", lambda: CFG) + project = _project("leetcode.com") + with pytest.raises(URLError): + browser.intervals(DAY, project) + assert sources.collect_by_source(DAY, project)["browser"] == [] diff --git a/focus_project/focus_project/tests/test_source_browser_io.py b/focus_project/focus_project/tests/test_source_browser_io.py new file mode 100644 index 0000000..733ef02 --- /dev/null +++ b/focus_project/focus_project/tests/test_source_browser_io.py @@ -0,0 +1,81 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""The browser adapter's plumbing: env-file config, HTTP, span maths.""" + +from __future__ import annotations + +import io +from typing import TYPE_CHECKING, Any + +from focus_project.sources import browser + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + +CFG = browser.Config("http://tracker.test", "tok") + + +def test_config_from_env_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + env = tmp_path / "env" + env.write_text("# c\nKT_URL=http://h:1\nKT_TOKEN='abc'\nJUNK\n") + monkeypatch.setenv("KUHYTRACK_ENV", str(env)) + assert browser._config() == browser.Config("http://h:1", "abc") + + +def test_config_missing_file_is_none( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setenv("KUHYTRACK_ENV", str(tmp_path / "absent")) + assert browser._config() is None + + +def test_config_without_url_is_none( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + env = tmp_path / "env" + env.write_text("KT_TOKEN=abc\n") + monkeypatch.setenv("KUHYTRACK_ENV", str(env)) + assert browser._config() is None + + +def test_get_json_sends_bearer_and_query(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, Any] = {} + + def fake_open(req: Any, timeout: float) -> io.BytesIO: # noqa: ANN401 + seen["url"] = req.full_url + seen["auth"] = req.get_header("Authorization") + seen["timeout"] = timeout + return io.BytesIO(b'{"ok": 1}') + + monkeypatch.setattr(browser, "_urlopen", fake_open) + assert browser._get_json(CFG, "/x", {"a": "b"}) == {"ok": 1} + assert seen == { + "url": "http://tracker.test/x?a=b", + "auth": "Bearer tok", + "timeout": 5, + } + + +def test_get_json_without_token_sends_no_header( + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: dict[str, Any] = {} + + def fake_open(req: Any, timeout: float) -> io.BytesIO: # noqa: ANN401 + seen["auth"] = req.get_header("Authorization") + return io.BytesIO(b"[]") + + monkeypatch.setattr(browser, "_urlopen", fake_open) + assert browser._get_json(browser.Config("http://t", ""), "/x") == [] + assert seen["auth"] is None + + +def test_intersect_and_merge() -> None: + merged = browser._merge([(5, 9), (0, 3), (2, 4), (9, 10)]) + assert merged == [(0, 4), (5, 10)] + assert browser._intersect(merged, [(3, 6), (8, 20)]) == [(3, 4), (5, 6), (8, 10)] + + +def test_blank_pattern_is_ignored() -> None: + assert browser._needles((" ", "LeetCode.com")) == (["leetcode.com"], ["leetcode"])