consultoria-as
8796cadb56
- Add Facturapi REST service (invoices, customers, orgs, cancel, downloads) - Add JSON payload builder for ingreso/egreso/pago/global invoices - Replace XML queue with Facturapi JSON queue (payload_unsigned, external_id) - Update invoicing blueprint with Facturapi config and download endpoints - Update global invoice service to use Facturapi payloads - Add migration v4.3_facturapi.sql and tenant rollout script - Update invoicing UI: payload preview, PDF/XML downloads, PAC status panel - Add FACTURAPI_USER_KEY to .env.example
211 lines
6.6 KiB
Python
211 lines
6.6 KiB
Python
# /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
|