Files
gti d21a9d1140 Add 'fiscal/' from commit '102ba84500dbd9655a0471288c3e278652593ca9'
git-subtree-dir: fiscal
git-subtree-mainline: d95165fc8f
git-subtree-split: 102ba84500
2026-09-29 14:43:23 -03:00

74 lines
4.0 KiB
Python

"""Validation of operator fiscal-context entries (facts the ERP does not know). Pure; shared by the Hub API, the importer CLI and the resolver."""
import json
SCOPES = ("COMPANY", "GROUP", "NCM", "PRODUCT")
ENUMS = {
"product_role": {"RESALE", "OWN_PRODUCTION", "INPUT", "SERVICE"},
"st_status": {"SUBJECT", "NOT_SUBJECT"},
"st_withheld": {"Y", "N"},
"icms_treatment": {"TAXED", "EXEMPT", "NOT_TAXED", "ST_WITHHELD", "SUSPENDED", "DEFERRED"},
# MIXED_OR_DYNAMIC: purchases are not predominantly intra- or interstate, so no single
# static entry CFOP is fiscally correct for all operations (section 0, AI semantic layer V1).
"entry_scope": {"INTRA", "INTER", "MIXED_OR_DYNAMIC"},
"rounding": {"A", "T"},
"piscofins": {"N", "S"},
"incidence": {"NENHUMA"},
"origin": {str(i) for i in range(9)},
"store_ipi_profile": {"POPULATE_REFERENCE_RATE", "LEAVE_NOT_APPLICABLE"},
# C000004.CONTRIBUINTE_IPI is not yet sent at heartbeat (no Agent-side reader exists for it): until that auto-discovery lands, this is the
# one-time operator answer field_contract's CST_IPI rule needs - never assumed "N" just because the Agent hasn't reported it.
"contribuinte_ipi": {"S", "N"},
}
PROFILE_KEYS = {"pis_in_cst", "pis_out_cst", "pis_rate", "cofins_in_cst", "cofins_out_cst", "cofins_rate"}
FREE_KEYS = {"ncm", "cest", "ex_tipi", "icms_rate", "credit_rate", "cst_ipi", "service_cfop", "pis_cofins_profile"}
ALLOWED_KEYS = set(ENUMS) | FREE_KEYS
PRODUCT_ONLY = {"ncm"}
COMPANY_ONLY = {"store_ipi_profile", "contribuinte_ipi"}
NCM_KEYS = {"cest", "ex_tipi"} # tied to the NCM tables: answered per NCM or per product # an NCM answer is always about one product
def validate_entry(entry):
"""Returns (normalized_entry, error). error is None when valid."""
scope = str(entry.get("scope", "")).strip().upper()
key = str(entry.get("key", "")).strip()
value = entry.get("value")
value = "" if value is None else str(value).strip()
scope_key = str(entry.get("scope_key") or "").strip()
if scope not in SCOPES:
return None, f"scope must be one of {SCOPES}"
if key not in ALLOWED_KEYS:
return None, f"unknown key {key!r}"
if scope != "COMPANY" and not scope_key:
return None, "scope_key is required for GROUP/NCM/PRODUCT scope"
if key in COMPANY_ONLY and scope != "COMPANY":
return None, f"{key} is a COMPANY-scope answer"
if key in PRODUCT_ONLY and scope != "PRODUCT":
return None, f"{key} is a PRODUCT-scope answer"
if key in NCM_KEYS and scope not in ("PRODUCT", "NCM"):
return None, f"{key} can only be answered for a PRODUCT or an NCM"
if not value:
return None, "value is empty"
if key in ENUMS and value.upper() not in ENUMS[key] and value not in ENUMS[key]:
return None, f"{key} must be one of {sorted(ENUMS[key])}"
if key in ENUMS:
value = value.upper() if key != "origin" else value
if key == "ncm" and not (value.isdigit() and len(value) == 8):
return None, "ncm must be 8 digits"
if key == "cest" and value.upper() != "NONE" and not (value.replace(".", "").isdigit() and len(value.replace(".", "")) == 7):
return None, "cest must be 7 digits or NONE"
if key in ("icms_rate", "credit_rate"):
try:
float(value.replace(",", "."))
except ValueError:
return None, f"{key} must be numeric"
if key == "pis_cofins_profile" and value.upper() not in ("NOT_APPLICABLE", "OPERATION_TIME"):
try:
prof = json.loads(value)
except ValueError:
return None, "pis_cofins_profile must be JSON, or exactly NOT_APPLICABLE / OPERATION_TIME"
if not isinstance(prof, dict) or not prof or set(prof) - PROFILE_KEYS:
return None, f"pis_cofins_profile keys must be a non-empty subset of {sorted(PROFILE_KEYS)}"
elif key == "pis_cofins_profile":
value = value.upper()
return {"scope": scope, "scope_key": scope_key if scope != "COMPANY" else "", "key": key, "value": value, "reason": (entry.get("reason") or "").strip() or None}, None