commit c48d6113ddd3b0de4169d9f7f36071b3f67c26ad Author: tegwick Date: Tue Oct 13 14:04:02 2026 +0200 Seeded repo diff --git a/INTENT.md b/INTENT.md new file mode 100644 index 0000000..903d9e8 --- /dev/null +++ b/INTENT.md @@ -0,0 +1,22 @@ +# INTENT + +`web-fetcher` exists to give research repositories a shared, trustworthy way +to pull data from the web and from public APIs. + +**Core commitments** + +1. **Provenance first.** No value without source, URL, retrieval time and + payload hash. +2. **Never invent.** Missing data is an explicit `not_found`, never a guess. + Language models are not a data source here. +3. **Reusable adapters.** Statistics APIs, CSV/JSON endpoints and + spreadsheets behind one small interface. New sources are added as adapters + with offline fixture tests. +4. **Replayable.** An optional content-addressed cache reproduces a run from + the exact bytes it used. +5. **Minimal footprint.** Standard library only, so it runs inside executors + (for example rein-aharness on railiance01) without dependency management. + +Out of scope: crawling, scraping behind logins, LLM-based extraction, +scheduling (activity-core owns *when*), and execution orchestration +(rein-aharness owns *how it runs*). diff --git a/README.md b/README.md new file mode 100644 index 0000000..7bf6aba --- /dev/null +++ b/README.md @@ -0,0 +1,48 @@ +# web-fetcher + +Reusable, provenance-first fetching of web and API data for research. + +Every fetch returns `Observation` records with the value **and** where it came +from: source, URL, the source's own publication or update time, retrieval +time, and the SHA-256 of the raw payload. Values are never estimated: a spec +yields a sourced value, or an explicit `not_found` or `error` record. An +optional content-addressed cache lets a research run be replayed from the +exact bytes it used. + +Standard library only. Python 3.10+. + +## Adapters + +| Adapter | Use | Key spec fields | +|---|---|---| +| `eurostat` | Eurostat dissemination API (JSON-stat 2.0) | `dataset`, `filters`, `split_by`, `last` | +| `http_json` | any JSON endpoint | `url`, `path` (dotted), `period_path` | +| `http_csv` | any CSV endpoint (e.g. FRED `fredgraph.csv`) | `url`, `filter`, `value_column`, `period_column` | +| `xlsx_wide` | spreadsheets with one column per period (e.g. Census BTOS) | `url`, `sheet`, `filter`, `period_regex` | +| `manual` | no machine-readable source; recorded as `not_found` with a note | `source` | + +## Usage + +```bash +python3 -m web_fetcher fetch specs.json [--cache .cache] [--only ID] > observations.jsonl +``` + +```python +from web_fetcher import run_spec +obs = run_spec({"id": "eu_ai", "adapter": "eurostat", "dataset": "isoc_eb_ai", + "filters": {"geo": ["EU27_2020"], "unit": "PC_ENT", "size_emp": "GE10", + "indic_is": "E_AI_TANY"}, "split_by": ["geo"]}) +``` + +## Tests + +```bash +python3 -B -m unittest discover -s tests +``` + +Tests run offline against fixtures in `tests/fixtures/`. + +## Consumers + +- `assembly-economics`: reality checkpoints (`checkpoints/sources.json`, + `checkpoints/run_checkpoint.py`). diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..94b5e55 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,16 @@ +[project] +name = "web-fetcher" +version = "0.1.0" +description = "Provenance-first fetching of web and API data for research" +requires-python = ">=3.10" +dependencies = [] + +[project.scripts] +web-fetcher = "web_fetcher.__main__:main" + +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +include = ["web_fetcher*"] diff --git a/tests/fixtures/btos_national.xlsx b/tests/fixtures/btos_national.xlsx new file mode 100644 index 0000000..9c7fbda Binary files /dev/null and b/tests/fixtures/btos_national.xlsx differ diff --git a/tests/fixtures/eurostat_isoc_eb_ai.json b/tests/fixtures/eurostat_isoc_eb_ai.json new file mode 100644 index 0000000..ac133f0 --- /dev/null +++ b/tests/fixtures/eurostat_isoc_eb_ai.json @@ -0,0 +1 @@ +{"version":"2.0","class":"dataset","label":"Artificial intelligence by size class of enterprise","source":"ESTAT","updated":"2026-06-15T11:00:00+0200","value":{"3":11.55,"4":19.75,"5":25.97,"0":8.06,"1":13.48,"2":19.95},"id":["freq","size_emp","nace_r2","indic_is","unit","geo","time"],"size":[1,1,1,1,1,2,3],"dimension":{"freq":{"label":"Time frequency","category":{"index":{"A":0},"label":{"A":"Annual"}}},"size_emp":{"label":"Size classes in number of persons employed","category":{"index":{"GE10":0},"label":{"GE10":"10 persons employed or more"}}},"nace_r2":{"label":"Statistical classification of economic activities in the European Community (NACE Rev. 2)","category":{"index":{"C10-S951_X_K":0},"label":{"C10-S951_X_K":"All activities (except agriculture, forestry and fishing, and mining and quarrying), without financial sector"}}},"indic_is":{"label":"Information society indicator","category":{"index":{"E_AI_TANY":0},"label":{"E_AI_TANY":"Enterprises using at least one of the AI technologies: AI_TTM, AI_TSR, AI_TNLG, AI_TIR, AI_TML, AI_TPA, AI_TAR, E_AI_TPVSG"}}},"unit":{"label":"Unit of measure","category":{"index":{"PC_ENT":0},"label":{"PC_ENT":"Percentage of enterprises"}}},"geo":{"label":"Geopolitical entity (reporting)","category":{"index":{"EU27_2020":0,"DE":1},"label":{"EU27_2020":"European Union - 27 countries (from 2020)","DE":"Germany"}}},"time":{"label":"Time","category":{"index":{"2023":0,"2024":1,"2025":2},"label":{"2023":"2023","2024":"2024","2025":"2025"}}}},"extension":{"lang":"EN","id":"ISOC_EB_AI","agencyId":"ESTAT","version":"1.0","datastructure":{"id":"ISOC_EB_AI","agencyId":"ESTAT","version":"45.0"},"annotation":[{"type":"CREATED","date":"2022-12-08T09:14:11+0100"},{"type":"DISSEMINATION_DOI_XML","title":"10.2908/ISOC_EB_AI2022-10-26"},{"type":"DISSEMINATION_OBJECT_TYPE","title":"DATASET"},{"type":"DISSEMINATION_TIMESTAMP_DATA","date":"2026-06-15T11:00:00+0200"},{"type":"DISSEMINATION_TIMESTAMP_GLOBAL","date":"2026-06-15T11:00:00+0200"},{"type":"DISSEMINATION_TIMESTAMP_PLANNED","date":"2026-06-15T11:00:00+0200"},{"type":"ESMS_HTML","title":"Explanatory texts (metadata)","href":"https://ec.europa.eu/eurostat/cache/metadata/en/isoc_e_esms.htm"},{"type":"ESMS_SDMX","title":"Explanatory texts (metadata)","href":"https://ec.europa.eu/eurostat/api/dissemination/files?file=metadata/isoc_e_esms.sdmx.zip"},{"type":"OBS_COUNT","title":"70756"},{"type":"OBS_PERIOD_OVERALL_LATEST","title":"2025"},{"type":"OBS_PERIOD_OVERALL_OLDEST","title":"2021"},{"type":"SOURCE_INSTITUTIONS","text":"Eurostat"},{"type":"UPDATE_DATA","date":"2026-06-15T11:00:00+0200"},{"type":"UPDATE_STRUCTURE","date":"2025-12-11T11:00:00+0100"}],"positions-with-no-data":{"freq":[],"size_emp":[],"nace_r2":[],"indic_is":[],"unit":[],"geo":[],"time":[]}}} \ No newline at end of file diff --git a/tests/test_web_fetcher.py b/tests/test_web_fetcher.py new file mode 100644 index 0000000..16a96d8 --- /dev/null +++ b/tests/test_web_fetcher.py @@ -0,0 +1,82 @@ +"""Offline tests (fixtures only). Usage: python3 -m unittest discover -s tests""" + +import json +import unittest +from pathlib import Path + +from web_fetcher.core import Fetched +from web_fetcher.registry import run_spec + +FIX = Path(__file__).parent / "fixtures" + + +def fake(body: bytes, url: str = "https://example.test/x"): + return lambda u: Fetched(url=u, body=body, retrieved_at="2026-10-12T00:00:00+00:00", + sha256="0" * 64, content_type="application/json") + + +class TestEurostat(unittest.TestCase): + def test_parses_values_per_geo_and_period(self): + body = (FIX / "eurostat_isoc_eb_ai.json").read_bytes() + spec = {"id": "eu_ai_use", "adapter": "eurostat", "dataset": "isoc_eb_ai", + "filters": {"geo": ["EU27_2020", "DE"], "unit": "PC_ENT", "size_emp": "GE10", + "indic_is": "E_AI_TANY"}, "split_by": ["geo"], "last": 3} + obs = run_spec(spec, get=fake(body)) + got = {(o.labels["geo"], o.period): o.value for o in obs} + self.assertEqual(got[("EU27_2020", "2025")], 19.95) + self.assertEqual(got[("DE", "2023")], 11.55) + self.assertTrue(all(o.status == "ok" and o.url and o.payload_sha256 for o in obs)) + self.assertEqual(obs[0].unit, "Percentage of enterprises") + + def test_url_contains_filters(self): + from web_fetcher.adapters.eurostat import build_url + url = build_url({"dataset": "isoc_eb_ai", "filters": {"geo": ["DE"], "unit": "PC_ENT"}, "last": 2}) + self.assertIn("isoc_eb_ai?", url) + self.assertIn("geo=DE", url) + self.assertIn("lastTimePeriod=2", url) + + +class TestXlsx(unittest.TestCase): + def test_btos_ai_use_latest_period(self): + body = (FIX / "btos_national.xlsx").read_bytes() + obs = run_spec({"id": "us_ai_use", "adapter": "xlsx_wide", "url": "https://x.test/National.xlsx", + "sheet": "Response Estimates", "filter": {"Question ID": "7", "Answer ID": "1"}, + "unit": "% of firms"}, get=fake(body)) + self.assertEqual(obs[0].status, "ok") + self.assertEqual(obs[0].period, "202620") + self.assertAlmostEqual(obs[0].value, 23.8) + + +class TestHttp(unittest.TestCase): + def test_json_path(self): + body = json.dumps({"data": {"rows": [{"v": 7.5, "p": "2026-Q3"}]}}).encode() + obs = run_spec({"id": "j", "adapter": "http_json", "url": "https://x.test", "path": "data.rows.0.v", + "period_path": "data.rows.0.p", "unit": "%"}, get=fake(body)) + self.assertEqual((obs[0].status, obs[0].value, obs[0].period), ("ok", 7.5, "2026-Q3")) + + def test_json_missing_path_is_not_found(self): + obs = run_spec({"id": "j", "adapter": "http_json", "url": "https://x.test", "path": "nope"}, + get=fake(b"{}")) + self.assertEqual(obs[0].status, "not_found") + + def test_csv_latest_matching_row(self): + body = b"geo,period,value\nUS,2026-08,8.1\nUS,2026-09,8.7\nDE,2026-09,9.9\n" + obs = run_spec({"id": "c", "adapter": "http_csv", "url": "https://x.test", "filter": {"geo": "US"}, + "value_column": "value", "period_column": "period"}, get=fake(body)) + self.assertEqual((obs[0].value, obs[0].period), (8.7, "2026-09")) + + def test_manual_and_unknown_adapters_never_raise(self): + self.assertEqual(run_spec({"id": "m", "adapter": "manual"})[0].status, "not_found") + self.assertEqual(run_spec({"id": "u", "adapter": "nope"})[0].status, "error") + + def test_fetch_error_becomes_error_observation(self): + from web_fetcher.core import FetchError + + def boom(url): + raise FetchError("offline") + obs = run_spec({"id": "e", "adapter": "http_json", "url": "https://x.test", "path": "a"}, get=boom) + self.assertEqual(obs[0].status, "error") + + +if __name__ == "__main__": + unittest.main() diff --git a/web_fetcher/__init__.py b/web_fetcher/__init__.py new file mode 100644 index 0000000..03a6eb5 --- /dev/null +++ b/web_fetcher/__init__.py @@ -0,0 +1,13 @@ +"""web-fetcher: reusable, provenance-first fetching of web and API data for research. + +Every fetch returns Observation records that say where a value came from +(URL), when it was retrieved, and a hash of the raw payload, so downstream +research can cite and re-verify it. Values are never estimated: a fetch either +yields a sourced value or an explicit `not_found` / `error` record. +""" + +from web_fetcher.core import Observation, fetch_bytes +from web_fetcher.registry import ADAPTERS, run_spec, run_specs + +__all__ = ["Observation", "fetch_bytes", "ADAPTERS", "run_spec", "run_specs"] +__version__ = "0.1.0" diff --git a/web_fetcher/__main__.py b/web_fetcher/__main__.py new file mode 100644 index 0000000..00ee670 --- /dev/null +++ b/web_fetcher/__main__.py @@ -0,0 +1,41 @@ +"""CLI: python -m web_fetcher fetch SPECS.json [--cache DIR] [--out FILE] + +SPECS.json is a JSON list of spec objects (or {"specs": [...]}). Output is +JSON Lines, one Observation per line. +""" + +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") + f.add_argument("--cache", default=None, help="content-addressed cache directory") + 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) + 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()) diff --git a/web_fetcher/adapters/__init__.py b/web_fetcher/adapters/__init__.py new file mode 100644 index 0000000..a5edce3 --- /dev/null +++ b/web_fetcher/adapters/__init__.py @@ -0,0 +1 @@ +"""Source adapters. Each exposes `fetch(spec, fetcher) -> list[Observation]`.""" diff --git a/web_fetcher/adapters/eurostat.py b/web_fetcher/adapters/eurostat.py new file mode 100644 index 0000000..3ad6bfd --- /dev/null +++ b/web_fetcher/adapters/eurostat.py @@ -0,0 +1,61 @@ +"""Eurostat dissemination API (JSON-stat 2.0). + +Spec fields: + dataset Eurostat dataset code, e.g. "isoc_eb_ai" + filters dict of dimension -> code or list of codes (all non-time dimensions + should be pinned to one code except those listed in `split_by`) + split_by dimensions that may vary; one observation per combination (e.g. ["geo"]) + last number of latest time periods to return (default 1) +""" + +from __future__ import annotations + +import itertools +import json +import urllib.parse +from typing import Any, Callable + +from web_fetcher.core import Fetched, Observation + +BASE = "https://ec.europa.eu/eurostat/api/dissemination/statistics/1.0/data/" + + +def build_url(spec: dict[str, Any]) -> str: + params: list[tuple[str, str]] = [("format", "JSON"), ("lang", "EN")] + for dim, val in (spec.get("filters") or {}).items(): + for v in (val if isinstance(val, list) else [val]): + params.append((dim, str(v))) + params.append(("lastTimePeriod", str(int(spec.get("last", 1))))) + return BASE + spec["dataset"] + "?" + urllib.parse.urlencode(params) + + +def parse(doc: dict[str, Any], spec: dict[str, Any], fetched: Fetched) -> list[Observation]: + ids = doc["id"] + dims = doc["dimension"] + cats = [sorted(dims[i]["category"]["index"], key=lambda k: dims[i]["category"]["index"][k]) for i in ids] + values = doc.get("value") or {} + unit_dim = "unit" if "unit" in ids else None + out: list[Observation] = [] + for n, combo in enumerate(itertools.product(*cats)): + v = values.get(str(n)) if isinstance(values, dict) else (values[n] if n < len(values) else None) + if v is None: + continue + coord = dict(zip(ids, combo)) + labels = {d: coord[d] for d in spec.get("split_by", [])} + out.append(Observation( + spec_id=spec["id"], status="ok", value=v, + unit=(dims[unit_dim]["category"]["label"].get(coord[unit_dim]) if unit_dim else None), + period=coord.get("time"), labels=labels, source=f"Eurostat {spec['dataset']}: {doc.get('label')}", + url=fetched.url, published=doc.get("updated"), retrieved_at=fetched.retrieved_at, + payload_sha256=fetched.sha256, + )) + if not out: + out.append(Observation(spec_id=spec["id"], status="not_found", url=fetched.url, + retrieved_at=fetched.retrieved_at, payload_sha256=fetched.sha256, + note="no values for the given filters")) + return out + + +def fetch(spec: dict[str, Any], get: Callable[[str], Fetched]) -> list[Observation]: + fetched = get(build_url(spec)) + return parse(json.loads(fetched.body), spec, fetched) diff --git a/web_fetcher/adapters/http.py b/web_fetcher/adapters/http.py new file mode 100644 index 0000000..bf296ac --- /dev/null +++ b/web_fetcher/adapters/http.py @@ -0,0 +1,97 @@ +r"""Generic HTTP adapters: JSON (dotted path) and CSV (filter + pick latest). + +json spec fields: url, path ("a.b.0.c"), unit?, period_path? +csv spec fields: url, filter (dict column -> value), value_column, + period_column?, unit?, delimiter? -> latest row by period +xlsx_wide fields: url, sheet, filter (dict header -> value), period_regex + (header pattern of period columns, default ^\d{6}$), unit? + -> value of the matching row in the latest period column +""" + +from __future__ import annotations + +import csv +import io +import json +from typing import Any, Callable + +from web_fetcher.core import Fetched, Observation + + +def _dig(obj: Any, path: str) -> Any: + for part in path.split("."): + if obj is None: + return None + if isinstance(obj, list): + try: + obj = obj[int(part)] + except (ValueError, IndexError): + return None + else: + obj = obj.get(part) if isinstance(obj, dict) else None + return obj + + +def fetch_json(spec: dict[str, Any], get: Callable[[str], Fetched]) -> list[Observation]: + f = get(spec["url"]) + doc = json.loads(f.body) + value = _dig(doc, spec["path"]) + period = _dig(doc, spec["period_path"]) if spec.get("period_path") else spec.get("period") + status = "ok" if value is not None else "not_found" + return [Observation(spec_id=spec["id"], status=status, value=value, unit=spec.get("unit"), + period=None if period is None else str(period), source=spec.get("source"), + url=f.url, retrieved_at=f.retrieved_at, payload_sha256=f.sha256, + note=None if value is not None else f"path {spec['path']} not found")] + + +def fetch_csv(spec: dict[str, Any], get: Callable[[str], Fetched]) -> list[Observation]: + f = get(spec["url"]) + text = f.body.decode(spec.get("encoding", "utf-8-sig"), errors="replace") + rows = list(csv.DictReader(io.StringIO(text), delimiter=spec.get("delimiter", ","))) + flt = spec.get("filter") or {} + rows = [r for r in rows if all(str(r.get(k, "")).strip() == str(v) for k, v in flt.items())] + pcol, vcol = spec.get("period_column"), spec["value_column"] + rows = [r for r in rows if str(r.get(vcol, "")).strip() not in ("", "NA", "N/A", "-")] + if not rows: + return [Observation(spec_id=spec["id"], status="not_found", url=f.url, retrieved_at=f.retrieved_at, + payload_sha256=f.sha256, source=spec.get("source"), note="no matching row")] + row = max(rows, key=lambda r: str(r.get(pcol, ""))) if pcol else rows[-1] + raw = str(row[vcol]).strip().replace("%", "") + try: + value: Any = float(raw) + except ValueError: + value = raw + return [Observation(spec_id=spec["id"], status="ok", value=value, unit=spec.get("unit"), + period=str(row.get(pcol)) if pcol else None, source=spec.get("source"), + labels={k: str(v) for k, v in flt.items()}, url=f.url, + retrieved_at=f.retrieved_at, payload_sha256=f.sha256)] + + +def fetch_xlsx_wide(spec: dict[str, Any], get: Callable[[str], Fetched]) -> list[Observation]: + import re + + from web_fetcher.xlsx import read_sheet + + f = get(spec["url"]) + rows = read_sheet(f.body, spec.get("sheet", 0)) + header, body = rows[0], rows[1:] + pat = re.compile(spec.get("period_regex", r"^\d{6}$")) + periods = [(i, h) for i, h in enumerate(header) if pat.match(h or "")] + flt = spec.get("filter") or {} + col = {h: i for i, h in enumerate(header)} + match = [r for r in body if all(i < len(r) and r[i].strip() == str(v) for h, v in flt.items() + for i in [col.get(h, 10**9)])] + base = dict(spec_id=spec["id"], url=f.url, retrieved_at=f.retrieved_at, payload_sha256=f.sha256, + source=spec.get("source"), labels={k: str(v) for k, v in flt.items()}) + if not match or not periods: + return [Observation(status="not_found", note="no matching row or period columns", **base)] + row = match[0] + for i, h in sorted(periods, key=lambda p: p[1], reverse=True): + raw = (row[i] if i < len(row) else "").strip().replace("%", "") + if raw not in ("", "S", "NA", "-"): + try: + value: Any = float(raw) + except ValueError: + value = raw + return [Observation(status="ok", value=value, unit=spec.get("unit"), period=h, **base)] + return [Observation(status="not_found", note="matching row has no value", **base)] diff --git a/web_fetcher/core.py b/web_fetcher/core.py new file mode 100644 index 0000000..9fc59a2 --- /dev/null +++ b/web_fetcher/core.py @@ -0,0 +1,93 @@ +"""Core types: Observation record and provenance-tracked HTTP fetching.""" + +from __future__ import annotations + +import hashlib +import json +import os +import time +import urllib.error +import urllib.request +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +USER_AGENT = "web-fetcher/0.1 (+https://forgejo.coulomb.social/coulomb/web-fetcher)" + + +@dataclass +class Observation: + """One fetched value with its provenance. status: ok | not_found | error.""" + + spec_id: str + status: str + value: Any = None + unit: str | None = None + period: str | None = None + labels: dict[str, str] = field(default_factory=dict) + source: str | None = None # dataset or source name + url: str | None = None + published: str | None = None # source-side update/publication timestamp + retrieved_at: str | None = None + payload_sha256: str | None = None + note: str | None = None + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + def to_json(self) -> str: + return json.dumps(self.to_dict(), ensure_ascii=False, sort_keys=True) + + +@dataclass +class Fetched: + url: str + body: bytes + retrieved_at: str + sha256: str + content_type: str | None + + +class FetchError(RuntimeError): + pass + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def fetch_bytes(url: str, *, timeout: float = 60.0, retries: int = 2, + cache_dir: str | os.PathLike | None = None, + headers: dict[str, str] | None = None) -> Fetched: + """GET `url` and return body plus provenance. Optional content-addressed cache. + + The cache stores each payload under its SHA-256 and an index entry per URL, + so a research run can be replayed offline from the exact bytes it used. + """ + hdrs = {"User-Agent": USER_AGENT, **(headers or {})} + last: Exception | None = None + for attempt in range(retries + 1): + try: + req = urllib.request.Request(url, headers=hdrs) + with urllib.request.urlopen(req, timeout=timeout) as resp: + body = resp.read() + ctype = resp.headers.get("Content-Type") + digest = hashlib.sha256(body).hexdigest() + fetched = Fetched(url=url, body=body, retrieved_at=_now(), sha256=digest, content_type=ctype) + if cache_dir: + _cache_store(Path(cache_dir), fetched) + return fetched + except (urllib.error.URLError, TimeoutError, OSError) as exc: + last = exc + if attempt < retries: + time.sleep(1.5 * (attempt + 1)) + raise FetchError(f"fetch failed for {url}: {last}") + + +def _cache_store(cache: Path, fetched: Fetched) -> None: + cache.mkdir(parents=True, exist_ok=True) + (cache / f"{fetched.sha256}.bin").write_bytes(fetched.body) + with (cache / "index.jsonl").open("a", encoding="utf-8") as fh: + fh.write(json.dumps({"url": fetched.url, "sha256": fetched.sha256, + "retrieved_at": fetched.retrieved_at}) + "\n") diff --git a/web_fetcher/registry.py b/web_fetcher/registry.py new file mode 100644 index 0000000..de64a7e --- /dev/null +++ b/web_fetcher/registry.py @@ -0,0 +1,42 @@ +"""Spec dispatch: run one or many fetch specs, never raising for a single failure.""" + +from __future__ import annotations + +from typing import Any, Callable + +from web_fetcher.adapters import eurostat, http +from web_fetcher.core import Fetched, FetchError, Observation, fetch_bytes + +ADAPTERS: dict[str, Callable[[dict[str, Any], Callable[[str], Fetched]], list[Observation]]] = { + "eurostat": eurostat.fetch, + "http_json": http.fetch_json, + "http_csv": http.fetch_csv, + "xlsx_wide": http.fetch_xlsx_wide, +} + + +def run_spec(spec: dict[str, Any], get: Callable[[str], Fetched] | None = None, + cache_dir: str | None = None) -> list[Observation]: + """Run one spec. Adapter errors become an `error` observation, not an exception.""" + sid = spec.get("id", "?") + kind = spec.get("adapter") + if kind == "manual": + return [Observation(spec_id=sid, status="not_found", source=spec.get("source"), + note="manual: no machine-readable source; collect at review")] + adapter = ADAPTERS.get(kind or "") + if adapter is None: + return [Observation(spec_id=sid, status="error", note=f"unknown adapter {kind!r}")] + getter = get or (lambda url: fetch_bytes(url, cache_dir=cache_dir)) + try: + return adapter(spec, getter) + except FetchError as exc: + return [Observation(spec_id=sid, status="error", note=str(exc)[:300])] + except (KeyError, ValueError, TypeError) as exc: + return [Observation(spec_id=sid, status="error", note=f"parse error: {exc!s}"[:300])] + + +def run_specs(specs: list[dict[str, Any]], **kw: Any) -> list[Observation]: + out: list[Observation] = [] + for spec in specs: + out.extend(run_spec(spec, **kw)) + return out diff --git a/web_fetcher/xlsx.py b/web_fetcher/xlsx.py new file mode 100644 index 0000000..b9965ab --- /dev/null +++ b/web_fetcher/xlsx.py @@ -0,0 +1,53 @@ +"""Minimal read-only XLSX reader (stdlib only): sheet -> list of rows of strings.""" + +from __future__ import annotations + +import io +import re +import zipfile +import xml.etree.ElementTree as ET + +_NS = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main", + "r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships"} +_T = "{%s}t" % _NS["m"] + + +def _col_index(ref: str) -> int: + letters = re.match(r"[A-Z]+", ref).group(0) + n = 0 + for ch in letters: + n = n * 26 + (ord(ch) - 64) + return n - 1 + + +def read_sheet(data: bytes, sheet: str | int = 0) -> list[list[str]]: + z = zipfile.ZipFile(io.BytesIO(data)) + shared: list[str] = [] + if "xl/sharedStrings.xml" in z.namelist(): + for si in ET.fromstring(z.read("xl/sharedStrings.xml")).findall("m:si", _NS): + shared.append("".join(t.text or "" for t in si.iter(_T))) + wb = ET.fromstring(z.read("xl/workbook.xml")) + sheets = [s for s in wb.find("m:sheets", _NS)] + if isinstance(sheet, str): + idx = next(i for i, s in enumerate(sheets) if s.get("name") == sheet) + else: + idx = sheet + rid = sheets[idx].get("{%s}id" % _NS["r"]) + rels = ET.fromstring(z.read("xl/_rels/workbook.xml.rels")) + target = next(r.get("Target") for r in rels if r.get("Id") == rid) + path = "xl/" + target.lstrip("/").removeprefix("xl/") + root = ET.fromstring(z.read(path)) + rows: list[list[str]] = [] + for r in root.find("m:sheetData", _NS).findall("m:row", _NS): + cells: dict[int, str] = {} + for c in r.findall("m:c", _NS): + v = c.find("m:v", _NS) + if v is None: + inline = c.find("m:is", _NS) + text = "".join(t.text or "" for t in inline.iter(_T)) if inline is not None else "" + else: + text = shared[int(v.text)] if c.get("t") == "s" else (v.text or "") + cells[_col_index(c.get("r"))] = text + width = max(cells) + 1 if cells else 0 + rows.append([cells.get(i, "") for i in range(width)]) + return rows