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
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""A7 CLI — intent + suggested answer for MercadoLibre buyer questions.
|
|
|
|
Uses ``services.question_ai``. Never sends anything unless ``--auto-answer`` is
|
|
given AND the tenant has ``meli_auto_answer_questions = true``.
|
|
|
|
Usage (from /home/Autopartes):
|
|
python3 scripts/suggest_meli_answer.py --tenant 28 --question 123
|
|
python3 scripts/suggest_meli_answer.py --tenant 28 --pending --no-llm
|
|
python3 scripts/suggest_meli_answer.py --tenant 28 --auto-answer
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'pos'))
|
|
|
|
|
|
def _load_env():
|
|
env_file = '/etc/nexus/pos.env'
|
|
if os.environ.get('MASTER_DB_URL') or not os.path.exists(env_file):
|
|
return
|
|
for line in open(env_file):
|
|
line = line.strip()
|
|
if not line or line.startswith('#') or '=' not in line:
|
|
continue
|
|
k, v = line.split('=', 1)
|
|
os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument('--tenant', type=int, required=True)
|
|
ap.add_argument('--question', type=int, help='local marketplace_questions.id')
|
|
ap.add_argument('--pending', action='store_true',
|
|
help='suggest for every unanswered question (limit --limit)')
|
|
ap.add_argument('--limit', type=int, default=20)
|
|
ap.add_argument('--no-llm', action='store_true', help='templates only, no LLM fallback')
|
|
ap.add_argument('--auto-answer', action='store_true',
|
|
help='send high-confidence template answers (requires tenant flag)')
|
|
args = ap.parse_args()
|
|
|
|
_load_env()
|
|
|
|
from tenant_db import get_tenant_conn
|
|
from services import question_ai, marketplace_external_service as meli
|
|
|
|
conn = get_tenant_conn(args.tenant)
|
|
try:
|
|
if args.auto_answer:
|
|
print(json.dumps(meli.auto_answer_questions(conn), ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
use_llm = not args.no_llm
|
|
if args.question:
|
|
result = question_ai.get_question_suggestion(conn, args.question, use_llm=use_llm)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
if args.pending:
|
|
cur = conn.cursor()
|
|
cur.execute("""
|
|
SELECT id FROM marketplace_questions
|
|
WHERE (status IS NULL OR status = 'unanswered')
|
|
ORDER BY question_date ASC NULLS LAST LIMIT %s
|
|
""", (args.limit,))
|
|
ids = [r[0] for r in cur.fetchall()]
|
|
cur.close()
|
|
for qid in ids:
|
|
result = question_ai.get_question_suggestion(conn, qid, use_llm=use_llm)
|
|
print(json.dumps({'question_id': qid, **result}, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
ap.error('use --question, --pending or --auto-answer')
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|