Spaces:
Running
Running
| """Build the certificate-format catalog from the parsed PDF artifact.""" | |
| from __future__ import annotations | |
| import argparse | |
| import gzip | |
| import json | |
| import shutil | |
| import sqlite3 | |
| import sys | |
| import tempfile | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from urllib.request import Request, urlopen | |
| ROOT = Path(__file__).resolve().parents[1] | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| from collections import defaultdict | |
| from sinetti_services.cert_format_catalog import ( | |
| FormatDoc, | |
| artifact_sha256, | |
| build_catalog, | |
| choose_bucket, | |
| load_format_docs, | |
| pick_exemplar, | |
| thumb_filename, | |
| type_key, | |
| write_catalog, | |
| ) | |
| from cert_pdf_models import PdfDocumentRef | |
| from cert_pdf_ocr import _render_pdf_pages | |
| from ins import _doc_operator_id | |
| def _utc_now() -> str: | |
| return datetime.now(timezone.utc).isoformat(timespec="seconds") | |
| def _iter_ins_records(value): | |
| if isinstance(value, list): | |
| for item in value: | |
| yield from _iter_ins_records(item) | |
| elif isinstance(value, dict): | |
| if ( | |
| "certificate_id" in value | |
| or "codiceDelCertificato" in value | |
| ): | |
| yield value | |
| else: | |
| for item in value.values(): | |
| yield from _iter_ins_records(item) | |
| def _load_ins_cb_map(path: Path) -> dict[str, str]: | |
| """Map INS operator app ids to the newest-issued non-empty issuing CB. | |
| Equal or missing issue dates keep the first non-empty CB encountered. | |
| """ | |
| if not path.exists(): | |
| return {} | |
| payload = json.loads(path.read_text(encoding="utf-8")) | |
| best: dict[str, tuple[str, str]] = {} | |
| for record in _iter_ins_records(payload): | |
| operator_id = _doc_operator_id(record) | |
| issuing_cb = str( | |
| record.get("issuing_cb") | |
| or record.get("organismoDiCertificazione") | |
| or "" | |
| ).strip() | |
| issued = str(record.get("issued") or "").strip() | |
| previous = best.get(operator_id) | |
| if operator_id and issuing_cb and ( | |
| previous is None or issued > previous[0] | |
| ): | |
| best[operator_id] = (issued, issuing_cb) | |
| return { | |
| operator_id: issuing_cb | |
| for operator_id, (_issued, issuing_cb) in best.items() | |
| } | |
| def _open_artifact(path: Path): | |
| if path.suffix != ".gz": | |
| return sqlite3.connect(path), None | |
| temp_dir = tempfile.TemporaryDirectory(prefix="sinetti_cert_formats_") | |
| sqlite_path = Path(temp_dir.name) / "details.sqlite" | |
| with gzip.open(path, "rb") as source, sqlite_path.open("wb") as target: | |
| shutil.copyfileobj(source, target) | |
| return sqlite3.connect(sqlite_path), temp_dir | |
| def _fetch_pdf_bytes(doc: FormatDoc) -> bytes: | |
| if doc.scheme == "REDcert" or doc.pdf_url.startswith("redcert://"): | |
| try: | |
| from redcert_pdf_store import load_hosted_pdf | |
| hosted = load_hosted_pdf(doc.certificate_id) | |
| if hosted: | |
| return bytes(hosted[0]) | |
| except Exception: | |
| pass | |
| from cert_pdf_parsers.redcert import fetch_redcert_pdf_bytes | |
| return fetch_redcert_pdf_bytes( | |
| PdfDocumentRef( | |
| scheme=doc.scheme, | |
| certificate_id=doc.certificate_id, | |
| pdf_url=doc.pdf_url or f"redcert://{doc.certificate_id}", | |
| ) | |
| ) | |
| if not doc.pdf_url.startswith(("http://", "https://")): | |
| return b"" | |
| request = Request( | |
| doc.pdf_url, | |
| headers={"User-Agent": "Sinetti cert-format catalog builder"}, | |
| ) | |
| with urlopen(request, timeout=30) as response: | |
| return response.read() | |
| def _exemplar_docs_by_type_key(docs: list[FormatDoc]) -> dict[str, FormatDoc]: | |
| grouped: dict[tuple[str, str, str], list[FormatDoc]] = defaultdict(list) | |
| for doc in docs: | |
| bucket_kind, bucket_id = choose_bucket( | |
| doc.scheme, | |
| doc.parse_status, | |
| list(doc.reason_codes), | |
| ins_cb=doc.ins_cb, | |
| ) | |
| grouped[(doc.scheme, bucket_kind, bucket_id)].append(doc) | |
| return { | |
| type_key(scheme, bucket_kind, bucket_id): pick_exemplar(bucket_docs) | |
| for (scheme, bucket_kind, bucket_id), bucket_docs in grouped.items() | |
| } | |
| def _add_thumbnails( | |
| catalog: dict, | |
| docs: list[FormatDoc], | |
| *, | |
| thumbs_dir: Path, | |
| ) -> None: | |
| exemplars = _exemplar_docs_by_type_key(docs) | |
| for entries in catalog["schemes"].values(): | |
| for entry in entries: | |
| try: | |
| doc = exemplars[entry["type_key"]] | |
| pdf_bytes = _fetch_pdf_bytes(doc) | |
| if not pdf_bytes: | |
| continue | |
| rendered, _diagnostics = _render_pdf_pages( | |
| pdf_bytes, | |
| pages=[1], | |
| dpi=110, | |
| ) | |
| if not rendered: | |
| continue | |
| filename = thumb_filename(entry["type_key"]) | |
| thumbs_dir.mkdir(parents=True, exist_ok=True) | |
| rendered[0][1].save(thumbs_dir / filename, "PNG") | |
| entry["thumbnail"] = f"cert_format_thumbs/{filename}" | |
| except Exception: | |
| continue | |
| def main(argv: list[str] | None = None) -> int: | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument( | |
| "--artifact", | |
| default=str(ROOT / "data" / "cert_pdf_details.sqlite.gz"), | |
| ) | |
| parser.add_argument( | |
| "--out", | |
| default=str(ROOT / "data" / "cert_format_catalog.json"), | |
| ) | |
| parser.add_argument( | |
| "--ins-index", | |
| default=str(ROOT / "data" / "ins" / "cert_pdfs.json"), | |
| ) | |
| parser.add_argument("--thumbs-dir", default="") | |
| parser.add_argument("--skip-thumbs", action="store_true") | |
| args = parser.parse_args(argv) | |
| artifact = Path(args.artifact) | |
| out = Path(args.out) | |
| conn, temp_dir = _open_artifact(artifact) | |
| try: | |
| docs = load_format_docs( | |
| conn, | |
| ins_cb_by_cert_id=_load_ins_cb_map(Path(args.ins_index)), | |
| ) | |
| finally: | |
| conn.close() | |
| if temp_dir is not None: | |
| temp_dir.cleanup() | |
| catalog = build_catalog(docs, built_at=_utc_now()) | |
| # Lets the freshness guard (tests/test_cert_format_catalog_fresh.py) prove this catalog | |
| # describes the committed details artifact, not an older parse. | |
| catalog["source_sha256"] = artifact_sha256(artifact) | |
| if not args.skip_thumbs: | |
| thumbs_dir = ( | |
| Path(args.thumbs_dir) | |
| if args.thumbs_dir | |
| else out.parent / "cert_format_thumbs" | |
| ) | |
| _add_thumbnails(catalog, docs, thumbs_dir=thumbs_dir) | |
| write_catalog(out, catalog) | |
| report = { | |
| "status": "ok", | |
| "types": sum(len(entries) for entries in catalog["schemes"].values()), | |
| "docs": len(docs), | |
| "path": str(out), | |
| } | |
| print(json.dumps(report, indent=2, sort_keys=True)) | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |