consultoria-as
a386ca5b4a
- F0 blindaje: Redis/Celery, pg_hba sin trust LAN, secretos a env 600, backups, webhook auth, jobs muertos - F1 caja: bugs POS, reembolsos al corte, apartados, CFDI, topes de descuento en cotizacion/apartado - F2/F7 IA de catalogo: autocategorizacion, dedup de SKUs con embeddings, OCR de facturas, descripciones, claves SAT, limpieza de fitment - F3 IA en el flujo: vision Yonke, preguntas MeLi, venta cruzada, anomalias, sinonimos, busqueda semantica (rerank/embeddings) - F4/F8 escala y cartera: /metrics+Prometheus+Grafana, pool, score de credito, churn, metricas multi-tenant - Nexus Yonke (M1-M6): modelo de datos 1-de-1, alta por donador, vitrina publica, MeLi usado, consignacion, garantia por dias, reportes, etiquetas - Marketplace B2B entre yonkes (catalogo + pedidos cross-tenant) - Seguridad: auth en /api/admin/* (35 endpoints), alertas por tenant - Migraciones v4.27/v4.28; timers expire-demos y rollup-metrics
239 lines
9.5 KiB
Python
239 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Flag/remove improbable QWEN fitment rows from ``inventory_vehicle_compat``.
|
|
|
|
The ``qwen_ai`` fitment backfill produced 20.7M rows on Estrada: 127,041
|
|
inventory items, average 162 vehicle links each and a hard cap of 200 — i.e.
|
|
almost every part was declared compatible with hundreds of vehicles. This
|
|
script finds those improbable rows and can **mark** them (reversible: set
|
|
``source='qwen_rejected'``, ``confidence=0``) or **delete** them.
|
|
|
|
Detectors (source = ``qwen_ai`` by default):
|
|
|
|
1. volume — items with >= ``--min-rows`` fitment rows (a single part rarely
|
|
fits hundreds of vehicles; the cap betrays completeness padding).
|
|
2. universal name — a normalized name shared by >= ``--name-min-items``
|
|
distinct inventory items (e.g. "BALATA DISCO") that still got a full
|
|
vehicle list.
|
|
3. semantic (opt-in, ``--semantic``) — QWEN embeddings cluster *name
|
|
variants* (cosine >= ``--similarity``) so "FILTRO AIRE" and "FILTRO DE
|
|
AIRE" are treated as one universal name. Degrades silently when the
|
|
provider is unavailable.
|
|
|
|
Safety: dry-run by default; nothing is written without ``--apply``.
|
|
Delete requires ``--apply --mode delete``; ``--mode mark`` is preferred.
|
|
|
|
Usage (as root, from /home/Autopartes):
|
|
python3 scripts/clean_fitment_qwen.py --tenant 28
|
|
python3 scripts/clean_fitment_qwen.py --tenant 28 --min-rows 120 --semantic
|
|
python3 scripts/clean_fitment_qwen.py --tenant 28 --apply --mode mark
|
|
python3 scripts/clean_fitment_qwen.py --tenant 28 --apply --mode delete --min-rows 150
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import unicodedata
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'pos'))
|
|
|
|
from tenant_db import get_tenant_conn # noqa: E402
|
|
|
|
|
|
def _norm_name(s):
|
|
s = unicodedata.normalize("NFD", str(s or ""))
|
|
s = s.encode("ascii", "ignore").decode("ascii")
|
|
out = []
|
|
for ch in s.upper():
|
|
out.append(ch if ch.isalnum() else " ")
|
|
return " ".join("".join(out).split())
|
|
|
|
|
|
def _collect_counts(cur, source, limit=None):
|
|
"""[(inventory_id, name, row_count)] for the given source."""
|
|
sql = """
|
|
SELECT c.inventory_id, i.name, count(*) AS n
|
|
FROM inventory_vehicle_compat c
|
|
JOIN inventory i ON i.id = c.inventory_id
|
|
WHERE c.source = %s
|
|
GROUP BY c.inventory_id, i.name
|
|
"""
|
|
if limit:
|
|
sql += " LIMIT %s"
|
|
cur.execute(sql, (source, limit))
|
|
else:
|
|
cur.execute(sql, (source,))
|
|
return cur.fetchall()
|
|
|
|
|
|
def _semantic_clusters(names, threshold=0.92, batch=64, timeout=30):
|
|
"""Union-find clusters of near-duplicate names via QWEN embeddings.
|
|
|
|
Returns {name: cluster_id} or {} when embeddings are unavailable.
|
|
"""
|
|
uniq = list(dict.fromkeys(n for n in names if n))
|
|
if len(uniq) < 2:
|
|
return {}
|
|
try:
|
|
from services import semantic_search as ss
|
|
if hasattr(ss, 'is_configured') and not ss.is_configured():
|
|
return {}
|
|
vectors = []
|
|
for i in range(0, len(uniq), batch):
|
|
out = ss.embed(uniq[i:i + batch], timeout=timeout)
|
|
if not out or len(out) != len(uniq[i:i + batch]):
|
|
return {}
|
|
vectors.extend(out)
|
|
except Exception:
|
|
return {}
|
|
|
|
parent = list(range(len(uniq)))
|
|
|
|
def find(x):
|
|
while parent[x] != x:
|
|
parent[x] = parent[parent[x]]
|
|
x = parent[x]
|
|
return x
|
|
|
|
def union(a, b):
|
|
ra, rb = find(a), find(b)
|
|
if ra != rb:
|
|
parent[rb] = ra
|
|
|
|
for i in range(len(uniq)):
|
|
for j in range(i + 1, len(uniq)):
|
|
sim = ss.cosine_similarity(vectors[i], vectors[j])
|
|
if sim is not None and sim >= threshold:
|
|
union(i, j)
|
|
|
|
return {uniq[i]: find(i) for i in range(len(uniq))}
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description='Clean improbable QWEN fitment rows.')
|
|
p.add_argument('--tenant', type=int, required=True)
|
|
p.add_argument('--source', default='qwen_ai')
|
|
p.add_argument('--min-rows', type=int, default=120,
|
|
help='flag items with >= N fitment rows')
|
|
p.add_argument('--name-min-items', type=int, default=8,
|
|
help='flag a normalized name shared by >= N inventory items')
|
|
p.add_argument('--min-name-vol', type=int, default=100,
|
|
help='only cluster names whose item count is >= this')
|
|
p.add_argument('--semantic', action='store_true',
|
|
help='merge name variants with embeddings (opt-in)')
|
|
p.add_argument('--similarity', type=float, default=0.92)
|
|
p.add_argument('--semantic-max', type=int, default=150,
|
|
help='max distinct names sent to the embedding provider')
|
|
p.add_argument('--mode', choices=['mark', 'delete'], default='mark')
|
|
p.add_argument('--apply', action='store_true', help='persist (default dry-run)')
|
|
p.add_argument('--limit', type=int, default=0, help='cap inventory ids scanned')
|
|
p.add_argument('--inventory-ids', default=None,
|
|
help='comma-separated inventory ids to force as suspects (targeted runs)')
|
|
p.add_argument('--show', type=int, default=12)
|
|
args = p.parse_args()
|
|
|
|
conn = get_tenant_conn(args.tenant)
|
|
try:
|
|
cur = conn.cursor()
|
|
print(f"tenant={args.tenant} source={args.source} min_rows={args.min_rows} "
|
|
f"name_min_items={args.name_min_items} semantic={args.semantic} "
|
|
f"mode={args.mode} apply={args.apply}")
|
|
|
|
if args.inventory_ids:
|
|
# Targeted run: skip the heavy detection scans, use given ids only.
|
|
rows = []
|
|
name_to_ids = {}
|
|
by_volume = by_name = by_semantic = set()
|
|
print(" targeted mode: --inventory-ids provided, detection skipped")
|
|
else:
|
|
rows = _collect_counts(cur, args.source, args.limit or None)
|
|
print(f" inventory items with source={args.source}: {len(rows)}")
|
|
|
|
by_volume = {iid for iid, _n, n in rows if n >= args.min_rows}
|
|
|
|
name_to_ids = {}
|
|
for iid, name, _n in rows:
|
|
name_to_ids.setdefault(_norm_name(name), set()).add(iid)
|
|
by_name = set()
|
|
for name, ids in name_to_ids.items():
|
|
if len(ids) >= args.name_min_items:
|
|
by_name.update(ids)
|
|
|
|
by_semantic = set()
|
|
if args.semantic:
|
|
vol_names = [name for name, ids in name_to_ids.items()
|
|
if len(ids) >= args.min_name_vol]
|
|
vol_names = vol_names[:args.semantic_max]
|
|
clusters = _semantic_clusters(vol_names, threshold=args.similarity)
|
|
if not clusters:
|
|
print(" [semantic] sin embeddings (degradado) — solo reglas")
|
|
else:
|
|
cluster_ids = {}
|
|
for name, cid in clusters.items():
|
|
cluster_ids.setdefault(cid, set()).update(name_to_ids.get(name, set()))
|
|
for cid, ids in cluster_ids.items():
|
|
if len(ids) >= args.name_min_items:
|
|
by_semantic.update(ids)
|
|
print(f" [semantic] names clustered={len(clusters)} "
|
|
f"extra_suspect_items={len(by_semantic - by_volume - by_name)}")
|
|
|
|
suspects = by_volume | by_name | by_semantic
|
|
if args.inventory_ids:
|
|
forced = {int(x) for x in args.inventory_ids.split(',') if x.strip()}
|
|
suspects |= forced
|
|
print(f" suspect inventory items: volume={len(by_volume)} "
|
|
f"universal_name={len(by_name)} semantic={len(by_semantic)} "
|
|
f"union={len(suspects)}")
|
|
|
|
if args.show:
|
|
print(" top inflated names:")
|
|
ranked = sorted(name_to_ids.items(), key=lambda kv: -len(kv[1]))[:args.show]
|
|
for name, ids in ranked:
|
|
print(f" {name[:48]:<48} items={len(ids):<5}")
|
|
|
|
# Count rows that would be affected
|
|
affected = 0
|
|
suspect_list = sorted(suspects)
|
|
for i in range(0, len(suspect_list), 1000):
|
|
chunk = suspect_list[i:i + 1000]
|
|
if not chunk:
|
|
continue
|
|
cur.execute(
|
|
f"SELECT count(*) FROM inventory_vehicle_compat "
|
|
f"WHERE source = %s AND inventory_id = ANY(%s)",
|
|
(args.source, chunk))
|
|
affected += cur.fetchone()[0]
|
|
total_src = sum(n for _i, _n, n in rows)
|
|
pct = (affected / total_src * 100) if total_src else 0.0
|
|
print(f" rows affected: {affected} / {total_src} ({pct:.1f}% of source={args.source})")
|
|
|
|
if not args.apply:
|
|
print(" DRY-RUN: nothing written. Re-run with --apply to persist.")
|
|
return 0
|
|
|
|
done = 0
|
|
for i in range(0, len(suspect_list), 1000):
|
|
chunk = suspect_list[i:i + 1000]
|
|
if not chunk:
|
|
continue
|
|
if args.mode == 'delete':
|
|
cur.execute(
|
|
"DELETE FROM inventory_vehicle_compat "
|
|
"WHERE source = %s AND inventory_id = ANY(%s)",
|
|
(args.source, chunk))
|
|
else:
|
|
cur.execute(
|
|
"UPDATE inventory_vehicle_compat "
|
|
"SET source = 'qwen_rejected', confidence = 0 "
|
|
"WHERE source = %s AND inventory_id = ANY(%s)",
|
|
(args.source, chunk))
|
|
done += cur.rowcount
|
|
conn.commit()
|
|
print(f" APPLIED mode={args.mode}: rows={done}")
|
|
return 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|