51 lines
2.3 KiB
Python
51 lines
2.3 KiB
Python
"""Per-product operational state and safe partial-processing metrics."""
|
|
|
|
from collections import Counter
|
|
import hashlib
|
|
import json
|
|
|
|
STATES = {
|
|
"COMPLETE", "CORRECTED", "BLOCKED_MISSING_EVIDENCE", "BLOCKED_CONFLICT",
|
|
"BLOCKED_COMPANY_CONTEXT", "BLOCKED_SCHEMA", "FAILED_TECHNICAL",
|
|
}
|
|
|
|
|
|
def blocker_fingerprint(tenant: str, company: str, product_code: str, field: str, reason_code: str, input_fingerprint: str, installation_id: str | int | None = None) -> str:
|
|
raw = {"tenant": tenant, "company": company, "product_code": product_code, "field": field, "reason_code": reason_code, "input_fingerprint": input_fingerprint}
|
|
if installation_id is not None:
|
|
raw["installation_id"] = str(installation_id)
|
|
return hashlib.sha256(json.dumps(raw, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
|
|
|
|
|
|
def classify_product_state(product: dict, decision: dict, *, write_status: str | None = None) -> dict:
|
|
status = str(decision.get("status") or "").upper()
|
|
if write_status == "FAILED":
|
|
state = "FAILED_TECHNICAL"
|
|
elif status in {"AMBIGUOUS", "CONFLICT", "NCM_CONFLICT"}:
|
|
state = "BLOCKED_CONFLICT"
|
|
elif status in {"INVALID_NCM", "LEGACY_NCM", "NOT_FOUND"}:
|
|
state = "BLOCKED_MISSING_EVIDENCE"
|
|
elif status in {"COMPANY_CONTEXT_REQUIRED", "COMPANY_CONTEXT_FAIL"}:
|
|
state = "BLOCKED_COMPANY_CONTEXT"
|
|
elif status in {"ERP_SCHEMA_LIMITATION", "UNSUPPORTED_SCHEMA"}:
|
|
state = "BLOCKED_SCHEMA"
|
|
elif write_status == "SUCCEEDED":
|
|
state = "CORRECTED"
|
|
else:
|
|
state = "COMPLETE"
|
|
return {"product_code": str(product.get("CODIGO") or ""), "state": state, "decision_status": status}
|
|
|
|
|
|
def should_retry_blocker(previous_input_fingerprint: str | None, current_input_fingerprint: str) -> bool:
|
|
return previous_input_fingerprint != current_input_fingerprint
|
|
|
|
|
|
def summarize_operational_states(states: list[dict]) -> dict:
|
|
counts = Counter(item.get("state") for item in states)
|
|
result = {state: counts.get(state, 0) for state in sorted(STATES)}
|
|
result["TOTAL_PRODUCTS"] = len(states)
|
|
result["SAFE_BLOCKED"] = sum(result[state] for state in STATES if state.startswith("BLOCKED_"))
|
|
result["AUTONOMOUSLY_PROCESSABLE_PRODUCTS"] = result["TOTAL_PRODUCTS"] - result["FAILED_TECHNICAL"]
|
|
result["AGENT_HEALTH"] = "HEALTHY" if result["FAILED_TECHNICAL"] == 0 else "DEGRADED"
|
|
return result
|