gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
76 lines
4.3 KiB
Python
76 lines
4.3 KiB
Python
"""READ-ONLY historical miner for product identity/fiscal evidence in the ERP databases (product master + NFC-e items + NF-e entry items + barcode tables).
|
|
|
|
python scripts/enrichment_history_miner.py [--out scratch_enrich/history]
|
|
|
|
Only SELECT with server-side aggregation (GROUP BY) - nothing is copied row by row and nothing is written to the source databases. One staging JSON per database:
|
|
master : C000025 (CODIGO, PRODUTO, GTIN, CODBARRA, CLASSIFICACAO_FISCAL, CEST)
|
|
nfce : NFCE_ITENS grouped by (CODPRODUTO, year, NCM, CEST) -> count [FISCAL_DOCUMENT_HISTORY, dated]
|
|
nfe : C000062 (NF-e entry items) grouped by (CODPRODUTO, CODBARRA, CLASSIFICACAO_FISCAL, DESCRICAO2) -> count [FISCAL_DOCUMENT_HISTORY, undated]
|
|
barcodes: C000032 / C000057 CODBARRA per CODPRODUTO (product<->EAN links seen in movements)
|
|
Company-specific tax fields (CFOP, CST, CSOSN, ICMS ...) are NOT read.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import tracemalloc
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts"))
|
|
from dotenv import load_dotenv # noqa: E402
|
|
|
|
load_dotenv(ROOT / ".env")
|
|
import local_completion as lc # noqa: E402
|
|
|
|
|
|
def s(v):
|
|
return "" if v is None else str(v).strip()
|
|
|
|
|
|
def mine(name, path, out_dir):
|
|
conn = lc.connect(path) # READ COMMITTED READ ONLY
|
|
cur = conn.cursor()
|
|
t0, stats = time.time(), {"database": name}
|
|
cur.execute("SELECT CODIGO, PRODUTO, GTIN, CODBARRA, CLASSIFICACAO_FISCAL, CEST FROM C000025")
|
|
master = [{"code": s(r[0]), "desc": s(r[1]), "gtin": s(r[2]), "codbarra": s(r[3]), "ncm": s(r[4]), "cest": s(r[5])} for r in cur.fetchall()]
|
|
cur.execute("SELECT COUNT(*) FROM NFCE_ITENS"); stats["nfce_rows_scanned"] = cur.fetchone()[0]
|
|
cur.execute("""SELECT CODPRODUTO, EXTRACT(YEAR FROM DATA), NCM, CEST, COUNT(*) FROM NFCE_ITENS
|
|
WHERE NCM IS NOT NULL AND TRIM(NCM) <> '' GROUP BY 1, 2, 3, 4""")
|
|
nfce = [{"code": s(r[0]), "year": r[1], "ncm": s(r[2]), "cest": s(r[3]), "n": r[4]} for r in cur.fetchall()]
|
|
cur.execute("SELECT COUNT(*) FROM C000062"); stats["nfe_rows_scanned"] = cur.fetchone()[0]
|
|
cur.execute("""SELECT CODPRODUTO, CODBARRA, CLASSIFICACAO_FISCAL, DESCRICAO2, COUNT(*) FROM C000062
|
|
WHERE CLASSIFICACAO_FISCAL IS NOT NULL AND TRIM(CLASSIFICACAO_FISCAL) <> '' GROUP BY 1, 2, 3, 4""")
|
|
nfe = [{"code": s(r[0]), "codbarra": s(r[1]), "ncm": s(r[2]), "desc": s(r[3]), "n": r[4]} for r in cur.fetchall()]
|
|
barcodes = []
|
|
for table in ("C000032", "C000057"):
|
|
cur.execute(f"SELECT COUNT(*) FROM {table}"); stats[f"{table.lower()}_rows_scanned"] = cur.fetchone()[0]
|
|
cur.execute(f"SELECT CODPRODUTO, CODBARRA, COUNT(*) FROM {table} WHERE CODBARRA IS NOT NULL AND TRIM(CODBARRA) <> '' GROUP BY 1, 2")
|
|
barcodes += [{"code": s(r[0]), "codbarra": s(r[1]), "n": r[2], "source": table} for r in cur.fetchall()]
|
|
conn.close()
|
|
years = sorted({r["year"] for r in nfce if r["year"]})
|
|
stats.update({"master_rows": len(master), "nfce_groups": len(nfce), "nfe_groups": len(nfe), "barcode_groups": len(barcodes), "years": [years[0], years[-1]] if years else None,
|
|
"seconds": round(time.time() - t0, 1)})
|
|
Path(out_dir).mkdir(parents=True, exist_ok=True)
|
|
(Path(out_dir) / f"{name}.json").write_text(json.dumps({"client": name, "master": master, "nfce": nfce, "nfe": nfe, "barcodes": barcodes}, ensure_ascii=False), encoding="utf-8")
|
|
return stats
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--out", default=str(ROOT / "scratch_enrich" / "history"))
|
|
args = ap.parse_args()
|
|
dbs = dict(lc.allowlist())
|
|
dbs["TESTESOFICIAIS"] = os.environ.get("FIREBIRD_DATABASE_PATH", str(ROOT / "TESTESOFICIAIS.FDB"))
|
|
tracemalloc.start()
|
|
report = [mine(n, p, args.out) for n, p in dbs.items() if Path(p).exists()]
|
|
peak = tracemalloc.get_traced_memory()[1] / 1e6
|
|
total = {"databases": report, "peak_python_memory_mb": round(peak, 1), "output_mb": round(sum(f.stat().st_size for f in Path(args.out).glob("*.json")) / 1e6, 2)}
|
|
(Path(args.out) / "_miner_report.json").write_text(json.dumps(total, indent=1), encoding="utf-8")
|
|
print(json.dumps(total, indent=1))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|