97 lines
5.4 KiB
Python
97 lines
5.4 KiB
Python
"""Pure capability detection / schema classification for SGV (C000025/C000004) databases. No I/O, no writes."""
|
|
|
|
WRITE_TARGETS = ["CLASSIFICACAO_FISCAL", "CFOP_ENTRADA", "CST_ENTRADA", "CFOP", "CST", "ALIQUOTA", "CEST", "SITUACAO_TRIBUTARIA", "COD_ENTRADA_COFINS",
|
|
"COD_SAIDA_COFINS", "ALIQ_COFINS", "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "CSOSN", "IPI", "CST_IPI", "IAT", "INCIDENCIA_PISCOFINS",
|
|
"PISCOFINS", "IPPT"]
|
|
ORIGIN_FIELD = "COD_ORIGEM_MERCADORIA"
|
|
|
|
# Firebird RDB$FIELD_TYPE -> family
|
|
FAMILY = {7: "int", 8: "int", 16: "int", 10: "float", 11: "float", 27: "float", 12: "date", 13: "time", 35: "timestamp", 14: "text", 37: "text", 261: "blob"}
|
|
# accepted families per field (UI edit controls take text or numbers; numeric columns may be int/float)
|
|
ACCEPT = {"ALIQUOTA": {"float", "int"}, "IPI": {"float", "int"}, "COD_ORIGEM_MERCADORIA": {"int"}}
|
|
DEFAULT_ACCEPT = {"text", "int"} # text columns; some ERPs keep codes such as CST_ENTRADA in SMALLINT
|
|
|
|
CAPABILITIES = {
|
|
"supports_ncm": ["CLASSIFICACAO_FISCAL"],
|
|
"supports_cfop_entry": ["CFOP_ENTRADA"],
|
|
"supports_cfop_exit": ["CFOP"],
|
|
"supports_cst_entry": ["CST_ENTRADA"],
|
|
"supports_cst_exit": ["CST"],
|
|
"supports_icms_rate": ["ALIQUOTA"],
|
|
"supports_cest": ["CEST"],
|
|
"supports_csosn": ["CSOSN"],
|
|
"supports_ipi": ["IPI", "CST_IPI"],
|
|
"supports_pis": ["COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS"],
|
|
"supports_cofins": ["COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS"],
|
|
"supports_origin": [ORIGIN_FIELD],
|
|
"supports_iat": ["IAT"],
|
|
"supports_incidence": ["INCIDENCIA_PISCOFINS"],
|
|
"supports_ippt": ["IPPT"],
|
|
}
|
|
|
|
|
|
def type_ok(field, type_code):
|
|
return FAMILY.get(type_code, "other") in ACCEPT.get(field, DEFAULT_ACCEPT)
|
|
|
|
|
|
def detect_capabilities(columns):
|
|
"""columns: {FIELD: rdb_type_code} of C000025. capability=true only if every required column exists with a usable type."""
|
|
out = {}
|
|
for cap, fields in CAPABILITIES.items():
|
|
out[cap] = all(f in columns and type_ok(f, columns[f]) for f in fields)
|
|
return out
|
|
|
|
|
|
def field_compatibility(columns, reference_types=None):
|
|
"""Per write target (+ origin): PRESENT_OK / PRESENT_TYPE_VARIATION / MISSING."""
|
|
res = {}
|
|
for f in WRITE_TARGETS + [ORIGIN_FIELD]:
|
|
if f not in columns:
|
|
res[f] = "MISSING"
|
|
elif not type_ok(f, columns[f]):
|
|
res[f] = "PRESENT_TYPE_INCOMPATIBLE"
|
|
elif reference_types and reference_types.get(f) is not None and reference_types[f] != columns[f]:
|
|
res[f] = "PRESENT_TYPE_VARIATION"
|
|
else:
|
|
res[f] = "PRESENT_OK"
|
|
return res
|
|
|
|
|
|
def classify_schema(has_c000004, has_c000025, fingerprint, reference_fingerprint, compat):
|
|
"""REFERENCE_SCHEMA_MATCH | COMPATIBLE_WITH_VARIATIONS | INCOMPATIBLE. A different hash is not a failure
|
|
when every required capability exists (origin is optional: it is only PROBABLE in the map)."""
|
|
if not (has_c000004 and has_c000025):
|
|
return "INCOMPATIBLE"
|
|
required = {f: v for f, v in compat.items() if f in WRITE_TARGETS}
|
|
if any(v in ("MISSING", "PRESENT_TYPE_INCOMPATIBLE") for v in required.values()):
|
|
return "INCOMPATIBLE"
|
|
if fingerprint == reference_fingerprint:
|
|
return "REFERENCE_SCHEMA_MATCH"
|
|
return "COMPATIBLE_WITH_VARIATIONS"
|
|
|
|
|
|
NAME_KEYWORDS = {"RESTAURANT": ("RESTAURANTE", "PIZZARIA", "LANCHONETE", "CHURRASC"), "RETAIL_MARKET": ("MERCADO", "SUPERMERC", "MERCEARIA", "ATACAD"),
|
|
"APPAREL": ("MODAS", "CONFEC", "BOUTIQUE", "ROUPAS"), "CONSTRUCTION": ("CONSTRU", "MATERIAIS"), "AUTO_REPAIR": ("OFICINA", "MECANICA", "AUTO PECAS", "AUTOPECAS")}
|
|
|
|
|
|
def infer_segment(signals, company_name=""):
|
|
"""Evidence-based segment guess from populated tables/flags (never from the file name).
|
|
signals: {"restaurant_tables_rows": int, "kitchen_products": int, "recipes": int, "service_orders": int, "vehicle_products": int, "grade_items": int, "nfce_items": int}"""
|
|
ev = []
|
|
score = {"RESTAURANT": 0, "AUTO_REPAIR": 0, "APPAREL": 0, "RETAIL_MARKET": 0, "CONSTRUCTION": 0}
|
|
import unicodedata
|
|
up = unicodedata.normalize("NFKD", company_name or "").encode("ascii", "ignore").decode().upper()
|
|
for seg_name, kws in NAME_KEYWORDS.items():
|
|
hit = [k for k in kws if k in up]
|
|
if hit:
|
|
score[seg_name] += 1; ev.append(f"company name in C000004 contains {hit[0]} (weak, name is data, not the file name)")
|
|
if signals.get("restaurant_tables_rows", 0) > 50: score["RESTAURANT"] += 2; ev.append(f"mesas/comandas rows={signals['restaurant_tables_rows']}")
|
|
if signals.get("kitchen_products", 0) > 0: score["RESTAURANT"] += 1; ev.append(f"PROD_COZINHA products={signals['kitchen_products']}")
|
|
if signals.get("recipes", 0) > 0: score["RESTAURANT"] += 1; ev.append(f"recipes={signals['recipes']}")
|
|
if signals.get("service_orders", 0) > 20: score["AUTO_REPAIR"] += 2; ev.append(f"ordens de serviço={signals['service_orders']}")
|
|
if signals.get("vehicle_products", 0) > 0: score["AUTO_REPAIR"] += 1; ev.append(f"veículos={signals['vehicle_products']}")
|
|
if signals.get("grade_items", 0) > 50: score["APPAREL"] += 2; ev.append(f"grade (tamanho/cor) rows={signals['grade_items']}")
|
|
if signals.get("nfce_items", 0) > 1000 and max(score.values()) == 0: score["RETAIL_MARKET"] += 1; ev.append("high-volume NFC-e retail without restaurant/OS/grade signals")
|
|
best = max(score, key=score.get)
|
|
return (best if score[best] > 0 else "UNKNOWN"), score, ev
|