tegwick
49de8d15b6
- .forgejo/workflows/test.yaml: offline tests on Python 3.10 and 3.13 - LICENSE: Target Revenue Source License V1C1 (org-wide preliminary licence) - pyproject: license field; AGENTS/SCOPE: CI and licence facts Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 153021@bnt-lap001 Assistant-Session: 46917be4-6400-4106-b8b7-61953a0caf24
44 lines
1.7 KiB
Python
44 lines
1.7 KiB
Python
"""CLI: python -m web_fetcher fetch SPECS.json [--cache DIR | --replay DIR] [--out FILE]
|
|
|
|
SPECS.json is a JSON list of spec objects (or {"specs": [...]}). Output is
|
|
JSON Lines, one Observation per line. --replay serves every URL from a cache
|
|
written by an earlier --cache run, without network access.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
|
|
from web_fetcher.registry import run_specs
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(prog="web_fetcher")
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
f = sub.add_parser("fetch", help="run fetch specs")
|
|
f.add_argument("specs")
|
|
src = f.add_mutually_exclusive_group()
|
|
src.add_argument("--cache", default=None, help="content-addressed cache directory")
|
|
src.add_argument("--replay", default=None, help="replay from a cache directory; no network")
|
|
f.add_argument("--out", default="-", help="output JSONL path (default stdout)")
|
|
f.add_argument("--only", action="append", default=[], help="run only these spec ids")
|
|
args = ap.parse_args(argv)
|
|
with open(args.specs, encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
specs = data["specs"] if isinstance(data, dict) else data
|
|
if args.only:
|
|
specs = [s for s in specs if s.get("id") in set(args.only)]
|
|
obs = run_specs(specs, cache_dir=args.cache, replay_dir=args.replay)
|
|
lines = "\n".join(o.to_json() for o in obs) + "\n"
|
|
if args.out == "-":
|
|
sys.stdout.write(lines)
|
|
else:
|
|
with open(args.out, "w", encoding="utf-8") as fh:
|
|
fh.write(lines)
|
|
return 0 if all(o.status != "error" for o in obs) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|