AnasSarkiz/muse-voice-remote

scripts/download-datasheets.py

"""Fetch official candidate datasheets without weakening TLS or inventing success."""
import concurrent.futures
import datetime
import hashlib
import json
from pathlib import Path
import urllib.request

root = Path(__file__).resolve().parent.parent
dest = root / "docs" / "datasheets"
dest.mkdir(parents=True, exist_ok=True)
sources = json.loads((root / "docs/datasheets.json").read_text())
previous = {r["part"]:r for r in json.loads((dest / "download-results.json").read_text())}

def fetch(item):
    name = item["part"].split(" / ")[0]
    result = {**previous.get(item["part"],{}), **item,
              "checked_at":datetime.datetime.now(datetime.timezone.utc).isoformat()}
    try:
        request=urllib.request.Request(item["url"],headers={"User-Agent":"Muse-Voice-Remote-datasheet-review/1.0"})
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = response.read(20 * 1024 * 1024 + 1)
        if len(payload) > 20 * 1024 * 1024:
            raise ValueError("PDF exceeds 20 MiB download limit")
        if not payload.startswith(b"%PDF-"):
            raise ValueError("Response is not a PDF; refusing to save as a datasheet")
        path = dest / (name + ".pdf")
        path.write_bytes(payload)
        result.pop("error",None)
        result.pop("last_attempt_error",None)
        result.update(status="downloaded", file=path.name, bytes=len(payload), sha256=hashlib.sha256(payload).hexdigest())
    except Exception as error:
        retained=previous.get(item["part"],{})
        path=dest / retained.get("file","missing.pdf")
        if path.is_file() and hashlib.sha256(path.read_bytes()).hexdigest()==retained.get("sha256"):
            result.update(status="retained",last_attempt_error=str(error))
        else:
            result.update(status="failed", error=str(error))
    return result

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(fetch, sources))
(dest / "download-results.json").write_text(json.dumps(results, indent=2) + "\n")
for item in results:
    print(item["part"] + ": " + item["status"] + (" — " + item["error"] if "error" in item else ""))
raise SystemExit(1 if any(item["status"] == "failed" for item in results) else 0)