Spaces:
Sleeping
Sleeping
File size: 4,590 Bytes
dd9584b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | from __future__ import annotations
import asyncio
import sys
from pathlib import Path
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from supabase import acreate_client
from src.ai.nodes.risk import risk_assessment_node
from src.config import settings
from src.services.validation_rules_svc import validation_rules_service
def _coerce_field(field: str, value: Any) -> Any:
if value is None:
return None
if field in {"gross_weight", "cif_value", "fob_value", "freight_value", "insurance_value"}:
try:
return float(str(value).replace(",", ""))
except Exception:
return value
if field == "total_packages":
try:
return int(float(str(value).replace(",", "")))
except Exception:
return value
return value
async def revalidate_batch(batch_id: str) -> dict[str, Any]:
supabase = await acreate_client(
settings.SUPABASE_URL,
settings.SUPABASE_SERVICE_KEY.get_secret_value(),
)
docs = (
await supabase.table("documents").select("*").eq("batch_id", batch_id).execute()
).data or []
fields = (
await supabase.table("extracted_fields").select("*").eq("batch_id", batch_id).execute()
).data or []
by_doc: dict[str, dict[str, Any]] = {}
combined: dict[str, Any] = {}
confidences: dict[str, float] = {}
for row in fields:
document_id = row.get("document_id")
name = row.get("ceisa_field")
if not document_id or not name:
continue
value = _coerce_field(name, row.get("normalized_value") or row.get("extracted_value"))
by_doc.setdefault(document_id, {})[name] = value
combined[name] = value
confidences[name] = float(row.get("confidence") or 0.0)
state = {
"batch_id": batch_id,
"company_id": "",
"documents": [
{
"doc_id": doc["id"],
"doc_type": doc.get("doc_type"),
"storage_path": doc.get("storage_path"),
"pages": [],
"extracted_data": by_doc.get(doc["id"], {}),
"quality_score": float(doc.get("quality_score") or 1.0),
"ocr_method": doc.get("ocr_engine_used"),
"error": doc.get("error_message"),
"ocr_candidates": {},
"ocr_conflicts": [],
"field_confidences": {},
}
for doc in docs
],
"combined_data": combined,
"validation_results": [],
"needs_human_review": False,
"risk_level": "UNKNOWN",
"customs_readiness_score": None,
"crs_grade": None,
"rejection_probability": None,
"risk_features": {},
"ocr_conflicts": [],
"field_confidences": confidences,
"steps": [],
}
validations, needs_review = validation_rules_service.evaluate(state)
state["validation_results"] = validations
state["needs_human_review"] = needs_review
risk = await risk_assessment_node(state) # type: ignore[arg-type]
await supabase.table("validation_results").delete().eq("batch_id", batch_id).execute()
if validations:
await supabase.table("validation_results").insert([
{
"batch_id": batch_id,
"rule_id": row.get("rule_id", "UNKNOWN"),
"rule_name": row.get("rule_name", row.get("message", "Validation")),
"severity": row.get("severity", "WARNING"),
"error_message": row.get("message"),
"affected_fields": row.get("affected_fields", []),
}
for row in validations
]).execute()
payload = {
"status": "review_ready" if risk.get("needs_human_review") else "validated",
"risk_level": risk.get("risk_level"),
"customs_readiness_score": risk.get("customs_readiness_score"),
"crs_grade": risk.get("crs_grade"),
"rejection_probability": risk.get("rejection_probability"),
}
await supabase.table("batches").update(payload).eq("id", batch_id).execute()
return {
**payload,
"validation_counts": {
severity: sum(1 for row in validations if row.get("severity") == severity)
for severity in {"PASS", "WARNING", "CRITICAL_FAIL"}
},
}
if __name__ == "__main__":
if len(sys.argv) != 2:
raise SystemExit("Usage: python /app/src/scripts/revalidate_batch.py <batch_id>")
print(asyncio.run(revalidate_batch(sys.argv[1])))
|