285 lines
16 KiB
Python
285 lines
16 KiB
Python
"""Central product Knowledge Base access (same PostgreSQL, existing kb_* tables) with per-field status.
|
|
|
|
Trust model (docs/product_enrichment.md):
|
|
- A provider value or a client observation is EVIDENCE. A field becomes VERIFIED_KB only when the official dataset confirms it AND at least two INDEPENDENT evidence
|
|
families agree with zero conflict (distinct providers each count once; client observations count once and only with >= 2 distinct clients).
|
|
- Human edits never enter the KB (HUMAN_OVERRIDDEN is per-installation protection, not knowledge). Company-specific fiscal decisions (CSOSN, CFOP, ...) never live here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from sqlalchemy import func
|
|
|
|
from ..models import KbKnowledge, KbProduct, KbPromotionAudit
|
|
from . import textnorm
|
|
|
|
UNVERIFIED, OBSERVED, CORROBORATED, EXTERNAL_CORROBORATED, VERIFIED_KB, CONFLICT, STALE = "UNVERIFIED", "OBSERVED", "CORROBORATED", "EXTERNAL_CORROBORATED", "VERIFIED_KB", "CONFLICT", "STALE"
|
|
EVIDENCE_ONLY = OBSERVED # legacy name
|
|
CLIENT_FAMILY = "CLIENT_DATA" # every client-derived lineage (master data + fiscal documents of the same ERP are NOT independent of each other)
|
|
RULE_VERSION = "kb-promotion-v2"
|
|
STRONG_DOCS, STRONG_YEARS = 20, 2
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def gtin_fingerprint(key: str) -> str:
|
|
return hashlib.sha256(f"gtin:{key}".encode()).hexdigest()
|
|
|
|
|
|
def find_by_gtin(session, key: str | None) -> KbProduct | None:
|
|
return session.scalar(select(KbProduct).where(KbProduct.gtin == key)) if key else None
|
|
|
|
|
|
NEUTRAL_EXTRAS = frozenset({"REFRIGERANTE", "REFRIG", "REFR", "BEBIDA", "PRODUTO", "PROD"}) # category nouns only: ZERO / DIET / a flavour / another size are NEVER neutral
|
|
|
|
|
|
def _tokset(text) -> frozenset:
|
|
return frozenset(t for t in textnorm.tokens(text) if t not in textnorm.MARKETING)
|
|
|
|
|
|
def alias_match(query: frozenset, alias: frozenset) -> bool:
|
|
"""Deterministic, conservative: same token set, or one set contains the other and the difference is only neutral category nouns (so "COCA COLA LATA 350 ML" = "REFRIGERANTE COCA COLA LATA 350 ML", but not "... ZERO ...")."""
|
|
if not query or not alias:
|
|
return False
|
|
if query == alias:
|
|
return True
|
|
if query < alias:
|
|
return (alias - query) <= NEUTRAL_EXTRAS
|
|
if alias < query:
|
|
return (query - alias) <= NEUTRAL_EXTRAS
|
|
return False
|
|
|
|
|
|
def name_candidates(session, description) -> list[KbProduct]:
|
|
"""Every KB product with an alias that matches the description under `alias_match` (exact fingerprint included). 0, 1 or several - the caller decides (unique VERIFIED only)."""
|
|
fp = textnorm.fingerprint(description)
|
|
if not fp:
|
|
return []
|
|
ids = set(session.scalars(select(KbKnowledge.kb_product_id).where(KbKnowledge.knowledge_key == "name_fingerprint", KbKnowledge.value == fp).distinct()))
|
|
query = _tokset(description)
|
|
anchor = max((t for t in query if not t.replace(".", "").isdigit() and len(t) > 2), key=len, default=None) # most selective informative token, then verify in Python
|
|
if anchor:
|
|
for row in session.execute(select(KbKnowledge.kb_product_id, KbKnowledge.value).where(KbKnowledge.knowledge_key == "alias", KbKnowledge.value.contains(anchor))):
|
|
if alias_match(query, _tokset(row[1])):
|
|
ids.add(row[0])
|
|
return [p for p in (session.get(KbProduct, i) for i in sorted(ids)) if p]
|
|
|
|
|
|
def upsert_product(session, key: str | None, description, *, brand=None, quantity=None) -> KbProduct:
|
|
if key:
|
|
from ..gtin_validator import validate_gtin
|
|
valid, _ = validate_gtin(key)
|
|
if not valid:
|
|
key = None # Regra C.6: GTIN inválido (DV ou prefixo errado) não entra na KB
|
|
product = find_by_gtin(session, key)
|
|
if product is None:
|
|
fp = gtin_fingerprint(key) if key else hashlib.sha256(f"name:{textnorm.fingerprint(description)}".encode()).hexdigest()
|
|
product = session.scalar(select(KbProduct).where(KbProduct.fingerprint == fp))
|
|
if product is None:
|
|
product = KbProduct(fingerprint=fp, gtin=key, identity_kind="GTIN" if key else "FINGERPRINT", normalized_description=textnorm.normalized(description))
|
|
session.add(product)
|
|
session.flush()
|
|
if description and not product.normalized_description:
|
|
product.normalized_description = textnorm.normalized(description)
|
|
if brand and not product.brand:
|
|
product.brand = brand
|
|
if quantity and not product.package_quantity:
|
|
product.package_quantity = str(quantity)[:32]
|
|
fp = textnorm.fingerprint(description)
|
|
norm = textnorm.normalized(description)
|
|
if norm:
|
|
add_alias(session, product, norm)
|
|
if fp:
|
|
add_evidence(session, product, "name_fingerprint", fp, "IDENTITY", family="IDENTITY", status=OBSERVED)
|
|
return product
|
|
|
|
|
|
def add_alias(session, product: KbProduct, normalized: str, cap: int = 20):
|
|
"""Non-fiscal aliases of the same GTIN (descriptions seen over the years). Helps name matching; never authority for a fiscal write."""
|
|
if not normalized or session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "alias", KbKnowledge.value == normalized[:200])):
|
|
return
|
|
if (session.scalar(select(func.count()).select_from(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "alias")) or 0) >= cap:
|
|
return
|
|
session.add(KbKnowledge(kb_product_id=product.id, knowledge_key="alias", value=normalized[:200], source="ALIAS", status=OBSERVED, evidence={"families": {}, "clients": [], "info": {}}))
|
|
|
|
|
|
def add_evidence(session, product: KbProduct, key: str, value: str, source: str, *, family: str, client_label: str | None = None, status: str = OBSERVED, info: dict | None = None,
|
|
lineage: dict | None = None) -> KbKnowledge:
|
|
"""lineage (client data only): {"master": 0/1, "docs": n, "years": {"2024": n}} of ONE client for this (GTIN, value); merged, never double counted."""
|
|
row = session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == key, KbKnowledge.value == value))
|
|
if row is None:
|
|
row = KbKnowledge(kb_product_id=product.id, knowledge_key=key, value=value, source=source[:32], status=status, evidence={"families": {}, "clients": [], "lineages": {}, "info": info or {}})
|
|
session.add(row)
|
|
session.flush()
|
|
evidence = dict(row.evidence or {})
|
|
families = dict(evidence.get("families") or {})
|
|
families[family] = _now().isoformat()
|
|
clients = list(evidence.get("clients") or [])
|
|
if client_label and client_label not in clients:
|
|
clients.append(client_label)
|
|
lineages = dict(evidence.get("lineages") or {})
|
|
if client_label and lineage:
|
|
cur = dict(lineages.get(client_label) or {"master": 0, "docs": 0, "years": {}})
|
|
cur["master"] = max(int(cur.get("master", 0)), int(lineage.get("master", 0)))
|
|
cur["docs"] = int(cur.get("docs", 0)) + int(lineage.get("docs", 0))
|
|
years = dict(cur.get("years") or {})
|
|
for y, n in (lineage.get("years") or {}).items():
|
|
years[str(y)] = int(years.get(str(y), 0)) + int(n)
|
|
cur["years"] = years
|
|
lineages[client_label] = cur
|
|
row.evidence = {**evidence, "families": families, "clients": clients, "lineages": lineages, "info": {**(evidence.get("info") or {}), **(info or {})}}
|
|
return row
|
|
|
|
|
|
def _lineage_summary(rows):
|
|
clients, docs, years = set(), 0, {}
|
|
for r in rows:
|
|
ev = r.evidence or {}
|
|
for c, lin in (ev.get("lineages") or {}).items():
|
|
clients.add(c)
|
|
docs += int(lin.get("docs", 0))
|
|
for y, n in (lin.get("years") or {}).items():
|
|
years[y] = years.get(y, 0) + int(n)
|
|
for c in ev.get("clients") or []:
|
|
clients.add(c)
|
|
return {"clients": sorted(clients), "docs": docs, "years": sorted(years), "last_year": max(int(y) for y in years) if years else None}
|
|
|
|
|
|
def independent_count(rows) -> tuple[int, list[str]]:
|
|
"""Independent families: each NCM-bearing provider = 1; ALL client-derived data = 1 and only when STRONG (>= STRONG_DOCS documents over >= STRONG_YEARS years, or >= 2 distinct clients)."""
|
|
names = sorted({f for r in rows for f in ((r.evidence or {}).get("families") or {}) if f.startswith("PROVIDER:")})
|
|
summary = _lineage_summary(rows)
|
|
has_client = any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in rows)
|
|
if has_client and (len(summary["clients"]) >= 2 or (summary["docs"] >= STRONG_DOCS and len(summary["years"]) >= STRONG_YEARS)):
|
|
names.append(CLIENT_FAMILY)
|
|
return len(names), names
|
|
|
|
|
|
def _official_period(session, value):
|
|
"""CURRENT_VALID / HISTORICAL_EXPIRED (exists in the dataset but not vigente today) / HISTORICAL_INVALID (never in the dataset)."""
|
|
from .resolver import ncm_exists_any
|
|
return "CURRENT_VALID" if ncm_exists_any(session, value, current=True) else ("HISTORICAL_EXPIRED" if ncm_exists_any(session, value, current=False) else "HISTORICAL_INVALID")
|
|
|
|
|
|
def _mark(row, state):
|
|
row.evidence = {**(row.evidence or {}), "info": {**((row.evidence or {}).get("info") or {}), "historical_state": state}}
|
|
|
|
|
|
OPERATOR_FAMILY = "OPERATOR_ATTESTATION"
|
|
|
|
|
|
def attest(session, gtin_key: str, ncm: str, operator: str, is_official, note: str = "") -> dict:
|
|
"""A human fiscal decision by an ADMIN (audited by the caller). Only a value that is ALREADY a candidate in the KB (some local/provider evidence) and vigente in the official dataset can be attested;
|
|
the dissenting candidates become STALE (REJECTED_BY_OPERATOR) and the product is promoted with the promotion audit naming the operator."""
|
|
product = find_by_gtin(session, gtin_key)
|
|
if product is None:
|
|
raise ValueError("GTIN_UNKNOWN_IN_KB")
|
|
row = session.scalar(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm", KbKnowledge.value == ncm))
|
|
if row is None:
|
|
raise ValueError("NCM_IS_NOT_A_CANDIDATE_OF_THIS_GTIN")
|
|
if not is_official(ncm):
|
|
raise ValueError("NCM_NOT_VIGENTE_IN_OFFICIAL_DATASET")
|
|
add_evidence(session, product, "ncm", ncm, f"operator:{operator}"[:32], family=OPERATOR_FAMILY, info={"attested_by": operator, "attested_at": _now().isoformat(), "note": note[:200]})
|
|
for other in session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == "ncm", KbKnowledge.value != ncm)):
|
|
if OPERATOR_FAMILY not in ((other.evidence or {}).get("families") or {}):
|
|
other.status = STALE
|
|
_mark(other, "REJECTED_BY_OPERATOR")
|
|
return evaluate(session, product, "ncm", is_official)
|
|
|
|
|
|
def evaluate(session, product: KbProduct, key: str, is_official) -> dict:
|
|
"""Recomputes and PERSISTS the status of one field. Never 'majority wins': two current-valid values => CONFLICT."""
|
|
rows = list(session.scalars(select(KbKnowledge).where(KbKnowledge.kb_product_id == product.id, KbKnowledge.knowledge_key == key)))
|
|
if not rows:
|
|
return {"value": None, "status": UNVERIFIED, "families": [], "candidates": [], "rejected": [], "detail": {}}
|
|
valid = [r for r in rows if is_official(r.value)]
|
|
rejected = [r.value for r in rows if not is_official(r.value)]
|
|
for r in rows:
|
|
if r not in valid:
|
|
state = _official_period(session, r.value) if key == "ncm" else "HISTORICAL_INVALID"
|
|
r.status = STALE if state == "HISTORICAL_EXPIRED" else OBSERVED
|
|
_mark(r, state)
|
|
attested = {r.value for r in valid if OPERATOR_FAMILY in ((r.evidence or {}).get("families") or {})}
|
|
if len(attested) == 1: # a human decision resolves the candidates: the others are recorded as rejected, not "outvoted"
|
|
winner = next(iter(attested))
|
|
for r in valid:
|
|
if r.value != winner:
|
|
r.status = STALE
|
|
_mark(r, "REJECTED_BY_OPERATOR")
|
|
valid = [r for r in valid if r.value == winner]
|
|
values = sorted({r.value for r in valid})
|
|
if len(values) > 1:
|
|
for r in valid:
|
|
r.status = CONFLICT
|
|
_mark(r, "HISTORICAL_CONFLICT")
|
|
return {"value": None, "status": CONFLICT, "families": [], "candidates": values, "rejected": rejected, "detail": {"clients": _lineage_summary(valid)["clients"]}}
|
|
if not values:
|
|
return {"value": None, "status": OBSERVED, "families": [], "candidates": [], "rejected": rejected, "detail": {}}
|
|
chosen = [r for r in valid if r.value == values[0]]
|
|
old = chosen[0].status
|
|
count, names = independent_count(chosen)
|
|
summary = _lineage_summary(chosen)
|
|
other_last = _lineage_summary([r for r in rows if r not in chosen])["last_year"]
|
|
superseded = other_last is not None and summary["last_year"] is not None and other_last >= summary["last_year"] # a different (expired) NCM is at least as recent: not stable
|
|
if superseded and CLIENT_FAMILY in names:
|
|
names.remove(CLIENT_FAMILY)
|
|
count -= 1
|
|
if any(OPERATOR_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen):
|
|
names = sorted(set(names) | {OPERATOR_FAMILY} | ({CLIENT_FAMILY} if any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen) else set()))
|
|
count = len(names)
|
|
providers = [n for n in names if n.startswith("PROVIDER:")]
|
|
already = any(r.status == VERIFIED_KB and r.validated_at for r in chosen)
|
|
has_client_data = any(CLIENT_FAMILY in ((r.evidence or {}).get("families") or {}) for r in chosen)
|
|
if count >= 2 or already:
|
|
status = VERIFIED_KB
|
|
elif providers and has_client_data:
|
|
status = EXTERNAL_CORROBORATED
|
|
elif CLIENT_FAMILY in names:
|
|
status = CORROBORATED
|
|
else:
|
|
status = OBSERVED
|
|
for r in chosen:
|
|
r.status = status
|
|
_mark(r, "HISTORICAL_CHANGED" if superseded else "HISTORICAL_STABLE_CURRENT")
|
|
if status == VERIFIED_KB and not r.validated_at:
|
|
r.validated_at = _now()
|
|
if status == VERIFIED_KB and old != VERIFIED_KB and key == "ncm":
|
|
from .resolver import dataset_versions
|
|
session.add(KbPromotionAudit(kb_product_id=product.id, gtin=product.gtin, field=key, value=values[0], old_status=old, new_status=status, families=names,
|
|
sources={"clients": summary["clients"], "documents": summary["docs"], "years": summary["years"], "providers": providers,
|
|
"operator": next((((r.evidence or {}).get("info") or {}).get("attested_by") for r in chosen if OPERATOR_FAMILY in ((r.evidence or {}).get("families") or {})), None)},
|
|
official_dataset=dataset_versions(session).get("NCM", {}), rule_version=RULE_VERSION))
|
|
return {"value": values[0], "status": status, "families": names, "candidates": values, "rejected": rejected, "detail": summary}
|
|
|
|
|
|
def revalidate(session, is_official) -> dict:
|
|
"""After the OFFICIAL NCM dataset changed: never delete knowledge. Rows whose NCM is no longer vigente become STALE (STALE_REVIEW_REQUIRED, audited when they were VERIFIED_KB);
|
|
rows still vigente keep their status (VERIFIED_KB stays valid while the NCM is vigente)."""
|
|
from .resolver import dataset_versions
|
|
stale = still = 0
|
|
versions = dataset_versions(session).get("NCM", {})
|
|
for row in session.scalars(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm")):
|
|
if is_official(row.value):
|
|
still += 1
|
|
if row.status == STALE:
|
|
row.status = OBSERVED # became vigente again: needs a fresh evaluate() to regain its level
|
|
continue
|
|
if row.status != STALE:
|
|
was = row.status
|
|
row.status = STALE
|
|
_mark(row, "HISTORICAL_EXPIRED")
|
|
row.evidence = {**(row.evidence or {}), "info": {**((row.evidence or {}).get("info") or {}), "stale_reason": "STALE_REVIEW_REQUIRED"}}
|
|
stale += 1
|
|
if was == VERIFIED_KB:
|
|
product = session.get(KbProduct, row.kb_product_id)
|
|
session.add(KbPromotionAudit(kb_product_id=row.kb_product_id, gtin=product.gtin if product else None, field="ncm", value=row.value, old_status=was, new_status=STALE,
|
|
families=sorted((row.evidence or {}).get("families") or {}), sources={"reason": "NCM_NO_LONGER_VIGENTE"}, official_dataset=versions, rule_version=RULE_VERSION))
|
|
session.flush()
|
|
return {"stale": stale, "still_vigente": still}
|