#!/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())