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
112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""A4 CLI — suggest a condition grade for a used (yonke) piece from a photo.
|
|
|
|
Uses ``services.part_vision`` (QWEN -> Hermes -> OpenRouter). When no vision
|
|
backend answers it prints the returned error and exits non-zero.
|
|
|
|
Usage (from /home/Autopartes):
|
|
python3 scripts/grade_piece_vision.py --tenant 28 --piece 300537
|
|
python3 scripts/grade_piece_vision.py --tenant 28 --piece 300537 --image /ruta/foto.jpg
|
|
python3 scripts/grade_piece_vision.py --image /ruta/foto.jpg --name "Faro" --dry-run
|
|
python3 scripts/grade_piece_vision.py --tenant 28 --piece 300537 --apply
|
|
"""
|
|
|
|
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, help='tenant id')
|
|
ap.add_argument('--piece', type=int, help='inventory id of the piece')
|
|
ap.add_argument('--image', help='image file path (overrides the stored photo)')
|
|
ap.add_argument('--name', help='piece name for the prompt')
|
|
ap.add_argument('--category', help='piece category for the prompt')
|
|
ap.add_argument('--condition', default='usado')
|
|
ap.add_argument('--apply', action='store_true',
|
|
help='persist the suggested grade to inventory.condition_grade')
|
|
args = ap.parse_args()
|
|
|
|
_load_env()
|
|
|
|
from services import part_vision
|
|
from services.image_service import DATA_DIR, STATIC_DIR
|
|
|
|
image_bytes = None
|
|
piece = None
|
|
conn = None
|
|
|
|
try:
|
|
if args.tenant and args.piece:
|
|
from tenant_db import get_tenant_conn
|
|
from services import yonke_service
|
|
conn = get_tenant_conn(args.tenant)
|
|
piece = yonke_service.get_piece(conn, args.piece)
|
|
if not piece:
|
|
print(json.dumps({'error': 'Piece not found'}))
|
|
return 1
|
|
|
|
if args.image:
|
|
with open(args.image, 'rb') as fh:
|
|
image_bytes = fh.read()
|
|
elif piece:
|
|
url = piece.get('image_url')
|
|
for img in piece.get('images') or []:
|
|
if img.get('is_primary'):
|
|
url = img.get('url') or url
|
|
break
|
|
idx = (url or '').find('/images/parts/')
|
|
if idx != -1:
|
|
rel = url[idx + len('/images/parts/'):]
|
|
for base in (DATA_DIR, STATIC_DIR):
|
|
path = os.path.join(base, rel)
|
|
if os.path.exists(path):
|
|
with open(path, 'rb') as fh:
|
|
image_bytes = fh.read()
|
|
break
|
|
|
|
if not image_bytes:
|
|
print(json.dumps({'error': 'No image available (use --image or a piece with a photo)'}))
|
|
return 1
|
|
|
|
result = part_vision.suggest_grade(
|
|
image_bytes=image_bytes,
|
|
name=args.name or (piece or {}).get('name'),
|
|
category=args.category,
|
|
condition=args.condition,
|
|
)
|
|
|
|
if args.apply and result.get('condition_grade_suggested') and conn and piece:
|
|
cur = conn.cursor()
|
|
cur.execute('UPDATE inventory SET condition_grade = %s WHERE id = %s',
|
|
(result['condition_grade_suggested'], piece['id']))
|
|
conn.commit()
|
|
cur.close()
|
|
result['applied'] = True
|
|
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if result.get('condition_grade_suggested') else 2
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|