"""Restricted production-context smoke tests for OCR and recovery schema.""" from __future__ import annotations import io import base64 import hashlib import json import re import time from collections import Counter from pathlib import Path from typing import Any from PIL import Image, ImageDraw from bridge_platform.storage.service import read_bytes, write_bytes from apps.wp_invoices.services.extractor import run_item_recovery_v2 from apps.wp_invoices.services.ocr_layout import extract_layout APP_ID = "wp_invoices" PIPELINE_VERSION = "unified-v2.3-ocr-layout" def run_smoke_tests(context: dict[str, Any]) -> dict[str, Any]: tenant_id = str(((context.get("tenant") or {}).get("tenant_id")) or "").strip() payload = context.get("payload") or {} invoice_id = str(payload.get("invoice_id") or "").strip() request_id = str(context.get("request_id") or "unknown") if not tenant_id or not re.fullmatch(r"inv_[a-f0-9]{32}", invoice_id): raise ValueError("invalid_input") test_mode = str(payload.get("test") or "all") if test_mode == "audit_report": return _read_audit_report(tenant_id, invoice_id, str(payload.get("read_request_id") or "")) read_request_id = str(payload.get("read_request_id") or "").strip() if read_request_id: return _read_smoke(tenant_id, read_request_id) root = Path("diagnostics") / "smoke" / request_id if test_mode == "forced_full": return _run_forced_full(context, tenant_id, invoice_id, payload) ocr_artifact = {"status": "not_run", "reason": "test_not_selected"} if test_mode in {"all", "ocr_layout"}: document, mimetype = _existing_document(tenant_id, invoice_id) ocr_started = time.monotonic() try: layout = extract_layout(document, mimetype) ocr_artifact = { "status": "success", "ocr_attempted": True, "ocr_success": True, "pipeline_version": PIPELINE_VERSION, "execution_mode": "ocr_layout", "duration_ms": round((time.monotonic() - ocr_started) * 1000, 2), "preprocessing": {"grayscale": True, "autocontrast": True, "minimum_width": 900, "maximum_height": 4000, "contrast_factor": 1.35, "tile_height": 500, "parallel_workers": 2, "omp_thread_limit": 1, "tesseract_psm": 6, "language": "eng"}, **layout, "stderr": "", } except Exception as exc: ocr_artifact = { "status": "error", "ocr_attempted": True, "ocr_success": False, "pipeline_version": PIPELINE_VERSION, "execution_mode": "visual_fallback", "duration_ms": round((time.monotonic() - ocr_started) * 1000, 2), "preprocessing": {"grayscale": True, "autocontrast": True, "minimum_width": 900, "maximum_height": 4000, "contrast_factor": 1.35, "tile_height": 500, "parallel_workers": 2, "omp_thread_limit": 1, "tesseract_psm": 6, "language": "eng"}, "engine": "tesseract", "error": _safe_error(exc), "lines": [], } _write_json(tenant_id, root / "08-ocr-layout.json", ocr_artifact) recovery_artifact = {"status": "not_run", "reason": "test_not_selected"} if test_mode in {"all", "recovery_schema"}: recovery_started = time.monotonic() try: synthetic = _synthetic_image() recovered = run_item_recovery_v2( data=synthetic, filename="synthetic-recovery-smoke.png", mimetype="image/png", engine="mini", context=context, supplier_hints="Synthetic smoke test. Return the single visible item only.", ) recovery_artifact = { "status": "success", "schema_accepted": True, "strict_parse_success": True, "pipeline_version": PIPELINE_VERSION, "duration_ms": round((time.monotonic() - recovery_started) * 1000, 2), "model": recovered.get("model"), "provider_request_id": recovered.get("provider_request_id"), "provider_response_id": recovered.get("provider_response_id"), "usage": recovered.get("usage") or {}, "parsed": recovered.get("parsed"), } except Exception as exc: recovery_artifact = { "status": "error", "schema_accepted": False, "strict_parse_success": False, "pipeline_version": PIPELINE_VERSION, "duration_ms": round((time.monotonic() - recovery_started) * 1000, 2), "error": _safe_error(exc), } _write_json(tenant_id, root / "10-recovery-response-format.json", recovery_artifact) summary = { "pipeline_version": PIPELINE_VERSION, "ocr_passed": ocr_artifact["status"] == "success" and bool(ocr_artifact.get("lines")), "recovery_schema_passed": recovery_artifact["status"] == "success", "artifact_root": f"diagnostics:smoke:{request_id}", } _write_json(tenant_id, root / "00-smoke-manifest.json", summary) return summary def _read_smoke(tenant_id: str, request_id: str) -> dict[str, Any]: root = Path("diagnostics") / "smoke" / request_id artifacts = {} for name in ("00-smoke-manifest.json", "08-ocr-layout.json", "10-recovery-response-format.json"): try: artifacts[name] = json.loads(read_bytes(tenant_id=tenant_id, app_id=APP_ID, relative_path=root / name)) except (FileNotFoundError, json.JSONDecodeError): artifacts[name] = {"status": "error", "error": {"code": "artifact_missing"}} manifest = artifacts["00-smoke-manifest.json"] return {"pipeline_version": PIPELINE_VERSION, "ocr_passed": bool(manifest.get("ocr_passed")), "recovery_schema_passed": bool(manifest.get("recovery_schema_passed")), "artifact_root": f"diagnostics:smoke:{request_id}", "artifacts": artifacts} def _run_forced_full(context: dict[str, Any], tenant_id: str, invoice_id: str, payload: dict[str, Any]) -> dict[str, Any]: ocr_ref = str(payload.get("ocr_smoke_request_id") or "") recovery_ref = str(payload.get("recovery_smoke_request_id") or "") ocr_manifest = _read_smoke(tenant_id, ocr_ref)["artifacts"]["00-smoke-manifest.json"] recovery_manifest = _read_smoke(tenant_id, recovery_ref)["artifacts"]["00-smoke-manifest.json"] if not ocr_manifest.get("ocr_passed") or not recovery_manifest.get("recovery_schema_passed"): raise ValueError("smoke_tests_not_passed") document, mimetype = _existing_document(tenant_id, invoice_id) extension = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp", "application/pdf": ".pdf"}[mimetype] digest = hashlib.sha256(document).hexdigest() forced_context = {**context, "payload": { "filename": f"aldi-forced{extension}", "content_type": mimetype, "file_bytes_base64": base64.b64encode(document).decode("ascii"), "idempotency_key": f"wp-1-{digest}", "engine": "mini", "force_uncached": True, }} from apps.wp_invoices.services.submission import submit_invoice_v2 result = submit_invoice_v2(forced_context) return { "pipeline_version": PIPELINE_VERSION, "ocr_passed": True, "recovery_schema_passed": True, "artifact_root": f"audits:{result['invoice_id']}:{context.get('request_id')}", "full_execution": {"invoice_id": result["invoice_id"], "status": result["status"], "review_status": result["result"].get("review_status"), "processing": result["result"].get("processing")}, } def _read_audit_report(tenant_id: str, invoice_id: str, request_id: str) -> dict[str, Any]: if not re.fullmatch(r"[a-f0-9-]{36}", request_id): raise ValueError("invalid_input") root = Path("audits") / invoice_id / request_id def stage(name: str) -> dict[str, Any]: value = json.loads(read_bytes(tenant_id=tenant_id, app_id=APP_ID, relative_path=root / name)) return value.get("data") or {} canonical = stage("04-primary-canonical.json") parsed = stage("03-primary-parsed.json") reconciliation = stage("05-primary-reconciliation.json") association = stage("09-visual-association.json") response = stage("07-wordpress-response.json") final_result = response.get("result") or {} items = canonical.get("items") or [] visual = association.get("visual_item_lines") or [] consumed = [line.get("consumed_by_item_id") for line in visual if line.get("consumed_by_item_id")] duplicates = sorted(key for key, count in Counter(consumed).items() if count > 1) incomplete = [item for item in items if item.get("complete") is False or item.get("missing_fields")] confirmed_money = sum( 1 for item in items if (((item.get("evidence") or {}).get("line_total") or {}).get("status") == "confirmed") ) tax_states = Counter( str((item.get("tax_code_evidence") or {}).get("evidence_state") or "missing") for item in items ) recovery = ((final_result.get("processing") or {}).get("recovery") or {}) subtotal = canonical.get("subtotal") item_sum = reconciliation.get("items_sum") expected = [3.49, 5.99, 7.99, 2.19, 7.49, 3.49, 4.11, 2.49, 3.99, 3.59, 1.96, 4.49, 0.95, 1.19, 2.29, 1.09, 1.09, 1.09, 2.99, 0.95, 1.39, 10.69, 5.99, 4.99, 3.49] parsed_items = parsed.get("items") or [] canonical_by_id = {f"item-{index + 1}": item for index, item in enumerate(items)} parsed_by_id = {f"item-{index + 1}": item for index, item in enumerate(parsed_items)} monetary_rows = [] for row_index, line in enumerate(visual): item_id = line.get("consumed_by_item_id") item = canonical_by_id.get(item_id) or {} semantic = parsed_by_id.get(item_id) or {} semantic_value = ((semantic.get("line_total") or {}).get("verbatim") if isinstance(semantic.get("line_total"), dict) else semantic.get("line_total")) selected = item.get("line_total") fixture = expected[row_index] if row_index < len(expected) else None monetary_rows.append({ "source_line_id": line.get("source_line_id"), "product_code": line.get("product_code"), "raw_ocr_text": line.get("raw_text"), "ocr_monetary_tokens": [{"raw": token, "x_y": None, "row_bbox": line.get("estimated_bounding_region"), "semantic_role": "line_total" if index == len(line.get("monetary_candidates") or []) - 1 else "unknown"} for index, token in enumerate(line.get("monetary_candidates") or [])], "semantic_model_amount": semantic_value, "canonical_selected_amount": selected, "evidence_state": (((item.get("evidence") or {}).get("line_total") or {}).get("status")), "expected_fixture_amount": fixture, "row_delta": None if fixture is None else round(float(selected or 0) - fixture, 2), }) row_delta_sum = round(sum(row["row_delta"] or 0 for row in monetary_rows), 2) return { "pipeline_version": PIPELINE_VERSION, "ocr_passed": True, "recovery_schema_passed": True, "artifact_root": f"audits:{invoice_id}:{request_id}", "audit_report": { "primary_item_count": len(items), "incomplete_item_count": len(incomplete), "incomplete_source_line_ids": [item.get("source_line") for item in incomplete], "unconsumed_primary_rows": [line.get("source_line_id") for line in association.get("unconsumed_source_lines") or []], "duplicate_source_line_consumption": duplicates, "item_sum": item_sum, "subtotal": subtotal, "subtotal_delta": None if item_sum is None or subtotal is None else round(float(item_sum) - float(subtotal), 2), "ocr_confirmed_monetary_values": confirmed_money, "tax_codes": {"confirmed": tax_states["confirmed"], "missing": tax_states["missing"], "conflicting": tax_states["conflicting"]}, "recovery_patch_operations": recovery.get("patch_operations") or [], "final_selected_result": recovery.get("selected_result"), "selection_reason": recovery.get("selection_reason"), "monetary_rows": monetary_rows, "row_delta_sum": row_delta_sum, "token_coordinate_limitation": "persisted_trace_contains_row_bbox_but_not_per-token_bbox", }, } def _existing_document(tenant_id: str, invoice_id: str) -> tuple[bytes, str]: for extension, mimetype in ((".jpg", "image/jpeg"), (".jpeg", "image/jpeg"), (".png", "image/png"), (".webp", "image/webp"), (".pdf", "application/pdf")): try: return read_bytes(tenant_id=tenant_id, app_id=APP_ID, relative_path=Path("documents") / invoice_id / f"original{extension}"), mimetype except FileNotFoundError: continue raise FileNotFoundError("stored_invoice_document_not_found") def _synthetic_image() -> bytes: image = Image.new("RGB", (700, 220), "white") draw = ImageDraw.Draw(image) draw.text((35, 45), "SYNTHETIC TEST ITEM", fill="black") draw.text((500, 45), "1.00 A", fill="black") stream = io.BytesIO(); image.save(stream, "PNG") return stream.getvalue() def _safe_error(exc: Exception) -> dict[str, Any]: body = getattr(exc, "body", None) provider = body.get("error") if isinstance(body, dict) and isinstance(body.get("error"), dict) else {} message = str(provider.get("message") or str(exc) or type(exc).__name__) message = re.sub(r"sk-[A-Za-z0-9_-]+", "[redacted]", message)[:2000] return {"exception_type": type(exc).__name__, "http_status": getattr(exc, "status_code", None), "provider_code": provider.get("code") or getattr(exc, "code", None), "provider_param": provider.get("param") or getattr(exc, "param", None), "message": message} def _write_json(tenant_id: str, path: Path, value: Any) -> None: write_bytes(tenant_id=tenant_id, app_id=APP_ID, relative_path=path, data=json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))