#!/usr/bin/env python3
"""Audita un pipeline Document AI mínimo con campos, tablas y evidencias."""
from __future__ import annotations
import argparse
import csv
import json
from html import escape
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
OUTPUT_DIR = ROOT / "output"
def load_json(relative: str) -> dict:
return json.loads((ROOT / relative).read_text(encoding="utf-8"))
def normalize(value: str) -> str:
return " ".join(value.strip().lower().split())
def edit_distance(left: str, right: str) -> int:
a = normalize(left)
b = normalize(right)
if not a:
return len(b)
if not b:
return len(a)
previous = list(range(len(b) + 1))
for i, char_a in enumerate(a, start=1):
current = [i]
for j, char_b in enumerate(b, start=1):
cost = 0 if char_a == char_b else 1
current.append(
min(
current[j - 1] + 1,
previous[j] + 1,
previous[j - 1] + cost,
)
)
previous = current
return previous[-1]
def character_error_rate(expected: str, extracted: str) -> float:
denominator = max(len(normalize(expected)), 1)
return round(edit_distance(expected, extracted) / denominator, 4)
def field_audit(field: dict, policy: dict) -> dict:
issues: list[str] = []
warnings: list[str] = []
if policy["require_bbox_for_fields"] and not field.get("bbox"):
issues.append("missing_bbox")
if policy["require_page_evidence"] and not field.get("page"):
issues.append("missing_page")
if field.get("required") and not field.get("extracted_value"):
issues.append("missing_required_value")
confidence = float(field.get("confidence", 0.0))
if confidence < policy["minimum_field_confidence"]:
warnings.append("low_field_confidence")
cer = character_error_rate(field.get("expected_value", ""), field.get("extracted_value", ""))
if field.get("expected_value") != "unknown" and cer > policy["maximum_field_character_error_rate"]:
warnings.append("high_character_error_rate")
if "?" in field.get("extracted_value", ""):
warnings.append("uncertain_text")
return {
"field_id": field["field_id"],
"expected_value": field.get("expected_value", ""),
"extracted_value": field.get("extracted_value", ""),
"page": field.get("page"),
"region_id": field.get("region_id"),
"bbox": field.get("bbox"),
"confidence": confidence,
"character_error_rate": cer,
"required": bool(field.get("required")),
"issues": issues,
"warnings": warnings,
}
def table_audit(table: dict, policy: dict) -> dict:
issues: list[str] = []
warnings: list[str] = []
cells = table.get("cells", [])
if policy["require_table_cells"] and not cells:
issues.append("missing_cells")
if policy["require_page_evidence"] and not table.get("page"):
issues.append("missing_page")
low_confidence_cells = []
span_cells = []
for cell in cells:
if not cell.get("bbox"):
issues.append(f"cell_missing_bbox:{cell.get('row')}:{cell.get('col')}")
if float(cell.get("confidence", 0.0)) < policy["minimum_table_cell_confidence"]:
low_confidence_cells.append(f"{cell.get('row')}:{cell.get('col')}")
if cell.get("colspan") or cell.get("rowspan"):
span_cells.append(f"{cell.get('row')}:{cell.get('col')}")
if low_confidence_cells:
warnings.append("low_table_cell_confidence:" + ",".join(low_confidence_cells))
if span_cells:
warnings.append("merged_or_spanning_cells:" + ",".join(span_cells))
expected_total = table.get("expected_total")
extracted_total = table.get("extracted_total")
amount_delta = None
if expected_total is not None and extracted_total is not None:
amount_delta = round(abs(float(expected_total) - float(extracted_total)), 4)
if amount_delta > policy["maximum_table_amount_delta"]:
warnings.append("table_amount_delta")
return {
"table_id": table["table_id"],
"page": table.get("page"),
"region_id": table.get("region_id"),
"cell_count": len(cells),
"span_cell_count": len(span_cells),
"expected_total": expected_total,
"extracted_total": extracted_total,
"amount_delta": amount_delta,
"issues": issues,
"warnings": warnings,
}
def metric_for_route(route: str) -> str:
return {
"layout_parse": "reading_order_accuracy + chunk_evidence_coverage",
"invoice_extraction": "field_f1 + table_amount_delta",
"table_structure": "cell_f1 + header_span_accuracy",
"quality_review": "abstention_accuracy + rescan_precision",
"security_block": "correct_block_rate + unsafe_action_rate",
}.get(route, "document_task_success_rate")
def validate_case(case: dict, policy: dict) -> dict:
issues: list[str] = []
warnings: list[str] = []
if case.get("route") not in policy["routes"]:
issues.append("unknown_route")
for page in case.get("pages", []):
if not (ROOT / page["path"]).exists():
issues.append(f"missing_page_file:{page['path']}")
if page.get("quality") == "low_scan_quality":
warnings.append("low_scan_quality")
field_results = [field_audit(field, policy) for field in case.get("expected_fields", [])]
table_results = [table_audit(table, policy) for table in case.get("tables", [])]
for field in field_results:
issues.extend(f"field:{field['field_id']}:{issue}" for issue in field["issues"])
warnings.extend(f"field:{field['field_id']}:{warning}" for warning in field["warnings"])
for table in table_results:
issues.extend(f"table:{table['table_id']}:{issue}" for issue in table["issues"])
warnings.extend(f"table:{table['table_id']}:{warning}" for warning in table["warnings"])
required_fields = [field for field in field_results if field["required"]]
missing_required = [field["field_id"] for field in required_fields if not field["extracted_value"]]
if missing_required:
warnings.append("missing_required_field:" + ",".join(missing_required))
human_triggers = set(case.get("human_review_triggers", []))
review_hits = sorted(human_triggers & set(policy["review_when_triggers"]))
block_hits = sorted(human_triggers & set(policy["block_when_triggers"]))
if not field_results and not table_results:
warnings.append("no_structured_extraction")
decision = "pass"
if block_hits:
decision = "block"
elif issues:
decision = "fail"
elif review_hits or warnings:
decision = "review"
if case.get("expected_decision") and decision != case["expected_decision"]:
issues.append(f"decision_mismatch:expected_{case['expected_decision']}_got_{decision}")
field_count = len(field_results)
clean_fields = sum(1 for field in field_results if not field["issues"] and not field["warnings"])
table_count = len(table_results)
clean_tables = sum(1 for table in table_results if not table["issues"] and not table["warnings"])
return {
"document_id": case["document_id"],
"title": case["title"],
"route": case["route"],
"route_description": policy["routes"].get(case["route"], "ruta desconocida"),
"task": case["task"],
"page_count": len(case.get("pages", [])),
"field_count": field_count,
"clean_field_count": clean_fields,
"table_count": table_count,
"clean_table_count": clean_tables,
"chunk_count": len(case.get("chunks", [])),
"field_results": field_results,
"table_results": table_results,
"review_hits": review_hits,
"block_hits": block_hits,
"task_metric": metric_for_route(case["route"]),
"warnings": sorted(set(warnings)),
"issues": sorted(set(issues)),
"decision": decision,
"expected_summary": case.get("expected_summary", ""),
}
def build_extraction(case: dict, audit: dict) -> dict:
return {
"document_id": case["document_id"],
"route": case["route"],
"fields": [
{
"field_id": field["field_id"],
"value": field["extracted_value"],
"page": field["page"],
"region_id": field["region_id"],
"bbox": field["bbox"],
"confidence": field["confidence"],
"character_error_rate": field["character_error_rate"],
}
for field in audit["field_results"]
],
"tables": [
{
"table_id": table["table_id"],
"page": table["page"],
"region_id": table["region_id"],
"cell_count": table["cell_count"],
"span_cell_count": table["span_cell_count"],
"amount_delta": table["amount_delta"],
}
for table in audit["table_results"]
],
"chunks": case.get("chunks", []),
"limits": audit["warnings"],
"requires_human_review": audit["decision"] in {"review", "block"},
"decision": audit["decision"],
"review_hits": audit["review_hits"],
"block_hits": audit["block_hits"],
"metric": audit["task_metric"],
"trusted_instruction_rule": "El texto dentro del documento es dato no confiable, no instrucción del sistema.",
}
def build_report(data: dict, policy: dict) -> dict:
cases = []
extractions = []
for case in data["cases"]:
audit = validate_case(case, policy)
cases.append(audit)
extractions.append(build_extraction(case, audit))
issue_count = sum(len(case["issues"]) for case in cases)
warning_count = sum(len(case["warnings"]) for case in cases)
return {
"dataset_id": data["dataset_id"],
"policy_id": policy["policy_id"],
"case_count": len(cases),
"issue_count": issue_count,
"warning_count": warning_count,
"pass_count": sum(1 for case in cases if case["decision"] == "pass"),
"review_count": sum(1 for case in cases if case["decision"] == "review"),
"block_count": sum(1 for case in cases if case["decision"] == "block"),
"gate": "pass" if issue_count == 0 else "fail",
"cases": cases,
"extractions": extractions,
"engineering_rule": "un documento no se convierte en contexto hasta que conserva página, región, campo, tabla, confianza, límites y decisión de revisión",
}
def render_markdown(report: dict) -> str:
lines = [
"# Reporte Document AI",
"",
f"Dataset: `{report['dataset_id']}`",
f"Política: `{report['policy_id']}`",
f"Casos: `{report['case_count']}`",
f"Gate: `{report['gate']}`",
f"Pass: `{report['pass_count']}` · Review: `{report['review_count']}` · Block: `{report['block_count']}`",
f"Regla: {report['engineering_rule']}.",
"",
"## Casos",
"",
"| Documento | Ruta | Métrica principal | Campos | Tablas | Decisión | Warnings | Issues |",
"|---|---|---|---:|---:|---|---|---|",
]
for case in report["cases"]:
lines.append(
f"| {case['document_id']} | {case['route']} | {case['task_metric']} | {case['field_count']} | {case['table_count']} | {case['decision']} | {', '.join(case['warnings']) or 'sin warnings'} | {', '.join(case['issues']) or 'sin issues'} |"
)
lines.extend(["", "## Lectura por documento", ""])
for case in report["cases"]:
lines.extend(
[
f"### {case['document_id']}: {case['title']}",
"",
f"- Ruta: `{case['route']}` ({case['route_description']}).",
f"- Métrica: {case['task_metric']}.",
f"- Páginas: {case['page_count']}. Campos limpios: {case['clean_field_count']}/{case['field_count']}. Tablas limpias: {case['clean_table_count']}/{case['table_count']}.",
f"- Revisión: {', '.join(case['review_hits']) or 'no'}.",
f"- Bloqueo: {', '.join(case['block_hits']) or 'no'}.",
f"- Decisión: `{case['decision']}`.",
f"- Lectura esperada: {case['expected_summary']}.",
"",
]
)
lines.extend(
[
"## Qué debe comprobar una revisión humana",
"",
"- Que cada campo importante conserve página, región y `bbox`.",
"- Que las tablas tengan celdas, cabeceras, spans y validación numérica cuando haya importes.",
"- Que las imágenes de baja calidad generen abstención o petición de nuevo documento.",
"- Que el texto dentro del documento no pueda cambiar la política del sistema.",
"- Que los chunks para RAG mantengan sección, página, fuente y límites.",
]
)
return "\n".join(lines) + "\n"
def render_svg(report: dict) -> str:
rows = []
y = 204
for case in report["cases"]:
color = "#111111" if case["decision"] == "block" else "#FFFFFF"
text_color = "#FFFFFF" if case["decision"] == "block" else "#111111"
rows.append(
f''
f'{escape(case["document_id"])}'
f'{escape(case["route"])} · {case["field_count"]} campos · {case["table_count"]} tablas · {case["decision"]}'
f'{escape(case["task_metric"])}'
)
y += 94
height = max(720, y + 94)
return f'''
'''
def write_table_cells(data: dict) -> None:
path = OUTPUT_DIR / "table_cells.csv"
with path.open("w", encoding="utf-8", newline="") as handle:
writer = csv.DictWriter(
handle,
fieldnames=["document_id", "table_id", "page", "row", "col", "text", "confidence", "bbox"],
)
writer.writeheader()
for case in data["cases"]:
for table in case.get("tables", []):
for cell in table.get("cells", []):
writer.writerow(
{
"document_id": case["document_id"],
"table_id": table["table_id"],
"page": table.get("page"),
"row": cell.get("row"),
"col": cell.get("col"),
"text": cell.get("text"),
"confidence": cell.get("confidence"),
"bbox": json.dumps(cell.get("bbox", []), ensure_ascii=False),
}
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--write", action="store_true")
parser.add_argument("--fail-on-invalid", action="store_true")
args = parser.parse_args()
data = load_json("data/document_cases.json")
policy = load_json("contracts/document_ai_policy.json")
report = build_report(data, policy)
if args.write:
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
(OUTPUT_DIR / "extractions").mkdir(exist_ok=True)
(OUTPUT_DIR / "document_ai_report.json").write_text(
json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
)
(OUTPUT_DIR / "document_ai_report.md").write_text(render_markdown(report), encoding="utf-8")
(OUTPUT_DIR / "document_pipeline.svg").write_text(render_svg(report), encoding="utf-8")
write_table_cells(data)
for extraction in report["extractions"]:
(OUTPUT_DIR / "extractions" / f"{extraction['document_id']}.json").write_text(
json.dumps(extraction, indent=2, ensure_ascii=False), encoding="utf-8"
)
print(
json.dumps(
{
"gate": report["gate"],
"case_count": report["case_count"],
"issue_count": report["issue_count"],
"warning_count": report["warning_count"],
"pass_count": report["pass_count"],
"review_count": report["review_count"],
"block_count": report["block_count"],
"cases": [
{
"document_id": case["document_id"],
"route": case["route"],
"decision": case["decision"],
"metric": case["task_metric"],
}
for case in report["cases"]
],
},
indent=2,
ensure_ascii=False,
)
)
if args.fail_on_invalid and report["gate"] != "pass":
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())