Spaces:
Running
Running
| #!/usr/bin/env python | |
| """Cloud OCR job: read scanned ISCC certificates the nightly cannot, publish one candidate release. | |
| Owner name: the morning OCR run. Since SIN-0645 it also has a REDcert lane (redcert_cloud_ocr): | |
| scanned REDcert certificates, read from the dataset's hosted REDcert PDF store, merged into the same | |
| candidate. Its default cap is `redcert_cloud_ocr.DEFAULT_MAX_CERTS`; an explicit zero disables it. | |
| Runs as its own Hugging Face Scheduled Job each morning after the nightly data refresh has | |
| published (registration: scripts/register_hf_cloud_ocr_job.py). Runs the DEPLOYED Space code, | |
| like every scheduled job, inside an isolated job workspace: | |
| 1. hydrate the newest dataset release into ./data (data_plane.boot_sync, materialize=True); | |
| 2. queue scanned ISCC certificates never OCR'd before (iscc_cloud_ocr.select_queue), capped; | |
| 3. download each PDF and OCR it with the pinned Tesseract (the same reader the PC uses); | |
| 4. offer only results that pass the approved quality rules to the existing merge tool; | |
| 5. rebuild structured + app data when anything was promoted, record every attempt in the | |
| ledger that rides the release, and publish ONE candidate release on top of the baseline. | |
| Production publication is explicitly opted in with SINETTI_PROD_AUTO_ACTIVATE; otherwise the | |
| publish remains a candidate. Complete Trust Review and exact revision verification gate it. | |
| `SINETTI_REHEARSAL=1` withholds the publish (data_plane.publish) and nothing else is outward. | |
| `--parity-ids` re-reads certificates the PC already OCR'd and reports differences; it never | |
| publishes. Decision brief: docs/superpowers/briefs/2026-09-29-cloud-ocr-job-brief.md. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import shutil | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import time | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| # Never .resolve(): in a Hugging Face snapshot every file is a symlink into the blob store, and a | |
| # resolved root lands there instead of in the checkout (same rule as scripts/daily_refresh.py). | |
| ROOT = Path(__file__).parent.parent | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| import cert_pdf_store # noqa: E402 | |
| import data_plane # noqa: E402 | |
| import iscc_cloud_ocr as ocr_job # noqa: E402 | |
| import redcert_cloud_ocr as redcert_lane # noqa: E402 | |
| import rehearsal # noqa: E402 | |
| import state_paths # noqa: E402 | |
| from sinetti_services.refresh_job_lifecycle import ( # noqa: E402 | |
| can_start_tail, | |
| remaining_budget_s, | |
| ) | |
| PRODUCER = "iscc_cloud_ocr" | |
| def iscc_lane_deadline(started_monotonic: float, now_monotonic: float, | |
| budget_s: int, reserve_s: int, | |
| iscc_selected: int, redcert_selected: int) -> float | None: | |
| """Absolute ISCC cutoff that reserves REDcert's proportional share of the OCR window. | |
| No bounded budget, an empty lane, or REDcert disabled keeps the historical single-lane | |
| behavior. The global publication reserve remains the hard cutoff for both lanes. | |
| """ | |
| if budget_s <= 0 or iscc_selected <= 0 or redcert_selected <= 0: | |
| return None | |
| elapsed = max(0.0, now_monotonic - started_monotonic) | |
| work_window = max(0.0, budget_s - elapsed - max(0, reserve_s)) | |
| share = iscc_selected / (iscc_selected + redcert_selected) | |
| return now_monotonic + work_window * share | |
| def can_start_ocr(*, now_monotonic: float, started_monotonic: float, budget_s: int, | |
| reserve_s: int, lane_deadline: float | None = None) -> bool: | |
| """Whether another OCR document may start under the lane and global cutoffs.""" | |
| if lane_deadline is not None and now_monotonic >= lane_deadline: | |
| return False | |
| if budget_s <= 0: | |
| return True | |
| remaining = remaining_budget_s(started_monotonic=started_monotonic, budget_s=budget_s, | |
| now_monotonic=now_monotonic) | |
| return can_start_tail(remaining_s=remaining, reserve_s=reserve_s) | |
| def review_candidate(*, data_dir: Path, state_root: Path) -> dict: | |
| """Catalogue and review the entire fixed certificate release; fail closed on lost evidence.""" | |
| from scripts.build_cert_format_catalog import main as build_catalog | |
| import sinetti_data | |
| from schemes import CERTIFICATION_SCHEMES, LOADED_CERTIFICATION_SCHEMES | |
| from sinetti_services import refresh_publication, trust_review_service | |
| try: | |
| if build_catalog([ | |
| "--artifact", str(data_dir / "cert_pdf_details.sqlite.gz"), | |
| "--out", str(data_dir / "cert_format_catalog.json"), | |
| "--ins-index", str(data_dir / "ins" / "cert_pdfs.json"), "--skip-thumbs", | |
| ]) != 0: | |
| raise ValueError("certificate-format catalog build failed") | |
| identity = refresh_publication.capture_review_identity(data_dir) | |
| groups = sinetti_data.load_app_certificate_groups(data_dir / "sinetti_app.sqlite.gz") | |
| review_schemes = tuple(scheme for scheme in CERTIFICATION_SCHEMES | |
| if scheme in LOADED_CERTIFICATION_SCHEMES or groups.get(scheme)) | |
| review = trust_review_service.safe_post_refresh_review( | |
| candidate_artifact=data_dir / "sinetti_app.sqlite.gz", | |
| details_artifact=data_dir / "cert_pdf_details.sqlite.gz", | |
| schemes=review_schemes, | |
| store_path=state_root / "trust_review.sqlite", | |
| ) | |
| if refresh_publication.capture_review_identity(data_dir) != identity: | |
| raise ValueError("candidate changed during review") | |
| review["publication_artifact_ids"] = identity["artifact_ids"] | |
| return review | |
| except Exception as exc: # noqa: BLE001 - a candidate may be saved, never falsely reviewed | |
| return {"status": "failed", "error": f"{type(exc).__name__}: {str(exc)[:300]}"} | |
| def activate_publication(**kwargs) -> dict: | |
| from sinetti_services.refresh_publication import publish_to_production | |
| return publish_to_production(**kwargs) | |
| def _env_int(name: str, default: int) -> int: | |
| try: | |
| return int(os.environ.get(name) or default) | |
| except (TypeError, ValueError): | |
| return default | |
| def _now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat(timespec="seconds") | |
| def engine_report(engine) -> dict: | |
| """Version + model checksum of the Tesseract the engine will call, and whether it is pinned.""" | |
| executable = engine._executable() # noqa: SLF001 -- same resolution the OCR itself uses | |
| if not executable: | |
| return {"executable": "", "version": "", "model_sha256": "", | |
| "problem": ocr_job.engine_pin_problem(version="", model_sha256="")} | |
| def _run(*args: str) -> str: | |
| done = subprocess.run([executable, *args], capture_output=True, text=True, timeout=60, | |
| check=False) | |
| return (done.stdout or "") + (done.stderr or "") | |
| version = ocr_job.tesseract_series(_run("--version")) | |
| model_dir = ocr_job.tessdata_dir(_run("--list-langs")) | |
| model = Path(model_dir) / "eng.traineddata" if model_dir else None | |
| model_sha = ocr_job.sha256_file(model) if model is not None and model.is_file() else "" | |
| return { | |
| "executable": executable, | |
| "version": version, | |
| "model_path": str(model or ""), | |
| "model_sha256": model_sha, | |
| "problem": ocr_job.engine_pin_problem(version=version, model_sha256=model_sha), | |
| } | |
| def download_pdfs(queue: list[dict], pdf_dir: Path, *, fetch=None) -> tuple[list[dict], list[dict]]: | |
| """(entries with a local PDF, failed download attempts). A failure is reported, never fatal.""" | |
| if fetch is None: | |
| from cert_pdf_download import fetch_url_bytes as fetch | |
| pdf_dir.mkdir(parents=True, exist_ok=True) | |
| ready, failed = [], [] | |
| for entry in queue: | |
| target = pdf_dir / f"{entry['certificate_id']}.pdf" | |
| try: | |
| data = fetch(entry["pdf_url"], timeout=90) | |
| if not data.startswith(b"%PDF"): | |
| raise ValueError("response is not a PDF") | |
| target.write_bytes(data) | |
| ready.append({**entry, "local_pdf": target.name}) | |
| except Exception as exc: # noqa: BLE001 -- one broken link must not stop the run | |
| failed.append({ | |
| "certificate_id": entry["certificate_id"], | |
| "disposition": ocr_job.DOWNLOAD_FAILED, | |
| "ocr_status": "", | |
| "pdf_sha256": entry.get("pdf_sha256", ""), | |
| "error": f"{type(exc).__name__}: {str(exc)[:200]}", | |
| }) | |
| return ready, failed | |
| def run_ocr(entries: list[dict], scratch: Path, engine, *, should_continue) -> tuple[dict, Path]: | |
| from cert_pdf_ocr import run_iscc_ocr_backlog | |
| manifest = scratch / "backlog.jsonl" | |
| manifest.write_text( | |
| "".join(json.dumps(ocr_job.backlog_manifest_row(entry, f"pdfs/{entry['local_pdf']}")) + "\n" | |
| for entry in entries), | |
| encoding="utf-8", | |
| ) | |
| ocr_db = scratch / "ocr.sqlite" | |
| report = run_iscc_ocr_backlog( | |
| manifest_path=manifest, | |
| pdf_root=scratch / "pdfs", | |
| root=scratch, | |
| output_db=ocr_db, | |
| artifact_path=scratch / "ocr.sqlite.gz", | |
| ocr_engine=engine, | |
| resume=False, | |
| should_continue=should_continue, | |
| ) | |
| return report, ocr_db | |
| def merge_and_rebuild(*, ocr_db: Path | None, accepted: set[str], scratch: Path, data_dir: Path, | |
| rebuild=None, validate=None, force_rebuild: bool = False, | |
| redcert: dict | None = None) -> dict: | |
| """Merge accepted results into the release's details DB; rebuild what reads it when changed. | |
| `redcert` ({"ocr_db", "accepted"}) is the REDcert lane's share, merged after the ISCC results | |
| with its own reason code and closing rule. `force_rebuild` (rehearsal only) rebuilds even when | |
| nothing was promoted, so a rehearsal on a day without recoveries still proves and times the | |
| rebuild in the cloud image.""" | |
| from cert_pdf_ocr_promote import merge_ocr_recoveries | |
| details_gz = data_dir / "cert_pdf_details.sqlite.gz" | |
| work_db = data_dir / "cert_pdf_details.sqlite" | |
| redcert_accepted = set((redcert or {}).get("accepted") or ()) | |
| result: dict = {"status": "skipped-nothing-accepted", "promoted_ids": [], | |
| "redcert_promoted_ids": []} | |
| promoted: list[str] = [] | |
| redcert_promoted: list[str] = [] | |
| if accepted or redcert_accepted: | |
| source = scratch / "details-base.sqlite" | |
| cert_pdf_store.gunzip_artifact(details_gz, source) | |
| result["status"] = "ok" | |
| if accepted: | |
| accepted_db = scratch / "accepted.sqlite" | |
| ocr_job.write_accepted_db(ocr_db, accepted_db, accepted) | |
| result["merge"] = merge_ocr_recoveries( | |
| source, accepted_db, work_db, | |
| reason_codes=frozenset({ocr_job.OCR_REASON_CODE}), replace=True, | |
| closed_empty_codes=ocr_job.CLOSED_EMPTY_CODES, | |
| ) | |
| promoted = sorted(ocr_job.promoted_ids(work_db, accepted)) | |
| source = scratch / "details-after-iscc.sqlite" | |
| work_db.replace(source) | |
| if redcert_accepted: | |
| redcert_db = scratch / "redcert-accepted.sqlite" | |
| ocr_job.write_accepted_db(redcert["ocr_db"], redcert_db, redcert_accepted) | |
| result["redcert_merge"] = merge_ocr_recoveries( | |
| source, redcert_db, work_db, | |
| reason_codes=frozenset({redcert_lane.OCR_REASON_CODE}), replace=True, | |
| closed_empty_codes=redcert_lane.CLOSED_EMPTY_CODES, | |
| ) | |
| redcert_promoted = sorted(redcert_lane.promoted_ids(work_db, redcert_accepted)) | |
| else: | |
| source.replace(work_db) | |
| result["promoted_ids"] = promoted | |
| result["redcert_promoted_ids"] = redcert_promoted | |
| if not promoted and not redcert_promoted: | |
| result["status"] = "skipped-nothing-promoted" | |
| if not promoted and not redcert_promoted: | |
| work_db.unlink(missing_ok=True) # the release keeps its own, unchanged, artifact | |
| if not force_rebuild: | |
| return result | |
| else: | |
| cert_pdf_store.gzip_artifact(work_db, details_gz) | |
| if rebuild is None: | |
| from scripts.run_cert_pdf_delta_refresh import rebuild_downstream_from_raw_artifact | |
| rebuild = rebuild_downstream_from_raw_artifact | |
| if validate is None: | |
| import sinetti_data_release | |
| validate = sinetti_data_release.validate_data_release | |
| rebuild_started = time.monotonic() | |
| downstream = rebuild( | |
| raw_work_db=work_db, | |
| raw_artifact=details_gz, | |
| structured_work_db=data_dir / "cert_pdf_structured.sqlite", | |
| structured_artifact=data_dir / "cert_pdf_structured.sqlite.gz", | |
| app_root=data_dir.parent, | |
| ) | |
| result["rebuild"] = { | |
| "minutes": round((time.monotonic() - rebuild_started) / 60, 1), | |
| "forced": bool(force_rebuild and not promoted and not redcert_promoted), | |
| "status": downstream.get("status"), | |
| "structured": downstream.get("structured"), | |
| "app_db": {k: (downstream.get("app_db") or {}).get(k) for k in ("status", "error")}, | |
| } | |
| if (downstream.get("app_db") or {}).get("status") != "ok": | |
| result["status"] = "error-rebuild" | |
| return result | |
| validation = validate( | |
| data_dir / "sinetti_app.sqlite.gz", | |
| manifest_path=data_dir / "sinetti_data_release.json", | |
| inspect_database=True, | |
| ) | |
| result["validation"] = {"ok": bool(validation.get("ok")), "reason": validation.get("reason")} | |
| if not validation.get("ok"): | |
| result["status"] = "error-validation" | |
| return result | |
| def build_arg_parser() -> argparse.ArgumentParser: | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--job-workspace", action="store_true", | |
| help="required: this checkout is a disposable job workspace whose ./data may " | |
| "be replaced by the newest dataset release") | |
| ap.add_argument("--data-repo-id", default=os.environ.get("SINETTI_DATA_REPO_ID", "")) | |
| ap.add_argument("--max-certs", type=int, | |
| default=_env_int("CLOUD_OCR_MAX_CERTS", ocr_job.DEFAULT_MAX_CERTS)) | |
| ap.add_argument("--redcert-max-certs", type=int, | |
| default=_env_int(redcert_lane.MAX_CERTS_ENV, redcert_lane.DEFAULT_MAX_CERTS), | |
| help="REDcert scans per run (SIN-0645); 0 keeps the REDcert lane off") | |
| ap.add_argument("--publish-reserve-s", type=int, | |
| default=_env_int("CLOUD_OCR_PUBLISH_RESERVE_S", ocr_job.DEFAULT_PUBLISH_RESERVE_S)) | |
| ap.add_argument("--dpi", type=int, default=300) | |
| ap.add_argument("--tesseract-cmd", default="tesseract") | |
| ap.add_argument("--parity-ids", default="", | |
| help="comma-separated certificates the PC already OCR'd: re-read them and " | |
| "report differences; never publishes") | |
| ap.add_argument("--retry-ids", default="", | |
| help="REHEARSAL ONLY: queue these scanned certificates even though they were " | |
| "tried before (proves the OCR path on a day with no new scans)") | |
| ap.add_argument("--rehearse-rebuild", action="store_true", | |
| help="REHEARSAL ONLY: run the structured/app rebuild even when nothing was " | |
| "promoted, to prove and time it in the cloud image") | |
| return ap | |
| def main(argv: list[str] | None = None) -> int: | |
| from sinetti_operator.registration import register_live_providers | |
| register_live_providers() # jobs never reach full_registration (SIN-0999) | |
| args = build_arg_parser().parse_args(argv) | |
| if rehearsal.is_rehearsal(): | |
| # A mounted production bucket may be present even during rehearsal. Keep every review | |
| # and final receipt off it, including early failures and independent state-path aliases. | |
| with tempfile.TemporaryDirectory(prefix="sinetti-ocr-rehearsal-state-") as temporary: | |
| return _run_job(args, state_root=Path(temporary)) | |
| return _run_job(args, state_root=state_paths.external_state_root() or ROOT / "data") | |
| def _run_job(args: argparse.Namespace, *, state_root: Path) -> int: | |
| from cert_pdf_ocr import TesseractCliOcrEngine | |
| started_mono = time.monotonic() | |
| budget_s = _env_int("SINETTI_JOB_BUDGET_S", 0) | |
| from sinetti_services import refresh_publication | |
| activation_enabled = refresh_publication.production_publication_enabled() | |
| effective_reserve = args.publish_reserve_s | |
| if activation_enabled: | |
| effective_reserve = max( | |
| effective_reserve, | |
| ocr_job.DEFAULT_PUBLISH_RESERVE_S + refresh_publication.publication_budget_s(), | |
| ) | |
| run_id = _now_iso() | |
| report: dict = { | |
| "producer": PRODUCER, "run_id": run_id, "started_at": run_id, | |
| "rehearsal": rehearsal.is_rehearsal(), | |
| "config": {"max_certs": args.max_certs, "redcert_max_certs": args.redcert_max_certs, | |
| "publish_reserve_s": effective_reserve, | |
| "prod_auto_activate": activation_enabled, | |
| "budget_s": budget_s, "dpi": args.dpi, | |
| "parity_ids": [c for c in args.parity_ids.split(",") if c.strip()]}, | |
| # SIN-1007: which code this run executed -- a pinned Space commit, or a jobs code release | |
| # (step 3) -- next to the commit the app ran when it started. | |
| "code": {key: os.environ.get(name) or "" for key, name in ( | |
| ("space_revision", "SINETTI_SPACE_REVISION"), ("app_revision", "SINETTI_APP_REVISION"), | |
| ("jobs_code_revision", "SINETTI_JOBS_CODE_REVISION"), | |
| ("jobs_code_release", "SINETTI_JOBS_CODE_RELEASE"), | |
| ("jobs_code_source", "SINETTI_JOBS_CODE_SOURCE"))}, | |
| } | |
| def finish(status: str, code: int) -> int: | |
| report["status"] = status | |
| report["finished_at"] = _now_iso() | |
| report["duration_s"] = round(time.monotonic() - started_mono, 1) | |
| # This final receipt is outside the candidate snapshot: it can record its real commit. | |
| ocr_job.write_json(state_root / "refresh" / "last_cloud_ocr_publication.json", report) | |
| print(json.dumps(report, indent=2, sort_keys=True, default=str), flush=True) | |
| return code | |
| retry_ids = [c.strip() for c in args.retry_ids.split(",") if c.strip()] | |
| if (retry_ids or args.rehearse_rebuild) and not rehearsal.is_rehearsal(): | |
| # Re-reading tried certificates is an owner decision (brief: out of scope), and a forced | |
| # rebuild publishes nothing new; both exist only to prove the path in a rehearsal. | |
| report["config"]["refused"] = "--retry-ids / --rehearse-rebuild need SINETTI_REHEARSAL=1" | |
| return finish("error-rehearsal-only-flag", 2) | |
| engine = TesseractCliOcrEngine(tesseract_cmd=args.tesseract_cmd, dpi=args.dpi) | |
| report["engine"] = engine_report(engine) | |
| print(f"engine: {json.dumps(report['engine'])}", flush=True) | |
| if report["engine"]["problem"]: | |
| return finish("error-engine-not-pinned", 1) | |
| if not args.job_workspace or not args.data_repo_id: | |
| report["baseline"] = {"status": "error-not-a-job-workspace", | |
| "error": "needs --job-workspace and SINETTI_DATA_REPO_ID"} | |
| return finish("error", 1) | |
| token = os.environ.get("HF_TOKEN", "") or os.environ.get("HF_JOB_TOKEN", "") | |
| data_dir = ROOT / "data" | |
| report["baseline"] = data_plane.boot_sync( | |
| repo_id=args.data_repo_id, token=token, data_dir=data_dir, materialize=True, | |
| ) | |
| baseline_revision = report["baseline"].get("hub_revision") | |
| if report["baseline"].get("status") != "ok" or not baseline_revision: | |
| return finish("error-baseline", 1) | |
| details_gz = data_dir / "cert_pdf_details.sqlite.gz" | |
| with tempfile.TemporaryDirectory(prefix="sinetti-cloud-ocr-") as tmp: | |
| scratch = Path(tmp) | |
| parity_ids = report["config"]["parity_ids"] | |
| if parity_ids: | |
| baseline_db = scratch / "details-release.sqlite" | |
| cert_pdf_store.gunzip_artifact(details_gz, baseline_db) | |
| queue = ocr_job.parity_queue(baseline_db, parity_ids) | |
| report["queue"] = {"parity": True, "selected": len(queue)} | |
| else: | |
| ledger = ocr_job.load_ledger(ROOT / ocr_job.LEDGER_REL) | |
| seed = ocr_job.load_seed(ROOT / ocr_job.SEED_REL) | |
| retry_set = set(retry_ids) | |
| attempted = {cid: pin for cid, pin in ocr_job.attempted_index(ledger, seed).items() | |
| if cid not in retry_set} | |
| queue, report["queue"] = ocr_job.select_queue( | |
| details_gz, attempted, | |
| max_certs=len(retry_ids) + 10**6 if retry_ids else args.max_certs, | |
| ) | |
| if retry_ids: | |
| wanted = set(retry_ids) | |
| queue = [entry for entry in queue if entry["certificate_id"] in wanted] | |
| queue = queue[:args.max_certs] | |
| report["queue"]["retry_ids"] = len(queue) | |
| report["queue"]["seed_ids"] = len(seed) | |
| redcert_ledger = ocr_job.load_ledger(ROOT / redcert_lane.LEDGER_REL) | |
| redcert_attempted = { | |
| cid: pin for cid, pin in ocr_job.attempted_index( | |
| redcert_ledger, set(), | |
| legacy_reader_version=redcert_lane.LEGACY_READER_VERSION).items() | |
| if cid not in retry_set} | |
| redcert_queue, redcert_stats = redcert_lane.select_queue( | |
| details_gz, redcert_attempted, | |
| max_certs=len(retry_ids) + 10**6 if retry_ids else args.redcert_max_certs, | |
| ) | |
| if retry_ids: | |
| redcert_queue = [entry for entry in redcert_queue | |
| if entry["certificate_id"] in retry_set][:args.redcert_max_certs] | |
| redcert_stats["retry_ids"] = len(redcert_queue) | |
| report["redcert"] = {"queue": redcert_stats} | |
| print(f"redcert queue: {json.dumps(redcert_stats)}", flush=True) | |
| print(f"queue: {json.dumps(report['queue'])}", flush=True) | |
| if parity_ids: | |
| redcert_queue = [] | |
| if queue: | |
| # The reader checks each queued PDF against the published register (SIN-0443). A | |
| # partial context cannot establish source identity, so refuse before any lane starts. | |
| queued_ids = sorted({str(entry["certificate_id"]) for entry in queue}) | |
| report["queue"]["register_context_required"] = len(queued_ids) | |
| report["queue"]["register_context_error"] = "" | |
| report["queue"]["register_context_missing_ids"] = [] | |
| report["queue"]["register_context_incomplete_fields"] = {} | |
| try: | |
| context = ocr_job.load_register_context( | |
| data_dir / "sinetti_app.sqlite.gz", set(queued_ids)) | |
| except Exception as exc: # noqa: BLE001 - report loader failures before processing | |
| report["queue"]["register_context"] = 0 | |
| report["queue"]["register_context_error"] = f"{type(exc).__name__}: {str(exc)[:200]}" | |
| return finish("error-register-context", 1) | |
| missing_ids = [cid for cid in queued_ids if cid not in context] | |
| incomplete_fields = { | |
| cid: [field for field in ocr_job.REGISTER_CONTEXT_FIELDS | |
| if not str((context.get(cid) or {}).get(field) or "").strip()] | |
| for cid in queued_ids if cid in context | |
| } | |
| incomplete_fields = {cid: fields for cid, fields in incomplete_fields.items() if fields} | |
| report["queue"]["register_context_missing_ids"] = missing_ids | |
| report["queue"]["register_context_incomplete_fields"] = incomplete_fields | |
| report["queue"]["register_context"] = len(context) | |
| # A certificate the published register does not describe completely (old backlog | |
| # entries that left the register, a blank holder or period) cannot be checked against | |
| # its PDF, so it is never read -- but one such entry must not stop the whole run: | |
| # v2026.10.6 aborted every morning on five such certificates. Skip and report them. | |
| skipped_ids = sorted(set(missing_ids) | set(incomplete_fields)) | |
| report["queue"]["register_context_skipped_ids"] = skipped_ids | |
| if skipped_ids: | |
| skipped = set(skipped_ids) | |
| queue = [entry for entry in queue if str(entry["certificate_id"]) not in skipped] | |
| print(f"register context: skipped {len(skipped_ids)} certificate(s) without complete " | |
| f"register context: {', '.join(skipped_ids[:10])}", flush=True) | |
| queue = ocr_job.attach_register_context(queue, context) | |
| if redcert_queue: | |
| redcert_queue, context_report = redcert_lane.attach_reviewed_register_context( | |
| redcert_queue, data_dir / "sinetti_app.sqlite.gz") | |
| report["redcert"]["queue"].update(context_report) | |
| if context_report["register_context_error"]: | |
| return finish("error-redcert-register-context", 1) | |
| if not queue and not redcert_queue: | |
| return finish("ok-nothing-to-do", 0) | |
| allocation_started = time.monotonic() | |
| iscc_deadline = iscc_lane_deadline( | |
| started_mono, allocation_started, budget_s, effective_reserve, | |
| len(queue), len(redcert_queue)) | |
| if iscc_deadline is not None: | |
| available = max(0.0, budget_s - (allocation_started - started_mono) | |
| - effective_reserve) | |
| report["config"]["ocr_budget_allocation"] = { | |
| "iscc_selected": len(queue), "redcert_selected": len(redcert_queue), | |
| "work_window_s": round(available, 1), | |
| "iscc_share_s": round(iscc_deadline - allocation_started, 1), | |
| } | |
| def should_continue() -> bool: | |
| return can_start_ocr( | |
| now_monotonic=time.monotonic(), started_monotonic=started_mono, | |
| budget_s=budget_s, reserve_s=effective_reserve) | |
| def iscc_should_continue() -> bool: | |
| return can_start_ocr( | |
| now_monotonic=time.monotonic(), started_monotonic=started_mono, | |
| budget_s=budget_s, reserve_s=effective_reserve, lane_deadline=iscc_deadline) | |
| ocr_db: Path | None = None | |
| results: list[dict] = [] | |
| download_failed: list[dict] = [] | |
| ocr_report: dict = {"status": "ok", "queued": 0} | |
| ocr_minutes = 0.0 | |
| if queue: | |
| ready, download_failed = download_pdfs(queue, scratch / "pdfs") | |
| report["download"] = {"ok": len(ready), "failed": len(download_failed), | |
| "failures": download_failed[:10]} | |
| ocr_started = time.monotonic() | |
| ocr_report, ocr_db = run_ocr( | |
| ready, scratch, engine, should_continue=iscc_should_continue) | |
| results = ocr_job.classify_results(ocr_db) if ocr_db.exists() else [] | |
| ocr_minutes = (time.monotonic() - ocr_started) / 60 | |
| report["ocr"] = { | |
| **{k: ocr_report.get(k) for k in ("status", "queued", "parsed", "no_materials", | |
| "failed", "stopped_no_time", "reason")}, | |
| "minutes": round(ocr_minutes, 1), | |
| "minutes_per_certificate": round(ocr_minutes / len(results), 2) if results else None, | |
| "dispositions": { | |
| name: sum(1 for r in results if r["disposition"] == name) | |
| for name in (ocr_job.ACCEPT, ocr_job.HOLD, ocr_job.UNREADABLE) | |
| }, | |
| } | |
| if ocr_report.get("status") != "ok": | |
| return finish("error-ocr", 1) | |
| if parity_ids: | |
| report["parity"] = ocr_job.parity_compare(baseline_db, ocr_db, parity_ids) | |
| return finish("ok-parity", 0) | |
| # REDcert lane (SIN-0645): same engine and time budget, after the ISCC lane. | |
| redcert_results: list[dict] = [] | |
| redcert_failed: list[dict] = [] | |
| redcert_ocr_db = scratch / "redcert-ocr.sqlite" | |
| if redcert_queue: | |
| redcert_pdfs = scratch / "redcert-pdfs" | |
| redcert_ready, redcert_failed, redcert_stale = redcert_lane.fetch_hosted_pdfs( | |
| redcert_queue, redcert_pdfs) | |
| report["redcert"]["download"] = { | |
| "ok": len(redcert_ready), "failed": len(redcert_failed), | |
| "stale": len(redcert_stale), "failures": redcert_failed[:10], | |
| "stale_files": redcert_stale[:10]} | |
| redcert_started = time.monotonic() | |
| redcert_report = redcert_lane.run_ocr( | |
| redcert_ready, redcert_pdfs, redcert_ocr_db, engine=engine, | |
| should_continue=should_continue) | |
| if redcert_ocr_db.exists(): | |
| redcert_results = redcert_lane.classify_results(redcert_ocr_db) | |
| report["redcert"]["ocr"] = { | |
| **redcert_report, | |
| "minutes": round((time.monotonic() - redcert_started) / 60, 1), | |
| "dispositions": { | |
| name: sum(1 for r in redcert_results if r["disposition"] == name) | |
| for name in (redcert_lane.ACCEPT, redcert_lane.HOLD, redcert_lane.UNREADABLE) | |
| }, | |
| } | |
| if redcert_report.get("status") != "ok": | |
| return finish("error-ocr", 1) | |
| redcert_accepted = {r["certificate_id"] for r in redcert_results | |
| if r["disposition"] == redcert_lane.ACCEPT} | |
| accepted = {r["certificate_id"] for r in results if r["disposition"] == ocr_job.ACCEPT} | |
| report["merge"] = merge_and_rebuild( | |
| ocr_db=ocr_db, accepted=accepted, scratch=scratch, data_dir=data_dir, | |
| force_rebuild=args.rehearse_rebuild, | |
| redcert={"ocr_db": redcert_ocr_db, "accepted": redcert_accepted}, | |
| ) | |
| if report["merge"]["status"].startswith("error"): | |
| return finish("error-merge", 1) | |
| promoted = set(report["merge"]["promoted_ids"]) | |
| reader = ocr_job.OCR_READER_VERSION | |
| attempts = [ | |
| {**r, "promoted": r["certificate_id"] in promoted, "reader": reader} for r in results | |
| ] + download_failed | |
| redcert_promoted = set(report["merge"]["redcert_promoted_ids"]) | |
| redcert_attempts = [ | |
| {**r, "promoted": r["certificate_id"] in redcert_promoted, | |
| "reader": redcert_lane.OCR_READER_VERSION} for r in redcert_results | |
| ] + redcert_failed | |
| report["attempts"] = len(attempts) + len(redcert_attempts) | |
| if not report["attempts"]: | |
| return finish("ok-nothing-attempted", 0) | |
| if attempts: | |
| ledger = ocr_job.record_attempts(ledger, attempts, run_id=run_id) | |
| ocr_job.write_json(ROOT / ocr_job.LEDGER_REL, ledger) | |
| report["ledger_entries"] = len(ledger["entries"]) | |
| if redcert_attempts: | |
| redcert_ledger = ocr_job.record_attempts(redcert_ledger, redcert_attempts, run_id=run_id) | |
| ocr_job.write_json(ROOT / redcert_lane.LEDGER_REL, redcert_ledger) | |
| report["redcert"]["ledger_entries"] = len(redcert_ledger["entries"]) | |
| # The candidate's internal receipt predates upload and cannot contain its own commit. | |
| ocr_job.write_json(ROOT / ocr_job.RECEIPT_REL, {**report, "status": "candidate-built"}) | |
| with tempfile.TemporaryDirectory(prefix="sinetti-ocr-candidate-") as tmp: | |
| candidate_root = Path(tmp) | |
| candidate_data_dir = candidate_root / "data" | |
| # Physical copies freeze snapshot symlinks before checks, upload and activation. | |
| shutil.copytree(data_dir, candidate_data_dir, copy_function=shutil.copyfile) | |
| report["trust_review"] = review_candidate(data_dir=candidate_data_dir, state_root=state_root) | |
| try: | |
| artifact_ids = report["trust_review"].get("publication_artifact_ids") | |
| if artifact_ids and refresh_publication.capture_review_identity(candidate_data_dir)["artifact_ids"] != artifact_ids: | |
| raise ValueError("candidate changed after review") | |
| report["publish"] = data_plane.publish( | |
| candidate_root, repo_id=args.data_repo_id, token=token, producer=PRODUCER, | |
| expected_parent=baseline_revision, | |
| ) | |
| except Exception as exc: # noqa: BLE001 - stale parent or mutation saves no release | |
| report["publish"] = {"status": "error", | |
| "error": f"{type(exc).__name__}: {str(exc)[:300]}"} | |
| if report["publish"].get("status") != "ok": | |
| return finish("error-publish", 1) | |
| # Flush independent refresh success before a restart/rollback can fail or time out. | |
| ocr_job.write_json(state_root / "refresh" / "last_cloud_ocr_publication.json", report) | |
| if activation_enabled and not report["publish"].get("rehearsal"): | |
| space = os.environ.get("HF_SPACE_REPO_ID", "") | |
| base_url = os.environ.get("SINETTI_PROD_AUTO_ACTIVATE_BASE_URL") or ( | |
| f"https://{space.replace('/', '-').lower()}.hf.space" if space else "" | |
| ) | |
| try: | |
| report["auto_activation"] = activate_publication( | |
| publish=report["publish"], review=report["trust_review"], | |
| candidate_data_dir=candidate_data_dir, state_root=state_root, | |
| space_repo_id=space, token=token, base_url=base_url, by=PRODUCER, | |
| remaining_s=remaining_budget_s(started_monotonic=started_mono, budget_s=budget_s, | |
| now_monotonic=time.monotonic()), | |
| ) | |
| except Exception as exc: # noqa: BLE001 - preserve successful refresh and surface failure | |
| report["auto_activation"] = {"status": "error", | |
| "reason": f"{type(exc).__name__}: {str(exc)[:300]}"} | |
| else: | |
| report["auto_activation"] = {"status": "skipped", "reason": "rehearsal-or-disabled"} | |
| return finish("ok", 0) | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |