watchers

1"""Shared watermark helper used by the three watcher scripts.2 3A watermark is just a JSON file that records the IDs we've seen on previous4runs, so the next run only emits items we haven't seen before.5 6Contract:7- First run: record all IDs from the fetched batch, emit nothing.8- Subsequent runs: emit items whose ID isn't in the stored set.9- Bounded: keep at most `max_seen` IDs (default 500).10- Atomic: write to a .tmp file and rename, so a crashed script can't11  leave a half-written state file that permanently breaks dedup.12 13Import and use from any custom watcher script:14 15    from _watermark import Watermark16 17    wm = Watermark.load("my-feed-name")18    new_items = wm.filter_new(fetched_items, id_key="id")19    wm.save()20"""21 22from __future__ import annotations23 24import json25import os26from pathlib import Path27from typing import Any, Dict, Iterable, List, Optional28 29 30def _state_dir() -> Path:31    """Where watermark files live — respects WATCHER_STATE_DIR override."""32    override = os.environ.get("WATCHER_STATE_DIR")33    if override:34        return Path(override)35    # Default: $HERMES_HOME/watcher-state/, falling back to ~/.hermes/watcher-state/.36    hermes_home = os.environ.get("HERMES_HOME") or str(Path.home() / ".hermes")37    return Path(hermes_home) / "watcher-state"38 39 40class Watermark:41    """Per-watcher state. Persisted to <state_dir>/<name>.json."""42 43    def __init__(self, name: str, *, max_seen: int = 500) -> None:44        if not name or not name.replace("-", "").replace("_", "").isalnum():45            raise ValueError(46                f"watermark name must be alphanumeric + '-'/'_' (got {name!r})"47            )48        self.name = name49        self.max_seen = max_seen50        self._path = _state_dir() / f"{name}.json"51        self._data: Dict[str, Any] = {"seen_ids": [], "first_run": True}52 53    @classmethod54    def load(cls, name: str, *, max_seen: int = 500) -> "Watermark":55        wm = cls(name, max_seen=max_seen)56        if wm._path.exists():57            try:58                wm._data = json.loads(wm._path.read_text(encoding="utf-8"))59                wm._data.setdefault("seen_ids", [])60                wm._data["first_run"] = False61            except (OSError, json.JSONDecodeError):62                # Corrupt state file — treat as a first run but don't crash.63                wm._data = {"seen_ids": [], "first_run": True}64        return wm65 66    @property67    def is_first_run(self) -> bool:68        return bool(self._data.get("first_run", True))69 70    @property71    def seen(self) -> List[str]:72        return list(self._data.get("seen_ids", []))73 74    def filter_new(75        self, items: Iterable[Dict[str, Any]], *, id_key: str = "id"76    ) -> List[Dict[str, Any]]:77        """Return items whose id isn't in the stored set.78 79        Side effect: updates the in-memory seen set with every id in the80        batch (so save() persists the full new watermark).  On first run,81        records every id but returns an empty list (baseline, no replay).82        """83        existing = set(str(x) for x in self._data.get("seen_ids", []))84        was_first_run = self.is_first_run85 86        new_items: List[Dict[str, Any]] = []87        batch_ids: List[str] = []88        for item in items:89            ident = item.get(id_key)90            if ident is None:91                continue92            ident_str = str(ident)93            batch_ids.append(ident_str)94            if ident_str in existing:95                continue96            if was_first_run:97                continue  # record but don't emit98            new_items.append(item)99 100        combined = list(existing) + [i for i in batch_ids if i not in existing]101        if len(combined) > self.max_seen:102            combined = combined[-self.max_seen:]103        self._data["seen_ids"] = combined104        self._data["first_run"] = False105        return new_items106 107    def save(self) -> None:108        self._path.parent.mkdir(parents=True, exist_ok=True)109        tmp = self._path.with_suffix(".tmp")110        tmp.write_text(111            json.dumps(self._data, indent=2, sort_keys=True),112            encoding="utf-8",113        )114        os.replace(tmp, self._path)115 116 117def format_items_as_markdown(118    items: List[Dict[str, Any]],119    *,120    title_key: str = "title",121    url_key: str = "url",122    body_key: Optional[str] = None,123    max_body_chars: int = 500,124) -> str:125    """Render a list of items as Markdown for cron delivery.126 127    One heading per item + its URL + optional snippet of body.  Output is128    empty string when items is empty — cron will then treat stdout as129    silent and skip delivery (existing behavior).130    """131    if not items:132        return ""133    lines: List[str] = []134    for item in items:135        title = (item.get(title_key) or "(no title)").strip()136        url = (item.get(url_key) or "").strip()137        lines.append(f"## {title}")138        if url:139            lines.append(url)140        if body_key:141            body = (item.get(body_key) or "").strip()142            if body:143                if len(body) > max_body_chars:144                    body = body[:max_body_chars].rstrip() + "…"145                lines.append("")146                lines.append(body)147        lines.append("")148    return "\n".join(lines).rstrip() + "\n"149