watchers

1#!/usr/bin/env python32"""Watch any JSON endpoint that returns a list of objects; dedup by ID field.3 4Usage (via cron with --no-agent):5 6    hermes cron create api-events \\7      --schedule "*/1 * * * *" --no-agent \\8      --script "$HERMES_HOME/skills/devops/watchers/scripts/watch_http_json.py" \\9      --script-args "--name api --url https://api.example.com/events \\10                     --id-field event_id --items-path data.events"11 12The response can be:13  - a top-level JSON list (default), or14  - a JSON object with a dotted ``--items-path`` pointing to the list.15 16Each item is deduped by ``--id-field`` (default "id").17 18Optional ``--header KEY:VALUE`` flags pass HTTP headers (repeatable).19"""20 21from __future__ import annotations22 23import argparse24import json25import sys26import urllib.error27import urllib.request28from pathlib import Path29 30sys.path.insert(0, str(Path(__file__).parent))31from _watermark import Watermark, format_items_as_markdown  # type: ignore32 33 34def _dig(obj, path: str):35    """Dotted-path lookup: _dig({'a':{'b':[1,2]}}, 'a.b') → [1,2]."""36    if not path:37        return obj38    cur = obj39    for part in path.split("."):40        if isinstance(cur, dict) and part in cur:41            cur = cur[part]42        else:43            return None44    return cur45 46 47def _parse_header(s: str):48    if ":" not in s:49        raise argparse.ArgumentTypeError(50            f"--header expects 'KEY: VALUE' (got {s!r})"51        )52    k, v = s.split(":", 1)53    return (k.strip(), v.strip())54 55 56def main() -> int:57    p = argparse.ArgumentParser(description="Poll a JSON endpoint.")58    p.add_argument("--name", required=True, help="Watcher name (used for state file)")59    p.add_argument("--url", required=True, help="JSON endpoint URL")60    p.add_argument("--id-field", default="id",61                   help="Field used to dedup items (default: 'id')")62    p.add_argument("--items-path", default="",63                   help="Dotted path to the list inside the JSON response (e.g. 'data.events')")64    p.add_argument("--title-field", default="title",65                   help="Field used as the item title in the rendered output (default: 'title')")66    p.add_argument("--url-field", default="url",67                   help="Field used as the item URL in the rendered output (default: 'url')")68    p.add_argument("--body-field", default="",69                   help="Optional body field to include as a snippet under each item")70    p.add_argument("--max", type=int, default=20,71                   help="Max new items to emit per tick (default: 20)")72    p.add_argument("--header", action="append", type=_parse_header, default=[],73                   metavar="KEY: VALUE",74                   help="HTTP header (repeatable)")75    p.add_argument("--timeout", type=float, default=20.0,76                   help="HTTP timeout in seconds (default: 20)")77    args = p.parse_args()78 79    req = urllib.request.Request(args.url, headers={"User-Agent": "Hermes-Watcher/1.0"})80    for k, v in args.header:81        req.add_header(k, v)82 83    try:84        with urllib.request.urlopen(req, timeout=args.timeout) as resp:85            raw = resp.read()86    except urllib.error.HTTPError as e:87        print(f"watch_http_json: HTTP {e.code} from {args.url}", file=sys.stderr)88        return 289    except (urllib.error.URLError, TimeoutError, OSError) as e:90        print(f"watch_http_json: network error: {e}", file=sys.stderr)91        return 292 93    try:94        data = json.loads(raw.decode("utf-8"))95    except (UnicodeDecodeError, json.JSONDecodeError) as e:96        print(f"watch_http_json: response is not valid JSON: {e}", file=sys.stderr)97        return 298 99    items = _dig(data, args.items_path) if args.items_path else data100    if not isinstance(items, list):101        print(102            f"watch_http_json: items_path={args.items_path!r} did not resolve to a list "103            f"(got {type(items).__name__})",104            file=sys.stderr,105        )106        return 2107 108    # Keep only dicts — skip any bare strings / numbers so filter_new doesn't crash.109    items = [i for i in items if isinstance(i, dict)]110 111    wm = Watermark.load(args.name)112    new_items = wm.filter_new(items, id_key=args.id_field)113    wm.save()114 115    if args.max > 0:116        new_items = new_items[: args.max]117 118    body_key = args.body_field or None119    output = format_items_as_markdown(120        new_items,121        title_key=args.title_field,122        url_key=args.url_field,123        body_key=body_key,124    )125    if output:126        sys.stdout.write(output)127    return 0128 129 130if __name__ == "__main__":131    sys.exit(main())132