84 lines
4.3 KiB
Python
84 lines
4.3 KiB
Python
"""Pure requiredness and fiscal completeness evaluation."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
VALID_CLASSIFICATIONS = {
|
|
"REQUIRED", "CONDITIONAL", "OPTIONAL_REFERENCE", "NOT_APPLICABLE", "ERP_NOT_USED"
|
|
}
|
|
|
|
|
|
def load_requiredness_matrix(path: str | Path) -> dict[str, Any]:
|
|
matrix = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
fields = matrix.get("fields", {})
|
|
if set(matrix.get("classifications", [])) != VALID_CLASSIFICATIONS:
|
|
raise ValueError("requiredness matrix classifications are incomplete")
|
|
if len(fields) != 23 or any(rule.get("classification") not in VALID_CLASSIFICATIONS for rule in fields.values()):
|
|
raise ValueError("requiredness matrix must define 23 valid field rules")
|
|
return matrix
|
|
|
|
|
|
def _has_value(value: Any) -> bool:
|
|
return value is not None and (not isinstance(value, str) or bool(value.strip()))
|
|
|
|
|
|
def _conditional_state(field: str, context: dict[str, Any]) -> str:
|
|
if field == "IPI":
|
|
return "REQUIRED" if context.get("contribuinte_ipi") == "S" and context.get("has_numeric_tipi") else "NOT_APPLICABLE"
|
|
if field == "EX_TIPI":
|
|
return "REQUIRED" if context.get("has_supported_ex") and context.get("product_matches_ex") else "NOT_APPLICABLE"
|
|
if field == "CEST":
|
|
return "REQUIRED" if context.get("has_official_cest") or context.get("st_subject") else "NOT_APPLICABLE"
|
|
if field in {"CFOP", "SITUACAO_TRIBUTARIA"}:
|
|
return "REQUIRED" if context.get("operation", "SALE") == "SALE" else "NOT_APPLICABLE"
|
|
if field in {"CFOP_ENTRADA", "CST_ENTRADA", "COD_ENTRADA_PIS", "COD_ENTRADA_COFINS"}:
|
|
return "REQUIRED" if context.get("operation") == "PURCHASE" and context.get("erp_consumes_entry_fields") else "NOT_APPLICABLE"
|
|
if field == "CST":
|
|
return "REQUIRED" if context.get("regime") == "NORMAL" and context.get("operation", "SALE") == "SALE" else "NOT_APPLICABLE"
|
|
if field == "CSOSN":
|
|
return "REQUIRED" if context.get("regime") == "SIMPLES" and context.get("operation", "SALE") == "SALE" else "NOT_APPLICABLE"
|
|
if field == "ALIQUOTA":
|
|
return "REQUIRED" if context.get("icms_taxed") else "NOT_APPLICABLE"
|
|
if field == "COD_ORIGEM_MERCADORIA":
|
|
return "REQUIRED" if context.get("origin_proven") else "NOT_APPLICABLE"
|
|
if field == "CST_IPI":
|
|
return "REQUIRED" if context.get("contribuinte_ipi") == "S" and context.get("erp_consumes_ipi_cst") else "NOT_APPLICABLE"
|
|
if field in {"COD_SAIDA_PIS", "COD_SAIDA_COFINS"}:
|
|
return "REQUIRED" if context.get("operation", "SALE") == "SALE" and context.get("piscofins_profile_proven") else "NOT_APPLICABLE"
|
|
if field in {"ALIQ_PIS", "ALIQ_COFINS"}:
|
|
return "REQUIRED" if context.get("piscofins_rate_applies") else "NOT_APPLICABLE"
|
|
if field in {"INCIDENCIA_PISCOFINS", "PISCOFINS"}:
|
|
return "REQUIRED" if context.get("piscofins_profile_proven") else "NOT_APPLICABLE"
|
|
return "NOT_APPLICABLE"
|
|
|
|
|
|
def evaluate_field(field: str, value: Any, context: dict[str, Any], matrix: dict[str, Any]) -> dict[str, Any]:
|
|
rule = matrix["fields"][field]
|
|
classification = rule["classification"]
|
|
if classification == "ERP_NOT_USED":
|
|
status = "ERP_NOT_SUPPORTED"
|
|
valid = True
|
|
elif classification == "NOT_APPLICABLE":
|
|
status = "NOT_APPLICABLE"
|
|
valid = True
|
|
elif classification == "OPTIONAL_REFERENCE":
|
|
status = "OPTIONAL_VALID" if _has_value(value) else "OPTIONAL_EMPTY"
|
|
valid = True
|
|
elif classification == "REQUIRED":
|
|
status = "REQUIRED_AND_VALID" if _has_value(value) else "REQUIRED_MISSING"
|
|
valid = _has_value(value)
|
|
else:
|
|
conditional = _conditional_state(field, context)
|
|
if conditional == "NOT_APPLICABLE":
|
|
status, valid = "CONDITIONAL_NOT_TRIGGERED", True
|
|
else:
|
|
status, valid = ("CONDITIONAL_AND_VALID", True) if _has_value(value) else ("REQUIRED_MISSING", False)
|
|
return {"field": field, "classification": classification, "status": status, "valid": valid, "value_present": _has_value(value), "source": rule["source"]}
|
|
|
|
|
|
def evaluate_product(product: dict[str, Any], context: dict[str, Any], matrix: dict[str, Any]) -> dict[str, Any]:
|
|
fields = {field: evaluate_field(field, product.get(field), context, matrix) for field in matrix["fields"]}
|
|
return {"FISCALLY_COMPLETE": all(result["valid"] for result in fields.values()), "fields": fields}
|