61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
"""GTIN KB audit (section 10) — validity, cross-tenant matches and conflicts.
|
|
|
|
Reads the seed payload (ground truth of what was observed per tenant) plus the KB tables.
|
|
"""
|
|
import json
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
|
|
from central.app.ai.fingerprint import validate_gtin
|
|
|
|
|
|
def main(payload_path="docs/runs/ai_kb_seed_payload.json", output="docs/GTIN_KB_AUDIT_V1.json") -> dict:
|
|
payload = json.loads(Path(payload_path).read_text(encoding="utf8"))
|
|
by_gtin = defaultdict(list)
|
|
invalid_gtin_raw = 0
|
|
for product in payload["products"]:
|
|
raw = product.get("gtin")
|
|
if raw is None:
|
|
continue
|
|
if validate_gtin(raw) is None:
|
|
invalid_gtin_raw += 1 # defensive: export already validates
|
|
continue
|
|
by_gtin[raw].append(product)
|
|
duplicated_same_product = sum(1 for entries in by_gtin.values() if len(entries) > 1)
|
|
conflicting = 0
|
|
cross_tenant = 0
|
|
conflict_examples = []
|
|
for gtin, entries in by_gtin.items():
|
|
ncms = {entry.get("ncm") for entry in entries if entry.get("ncm")}
|
|
databases = {obs["database"] for entry in entries for obs in entry["observations"]}
|
|
if len(ncms) > 1:
|
|
conflicting += 1
|
|
if len(conflict_examples) < 10:
|
|
conflict_examples.append({"gtin": gtin, "ncms": sorted(ncms)})
|
|
if len(databases) > 1:
|
|
cross_tenant += 1
|
|
report = {
|
|
"artifact": "GTIN_KB_AUDIT_V1",
|
|
"gtin_valid": len(by_gtin),
|
|
"gtin_invalid_raw": invalid_gtin_raw,
|
|
"gtin_duplicated_same_product": duplicated_same_product,
|
|
"gtin_conflicting_products": conflicting,
|
|
"gtin_cross_tenant_matches": cross_tenant,
|
|
"conflict_examples": conflict_examples,
|
|
"policy": "same GTIN across tenants is STRONG evidence for sharing global product knowledge; tenant fiscal context is never shared",
|
|
}
|
|
Path(output).write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return report
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import argparse
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--payload-path", default="/tmp/ai_kb_seed_payload.json")
|
|
parser.add_argument("--output", default="/tmp/gtin_audit.json")
|
|
args = parser.parse_args()
|
|
print(json.dumps(main(args.payload_path, args.output), indent=2))
|