Spaces:
Running
Running
| """Sidecar freshness metadata for the bundled certificate snapshots. | |
| Each cert bundle — data/iscc/snapshot.csv.gz, data/redcert/certs.json.gz, data/ins/certs.json.gz | |
| — gets a sibling ``<stem>.meta.json`` written AT FETCH TIME by the refresh pipeline. It records | |
| when the data was actually downloaded from the registry, which is the honest age the staleness | |
| banner reads. Load time (app boot) and file modification time (reset by every deploy) both lie | |
| about age; this file does not. Pure I/O helper — no Streamlit, no app imports. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| def meta_path_for(bundle_path) -> Path: | |
| """The sidecar meta path for a bundle: snapshot.csv.gz -> snapshot.meta.json, | |
| certs.json.gz -> certs.meta.json (everything from the first dot is treated as suffix).""" | |
| p = Path(bundle_path) | |
| stem = p.name.split(".")[0] | |
| return p.with_name(f"{stem}.meta.json") | |
| def write_meta( | |
| bundle_path, *, rows: int, source: str, fetched_at: str | None = None, | |
| source_last_modified: str | None = None, | |
| source_etag: str | None = None, | |
| source_content_length: int | None = None, | |
| ) -> Path: | |
| """Write the sidecar meta next to ``bundle_path``. ``fetched_at`` defaults to now (UTC ISO) | |
| and records when SINETTI pulled the data. ``source_last_modified`` is the OWNER's own | |
| publish/revision date (e.g. an HTTP Last-Modified header on the live file) when known -- | |
| None when the source gives no such signal, never guessed.""" | |
| fetched_at = fetched_at or datetime.now(timezone.utc).isoformat() | |
| mp = meta_path_for(bundle_path) | |
| mp.parent.mkdir(parents=True, exist_ok=True) | |
| mp.write_text( | |
| json.dumps( | |
| { | |
| "fetched_at": fetched_at, | |
| "rows": int(rows), | |
| "source": source, | |
| "source_last_modified": source_last_modified or None, | |
| # Secondary probe signals, written only when the publisher supplies them. | |
| # They exist so the next run can ask "has this moved?" without downloading: | |
| # an ETag where one is offered, and Content-Length as the last resort for a | |
| # publisher (EPA's CDX endpoint) that offers neither date nor validator. | |
| "source_etag": source_etag or None, | |
| "source_content_length": ( | |
| int(source_content_length) if source_content_length else None | |
| ), | |
| }, | |
| indent=2, | |
| ), | |
| encoding="utf-8", | |
| ) | |
| return mp | |
| def read_meta(bundle_path) -> dict | None: | |
| """Read the sidecar meta, or None if it is missing, unparseable, or has no fetched_at (a meta | |
| without a timestamp tells us nothing about age, so it is treated as absent).""" | |
| mp = meta_path_for(bundle_path) | |
| if not mp.exists(): | |
| return None | |
| try: | |
| d = json.loads(mp.read_text(encoding="utf-8")) | |
| except Exception: | |
| return None | |
| if isinstance(d, dict) and d.get("fetched_at"): | |
| return d | |
| return None | |