tegwick
96fe5b022b
- --replay DIR / replay_dir: offline replay from the cache, hash-verified - ok values without source become error; Last-Modified recorded as published - latest period by parsed order (web_fetcher/periods.py); mixed formats refused - non-numeric values not_found unless value_type: string - retry only connection errors, timeouts, 429/5xx; honour Retry-After - opt-in live smoke test; 18 offline tests - CHANGELOG, README install docs, SCOPE update, capability D2/A3/C2/R2 - WEBFETCH-WP-0003 (repo hygiene) added 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())
|