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
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the A5 business anomaly scan for one or all tenants.
|
|
|
|
Reports, per tenant, the anomalies detected by services.anomaly_detector:
|
|
atypical discounts (per cajero / per SKU), sales below cost, repeated
|
|
cancellations and off-hours tickets.
|
|
|
|
Usage (as postgres):
|
|
MASTER_DB_NAME=nexus_autopartes python3 scripts/run_anomaly_scan.py [db_name|all] [days]
|
|
python3 scripts/run_anomaly_scan.py tenant_refaccionaria_atlas 30
|
|
python3 scripts/run_anomaly_scan.py all 90 --json
|
|
|
|
Exit code is 1 when at least one critical/high finding is reported, so it can
|
|
be wired into cron/Alertmanager; 0 otherwise.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'pos'))
|
|
|
|
import psycopg2 # noqa: E402
|
|
from services.anomaly_detector import detect_anomalies # noqa: E402
|
|
|
|
MASTER_DB = os.environ.get('MASTER_DB_NAME', 'nexus_autoparts')
|
|
|
|
SEVERITY_ORDER = {'critical': 0, 'high': 1, 'medium': 2, 'low': 3}
|
|
|
|
|
|
def _tenant_dbs():
|
|
conn = psycopg2.connect(dbname=MASTER_DB)
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT db_name FROM tenants WHERE is_active = true ORDER BY db_name")
|
|
dbs = [r[0] for r in cur.fetchall()]
|
|
cur.close()
|
|
finally:
|
|
conn.close()
|
|
return dbs
|
|
|
|
|
|
def _print_report(db_name, report):
|
|
sample = report['sample']
|
|
cold = ' (COLD START: poca historia)' if report['cold_start'] else ''
|
|
print(f"\n=== {db_name}{cold} ===")
|
|
print(f" ventana: {report['window_days']}d | sale_items={sample['sale_items']} "
|
|
f"sales={sample['sales']} empleados={sample['employees']}")
|
|
if not report['employees'] and not report['skus']:
|
|
print(" sin anomalías")
|
|
return
|
|
|
|
for entry in report['employees']:
|
|
print(f" [empleado {entry['employee_id']} {entry['employee_name']}] "
|
|
f"score={entry['score']}")
|
|
for a in sorted(entry['anomalies'], key=lambda x: SEVERITY_ORDER.get(x['severity'], 9)):
|
|
print(f" - {a['severity']:8s} {a['type']:20s} {a['detail']}")
|
|
for entry in report['skus']:
|
|
types = ', '.join(sorted({a['type'] for a in entry['anomalies']}))
|
|
print(f" [sku {entry['inventory_id']} {entry['name']}] score={entry['score']} -> {types}")
|
|
|
|
|
|
def main():
|
|
args = [a for a in sys.argv[1:] if not a.startswith('--')]
|
|
as_json = '--json' in sys.argv
|
|
target = args[0] if args else 'all'
|
|
days = int(args[1]) if len(args) > 1 else 90
|
|
|
|
dbs = _tenant_dbs() if target == 'all' else [target]
|
|
reports = {}
|
|
exit_code = 0
|
|
|
|
for db in dbs:
|
|
try:
|
|
conn = psycopg2.connect(dbname=db)
|
|
except Exception as e:
|
|
print(f"[{db}] ERROR de conexión: {e}", file=sys.stderr)
|
|
exit_code = 2
|
|
continue
|
|
try:
|
|
report = detect_anomalies(conn, days=days)
|
|
except Exception as e:
|
|
print(f"[{db}] ERROR: {e}", file=sys.stderr)
|
|
exit_code = 2
|
|
continue
|
|
finally:
|
|
conn.close()
|
|
|
|
reports[db] = report
|
|
if not as_json:
|
|
_print_report(db, report)
|
|
if report['summary']['by_type'] and any(
|
|
e['anomalies'] and min(
|
|
SEVERITY_ORDER.get(a['severity'], 9) for a in e['anomalies']) <= 1
|
|
for e in report['employees'] + report['skus']):
|
|
exit_code = max(exit_code, 1)
|
|
|
|
if as_json:
|
|
print(json.dumps(reports, indent=2, default=str))
|
|
|
|
return exit_code
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|