132 lines
7.6 KiB
Python
132 lines
7.6 KiB
Python
"""Conservative deterministic NCM classification."""
|
|
import re
|
|
import unicodedata
|
|
|
|
ENGINE_VERSION = "0.3.0"
|
|
DECISION_VERSION = "mission-04c-ipi-ex-split"
|
|
|
|
_NT_MARKERS = {"NT", "N/T", "NAO TRIBUTADO", "NÃO TRIBUTADO"}
|
|
_ZERO_MARKERS = {"0", "0,00", "0.00"}
|
|
|
|
|
|
def classify_tipi_rate(entry):
|
|
"""Classify why a TIPI base entry has no usable numeric rate; never guess a value."""
|
|
if entry.ex:
|
|
return "EX_SPECIFIC"
|
|
text = entry.ipi_rate_text
|
|
if text is None:
|
|
return "SOURCE_BLANK"
|
|
normalized = text.strip().upper()
|
|
if normalized in _NT_MARKERS:
|
|
return "NT_OR_NON_TAXED"
|
|
if entry.ipi_rate == 0 or normalized in _ZERO_MARKERS:
|
|
return "EXPLICIT_ZERO"
|
|
return "OTHER"
|
|
|
|
def normalize_code(value): return "".join(char for char in str(value or "") if char.isdigit())
|
|
def normalize_description(value): return " ".join(re.sub(r"[^A-Z0-9]+", " ", unicodedata.normalize("NFKD", str(value or "")).encode("ascii", "ignore").decode().upper()).split())
|
|
|
|
# Portuguese function words with no fiscal-descriptive content. Without this, a
|
|
# product description sharing only "DE"/"E"/"COM" with an official CEST text
|
|
# scores a false-positive nonzero overlap that has nothing to do with the product.
|
|
_STOPWORDS = {
|
|
"DE", "DA", "DO", "DAS", "DOS", "E", "OU", "A", "O", "AS", "OS",
|
|
"EM", "NO", "NA", "NOS", "NAS", "PARA", "POR", "COM", "SEM",
|
|
"MESMO", "OUTRO", "OUTRA", "OUTROS", "OUTRAS", "QUE",
|
|
}
|
|
|
|
|
|
def description_tokens(value):
|
|
return set(normalize_description(value).split()) - _STOPWORDS
|
|
|
|
class Classifier:
|
|
def __init__(self, repository): self.repository = repository
|
|
def classify(self, product):
|
|
entries = self.repository.active_entries(); code = normalize_code(product.get("current_ncm"))
|
|
exact = next((entry for entry in entries if entry.code == code and len(code) == 8), None)
|
|
legacy = self.repository.legacy_ncm(code) if len(code) == 8 and hasattr(self.repository, "legacy_ncm") else None
|
|
if exact:
|
|
status, candidate = "EXACT_NCM", exact
|
|
elif legacy:
|
|
status, candidate = "LEGACY_NCM", None
|
|
else:
|
|
status, candidate = ("INVALID_NCM", None) if code else self._candidate(entries, product.get("description"))
|
|
matched_code = candidate.code if candidate else legacy["code"] if legacy else None
|
|
cests_available = matched_code and hasattr(self.repository, "cest_for_ncm")
|
|
cests = self.repository.cest_for_ncm(matched_code) if cests_available else None
|
|
cests = list({item.code: item for item in cests or []}.values())
|
|
scores = {item.code: self._description_score(product.get("description"), item.description) for item in cests}
|
|
if legacy:
|
|
cest_status = "LEGACY_NCM"
|
|
elif not cests:
|
|
cest_status = "NO_CEST_MAPPING_FOUND"
|
|
elif max(scores.values()) < .5:
|
|
cest_status = "CEST_DESCRIPTION_MISMATCH"
|
|
elif len(cests) == 1:
|
|
cest_status = "CEST_CONFIRMED"
|
|
else:
|
|
cest_status = "CEST_CANDIDATES"
|
|
resolved_cest = cests[0].code if cest_status == "CEST_CONFIRMED" else None
|
|
ipi = self._resolve_ipi(matched_code, legacy)
|
|
mva = self._resolve_mva(matched_code, legacy, resolved_cest)
|
|
return {"status": status, "cest_status": cest_status, "ipi_status": ipi["status"], "ba_st_mva_status": mva["status"], "classification": {"ncm": matched_code, "ncm_description": candidate.description if candidate else None, "cest": [item.code for item in cests] if cests_available else None, "cest_candidates": [{"code": item.code, "official_description": item.description, "description_score": scores[item.code]} for item in cests], "cfop": None, "cst": None, "csosn": None, "pis_cofins": None, "ipi": ipi["resolved"], "ipi_candidates": ipi["candidates"], "ba_st_mva": mva["resolved"], "ba_st_mva_candidates": mva["candidates"], "ibs_cbs": None}, "legacy_ncm": legacy, "reason_codes": ["FISCAL_DETERMINATION_NOT_CONFIGURED"], "can_auto_apply": False, "engine_version": ENGINE_VERSION, "decision_version": DECISION_VERSION, "dataset_versions": self.repository.dataset_versions()}
|
|
|
|
def _resolve_ipi(self, matched_code, legacy):
|
|
if legacy:
|
|
return {"status": "LEGACY_NCM", "resolved": None, "candidates": []}
|
|
if not matched_code or not hasattr(self.repository, "tipi_for_ncm"):
|
|
return {"status": "NO_TIPI_MAPPING_FOUND", "resolved": None, "candidates": []}
|
|
entries = self.repository.tipi_for_ncm(matched_code)
|
|
if not entries:
|
|
return {"status": "NO_TIPI_MAPPING_FOUND", "resolved": None, "candidates": []}
|
|
base = next((entry for entry in entries if not entry.ex), None)
|
|
ex_variants = [entry for entry in entries if entry.ex]
|
|
candidates = [{"ncm": entry.ncm, "ex": entry.ex, "description": entry.description, "rate": entry.ipi_rate_text} for entry in ex_variants]
|
|
if base:
|
|
if base.ipi_rate is not None:
|
|
rate_kind = "NUMERIC"
|
|
elif base.ipi_rate_text is not None:
|
|
rate_kind = "NON_NUMERIC"
|
|
else:
|
|
rate_kind = "BLANK"
|
|
resolved = {
|
|
"ncm": base.ncm,
|
|
"ex": base.ex,
|
|
"description": base.description,
|
|
"rate": base.ipi_rate_text,
|
|
"rate_numeric": base.ipi_rate,
|
|
"rate_kind": rate_kind,
|
|
}
|
|
if rate_kind != "NUMERIC":
|
|
resolved["rate_classification"] = classify_tipi_rate(base)
|
|
status = "IPI_CANDIDATES" if ex_variants else "IPI_RESOLVED"
|
|
return {"status": status, "resolved": resolved, "candidates": candidates}
|
|
return {"status": "IPI_CANDIDATES", "resolved": None, "candidates": candidates}
|
|
|
|
def _resolve_mva(self, matched_code, legacy, resolved_cest):
|
|
if legacy:
|
|
return {"status": "LEGACY_NCM", "resolved": None, "candidates": []}
|
|
if not matched_code or not hasattr(self.repository, "ba_st_mva_for_ncm"):
|
|
return {"status": "NO_MVA_MAPPING_FOUND", "resolved": None, "candidates": []}
|
|
entries = self.repository.ba_st_mva_for_ncm(matched_code)
|
|
if not entries:
|
|
return {"status": "NO_MVA_MAPPING_FOUND", "resolved": None, "candidates": []}
|
|
candidates = [{"item": entry.item, "cest": entry.cest, "mva_original": entry.mva_original, "mva_adjusted": entry.mva_adjusted, "description": entry.description} for entry in entries]
|
|
if len(entries) == 1:
|
|
return {"status": "MVA_RESOLVED", "resolved": candidates[0], "candidates": candidates}
|
|
if resolved_cest:
|
|
by_cest = [item for item in candidates if item["cest"] == resolved_cest]
|
|
if len(by_cest) == 1:
|
|
return {"status": "MVA_RESOLVED", "resolved": by_cest[0], "candidates": candidates}
|
|
return {"status": "MVA_CANDIDATES", "resolved": None, "candidates": candidates}
|
|
def _candidate(self, entries, description):
|
|
tokens = set(normalize_description(description).split())
|
|
scores = sorted(((len(tokens & set(normalize_description(item.description).split())) / len(tokens or {"_"}), item) for item in entries), reverse=True, key=lambda row: row[0])
|
|
if not scores or scores[0][0] == 0: return "NOT_FOUND", None
|
|
return ("HIGH_CONFIDENCE_CANDIDATE", scores[0][1]) if scores[0][0] >= .90 and (len(scores) == 1 or scores[0][0] - scores[1][0] >= .15) else ("AMBIGUOUS", None)
|
|
|
|
def _description_score(self, product_description, official_description):
|
|
product_tokens = description_tokens(product_description)
|
|
official_tokens = description_tokens(official_description)
|
|
return len(product_tokens & official_tokens) / len(product_tokens or {"_"})
|