Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import httpx, hashlib, json, os, mimetypes | |
| from datetime import datetime | |
| from pydantic import HttpUrl | |
| from .models import SourceMeta | |
| def _sha256_bytes(b: bytes) -> str: | |
| return hashlib.sha256(b).hexdigest() | |
| def fetch_url(url: str, out_dir: str = "data") -> str: | |
| os.makedirs(out_dir, exist_ok=True) | |
| ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ") | |
| filename_base = url.replace("://", "_").replace("/", "_") | |
| target_path = os.path.join(out_dir, f"{filename_base}") | |
| manifest_path = target_path + ".json" | |
| with httpx.Client(follow_redirects=True, timeout=60) as client: | |
| r = client.get(url) | |
| r.raise_for_status() | |
| media_type = r.headers.get("content-type", "").split(";")[0] or mimetypes.guess_type(url)[0] or "application/octet-stream" | |
| b = r.content | |
| sha = _sha256_bytes(b) | |
| # write raw artifact | |
| ext = mimetypes.guess_extension(media_type) or ".bin" | |
| artifact_path = target_path + ext | |
| with open(artifact_path, "wb") as f: | |
| f.write(b) | |
| # write manifest | |
| meta = { | |
| "url": url, | |
| "retrieved_at": ts, | |
| "status_code": r.status_code, | |
| "redirect_chain": [str(h.request.url) for h in r.history] + [str(r.request.url)], | |
| "etag": r.headers.get("etag"), | |
| "last_modified": r.headers.get("last-modified"), | |
| "media_type": media_type, | |
| "sha256": sha, | |
| "bytes": len(b), | |
| "artifact_path": artifact_path, | |
| } | |
| with open(manifest_path, "w") as f: | |
| json.dump(meta, f, indent=2) | |
| return manifest_path | |
| def manifest_to_source(meta_path: str, title: str | None = None, jurisdiction: str = "OTHER") -> SourceMeta: | |
| with open(meta_path) as f: | |
| m = json.load(f) | |
| return SourceMeta( | |
| jurisdiction=jurisdiction, title=title, url=m["url"], | |
| media_type=m["media_type"], etag=m.get("etag"), last_modified=m.get("last_modified"), | |
| sha256=m.get("sha256") | |
| ) |