FASE 4-5-6: Infraestructura, CRM, Service Orders, Notificaciones, Ahorro, Logistica, API Publica
FASE 4: - Redis cache de stock con fallback graceful - Multi-moneda (MXN/USD) con contabilidad en MXN - Proveedores y ordenes de compra completo - Meilisearch 1.5M+ partes indexadas - Metabase KPIs con dashboard auto-generado FASE 5: - CRM mejorado: activities, tags, loyalty program, analytics - Imagenes de partes: upload, resize, thumbnails WebP - Ordenes de servicio Kanban: received->diagnosis->repair->ready->delivered - Garantias/RMA, alertas de reorden, multi-sucursal - Stubs BNPL (APLAZO) y ERP Sync (Aspel/Contpaqi) FASE 6: - Notificaciones automaticas: push/WhatsApp/email/in-app - Reportes de ahorro vs retail_price - Logistica + tracking: DHL, FedEx, Estafeta, 99min, Uber - API Publica: API keys, rate limiting, catalog search Migraciones: v1.9-v3.0 Tests: 93/93 pasando Backup: nexus_backup_20260427_045859.tar.gz
This commit is contained in:
Nexus Dev
273
pos/services/warranty_engine.py
Normal file
273
pos/services/warranty_engine.py
Normal file
@@ -0,0 +1,273 @@
|
||||
"""Warranty / RMA engine (Mejora #10).
|
||||
|
||||
Registers warranties at sale time and manages the claim lifecycle.
|
||||
|
||||
Tables:
|
||||
warranties — one row per warranted item sold
|
||||
warranty_claims — one row per claim filed
|
||||
"""
|
||||
|
||||
from datetime import date, timedelta
|
||||
from services.audit import log_action
|
||||
|
||||
|
||||
def register_warranty(conn, sale_id, sale_item_id, inventory_id,
|
||||
customer_id, warranty_months, supplier_id=None,
|
||||
part_number=None, name=None, notes=None):
|
||||
"""Register a warranty for a sold item.
|
||||
|
||||
Args:
|
||||
warranty_months: int (e.g., 12, 24, 36)
|
||||
Returns:
|
||||
int: warranty_id
|
||||
"""
|
||||
if not warranty_months or warranty_months <= 0:
|
||||
raise ValueError("warranty_months must be a positive integer")
|
||||
|
||||
start = date.today()
|
||||
end = start + timedelta(days=30 * warranty_months)
|
||||
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
INSERT INTO warranties
|
||||
(sale_id, sale_item_id, inventory_id, customer_id, supplier_id,
|
||||
part_number, name, warranty_months, start_date, end_date, notes)
|
||||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
RETURNING id
|
||||
""", (
|
||||
sale_id, sale_item_id, inventory_id, customer_id, supplier_id,
|
||||
part_number, name, warranty_months, start, end, notes
|
||||
))
|
||||
w_id = cur.fetchone()[0]
|
||||
cur.close()
|
||||
|
||||
log_action(conn, 'WARRANTY_REGISTER', 'warranty', w_id,
|
||||
new_value={'months': warranty_months, 'end_date': str(end)})
|
||||
return w_id
|
||||
|
||||
|
||||
def create_claim(conn, warranty_id, reason, employee_id=None, notes=None):
|
||||
"""File a warranty claim.
|
||||
|
||||
Args:
|
||||
warranty_id: int
|
||||
reason: str (min 10 chars)
|
||||
Returns:
|
||||
int: claim_id
|
||||
"""
|
||||
if not reason or len(reason.strip()) < 10:
|
||||
raise ValueError("Claim reason must be at least 10 characters")
|
||||
|
||||
cur = conn.cursor()
|
||||
# Verify warranty exists and is active
|
||||
cur.execute("SELECT status FROM warranties WHERE id = %s", (warranty_id,))
|
||||
row = cur.fetchone()
|
||||
if not row:
|
||||
raise ValueError("Warranty not found")
|
||||
if row[0] != 'active':
|
||||
raise ValueError(f"Cannot claim a warranty with status '{row[0]}'")
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO warranty_claims
|
||||
(warranty_id, reason, employee_id, notes)
|
||||
VALUES (%s, %s, %s, %s)
|
||||
RETURNING id
|
||||
""", (warranty_id, reason, employee_id, notes))
|
||||
claim_id = cur.fetchone()[0]
|
||||
cur.close()
|
||||
|
||||
log_action(conn, 'WARRANTY_CLAIM', 'warranty_claim', claim_id,
|
||||
new_value={'warranty_id': warranty_id, 'reason': reason})
|
||||
return claim_id
|
||||
|
||||
|
||||
def resolve_claim(conn, claim_id, resolution, diagnosis=None,
|
||||
replacement_inventory_id=None, refund_amount=None,
|
||||
labor_cost=None, supplier_rma_number=None, notes=None):
|
||||
"""Resolve a warranty claim.
|
||||
|
||||
Args:
|
||||
resolution: 'approved'|'rejected'|'repaired'|'replaced'|'refunded'
|
||||
"""
|
||||
if resolution not in ('approved', 'rejected', 'repaired', 'replaced', 'refunded'):
|
||||
raise ValueError(f"Invalid resolution: {resolution}")
|
||||
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
UPDATE warranty_claims
|
||||
SET resolution = %s, diagnosis = COALESCE(%s, diagnosis),
|
||||
replacement_inventory_id = COALESCE(%s, replacement_inventory_id),
|
||||
refund_amount = COALESCE(%s, refund_amount),
|
||||
labor_cost = COALESCE(%s, labor_cost),
|
||||
supplier_rma_number = COALESCE(%s, supplier_rma_number),
|
||||
notes = COALESCE(notes || ' | ', '') || %s,
|
||||
status = 'resolved', resolved_at = NOW()
|
||||
WHERE id = %s AND status != 'closed'
|
||||
""", (
|
||||
resolution, diagnosis, replacement_inventory_id,
|
||||
refund_amount, labor_cost, supplier_rma_number,
|
||||
notes or 'Resuelto', claim_id
|
||||
))
|
||||
updated = cur.rowcount > 0
|
||||
|
||||
# If replaced or refunded, mark warranty as claimed
|
||||
if updated and resolution in ('replaced', 'refunded', 'approved'):
|
||||
cur.execute("""
|
||||
UPDATE warranties SET status = 'claimed'
|
||||
WHERE id = (SELECT warranty_id FROM warranty_claims WHERE id = %s)
|
||||
""", (claim_id,))
|
||||
|
||||
cur.close()
|
||||
return updated
|
||||
|
||||
|
||||
def close_claim(conn, claim_id):
|
||||
"""Close a resolved claim (final status)."""
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
UPDATE warranty_claims
|
||||
SET status = 'closed'
|
||||
WHERE id = %s AND status = 'resolved'
|
||||
""", (claim_id,))
|
||||
updated = cur.rowcount > 0
|
||||
cur.close()
|
||||
return updated
|
||||
|
||||
|
||||
def get_warranty(conn, warranty_id):
|
||||
"""Get warranty detail."""
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT w.id, w.sale_id, w.sale_item_id, w.inventory_id, w.customer_id,
|
||||
w.supplier_id, w.part_number, w.name, w.warranty_months,
|
||||
w.start_date, w.end_date, w.status, w.notes, w.created_at,
|
||||
c.name as customer_name, s.name as supplier_name
|
||||
FROM warranties w
|
||||
LEFT JOIN customers c ON w.customer_id = c.id
|
||||
LEFT JOIN suppliers s ON w.supplier_id = s.id
|
||||
WHERE w.id = %s
|
||||
""", (warranty_id,))
|
||||
row = cur.fetchone()
|
||||
cur.close()
|
||||
if not row:
|
||||
return None
|
||||
return {
|
||||
'id': row[0], 'sale_id': row[1], 'sale_item_id': row[2],
|
||||
'inventory_id': row[3], 'customer_id': row[4], 'supplier_id': row[5],
|
||||
'part_number': row[6], 'name': row[7], 'warranty_months': row[8],
|
||||
'start_date': str(row[9]), 'end_date': str(row[10]),
|
||||
'status': row[11], 'notes': row[12], 'created_at': str(row[13]),
|
||||
'customer_name': row[14], 'supplier_name': row[15],
|
||||
}
|
||||
|
||||
|
||||
def list_warranties(conn, customer_id=None, status=None, limit=50, offset=0):
|
||||
"""List warranties."""
|
||||
cur = conn.cursor()
|
||||
filters = []
|
||||
params = []
|
||||
if customer_id:
|
||||
filters.append("w.customer_id = %s")
|
||||
params.append(customer_id)
|
||||
if status:
|
||||
filters.append("w.status = %s")
|
||||
params.append(status)
|
||||
where = "WHERE " + " AND ".join(filters) if filters else ""
|
||||
|
||||
cur.execute(f"""
|
||||
SELECT w.id, w.part_number, w.name, w.warranty_months,
|
||||
w.start_date, w.end_date, w.status,
|
||||
c.name as customer_name
|
||||
FROM warranties w
|
||||
LEFT JOIN customers c ON w.customer_id = c.id
|
||||
{where}
|
||||
ORDER BY w.end_date ASC
|
||||
LIMIT %s OFFSET %s
|
||||
""", params + [limit, offset])
|
||||
rows = cur.fetchall()
|
||||
cur.close()
|
||||
return [{
|
||||
'id': r[0], 'part_number': r[1], 'name': r[2],
|
||||
'warranty_months': r[3], 'start_date': str(r[4]),
|
||||
'end_date': str(r[5]), 'status': r[6],
|
||||
'customer_name': r[7],
|
||||
} for r in rows]
|
||||
|
||||
|
||||
def get_claim(conn, claim_id):
|
||||
"""Get claim detail."""
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT wc.id, wc.warranty_id, wc.claim_date, wc.reason, wc.diagnosis,
|
||||
wc.resolution, wc.replacement_inventory_id, wc.refund_amount,
|
||||
wc.labor_cost, wc.status, wc.supplier_rma_number, wc.notes,
|
||||
wc.created_at, wc.resolved_at,
|
||||
w.part_number, w.name, w.customer_id, c.name as customer_name
|
||||
FROM warranty_claims wc
|
||||
JOIN warranties w ON wc.warranty_id = w.id
|
||||
LEFT JOIN customers c ON w.customer_id = c.id
|
||||
WHERE wc.id = %s
|
||||
""", (claim_id,))
|
||||
row = cur.fetchone()
|
||||
cur.close()
|
||||
if not row:
|
||||
return None
|
||||
return {
|
||||
'id': row[0], 'warranty_id': row[1], 'claim_date': str(row[2]),
|
||||
'reason': row[3], 'diagnosis': row[4], 'resolution': row[5],
|
||||
'replacement_inventory_id': row[6], 'refund_amount': float(row[7]) if row[7] else None,
|
||||
'labor_cost': float(row[8]) if row[8] else None,
|
||||
'status': row[9], 'supplier_rma_number': row[10], 'notes': row[11],
|
||||
'created_at': str(row[12]), 'resolved_at': str(row[13]) if row[13] else None,
|
||||
'part_number': row[14], 'name': row[15],
|
||||
'customer_id': row[16], 'customer_name': row[17],
|
||||
}
|
||||
|
||||
|
||||
def list_claims(conn, status=None, warranty_id=None, limit=50, offset=0):
|
||||
"""List warranty claims."""
|
||||
cur = conn.cursor()
|
||||
filters = []
|
||||
params = []
|
||||
if status:
|
||||
filters.append("wc.status = %s")
|
||||
params.append(status)
|
||||
if warranty_id:
|
||||
filters.append("wc.warranty_id = %s")
|
||||
params.append(warranty_id)
|
||||
where = "WHERE " + " AND ".join(filters) if filters else ""
|
||||
|
||||
cur.execute(f"""
|
||||
SELECT wc.id, wc.claim_date, wc.reason, wc.resolution, wc.status,
|
||||
w.part_number, w.name, c.name as customer_name
|
||||
FROM warranty_claims wc
|
||||
JOIN warranties w ON wc.warranty_id = w.id
|
||||
LEFT JOIN customers c ON w.customer_id = c.id
|
||||
{where}
|
||||
ORDER BY wc.created_at DESC
|
||||
LIMIT %s OFFSET %s
|
||||
""", params + [limit, offset])
|
||||
rows = cur.fetchall()
|
||||
cur.close()
|
||||
return [{
|
||||
'id': r[0], 'claim_date': str(r[1]), 'reason': r[2],
|
||||
'resolution': r[3], 'status': r[4],
|
||||
'part_number': r[5], 'name': r[6], 'customer_name': r[7],
|
||||
} for r in rows]
|
||||
|
||||
|
||||
def expire_warranties(conn):
|
||||
"""Batch-update warranties whose end_date has passed to 'expired'.
|
||||
|
||||
Should be run periodically (e.g., daily via cron).
|
||||
Returns number of warranties expired.
|
||||
"""
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
UPDATE warranties
|
||||
SET status = 'expired'
|
||||
WHERE status = 'active' AND end_date < CURRENT_DATE
|
||||
""")
|
||||
count = cur.rowcount
|
||||
cur.close()
|
||||
return count
|
||||
Reference in New Issue
Block a user