gtiandClaude Sonnet 5
b15dbea213
- central/app/unified_fiscal_coverage.py: pure rules (PIS/COFINS NOT_APPLICABLE/REQUIRED/OPTIONAL/NEEDS_REVIEW, RTC only when the ERP switch is on) - agent: read-only loader (unified_fiscal.py) replaces the legacy/expanded module; findings feed the single operational state, coverage and blocker queue - hub: BLOCKED_FISCAL_DATA blockers (PIS/COFINS/RTC) share the queue; stale ones close on the next full pass; dashboard breakdown gains CEST/PIS/COFINS/RTC - panel: legacy/expanded coverage block removed; discreet 'Reforma Tributaria: Ativa/Desativada' - docs: unified_fiscal_completeness, pis_cofins_applicability, rtc_pis_cofins_model + official source manifests; read-only E2E script Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
60 lines
3.0 KiB
Python
60 lines
3.0 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", "BLOCKED_FISCAL_DATA", "FAILED_TECHNICAL", "NEEDS_REVIEW",
|
|
}
|
|
|
|
|
|
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, fiscal: dict | None = None) -> dict:
|
|
"""`fiscal` = unified_fiscal_coverage.evaluate_product(...): PIS/COFINS + (only when the ERP has RTC on) cClassTrib. One definition of COMPLETE."""
|
|
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"
|
|
fiscal = fiscal or {}
|
|
if state in ("COMPLETE", "CORRECTED"):
|
|
if fiscal.get("blockers"):
|
|
state = "BLOCKED_FISCAL_DATA"
|
|
elif fiscal.get("needs_review"):
|
|
state = "NEEDS_REVIEW"
|
|
return {"product_code": str(product.get("CODIGO") or ""), "state": state, "decision_status": status, "fiscal_blockers": list(fiscal.get("blockers") or [])}
|
|
|
|
|
|
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["UNCLASSIFIED"] = result["NEEDS_REVIEW"]
|
|
result["COMPLETE_PRODUCTS"] = result["COMPLETE"] + result["CORRECTED"]
|
|
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
|