# /home/Autopartes/pos/services/global_invoice.py """Global invoice (Factura Global) service. Groups cash sales (PUE, <= $2,000, no individual CFDI) into a single monthly CFDI with InformacionGlobal per SAT requirements. """ from datetime import datetime from decimal import Decimal from services.cfdi_facturapi_builder import build_global_invoice_payload from services.cfdi_queue import enqueue_cfdi, _generate_provisional_folio def get_eligible_sales(conn, year, month, branch_id=None, max_total=2000): """Find sales eligible for global invoicing. Criteria: - Payment method: PUE (paid in full) - Total <= max_total - No individual CFDI stamped - Not already included in a global invoice - Created in the given year/month - Optionally filtered by branch_id Returns: list of sale dicts with items """ cur = conn.cursor() # Find eligible sale IDs sql = """ SELECT s.id FROM sales s WHERE s.metodo_pago_sat = 'PUE' AND s.total <= %s AND s.status = 'completed' AND s.global_invoiced_at IS NULL AND EXTRACT(YEAR FROM s.created_at) = %s AND EXTRACT(MONTH FROM s.created_at) = %s AND NOT EXISTS ( SELECT 1 FROM cfdi_queue c WHERE c.sale_id = s.id AND c.status = 'stamped' ) """ params = [max_total, year, month] if branch_id: sql += " AND s.branch_id = %s" params.append(branch_id) sql += " ORDER BY s.created_at ASC" cur.execute(sql, params) sale_ids = [r[0] for r in cur.fetchall()] if not sale_ids: cur.close() return [] # Load sale details with items sales = [] for sale_id in sale_ids: cur.execute(""" SELECT id, branch_id, customer_id, employee_id, sale_type, payment_method, subtotal, discount_total, tax_total, total, metodo_pago_sat, forma_pago_sat, status, created_at FROM sales WHERE id = %s """, (sale_id,)) row = cur.fetchone() if not row: continue sale = { 'id': row[0], 'branch_id': row[1], 'customer_id': row[2], 'employee_id': row[3], 'sale_type': row[4], 'payment_method': row[5], 'subtotal': float(row[6]) if row[6] else 0, 'discount_total': float(row[7]) if row[7] else 0, 'tax_total': float(row[8]) if row[8] else 0, 'total': float(row[9]) if row[9] else 0, 'metodo_pago_sat': row[10] or 'PUE', 'forma_pago_sat': row[11] or '01', 'status': row[12], 'created_at': str(row[13]), 'items': [], } cur.execute(""" SELECT id, inventory_id, part_number, name, quantity, unit_price, unit_cost, discount_pct, discount_amount, tax_rate, tax_amount, subtotal, clave_prod_serv, clave_unidad FROM sale_items WHERE sale_id = %s ORDER BY id """, (sale_id,)) for r in cur.fetchall(): sale['items'].append({ 'id': r[0], 'inventory_id': r[1], 'part_number': r[2], 'name': r[3], 'quantity': r[4], 'unit_price': float(r[5]) if r[5] else 0, 'unit_cost': float(r[6]) if r[6] else 0, 'discount_pct': float(r[7]) if r[7] else 0, 'discount_amount': float(r[8]) if r[8] else 0, 'tax_rate': float(r[9]) if r[9] else 0.16, 'tax_amount': float(r[10]) if r[10] else 0, 'subtotal': float(r[11]) if r[11] else 0, 'clave_prod_serv': r[12] or '25174800', 'clave_unidad': r[13] or 'H87', }) sales.append(sale) cur.close() return sales def generate_global_invoice(conn, tenant_config, year, month, branch_id=None, max_total=2000, employee_id=None): """Generate a global invoice for the given month. Args: conn: psycopg2 connection tenant_config: dict with rfc, razon_social, regimen_fiscal, cp, serie year: int month: int branch_id: optional branch filter max_total: max sale total to include (default $2,000) employee_id: optional employee ID for audit Returns: dict: {id, status, sales_count, total, xml, provisional_folio} or {error, message} if no eligible sales """ sales = get_eligible_sales(conn, year, month, branch_id, max_total) if not sales: return {'error': 'NO_ELIGIBLE_SALES', 'message': f'No hay ventas elegibles para factura global de {month:02d}/{year}'} payload = build_global_invoice_payload(sales, tenant_config, year, month) # Enqueue with sale_id=NULL (global invoice) result = enqueue_cfdi(conn, None, 'ingreso', payload) cfdi_id = result['id'] cur = conn.cursor() # Link sales to global invoice for sale in sales: cur.execute(""" INSERT INTO global_invoice_sales (global_invoice_id, sale_id) VALUES (%s, %s) ON CONFLICT DO NOTHING """, (cfdi_id, sale['id'])) # Mark sale as globally invoiced cur.execute(""" UPDATE sales SET global_invoiced_at = NOW() WHERE id = %s """, (sale['id'],)) conn.commit() cur.close() return { 'id': cfdi_id, 'status': 'pending', 'sales_count': len(sales), 'total': sum(s['total'] for s in sales), 'provisional_folio': result['provisional_folio'], 'payload': payload, } def get_global_invoice_status(conn, cfdi_id): """Get status of a global invoice including linked sales.""" cur = conn.cursor() cur.execute(""" SELECT id, status, uuid_fiscal, provisional_folio, error_message, created_at, stamped_at FROM cfdi_queue WHERE id = %s """, (cfdi_id,)) row = cur.fetchone() if not row: cur.close() return None result = { 'id': row[0], 'status': row[1], 'uuid_fiscal': row[2], 'provisional_folio': row[3], 'error_message': row[4], 'created_at': str(row[5]), 'stamped_at': str(row[6]) if row[6] else None, 'sales': [], } cur.execute(""" SELECT s.id, s.total, s.created_at FROM global_invoice_sales gis JOIN sales s ON s.id = gis.sale_id WHERE gis.global_invoice_id = %s ORDER BY s.created_at ASC """, (cfdi_id,)) for r in cur.fetchall(): result['sales'].append({ 'id': r[0], 'total': float(r[1]) if r[1] else 0, 'created_at': str(r[2]), }) cur.close() return result