Merge remote-tracking branch 'origin/development' into feature/Invoice-movements

# Conflicts:
#	backend/api/v1/modules/a76/reports/importacion/facturas/usa/service.py
#	backend/core/celery_app.py
#	frontend/src/lib/api/dashboard/a76/general_catalogs/company.ts
This commit is contained in:
Galindo97
2026-02-20 10:09:19 -06:00
127 changed files with 17010 additions and 9437 deletions

View File

@@ -17,7 +17,7 @@ from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
from api.v1.modules.a76.pedmientos.models.pedimento_validation import PedimentoValidation
from api.v1.modules.a76.customs_brokers.models import CustomsBroker, CustomsBrokerPersonnel
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import ClientProvider, ClientProviderAddress, ClientProviderPrograms
# --- SCHEMAS FOR TEMPLATE CONTEXT ---
@@ -186,7 +186,7 @@ class AvisoConsolidadoExportacionService:
logistics = header.logistics
# Fetch Items associated with this invoice (MOVED UP FOR WEIGHT CALCULATION)
items = db.query(Item).filter(Item.invoice_id == invoice_id).all()
items = db.query(LineItem).filter(LineItem.invoice_id == invoice_id).all()
# Peso Bruto
peso_bruto_val = "0.0"
@@ -194,14 +194,12 @@ class AvisoConsolidadoExportacionService:
# Calculate sum from items first
if items:
for item in items:
if item.lines:
for line in item.lines:
if line.quantity and line.quantity.gross_weight:
try:
calculated_gross_weight += float(line.quantity.gross_weight)
except (ValueError, TypeError):
pass
for item in items:
if item.quantity and item.quantity.gross_weight:
try:
calculated_gross_weight += float(item.quantity.gross_weight)
except (ValueError, TypeError):
pass
if financials and financials.gross_weight and float(financials.gross_weight) > 0:
peso_bruto_val = f"{financials.gross_weight:,.2f}"
@@ -379,20 +377,19 @@ class AvisoConsolidadoExportacionService:
cant_total = 0.0
if items:
for item in items:
if item.lines:
for line in item.lines:
# Priority: Quantity (UMA or Standard)
q = 0.0
if line.quantity:
try:
if line.quantity.quantity_uma is not None:
q = float(line.quantity.quantity_uma)
elif line.quantity.quantity is not None:
q = float(line.quantity.quantity)
except (ValueError, TypeError):
q = 0.0
cant_total += q
for item in items:
for line in item:
# Priority: Quantity (UMA or Standard)
q = 0.0
if line.quantity:
try:
if line.quantity.quantity_uma is not None:
q = float(line.quantity.quantity_uma)
elif line.quantity.quantity is not None:
q = float(line.quantity.quantity)
except (ValueError, TypeError):
q = 0.0
cant_total += q
# Format: 15 chars, 3 decimals? Actually Clarion LINEPRINT usually just prints the text.
# Clarion 'CLIP(FORMAT(Loc:CantTotal,@n015.3))' removes spaces.

View File

@@ -15,10 +15,9 @@ from datetime import datetime
# --- MODELOS (Imported from system for Header info) ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
from api.v1.modules.core.tenants.models import Tenant
@@ -40,8 +39,8 @@ class FIFOAssignmentService:
Returns a list of calculated discharges.
"""
# 1. Get Export Lines
export_lines = db.query(LineItem).join(Item).filter(
Item.invoice_id == invoice_id
export_lines = db.query(LineItem).join(LineItem).filter(
LineItem.invoice_id == invoice_id
).options(
joinedload(LineItem.quantity),
joinedload(LineItem.description),
@@ -58,7 +57,7 @@ class FIFOAssignmentService:
if qty_needed <= 0:
continue
part_number = exp_line.part_number
part_number = exp_line.part_number_id
if not part_number:
self._log(f"Skipping line {exp_line.id}, no part number")
continue
@@ -79,10 +78,10 @@ class FIFOAssignmentService:
# 2. Find Import Candidates (FIFO order by payment date)
# Use outerjoin for pedimento dates to avoid filtering out candidates with missing dates
candidates = db.query(LineItem).join(Item).join(InvoiceHeader)\
candidates = db.query(LineItem).join(InvoiceHeader)\
.join(InvoiceComplianceMx).join(InvoiceComplianceMx.pedimento).outerjoin(Pedimentos.pedimento_dates)\
.filter(
LineItem.part_number == part_number,
LineItem.part_number_id == part_number,
InvoiceHeader.operation_type == 'imp', # Assuming 'imp' is the value for Import based on Enum
).order_by(
PedimentoDates.payment_date.asc()
@@ -90,7 +89,7 @@ class FIFOAssignmentService:
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.financial),
joinedload(LineItem.item).joinedload(Item.invoice).joinedload(InvoiceHeader.compliance_mx).joinedload(InvoiceComplianceMx.pedimento).joinedload(Pedimentos.pedimento_dates)
joinedload(LineItem.item).joinedload(LineItem.invoice).joinedload(InvoiceHeader.compliance_mx).joinedload(InvoiceComplianceMx.pedimento).joinedload(Pedimentos.pedimento_dates)
).all()
self._log(f"Found {len(candidates)} candidates for {part_number}")
@@ -259,16 +258,15 @@ class DescargaReportService:
# --- 2. Obtener Líneas de Exportación (Lo que necesitamos cubrir) ---
if progress_callback: progress_callback(20, "Obteniendo items a exportar...")
export_lines = db.query(LineItem).filter(
LineItem.item_id == Item.id,
Item.invoice_id == invoice_id
export_lines = db.query(LineItem).filter(
LineItem.invoice_id == invoice_id
).options(
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.description),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.part_info)
).join(Item).all()
).join(LineItem).all()
items_reporte = []

View File

@@ -0,0 +1,356 @@
from typing import List, Dict, Any, Tuple
from sqlalchemy.orm import Session, joinedload
from .schemas import Mainx30GenerationRequest, ErrorValidacion
# --- MODELOS A76 ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
class ScaiiProcessor:
def __init__(self):
self.cuenta_partidas = 0
self.cuenta_facturas = 0
self.valor_total_factura = 0.0
self.peso_bruto_factura = 0.0
self.peso_neto_factura = 0.0
self.errores: List[ErrorValidacion] = []
def _obtener_datos_cliente(self, cliente: ClientProvider) -> dict:
"""Extrae de manera segura los datos del cliente/dirección"""
# Determine Tax ID: RFC for MX, Tax ID for others
address = cliente.address
pais_raw = (address.country or "MX").upper()
pais = "MX"
if pais_raw in ["MEXICO", "MEX", "MX"]:
pais = "MX"
elif pais_raw in ["USA", "US", "UNITED STATES"]:
pais = "US"
else:
pais = pais_raw[:2]
tax_id = ""
if pais == "MX":
tax_id = cliente.rfc or ""
else:
# Try generic tax_id field if exists, else generic field or RFC as fallback
# Providing a fallback to extra_information or web_key if needed, but per model inspection:
# We don't see a specific 'tax_id' field in ClientProvider model snippet.
# We see 'rfc'. Let's use RFC as generic holder or look for 'tax_id' if I missed it.
# Re-reading model: rfc is the only obvious one.
# Let's use RFC field for foreign tax id too unless instructed otherwise.
tax_id = cliente.rfc or ""
data = {
"nombre": (cliente.name or "")[:39],
"tax_id": tax_id[:15],
"broker": "", "calle": "", "cp": "", "ciudad": "", "estado": "", "pais": pais, "tel": ""
}
if cliente.programs:
data["broker"] = (cliente.programs.broker or "")[:6]
if address:
calle_comp = f"{address.streets or ''} {address.exterior_number or ''}".strip()
data["calle"] = calle_comp[:35]
data["cp"] = (address.postal_code or "")[:9]
data["ciudad"] = (address.city or "")[:20]
data["estado"] = (address.state or "")[:2].upper()
data["tel"] = (address.phone or "")[:15] # Remove default "000000"
return data
def procesar_facturas(
self, db: Session, manifiesto: str, empresa_dict: Dict[str, Any], request: Mainx30GenerationRequest
) -> Tuple[List[str], List[ErrorValidacion]]:
lineas = []
self.errores = []
# 1. Traer Facturas del Manifiesto
facturas = db.query(InvoiceHeader).join(
InvoiceComplianceMx, InvoiceHeader.id == InvoiceComplianceMx.invoice_id
).options(
joinedload(InvoiceHeader.financials),
joinedload(InvoiceHeader.compliance_mx)
).filter(
InvoiceComplianceMx.manifest_number == manifiesto
).all()
for factura in facturas:
self.cuenta_facturas += 1
f_val_total = 0.0
f_pb = 0.0
f_pn = 0.0
f_consec_partidas = 0
# --- MF20 / MF22: Per-Invoice Header at Manifest Level ---
# Sample: MF20AAK22-001 I10900 1234 1234
# Invoice(15) + Type(1?) + Port(5?) + ...
entry_port = manifiesto.replace("-", "")[:4] # or from manifest object if available here?
# Manifiesto passed to this method is just a string 'manifest_number'.
# We need to query manifest or pass it.
# Actually, `manifiesto` arg is just the number string.
# But we can pass the entry_port from service.py in empresa_dict or request?
# Let's check service.py.
# Assuming it is in empresa_dict for now (I will add it next step)
# --- MF20 / MF22: Per-Invoice Header at Manifest Level ---
# Sample: MF20AAK22-001 I10900 1234 1234
port_code = empresa_dict.get('entry_port', '')[:4]
manufacturer_id = empresa_dict.get('manufacturer_id', '')[:10]
# Constructing line to match sample length/spacing
lineas.append(
f"MF20"
f"{factura.invoice_number[:15]:<15}"
f"I{manufacturer_id:<15}"
f"{port_code:<20}"
f"{port_code:<4}"
)
lineas.append(f"MF22")
self.cuenta_partidas += 2
# --- IV01: Header de Factura ---
flete = float(factura.financials.freight) if factura.financials and factura.financials.freight else 0.0
fecha_str = factura.invoice_date.strftime("%y%m%d") if factura.invoice_date else "000000"
s_rfc = ""
c_rfc = ""
if factura.compliance_mx:
if factura.compliance_mx.provider_id:
s_obj = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if s_obj: s_rfc = s_obj.rfc or ""
if factura.compliance_mx.sold_to_id:
c_obj = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.sold_to_id).first()
if c_obj: c_rfc = c_obj.rfc or ""
lineas.append(
f"IV01{factura.invoice_number[:15]:<15}"
f"{fecha_str}01 " # 6 + 3 = 9
f"{port_code:<11}" # Port (Use same as MF20)
f"{empresa_dict.get('broker', '')[:6]:<15}" # Broker
f"{s_rfc[:12]:<12}{c_rfc[:12]:<12}"
)
self.cuenta_partidas += 1
# --- IV02: Company Name ---
nombre_empresa = empresa_dict.get('nombre_empresa', '')[:40]
lineas.append(f"IV02 {nombre_empresa:<40}")
self.cuenta_partidas += 1
# --- IV10: Goods Description & Contact ---
# Dynamic Description from Invoice (observation_en or observation_es)
desc_global = (factura.observation_en or factura.observation_es or "")[:30]
contacto = empresa_dict.get('responsable', '')[:30]
lineas.append(f"IV10 {desc_global:<30}{contacto:<30}")
self.cuenta_partidas += 1
# --- IV11: Headers ---
lineas.append(f"IV11H")
lineas.append(f"IV11F")
self.cuenta_partidas += 2
# --- DATOS DE DIRECCIONES (S, C, T, I) ---
# Shipper (S) -> Proveedor de la factura
if factura.compliance_mx and factura.compliance_mx.provider_id:
s_cliente = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if s_cliente:
s_data = self._obtener_datos_cliente(s_cliente)
calle_cp = f"{s_data['calle']} {s_data['cp']}".strip()
lineas.append(f"IV12S {s_data['nombre'][:39]:<39}")
lineas.append(f"IV13S {calle_cp[:35]:<35}")
lineas.append(f"IV14S{s_data['ciudad'][:20]:<20}{s_data['estado'][:2]}{s_data['pais'][:2]}{s_data['tel'][:15]:<15}{s_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# Consignee / Vendido A (C)
c_data = None
if factura.compliance_mx and factura.compliance_mx.sold_to_id:
c_cliente = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.sold_to_id).first()
if c_cliente:
c_data = self._obtener_datos_cliente(c_cliente)
calle_cp = f"{c_data['calle']} {c_data['cp']}".strip()
lineas.append(f"IV12C {c_data['nombre'][:39]:<39}")
lineas.append(f"IV13C {calle_cp[:35]:<35}")
lineas.append(f"IV14C{c_data['ciudad'][:20]:<20}{c_data['estado'][:2]}{c_data['pais'][:2]}{c_data['tel'][:15]:<15}{c_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# Ship To / Enviado A (T)
if factura.compliance_mx and factura.compliance_mx.shipped_to_id:
t_cliente = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.shipped_to_id).first()
if t_cliente:
t_data = self._obtener_datos_cliente(t_cliente)
calle_cp = f"{t_data['calle']} {t_data['cp']}".strip()
lineas.append(f"IV12T {t_data['nombre'][:39]:<39}")
lineas.append(f"IV13T {calle_cp[:35]:<35}")
lineas.append(f"IV14T{t_data['ciudad'][:20]:<20}{t_data['estado'][:2]}{t_data['pais'][:2]}{t_data['tel'][:15]:<15}{t_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# Importer (I) - Sample shows it same as Consignee or Importer
if c_data:
# Reuse c_data calculation or re-fetch if needed. Reusing c_data structure.
calle_cp = f"{c_data['calle']} {c_data['cp']}".strip()
lineas.append(f"IV12I {c_data['nombre'][:39]:<39}")
lineas.append(f"IV13I {calle_cp[:35]:<35}")
lineas.append(f"IV14I{c_data['ciudad'][:20]:<20}{c_data['estado'][:2]}{c_data['pais'][:2]}{c_data['tel'][:15]:<15}{c_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# --- PARTIDAS (DETALLE IV20-IV27) ---
items_query = db.query(LineItem).filter(
LineItem.invoice_id == factura.id
).options(
joinedload(LineItem.part_info),
joinedload(LineItem.description),
joinedload(LineItem.financial),
joinedload(LineItem.quantity), # Added quantity relation
joinedload(LineItem.customs),
joinedload(LineItem.unit_of_measure_info)
).all()
for line in items_query:
f_consec_partidas += 1
part_num = line.part_info.part_number if line.part_info else "S/N"
po_num = factura.purchase_order or ""
desc = ""
if line.description:
desc = line.description.description_english or line.description.description_spanish or ""
# --- OBTENCIÓN DE DATOS DE LINEFINANCIAL / LINEQUANTITY ---
qty = 0.0; pb = 0.0; pn = 0.0; val_usd = 0.0
val_no_duty = 0.0; val_packing = 0.0
if line.quantity:
qty = float(line.quantity.quantity or 0.0)
pb = float(line.quantity.gross_weight or 0.0)
pn = float(line.quantity.net_weight or 0.0)
if pb == 0 and pn > 0: pb = pn
if line.financial:
val_usd = float(line.financial.value_usd or 0.0)
val_no_duty = float(line.financial.exempt_amount_usd or 0.0) # IV24
val_packing = float(line.financial.value_us_packing_usd or 0.0) # IV26
f_pb += pb
f_pn += pn
f_val_total += val_usd # Assuming Total Invoice Value is sum of line.value_usd
# Aduanas
hts_ame = ""
pais_orig = "MX"
if line.customs:
raw_hts = line.customs.american_fraction or line.customs.fraction or ""
hts_ame = raw_hts.replace(".", "").strip()
pais_orig = (line.customs.origin_country or "MX")[:2]
# UM
um_ame = "PC"
if line.unit_of_measure_info:
um_ame = line.unit_of_measure_info.american_code or "PC"
# Escritura (Igual que el Clarion)
lineas.append(f"IV20{f_consec_partidas:03d} {part_num[:25]:<25}A {po_num[:20]:<20}")
lineas.append(f"IV21{' ':21}{desc[:50]:<50}")
# IV22: Fix alignment based on sample
# Sample: N0000002235PCS000050000CN0000010000000000000000 000000549000000408
# HTS(10?) + Val(10) + UM(3) + Cant(9) + Pais(2) + ...
v_int = int(round(val_usd * 100))
q_int = int(round(qty * 1000)) # Sample 000050000 for 50? 50 * 1000 = 50000.
pb_int = int(round(pb * 100))
pn_int = int(round(pn * 100))
lineas.append(
f"IV22 N" # 10 spaces + N
f"{v_int:010d}" # Value (integer 10)
f"{um_ame[:3]:<3}" # UM (3)
f"{q_int:09d}" # Qty (integer 9)
f"{pais_orig[:2]:<2}" # Pais (2)
f"0000010000000000000000 " # Fixed (23 with space)
f"{pb_int:010d}" # Peso Bruto (10 chars)
f"{pn_int:010d}" # Peso Neto (10 chars)
)
# IV24 (No Duty / Exempt)
# Dynamic Logic: Use exempt_amount_usd if > 0
v_nd_int = int(round(val_no_duty * 100))
# IV24 uses same UM and Qty layout as IV22 but for NoDuty portion?
# Sample shows just value and then mostly zeros?
# Sample: IV24 0000000000 000000000 0000000000000000000000
# We will use v_nd_int. If 0, it renders as 0000000000.
if v_nd_int > 0:
# If there IS a No Duty value, we should probably output it.
# Format seems to start at same pos as IV22 Value?
# IV22 starts value at col 20 (approx).
# IV24 starts value at col 20 (approx).
# IV24 {Val} {Qty?} ...
# Given sample: `IV24 0000000000 000000000 ...`
# It looks like: Prefix(15) + Val(10) + Space(3) + Qty??(9) + ...
# Let's mimic structure
lineas.append(f"IV24 {v_nd_int:010d} {0:09d} 0000000000000000000000")
else:
lineas.append(f"IV24 {0:010d} {0:09d} 0000000000000000000000")
# IV26 (Packing)
# Dynamic Logic: Use value_us_packing_usd
v_p_int = int(round(val_packing * 100))
if v_p_int > 0:
lineas.append(f"IV26 {v_p_int:010d} {0:09d} 0000000000000000000000")
else:
lineas.append(f"IV26 {0:010d} {0:09d} 0000000000000000000000")
# IV27 (Unit Costs)
# Sample: IV27 000000000000000000000000000000000000000000000000000000000000000000
# If we have distinct values, maybe we should calculate unit costs?
# But legacy sample shows all zeros.
# Calculating separate unit costs for Duty/NoDuty/Packing:
c_u_d = val_usd / qty if qty > 0 else 0
c_u_nd = val_no_duty / qty if qty > 0 else 0
c_u_p = val_packing / qty if qty > 0 else 0
# If user wants NO HARDCODING, maybe we should populate this?
# But sample had 0s. Let's populate specific costs if values exist, else 0.
# Format: IV27 + 10 spaces + CostDuty(11) + CostNoDuty(11) + CostPacking(11) + ...
# Based on legacy Clarion: `FORMAT(Left(Loc:CostoUDuty),@n011v5)`
cud_int = int(round(c_u_d * 100000))
cund_int = int(round(c_u_nd * 100000))
cup_int = int(round(c_u_p * 100000))
lineas.append(f"IV27 {cud_int:011d}{cund_int:011d}{cup_int:011d}000000000000000000000000000000000")
self.cuenta_partidas += 6
# --- TOTALES FACTURA ---
# Sample: IV900000700000000000000000063320000005348
# IV90 + CantPartidas(5) + ValTotal(12) + PesoBruto(10) + PesoNeto(10)
f_val_int = int(round(f_val_total * 100))
f_pb_int = int(round(f_pb * 100))
f_pn_int = int(round(f_pn * 100))
lineas.append(f"IV90{f_consec_partidas:05d}{f_val_int:012d}{f_pb_int:010d}{f_pn_int:010d}")
self.cuenta_partidas += 1
self.valor_total_factura += f_val_total
self.peso_bruto_factura += f_pb
self.peso_neto_factura += f_pn
return lineas, self.errores
def _agregar_error(self, partida, id_err, desc, sol, tipo):
self.errores.append(ErrorValidacion(partida=partida, linea=0, descripcion=f"[ {id_err} ] {desc}", soluciones=sol, identificador=tipo))
# (Dummy Processors para que no truene el Service)
class ScafDefProcessor:
def __init__(self): self.cuenta_partidas=0; self.cuenta_facturas=0; self.valor_total_factura=0; self.peso_bruto_factura=0; self.peso_neto_factura=0
def procesar_facturas(self, db, manifiesto, empresa_dict, request): return [], []
class ScafTempProcessor:
def __init__(self): self.cuenta_partidas=0; self.cuenta_facturas=0; self.valor_total_factura=0; self.peso_bruto_factura=0; self.peso_neto_factura=0
def procesar_facturas(self, db, manifiesto, empresa_dict, request): return [], []

View File

@@ -0,0 +1,43 @@
from typing import Dict, Any
from fastapi import APIRouter, Depends, Body
from celery.result import AsyncResult
from core.celery_app import celery_app
from core.security import get_current_user
from .task import generar_transmission_file_async
from .schemas import Mainx30GenerationRequest
router = APIRouter()
@router.get("/tasks/{task_id}")
async def get_task_status(
task_id: str,
current_user: Dict[str, Any] = Depends(get_current_user)
):
task_result = AsyncResult(task_id, app=celery_app)
response = {
"task_id": task_id,
"state": task_result.state,
"result": None,
"info": None
}
if task_result.state == 'FAILURE':
response["result"] = str(task_result.result)
elif task_result.state == 'SUCCESS':
response["result"] = task_result.result
elif task_result.state == 'PROCESSING':
# Ensure info is serializable
response["info"] = task_result.info
return response
@router.post("/generate")
async def trigger_generation(
request: Mainx30GenerationRequest,
current_user: Dict[str, Any] = Depends(get_current_user)
):
tenant_id = current_user.get("tenant_id")
# Pass request as dict to Celery task
task = generar_transmission_file_async.delay(request.model_dump(), tenant_id)
return {"task_id": task.id, "message": "Generación iniciada"}

View File

@@ -0,0 +1,71 @@
from typing import List, Optional, Any
from pydantic import BaseModel, Field
class Mainx30GenerationRequest(BaseModel):
"""
Schema for the Mainx30 file generation request
"""
manifiestos: List[str] = Field(..., description="Lista de números de manifiesto a procesar")
nomenclatura_factura: bool = Field(False, description="Usar nomenclatura basada en factura")
consolidar_rbs: bool = Field(False, description="Consolidar por fracción RB System")
emanifest_fast_blanco: bool = Field(False, description="E-Manifest y FAST en blanco")
no_enviar_emanifest: bool = Field(False, description="No enviar E-Manifest")
consolidar_partidas: bool = Field(False, description="Consolidar partidas (XML OPTIMA Y RBS2)")
main_x40_emanifest: bool = Field(False, description="Main X40 E-Manifest")
main_x30_fedex: bool = Field(False, description="Main X30 (FEDEX)")
iv11: bool = Field(False, description="IV11")
iv42: bool = Field(False, description="IV42")
class ErrorValidacion(BaseModel):
"""
Schema for validation errors during file generation
"""
partida: int
linea: int
descripcion: str
soluciones: str
identificador: str
campos: str = ""
campos2: str = ""
class Mainx30Response(BaseModel):
"""
Schema for the generation response
"""
success: bool
message: str
task_id: Optional[str] = None
archivo_generado: Optional[str] = None
ruta_archivo: Optional[str] = None
content: Optional[str] = None
# Statistics
cuenta_partidas: int = 0
valor_total: float = 0.0
flete_total: float = 0.0
peso_bruto_total: float = 0.0
peso_neto_total: float = 0.0
cuenta_facturas: int = 0
# Validation
errores: List[ErrorValidacion] = []
tiene_inconsistencias: bool = False
class BrokerValidationResult(BaseModel):
es_valido: bool
mensaje_error: Optional[str] = None
broker_cliente: Optional[str] = None
class EmpresaDatos(BaseModel):
broker: str
responsable: str
rfc: str
tiene_linea_express: str
nombre_empresa: str = "AAKRON RULE CORPORATION"
manufacturer_id: str = "I10900"
ftp_key: str = "00SCSI"
class ConfiguracionSistema(BaseModel):
path_arch_transmision: str
utilizar_nombre_generico_mainx30: bool
utilizar_codigo_broker_cliente: bool

View File

@@ -0,0 +1,300 @@
import os
import tempfile
from datetime import date, datetime
from typing import List, Tuple, Optional
from pathlib import Path
from sqlalchemy.orm import Session
from fastapi import HTTPException
from .schemas import (
Mainx30GenerationRequest, Mainx30Response, ErrorValidacion,
EmpresaDatos, ConfiguracionSistema
)
# --- HELPERS ---
def fecha_clarion_a_iso(clarion_date):
"""Convierte fecha Clarion (días desde 1800-12-28) a ISO YYYY-MM-DD"""
if not clarion_date: return "1900-01-01"
try:
from datetime import date, timedelta
base_date = date(1800, 12, 28)
delta = timedelta(days=int(clarion_date))
return (base_date + delta).isoformat()
except:
return "1900-01-01"
# --- MODELOS A76 ---
from api.v1.modules.a76.manifests.manifest.models import Manifest
from api.v1.modules.a76.general_catalogs.company.models import Company as GEmpresa
# --- PROCESADORES ---
from .processors import ScaiiProcessor, ScafDefProcessor, ScafTempProcessor
class Mainx30Service:
def __init__(self):
self.errores_validacion: List[ErrorValidacion] = []
self.cuenta_partidas = 0
self.cuenta_facturas = 0
self.valor_total = 0.0
self.flete_total = 0.0
self.peso_bruto_total = 0.0
self.peso_neto_total = 0.0
def generar_mainx30_expo(
self,
db: Session,
request: Mainx30GenerationRequest,
task_instance=None
) -> Mainx30Response:
try:
self._inicializar_variables()
fecha_transmision = date.today().strftime("%y%m%d")
config_sistema = self._obtener_configuracion_sistema(db)
datos_empresa = self._obtener_datos_empresa(db)
self._validar_datos_empresa(datos_empresa)
if not request.manifiestos:
raise HTTPException(status_code=400, detail="No se seleccionaron manifiestos")
nombre_archivo = self._generar_nombre_archivo(config_sistema, request, request.manifiestos[0])
lineas_archivo = []
# Línea A
lineas_archivo.append(self._generar_linea_a(fecha_transmision, datos_empresa))
for manifiesto_num in request.manifiestos:
if task_instance:
task_instance.update_state(state='PROCESSING', meta={'status': f'Procesando {manifiesto_num}'})
lineas_manifiesto = self._procesar_manifiesto(
db, manifiesto_num, datos_empresa, fecha_transmision, request
)
lineas_archivo.extend(lineas_manifiesto)
# Línea Z
lineas_archivo.append(f"Z {self.cuenta_partidas:05d}")
ruta_completa = os.path.join("api/v1/modules/reports/generated", nombre_archivo)
self._escribir_archivo(ruta_completa, lineas_archivo)
return Mainx30Response(
success=len(self.errores_validacion) == 0,
message=self._generar_mensaje_resultado(ruta_completa),
archivo_generado=nombre_archivo,
ruta_archivo=ruta_completa,
cuenta_partidas=self.cuenta_partidas,
valor_total=self.valor_total,
flete_total=self.flete_total,
peso_bruto_total=self.peso_bruto_total,
peso_neto_total=self.peso_neto_total,
cuenta_facturas=self.cuenta_facturas,
errores=self.errores_validacion,
tiene_inconsistencias=len(self.errores_validacion) > 0,
content="\r\n".join(lineas_archivo)
)
except Exception as e:
import traceback
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Error generando Mainx30: {str(e)}")
def _inicializar_variables(self):
self.errores_validacion = []
self.cuenta_partidas = 1 # Empieza en 1 por la línea A
self.valor_total = 0.0
self.flete_total = 0.0
self.peso_bruto_total = 0.0
self.peso_neto_total = 0.0
self.cuenta_facturas = 0
def _procesar_manifiesto(
self,
db: Session,
manifiesto_num: str,
datos_empresa: EmpresaDatos,
fecha_transmision: str,
request: Mainx30GenerationRequest
) -> List[str]:
lineas = []
# --- TABLA A76: MANIFEST ---
manifiesto = db.query(Manifest).filter(
Manifest.manifest_number == manifiesto_num
).first()
if not manifiesto:
self._agregar_error_validacion(0, "MF", f"Manifiesto {manifiesto_num} no encontrado.", "Verificar BD", "MANIFIESTO")
return lineas
persona_cargo = manifiesto.person_in_charge or ""
if not persona_cargo:
self._agregar_error_validacion(0, "MF03", "Falta Persona a Cargo", "Capturar en Manifiesto", "MANIFIESTO")
num_manifiesto_clean = manifiesto_num.replace("-", "")
# Fecha en formato yyMMdd. Asumimos entry_date almacena Clarion Date o Timestamp.
fecha_entrada_str = "000000"
if manifiesto.entry_date:
try:
# Si es Clarion Date
fecha_iso = fecha_clarion_a_iso(manifiesto.entry_date)
fecha_entrada_str = datetime.strptime(fecha_iso, "%Y-%m-%d").strftime("%y%m%d")
except: pass
firms_code = manifiesto.entry_port_loc or ""
entry_port = manifiesto.entry_port or "000"
# MF01
# Sample Clarion: MF01AKR 1234 1234 2602061233026021345
# Layout:
# MF01 (4)
# Broker (6) -> "AKR "
# Port Ent (5) -> "1234 "
# Port Sal (5) -> "1234 "
# FecEnt (6) -> "260206"
# 12 (2) -> Prefix?
# 3 (1) -> Digit 3?
# 30 (2) -> Constant?
# FecTrans (6) -> "260213"
# Manifiesto (15?) -> "45 " (Sample has '45' at end, maybe manifest is '45'?)
# Let's align with sample string length and fields.
# "MF01"
# Broker: Left aligned 6 chars
# Port1: Left aligned 5 chars
# Port2: Left aligned 5 chars
# Date1: 6 chars
# "12330" (Hardcoded sequence based on sample analysis vs previous logic)
# Date2: 6 chars
# Manifest: Left aligned 15 chars? Sample "45" is at end.
# Re-analyzing sample: "MF01AKR 1234 1234 2602061233026021345"
# Length: 4+6+5+5+6+2+1+2+6+2 = 39? No.
# AKR : 6
# 1234 : 5
# 1234 : 5
# 260206: 6
# 12: 2
# 3: 1
# 30: 2
# 260213: 6
# 45: 2?
# Total: 4+6+5+5+6+5+6+2 = 39 chars displayed.
# My generated was: MF01123 000 000 0001011230260213123456879
# It was way off.
man_clean = num_manifiesto_clean[:15]
lineas.append(
f"MF01{datos_empresa.broker:<6}"
f"{entry_port:<5}"
f"{entry_port:<5}"
f"{fecha_entrada_str}"
f"12330{fecha_transmision}" # Fixed sequence "12330" inferred from sample
f"{man_clean:<15}"
)
self.cuenta_partidas += 1
# MF03
# Sample: MF03FRANCISCO 1234
# MF03 (4)
# Person (Top Left?)
# Sample: "MF03FRANCISCO 1234"
# It seems "FRANCISCO" is right after MF03. That's the PERSON.
# "1234" is the Gafete/License.
# My previous code put Carrier first: "MF03TRUCK Lopez Doriga..."
# Correct mapping: MF03 + Person(Included Name) + License
# Let's follow sample:
# MF03 + Person(15?) + License(15?)
# MF03
transportista = manifiesto.carrier_code or ""
persona = persona_cargo or ""
# 'driver_license' attribute does not exist in Manifest model.
# Using 'transport_code' or similar as fallback for license/gafete.
licencia = manifiesto.transport_code or ""
lineas.append(
f"MF03{persona[:15]:<15} {licencia[:15]:<15}"
)
self.cuenta_partidas += 1
# --- PROCESAR FACTURAS ---
empresa_dict = {
'broker': datos_empresa.broker,
'responsable': datos_empresa.responsable,
'rfc': datos_empresa.rfc,
'nombre_empresa': datos_empresa.nombre_empresa,
'entry_port': entry_port,
'manufacturer_id': datos_empresa.manufacturer_id
}
processor = ScaiiProcessor()
l_facturas, e_facturas = processor.procesar_facturas(db, manifiesto_num, empresa_dict, request)
lineas.extend(l_facturas)
self.errores_validacion.extend(e_facturas)
# Actualizar acumuladores Globales
self.cuenta_partidas += processor.cuenta_partidas
self.cuenta_facturas += processor.cuenta_facturas
self.valor_total += processor.valor_total_factura
self.peso_bruto_total += processor.peso_bruto_factura
self.peso_neto_total += processor.peso_neto_factura
# MF80 (Totales Manifiesto)
# Sample: MF80000000000000000200000001099200000002000000009736
# MF80 (4) + Val(12) + CantFact(4) + PB(12) + Flete(8) + PN(12)
# Importante: El sample muestra que los totales NO tienen puntos y son enteros (centavos).
val_int = int(round(processor.valor_total_factura * 100))
pb_int = int(round(processor.peso_bruto_factura * 100))
pn_int = int(round(processor.peso_neto_factura * 100))
flete_int = 0 # Flete total
lineas.append(
f"MF80{val_int:012d}"
f"{processor.cuenta_facturas:04d}"
f"{pb_int:012d}"
f"{flete_int:08d}"
f"{pn_int:012d}"
)
self.cuenta_partidas += 1
return lineas
# (Mantenemos los métodos auxiliares: _obtener_configuracion_sistema, _obtener_datos_empresa, _escribir_archivo, etc.)
def _obtener_configuracion_sistema(self, db): return ConfiguracionSistema(path_arch_transmision="/tmp", utilizar_nombre_generico_mainx30=True, utilizar_codigo_broker_cliente=False)
def _obtener_datos_empresa(self, db):
empresa = db.query(GEmpresa).first()
if not empresa:
# Fallback safe defaults if no company config found
return EmpresaDatos(broker="", responsable="", rfc="", tiene_linea_express="N", nombre_empresa="", manufacturer_id="", ftp_key="")
return EmpresaDatos(
broker=(empresa.broker_company or "")[:5],
responsable=(empresa.responsible or "")[:30],
rfc=(empresa.rfc or "")[:13],
tiene_linea_express=empresa.has_express_line or "N",
nombre_empresa=(empresa.name or "")[:40],
manufacturer_id=(empresa.manufacturer_id or "")[:10],
ftp_key=(empresa.ftp_key or "")[:10]
)
def _validar_datos_empresa(self, datos): pass
def _generar_nombre_archivo(self, c, r, m): return f"{m}_Mainx30.dat"
def _generar_linea_a(self, f, d):
# Sample: A 26021203AKR AKR 00SCSI
broker = d.broker.strip()[:6]
# Use ftp_key (password?)
password = (d.ftp_key or "00SCSI")[:6]
return f"A {f}03{broker:<6}{broker:<10}{password}"
def _escribir_archivo(self, ruta, lineas):
Path(ruta).parent.mkdir(parents=True, exist_ok=True)
with open(ruta, 'w', encoding='latin-1') as f: f.write('\r\n'.join(lineas))
def _generar_mensaje_resultado(self, nombre): return f"Generado: {nombre}"
def _agregar_error_validacion(self, partida, id_err, desc, sol, tipo):
self.errores_validacion.append(ErrorValidacion(partida=partida, linea=0, descripcion=desc, soluciones=sol, identificador=tipo))

View File

@@ -0,0 +1,46 @@
from celery import Task
from core.celery_app import celery_app
from core.celery_app import celery_app
from core.database import get_core_db as get_db
from .service import Mainx30Service
from .schemas import Mainx30GenerationRequest, Mainx30Response
@celery_app.task(name="generar_transmission_file_async", bind=True)
def generar_transmission_file_async(self, request_data: dict, tenant_id: int):
"""
Generates the transmission .dat file asynchronously using Mainx30Service
"""
try:
# Re-create db session for task
# Using next(get_db()) is a common pattern for obtaining a session in tasks
# but ensure context management
db = next(get_db())
# Deserialize request
request = Mainx30GenerationRequest(**request_data)
service = Mainx30Service()
response = service.generar_mainx30_expo(db, request, task_instance=self)
# Return result as dict for Celery serialization
# Ensure we return valid JSON serializable dict
result = response.model_dump()
# If we returned content directly, encode it if it's bytes (it's str here)
if response.content:
import base64
# Mainx30Service returns content as string with \r\n
encoded_content = base64.b64encode(response.content.encode('utf-8')).decode('utf-8')
# Add to result to match expected format by frontend dialog
result['content'] = encoded_content
result['file_name'] = response.archivo_generado
result['media_type'] = "text/plain"
return result
except Exception as e:
import traceback
traceback.print_exc()
self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': str(e)})
# Re-raise to mark task as failed in Celery
raise e

View File

@@ -17,7 +17,7 @@ from api.v1.modules.a76.invoices.models import (
)
from api.v1.modules.a76.items.line_financials.models import LineFinancial
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider,
ClientProviderAddress,
@@ -27,7 +27,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -488,8 +488,7 @@ class ConsolidadoImportacionMexService:
lines = (
db.query(LineItem)
.join(Item, LineItem.item_id == Item.id)
.filter(Item.invoice_id.in_(target_invoice_ids))
.filter(LineItem.invoice_id.in_(target_invoice_ids))
.all()
)
@@ -538,7 +537,7 @@ class ConsolidadoImportacionMexService:
.filter(LineFinancial.item_line_id == line.id)
.first()
)
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
# --- Resolver Identificadores (MOVED INSIDE MAIN LOOP) ---
us_fraction_raw = ""
@@ -565,7 +564,7 @@ class ConsolidadoImportacionMexService:
# --- Multi-Currency Normalization Logic ---
# Determine Line Currency context
# Use manual lookup instead of specific attribute
invoice_id = line.item.invoice_id if line.item else None
invoice_id = line.invoice_id
line_invoice = invoice_map.get(invoice_id) if invoice_id else None
line_currency_is_mxn = False

View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics, InvoiceComplianceMx
from api.v1.modules.a76.items.line_financials.models import LineFinancial
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
)
@@ -21,7 +20,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -314,8 +313,8 @@ class ConsolidadoImportacionMexService:
# NOT consolidating all invoices from the same Pedimento.
target_invoice_ids = [header.id]
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(
Item.invoice_id.in_(target_invoice_ids)
lines = db.query(LineItem).filter(
LineItem.invoice_id.in_(target_invoice_ids)
).all()
partidas_list = []
@@ -350,7 +349,7 @@ class ConsolidadoImportacionMexService:
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
# --- Resolver Identificadores (MOVED INSIDE MAIN LOOP) ---
us_fraction_raw = ""
@@ -371,7 +370,7 @@ class ConsolidadoImportacionMexService:
# --- Multi-Currency Normalization Logic ---
# Determine Line Currency context
# Use manual lookup instead of specific attribute
invoice_id = line.item.invoice_id if line.item else None
invoice_id = line.invoice_id
line_invoice = invoice_map.get(invoice_id) if invoice_id else None
line_currency_is_mxn = False

View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.items.line_financials.models import LineFinancial
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider,
ClientProviderAddress,
@@ -23,7 +22,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -36,6 +35,7 @@ from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models impor
# --- MODELO DE UNIDADES DE MEDIDA ---
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.packages.models import Package
# --- SCHEMAS ---
from .schemas import (
@@ -428,9 +428,8 @@ class FacturaImportacionMexService:
if progress_callback:
progress_callback(50, "Procesando partidas...")
lines = (
db.query(LineItem)
.join(Item, LineItem.item_id == Item.id)
.filter(Item.invoice_id == header.id)
db.query(LineItem)
.filter(LineItem.invoice_id == header.id)
.all()
)
partidas_list = []
@@ -446,10 +445,10 @@ class FacturaImportacionMexService:
.filter(LineFinancial.item_line_id == line.id)
.first()
)
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
desc_final = "S/D"
num_parte_final = str(line.part_number or "S/N")
num_parte_final = str(line.part_number_id or "S/N")
fraccion_raw = ""
origen_final = "MEX"
@@ -539,7 +538,7 @@ class FacturaImportacionMexService:
if uom:
unidad_desc = uom.description or uom.code
else:
unidad_desc = ""
unidad_desc = ""
partidas_list.append(
PartidaSchema(
@@ -558,7 +557,7 @@ class FacturaImportacionMexService:
if qty and qty.package_quantity
else 0
),
clave_bultos=(qty.package_key or "") if qty else "",
clave_bultos=(qty.package_info.key if (qty and qty.package_info) else ""),
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(
qty.gross_weight if qty else 0

View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.items.line_financials.models import LineFinancial
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
)
@@ -21,7 +20,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -227,16 +226,16 @@ class FacturaImportacionMexService:
)
if progress_callback: progress_callback(50, "Procesando partidas...")
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(Item.invoice_id == header.id).all()
lines = db.query(LineItem).filter(LineItem.invoice_id == header.id).all()
partidas_list = []
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
desc_final = "S/D"
num_parte_final = str(line.part_number or "S/N")
num_parte_final = str(line.part_number_id or "S/N")
fraccion_raw = ""
origen_final = "MEX"
@@ -316,7 +315,7 @@ class FacturaImportacionMexService:
cantidad_importacion=self.formatear_numero(qty.quantity if qty else 0),
unidad_medida=qty.weight_unit if qty else "PZA",
cantidad_bultos=int(qty.package_quantity) if qty and qty.package_quantity else 0,
clave_bultos=(qty.package_key or "") if qty else "",
clave_bultos=(qty.package_info.key if (qty and qty.package_info) else ""),
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(qty.gross_weight if qty else 0),
valor_costo_unitario=self.formatear_numero(v_unitario),

View File

@@ -10,18 +10,19 @@ from fastapi import HTTPException
from sqlalchemy.orm import Session
# --- MODELOS ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.items.line_financials.models import LineFinancial
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
ClientProvider,
ClientProviderAddress,
ClientProviderPrograms,
)
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -30,32 +31,38 @@ from api.v1.modules.a76.transportation.trailers.models import Trailer
from api.v1.modules.a76.transportation.drivers.models import Driver
# --- MODELO DE FRACCIONES ---
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import TariffFraction
from api.v1.modules.a76.general_catalogs.fractions.tariff_fractions.models import (
TariffFraction,
)
# --- SCHEMAS ---
# Reuse schemas from neighbor package as they fit the same data structure
from ..mex.schemas import (
ClienteSchema, PartidaSchema, TotalesSchema,
FacturaSchema, FacturaImportacionCompleta
ClienteSchema,
PartidaSchema,
TotalesSchema,
FacturaSchema,
FacturaImportacionCompleta,
)
class FacturaImportacionUsaService:
def __init__(self):
self.template_dir = Path(__file__).parent.parent / "templates"
self.jinja_env = Environment(
loader=FileSystemLoader(self.template_dir),
autoescape=select_autoescape(['html', 'xml'])
autoescape=select_autoescape(["html", "xml"]),
)
self.template = self.jinja_env.get_template('factura_usa_ver.html')
self.template = self.jinja_env.get_template("factura_usa_ver.html")
def _get_document_title(self, invoice_type: str, is_american: bool = True) -> str:
"""
Determina el título del documento basado en el tipo de factura.
Args:
invoice_type: Tipo de factura (TEM, DEF, MEX, CR)
is_american: Si es factura americana (True) o mexicana (False)
Returns:
Título formateado para la factura
"""
@@ -66,7 +73,7 @@ class FacturaImportacionUsaService:
"TEM": "Importación Temporal",
"CR": "Importación de Cambio de Régimen",
}
# Mapeo para facturas americanas
american_titles = {
"MEX": "Mexican Purchases Import Invoice",
@@ -74,18 +81,18 @@ class FacturaImportacionUsaService:
"TEM": "Temporary Importation",
"CR": "Regime Change Importation",
}
# Seleccionar el mapa correcto
titles = american_titles if is_american else mexican_titles
# Obtener el título (normalizar a mayúsculas)
invoice_type_upper = invoice_type.upper() if invoice_type else ""
title = titles.get(invoice_type_upper, "")
# Fallback a genéricos si no se encuentra
if not title:
return "Commercial Invoice" if is_american else "Factura de Importación"
return title
def _get_wkhtmltopdf_config(self):
@@ -108,18 +115,37 @@ class FacturaImportacionUsaService:
return fraccion_raw
return f"{fraccion_raw[:4]}.{fraccion_raw[4:6]}.{fraccion_raw[6:]}"
def _obtener_datos_cliente(self, db: Session, client_id: int, rol: str) -> ClienteSchema:
def _obtener_datos_cliente(
self, db: Session, client_id: int, rol: str
) -> ClienteSchema:
main = db.query(ClientProvider).filter(ClientProvider.id == client_id).first()
if not main:
return ClienteSchema(header=rol, nombre="Unknown", direccion="", tax_id="", codigo_postal="", ciudad="", estado="", pais="USA")
addr = db.query(ClientProviderAddress).filter(ClientProviderAddress.client_id == client_id).first()
prog = db.query(ClientProviderPrograms).filter(ClientProviderPrograms.client_id == client_id).first()
return ClienteSchema(
header=rol,
nombre="Unknown",
direccion="",
tax_id="",
codigo_postal="",
ciudad="",
estado="",
pais="USA",
)
addr = (
db.query(ClientProviderAddress)
.filter(ClientProviderAddress.client_id == client_id)
.first()
)
prog = (
db.query(ClientProviderPrograms)
.filter(ClientProviderPrograms.client_id == client_id)
.first()
)
return ClienteSchema(
header=rol,
nombre=(main.name or main.short_name) or "N/A",
direccion=(addr.streets or "") if addr else "",
direccion=(addr.streets or "") if addr else "",
num_exterior=(addr.exterior_number or "") if addr else "",
num_interior=(addr.interior_number or "") if addr else "",
colonia=(addr.neighborhood or "") if addr else "",
@@ -127,58 +153,128 @@ class FacturaImportacionUsaService:
ciudad=(addr.city or "") if addr else "",
estado=(addr.state or "") if addr else "",
pais=(addr.country or "USA") if addr else "USA",
tax_id=prog.tax_id if (prog and prog.tax_id) else (getattr(main, 'rfc', "") or ""),
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=prog.prosec_authorization if (prog and prog.prosec and prog.prosec_authorization) else "",
reg_emp=prog.val_certified_company_registry if (prog and hasattr(prog, 'val_certified_company_registry')) else (
prog.certified_company_registry if (prog and prog.certified_company_registry) else ""
tax_id=(
prog.tax_id
if (prog and prog.tax_id)
else (getattr(main, "rfc", "") or "")
),
programa="IMMEX" if (prog and prog.program) else "",
autorizacion=prog.program_number if prog else "",
prosec=(
prog.prosec_authorization
if (prog and prog.prosec and prog.prosec_authorization)
else ""
),
reg_emp=(
prog.val_certified_company_registry
if (prog and hasattr(prog, "val_certified_company_registry"))
else (
prog.certified_company_registry
if (prog and prog.certified_company_registry)
else ""
)
),
cert=(
prog.is_certified_company
if (prog and prog.is_certified_company)
else ""
),
cert=prog.is_certified_company if (prog and prog.is_certified_company) else ""
)
def obtener_datos(self, db: Session, invoice_id: int, company_id: int, progress_callback: Optional[Callable] = None, currency_code: str = 'ORIGINAL') -> FacturaImportacionCompleta:
def obtener_datos(
self,
db: Session,
invoice_id: int,
company_id: int,
progress_callback: Optional[Callable] = None,
currency_code: str = "ORIGINAL",
) -> FacturaImportacionCompleta:
try:
if progress_callback: progress_callback(10, "Searching invoice...")
header = db.query(InvoiceHeader).filter(InvoiceHeader.id == invoice_id, InvoiceHeader.company_id == company_id).first()
if not header: raise HTTPException(status_code=404, detail="Invoice not found")
if progress_callback:
progress_callback(10, "Searching invoice...")
header = (
db.query(InvoiceHeader)
.filter(
InvoiceHeader.id == invoice_id,
InvoiceHeader.company_id == company_id,
)
.first()
)
if not header:
raise HTTPException(status_code=404, detail="Invoice not found")
compliance = header.compliance_mx
compliance = header.compliance_mx
logistics = header.logistics if header.logistics else None
financials = header.financials if header.financials else None
if progress_callback: progress_callback(20, "Fetching entry data...")
pedimento_id = compliance.pedimento_id if (compliance and compliance.pedimento_id) else header.related_doc_id
pedimento = db.query(Pedimentos).filter(Pedimentos.id == pedimento_id).first() if pedimento_id else None
if progress_callback: progress_callback(30, "Fetching client and supplier...")
if progress_callback:
progress_callback(20, "Fetching entry data...")
pedimento_id = (
compliance.pedimento_id
if (compliance and compliance.pedimento_id)
else header.related_doc_id
)
pedimento = (
db.query(Pedimentos).filter(Pedimentos.id == pedimento_id).first()
if pedimento_id
else None
)
if progress_callback:
progress_callback(30, "Fetching client and supplier...")
proveedor_id = compliance.provider_id if compliance else None
cliente_proveedor = self._obtener_datos_cliente(db, proveedor_id, "Supplier:") if proveedor_id else ClienteSchema(header="Supplier", nombre="Unassigned", direccion="", tax_id="", codigo_postal="", ciudad="", estado="", pais="")
cliente_proveedor = (
self._obtener_datos_cliente(db, proveedor_id, "Supplier:")
if proveedor_id
else ClienteSchema(
header="Supplier",
nombre="Unassigned",
direccion="",
tax_id="",
codigo_postal="",
ciudad="",
estado="",
pais="",
)
)
nombre_agente = ""
if compliance and compliance.customs_broker_id:
broker = db.query(CustomsBroker).filter(CustomsBroker.id == compliance.customs_broker_id).first()
if broker: nombre_agente = broker.name
broker = (
db.query(CustomsBroker)
.filter(CustomsBroker.id == compliance.customs_broker_id)
.first()
)
if broker:
nombre_agente = broker.name
company = db.query(Company).filter(Company.id == header.company_id).first()
# Datos Default (Company/Importer)
cliente_default = ClienteSchema(
header="Importer / Consignee:",
nombre=getattr(company, 'name', "Local Company"),
nombre=getattr(company, "name", "Local Company"),
direccion="FISCAL ADDRESS",
num_exterior="", colonia="", codigo_postal="", ciudad="", estado="", pais="MEX",
tax_id=getattr(company, 'rfc', ""),
programa=getattr(company, 'program', "IMMEX"), autorizacion=getattr(company, 'program_number', "")
num_exterior="",
colonia="",
codigo_postal="",
ciudad="",
estado="",
pais="MEX",
tax_id=getattr(company, "rfc", ""),
programa=getattr(company, "program", "IMMEX"),
autorizacion=getattr(company, "program_number", ""),
)
# Left Side Logic (Sold To)
cliente_vendido = cliente_default
if compliance and compliance.sold_to_id:
# Force English header for American Invoice
clean_header = "Sold To:"
clean_header = "Sold To:"
# raw_header = compliance.sold_to_header or "SOLD_TO"
# clean_header = raw_header.replace("_", " ").title() + ":"
cliente_vendido = self._obtener_datos_cliente(db, compliance.sold_to_id, clean_header)
cliente_vendido = self._obtener_datos_cliente(
db, compliance.sold_to_id, clean_header
)
# Right Side Logic (Shipped To)
cliente_enviado = cliente_default
if compliance and compliance.shipped_to_id:
@@ -186,26 +282,37 @@ class FacturaImportacionUsaService:
clean_header_shipped = "Shipped To:"
# raw_header_shipped = compliance.shipped_to_header or "SHIPPED_TO"
# clean_header_shipped = raw_header_shipped.replace("_", " ").title() + ":"
# Fetch client data
cliente_enviado = self._obtener_datos_cliente(db, compliance.shipped_to_id, clean_header_shipped)
remesa_valor = str(compliance.remesa) if (compliance and compliance.remesa) else ""
acuse_valor = str(compliance.edocument) if (compliance and compliance.edocument) else "N/A"
# Fetch client data
cliente_enviado = self._obtener_datos_cliente(
db, compliance.shipped_to_id, clean_header_shipped
)
remesa_valor = (
str(compliance.remesa) if (compliance and compliance.remesa) else ""
)
acuse_valor = (
str(compliance.edocument)
if (compliance and compliance.edocument)
else "N/A"
)
patente_val = ""
if pedimento and pedimento.license:
patente_val = pedimento.license
elif 'broker' in locals() and broker and broker.license:
elif "broker" in locals() and broker and broker.license:
patente_val = broker.license
# --- Transport Data Fetching ---
transporte_txt = str(logistics.transport_type) if (logistics and logistics.transport_type) else ""
transporte_txt = (
str(logistics.transport_type)
if (logistics and logistics.transport_type)
else ""
)
num_transporte_val = (logistics.trailer_num or "") if logistics else ""
# Init values
placas_val = (logistics.license_plate or "") if logistics else "" # Plates
placas_val = (logistics.license_plate or "") if logistics else "" # Plates
placas_remolque_val = ""
transportista_val = (logistics.carrier_id or "") if logistics else ""
caat_val = ""
@@ -215,53 +322,89 @@ class FacturaImportacionUsaService:
if logistics:
# 1. Transporter (CAAT / SCAC)
if logistics.carrier_id:
transporter_obj = db.query(Transporter).filter(Transporter.transporter_key == logistics.carrier_id).first()
transporter_obj = (
db.query(Transporter)
.filter(Transporter.transporter_key == logistics.carrier_id)
.first()
)
if transporter_obj:
caat_val = transporter_obj.caat_code or ""
scac_val = transporter_obj.transport_code or "" # Mapping transport_code to SCAC
scac_val = (
transporter_obj.transport_code or ""
) # Mapping transport_code to SCAC
transportista_val = transporter_obj.name or logistics.carrier_id
# 2. Vehicle (Plates)
if logistics.transport_id:
veh_obj = db.query(Vehicle).filter(Vehicle.vehicle_key == logistics.transport_id).first()
veh_obj = (
db.query(Vehicle)
.filter(Vehicle.vehicle_key == logistics.transport_id)
.first()
)
if veh_obj:
placas_val = veh_obj.plate_number or placas_val
elif logistics.vehicle_num:
veh_obj = db.query(Vehicle).filter(Vehicle.vehicle_key == logistics.vehicle_num).first()
if veh_obj:
placas_val = veh_obj.plate_number or placas_val
placas_val = veh_obj.plate_number or placas_val
elif logistics.vehicle_num:
veh_obj = (
db.query(Vehicle)
.filter(Vehicle.vehicle_key == logistics.vehicle_num)
.first()
)
if veh_obj:
placas_val = veh_obj.plate_number or placas_val
# 3. Trailer
if logistics.trailer_num:
trl_obj = db.query(Trailer).filter(Trailer.trailer_number == logistics.trailer_num).first()
trl_obj = (
db.query(Trailer)
.filter(Trailer.trailer_number == logistics.trailer_num)
.first()
)
if trl_obj:
placas_remolque_val = trl_obj.plate_number or ""
# 4. Driver (License)
if logistics.carrier_id and logistics.driver_name:
drv_obj = db.query(Driver).filter(
Driver.transporter_key == logistics.carrier_id,
Driver.driver_name == logistics.driver_name
).first()
drv_obj = (
db.query(Driver)
.filter(
Driver.transporter_key == logistics.carrier_id,
Driver.driver_name == logistics.driver_name,
)
.first()
)
if drv_obj:
licencia_cond_val = drv_obj.license_number or ""
licencia_cond_val = drv_obj.license_number or ""
# Determine Currency
moneda_final = getattr(header, 'currency', "USD") or "USD"
if currency_code == 'MXN':
moneda_final = 'MXN'
elif currency_code == 'USD':
moneda_final = 'USD'
moneda_final = getattr(header, "currency", "USD") or "USD"
if currency_code == "MXN":
moneda_final = "MXN"
elif currency_code == "USD":
moneda_final = "USD"
factura_schema = FacturaSchema(
numero=header.invoice_number or "N/A",
titulo_documento=self._get_document_title(header.invoice_type or "", is_american=True),
titulo_documento=self._get_document_title(
header.invoice_type or "", is_american=True
),
fecha=str(header.invoice_date) if header.invoice_date else "",
tipo_cambio=float(financials.exchange_rate) if (financials and financials.exchange_rate) else (float(pedimento.exchange_rate) if pedimento and pedimento.exchange_rate else 1.0),
tipo_cambio=(
float(financials.exchange_rate)
if (financials and financials.exchange_rate)
else (
float(pedimento.exchange_rate)
if pedimento and pedimento.exchange_rate
else 1.0
)
),
moneda=moneda_final,
incoterm=(logistics.incoterm or "") if logistics else "",
observaciones=header.observation_es or header.observation_en or "",
pedimento=f"{pedimento.year} {pedimento.customs_office[:2] if pedimento.customs_office else ''} {pedimento.license} {pedimento.pedimento_number}" if pedimento else "",
pedimento=(
f"{pedimento.year} {pedimento.customs_office[:2] if pedimento.customs_office else ''} {pedimento.license} {pedimento.pedimento_number}"
if pedimento
else ""
),
clave_pedimento=pedimento.pedimento_code if pedimento else "",
regimen=header.document_type or "",
patente=patente_val,
@@ -274,62 +417,87 @@ class FacturaImportacionUsaService:
caat=caat_val,
scac=scac_val,
licencia_conductor=licencia_cond_val,
aduana=compliance.aduana if (compliance and compliance.aduana) else (pedimento.customs_office[:2] if (pedimento and pedimento.customs_office) else ""),
aduana=(
compliance.aduana
if (compliance and compliance.aduana)
else (
pedimento.customs_office[:2]
if (pedimento and pedimento.customs_office)
else ""
)
),
precinto=(logistics.seal_number or "") if logistics else "",
destino=(logistics.destination_goods or "") if logistics else "",
remesa=remesa_valor, acuse_electronico=acuse_valor
remesa=remesa_valor,
acuse_electronico=acuse_valor,
)
if progress_callback: progress_callback(50, "Processing items...")
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(Item.invoice_id == header.id).all()
if progress_callback:
progress_callback(50, "Processing items...")
lines = db.query(LineItem).filter(LineItem.invoice_id == header.id).all()
partidas_list = []
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
fin = db.query(LineFinancial).filter(LineFinancial.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
qty = (
db.query(LineQuantity)
.filter(LineQuantity.item_line_id == line.id)
.first()
)
fin = (
db.query(LineFinancial)
.filter(LineFinancial.item_line_id == line.id)
.first()
)
part_master = (
db.query(Part).filter(Part.id == line.part_number_id).first()
)
desc_final = "N/D"
num_parte_final = str(line.part_number or "N/A")
fraccion_raw = ""
num_parte_final = str(line.part_number_id or "N/A")
fraccion_raw = ""
origen_final = "MEX"
if part_master:
# Prefer English description if available, else Spanish
desc_final = part_master.description_english or part_master.description_spanish or "No Desc."
desc_final = (
part_master.description_english
or part_master.description_spanish
or "No Desc."
)
num_parte_final = part_master.part_number
# Prefer US Fraction (HTS) if available
fraccion_raw = part_master.us_fraction if part_master.us_fraction else ""
fraccion_raw = (
part_master.us_fraction if part_master.us_fraction else ""
)
if part_master.fa_data and part_master.fa_data.origin_country:
origen_final = part_master.fa_data.origin_country
# FRACTION LOGIC: Use US Fraction (us_fraction) if available, otherwise blank
fraccion_imprimir = ""
# Check part master US fraction
if part_master and part_master.us_fraction:
fraccion_imprimir = part_master.us_fraction.strip()
# Optional: Format if needed, but raw is usually fine for US HTS
# If valid US fraction logic requires looking up in DB, we could add that here.
# For now, per requirement: "Si no tiene, pues de queda en blanco"
# Default "General" and "0%" if no specific logic for US duties yet
preferencia_txt = "General"
preferencia_txt = "General"
advalorem_txt = "0%"
# Prioritize USD for American Invoice logic if available?
# Sticking to same logic as Mex for now but could prioritize USD columns.
# Actually, duplicate logic from mex service for now to ensure consistency.
v_unitario = 0.0
v_total = 0.0
if fin:
is_mxn = (factura_schema.moneda == 'MXN')
is_mxn = factura_schema.moneda == "MXN"
if is_mxn:
v_unitario = float(fin.unit_cost_commercial_mxn or 0.0)
v_total = float(fin.value_commercial_mxn or 0.0)
@@ -338,13 +506,13 @@ class FacturaImportacionUsaService:
v_total = float(fin.value_commercial_usd or 0.0)
if not v_unitario:
v_unitario = float(fin.commercial_unit_cost or 0.0)
v_unitario = float(fin.commercial_unit_cost or 0.0)
if not v_total:
v_total = float(fin.total_commercial_value or 0.0)
v_total = float(fin.total_commercial_value or 0.0)
cantidad = float(qty.quantity) if (qty and qty.quantity) else 0.0
if cantidad > 0:
if v_unitario > 0 and v_total == 0:
v_total = v_unitario * cantidad
@@ -352,39 +520,59 @@ class FacturaImportacionUsaService:
v_unitario = v_total / cantidad
# UOM Mapping for English context
uom_raw = qty.weight_unit if qty else "PCS"
if uom_raw == "PZA": uom_raw = "PCS"
uom_raw = line.unit_of_measure_info.code if line.unit_of_measure_info else "PCS"
if uom_raw == "PZA":
uom_raw = "PCS"
partidas_list.append(PartidaSchema(
numero_parte=num_parte_final,
descripcion=desc_final,
fraccion=fraccion_imprimir,
origen=origen_final,
advalorem=advalorem_txt,
preferencia=preferencia_txt,
cantidad_importacion=self.formatear_numero(qty.quantity if qty else 0),
unidad_medida=uom_raw,
cantidad_bultos=int(qty.package_quantity) if qty and qty.package_quantity else 0,
clave_bultos=(qty.package_key or "") if qty else "",
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(qty.gross_weight if qty else 0),
valor_costo_unitario=self.formatear_numero(v_unitario),
valor_total=self.formatear_numero(v_total)
))
partidas_list.append(
PartidaSchema(
numero_parte=num_parte_final,
descripcion=desc_final,
fraccion=fraccion_imprimir,
origen=origen_final,
advalorem=advalorem_txt,
preferencia=preferencia_txt,
cantidad_importacion=self.formatear_numero(
qty.quantity if qty else 0
),
unidad_medida=uom_raw,
cantidad_bultos=(
int(qty.package_quantity)
if qty and qty.package_quantity
else 0
),
clave_bultos=(
qty.package_info.key if (qty and qty.package_info) else ""
),
peso_neto=self.formatear_numero(qty.net_weight if qty else 0),
peso_bruto=self.formatear_numero(
qty.gross_weight if qty else 0
),
valor_costo_unitario=self.formatear_numero(v_unitario),
valor_total=self.formatear_numero(v_total),
)
)
totales = self.calcular_totales(partidas_list, Decimal(factura_schema.tipo_cambio))
totales = self.calcular_totales(
partidas_list, Decimal(factura_schema.tipo_cambio)
)
return FacturaImportacionCompleta(
cliente_proveedor=cliente_proveedor, cliente_vendido=cliente_vendido,
cliente_enviado=cliente_enviado, factura=factura_schema,
partidas=partidas_list, totales=totales
cliente_proveedor=cliente_proveedor,
cliente_vendido=cliente_vendido,
cliente_enviado=cliente_enviado,
factura=factura_schema,
partidas=partidas_list,
totales=totales,
)
except Exception as e:
print(f"Error Service A76 USA: {e}")
raise HTTPException(status_code=500, detail=f"Error: {str(e)}")
def calcular_totales(self, partidas: List[PartidaSchema], tipo_cambio: Decimal) -> TotalesSchema:
def calcular_totales(
self, partidas: List[PartidaSchema], tipo_cambio: Decimal
) -> TotalesSchema:
cant = sum(p.cantidad_importacion for p in partidas)
valor = sum(p.valor_total for p in partidas)
peso_n = sum(p.peso_neto for p in partidas)
@@ -392,22 +580,38 @@ class FacturaImportacionUsaService:
bultos = sum(p.cantidad_bultos for p in partidas)
claves = [p.clave_bultos for p in partidas if p.clave_bultos]
clave_comun = max(set(claves), key=claves.count) if claves else ""
# if bultos > 1 and clave_comun and not clave_comun.endswith("S"): clave_comun += "S"
# if bultos > 1 and clave_comun and not clave_comun.endswith("S"): clave_comun += "S"
# Don't pluralize strictly in English without logic, kept simple.
tc = float(tipo_cambio) if tipo_cambio else 1.0
return TotalesSchema(
cantidad_total=self.formatear_numero(cant), bultos_total=bultos, clave_bultos=clave_comun,
peso_neto_total=self.formatear_numero(peso_n), peso_bruto_total=self.formatear_numero(peso_b),
valor_total_total=self.formatear_numero(valor), valor_total_dolares=self.formatear_numero(valor / tc if tc > 0 else 0)
return TotalesSchema(
cantidad_total=self.formatear_numero(cant),
bultos_total=bultos,
clave_bultos=clave_comun,
peso_neto_total=self.formatear_numero(peso_n),
peso_bruto_total=self.formatear_numero(peso_b),
valor_total_total=self.formatear_numero(valor),
valor_total_dolares=self.formatear_numero(valor / tc if tc > 0 else 0),
)
def generar_factura_completa(self, db: Session, invoice_id: int, company_id: int, formato: str = "pdf", progress_callback: Optional[Callable] = None, currency_code: str = 'ORIGINAL') -> Tuple[bytes, str, str]:
if progress_callback: progress_callback(5, "Starting report service...")
datos = self.obtener_datos(db, invoice_id, company_id, progress_callback, currency_code)
if progress_callback: progress_callback(80, "Rendering template...")
def generar_factura_completa(
self,
db: Session,
invoice_id: int,
company_id: int,
formato: str = "pdf",
progress_callback: Optional[Callable] = None,
currency_code: str = "ORIGINAL",
) -> Tuple[bytes, str, str]:
if progress_callback:
progress_callback(5, "Starting report service...")
datos = self.obtener_datos(
db, invoice_id, company_id, progress_callback, currency_code
)
if progress_callback:
progress_callback(80, "Rendering template...")
# LOGO LOGIC
logo_b64 = None
try:
@@ -422,26 +626,48 @@ class FacturaImportacionUsaService:
if target_path.exists():
with open(target_path, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
encoded_string = base64.b64encode(image_file.read()).decode(
"utf-8"
)
mime = "image/png"
if target_path.suffix.lower() in ['.jpg', '.jpeg']: mime = "image/jpeg"
if target_path.suffix.lower() in [".jpg", ".jpeg"]:
mime = "image/jpeg"
logo_b64 = f"data:{mime};base64,{encoded_string}"
except Exception as e:
print(f"Error loading logo: {e}")
context = {
'cliente_proveedor': datos.cliente_proveedor.model_dump(), 'cliente_vendido': datos.cliente_vendido.model_dump(),
'cliente_enviado': datos.cliente_enviado.model_dump(), 'factura': datos.factura.model_dump(),
'partidas': [p.model_dump() for p in datos.partidas], 'totales': datos.totales.model_dump(),
'logo_b64': logo_b64
"cliente_proveedor": datos.cliente_proveedor.model_dump(),
"cliente_vendido": datos.cliente_vendido.model_dump(),
"cliente_enviado": datos.cliente_enviado.model_dump(),
"factura": datos.factura.model_dump(),
"partidas": [p.model_dump() for p in datos.partidas],
"totales": datos.totales.model_dump(),
"logo_b64": logo_b64,
}
html_content = self.template.render(**context)
nombre = f"Commercial_Invoice_{datos.factura.numero}.{formato}"
if formato == "html": return html_content.encode('utf-8'), nombre, "text/html"
if progress_callback: progress_callback(90, "Generating PDF...")
options = {'page-size': 'Letter', 'margin-top': '0.5in', 'margin-right': '0.5in', 'margin-bottom': '0.5in', 'margin-left': '0.5in', 'encoding': "UTF-8", 'enable-local-file-access': None}
pdf = pdfkit.from_string(html_content, False, options=options, configuration=self._get_wkhtmltopdf_config())
if progress_callback: progress_callback(100, "Completed")
if formato == "html":
return html_content.encode("utf-8"), nombre, "text/html"
if progress_callback:
progress_callback(90, "Generating PDF...")
options = {
"page-size": "Letter",
"margin-top": "0.5in",
"margin-right": "0.5in",
"margin-bottom": "0.5in",
"margin-left": "0.5in",
"encoding": "UTF-8",
"enable-local-file-access": None,
}
pdf = pdfkit.from_string(
html_content,
False,
options=options,
configuration=self._get_wkhtmltopdf_config(),
)
if progress_callback:
progress_callback(100, "Completed")
return pdf, nombre, "application/pdf"

View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import Session
# --- MODELOS ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceLogistics
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.line_items.models import LineItem
from api.v1.modules.a76.items.line_customs.models import LineCustom
from api.v1.modules.a76.clients_and_providers.models import (
ClientProvider, ClientProviderAddress, ClientProviderPrograms
@@ -22,7 +21,7 @@ from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.pedmientos.models import Pedimentos
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.items.models import Item
from api.v1.modules.a76.items.models import LineItem
# --- TRANSPORTATION MODELS ---
from api.v1.modules.a76.transportation.transporters.models import Transporter
@@ -246,40 +245,40 @@ class PackingListService:
)
if progress_callback: progress_callback(50, "Procesando partidas...")
lines = db.query(LineItem).join(Item, LineItem.item_id == Item.id).filter(Item.invoice_id == header.id).all()
lines = db.query(LineItem).filter(LineItem.invoice_id == header.id).all()
partidas_list = []
for line in lines:
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
weight_type = db.query(InvoiceLogistics.weight_type).filter(InvoiceLogistics.invoice_id == line.invoice_id).scalar()
qty = db.query(LineQuantity).filter(LineQuantity.item_line_id == line.id).first()
# --- WEIGHT CALCULATION LOGIC ---
peso_neto_kg = 0.0
peso_bruto_kg = 0.0
peso_neto_lb = 0.0
peso_bruto_lb = 0.0
peso_bruto_lb = 0.0
if qty:
raw_net = float(qty.net_weight or 0)
raw_gross = float(qty.gross_weight or 0)
unit = (qty.weight_unit or "KG").upper()
if unit == "LB" or unit == "LBS":
peso_neto_lb = raw_net
peso_bruto_lb = raw_gross
peso_neto_kg = raw_net / 2.20462
peso_bruto_kg = raw_gross / 2.20462
else: # Default KG
peso_neto_kg = raw_net
peso_bruto_kg = raw_gross
peso_neto_lb = raw_net * 2.20462
peso_bruto_lb = raw_gross * 2.20462
raw_net = float(qty.net_weight or 0)
raw_gross = float(qty.gross_weight or 0)
unit = (weight_type or "KGS").upper()
if unit == "LBS":
peso_neto_lb = raw_net
peso_bruto_lb = raw_gross
peso_neto_kg = raw_net / 2.20462
peso_bruto_kg = raw_gross / 2.20462
else: # Default KG
peso_neto_kg = raw_net
peso_bruto_kg = raw_gross
peso_neto_lb = raw_net * 2.20462
peso_bruto_lb = raw_gross * 2.20462
# --------------------------------
custom_obj = db.query(LineCustom).filter(LineCustom.item_line_id == line.id).first()
part_master = db.query(Part).filter(Part.id == line.part_number).first()
part_master = db.query(Part).filter(Part.id == line.part_number_id).first()
desc_final = "S/D"
num_parte_final = str(line.part_number or "S/N")
num_parte_final = str(line.part_number_id or "S/N")
fraccion_raw = ""
origen_final = "MEX"
uom_comercial = "PZA" # Default UOM

View File

@@ -0,0 +1,369 @@
from typing import List, Dict, Any, Tuple
from sqlalchemy.orm import Session, joinedload
from datetime import datetime
from .schemas import Mainx30GenerationRequest, ErrorValidacion
# --- MODELOS A76 ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
class ScaiiProcessor:
def __init__(self):
self.cuenta_partidas = 0
self.cuenta_facturas = 0
self.valor_total_factura = 0.0
self.flete_total = 0.0
self.peso_bruto_factura = 0.0
self.peso_neto_factura = 0.0
self.errores: List[ErrorValidacion] = []
def _obtener_datos_cliente(self, cliente: ClientProvider) -> dict:
"""Extrae de manera segura los datos del cliente/dirección"""
address = cliente.address
pais_raw = (address.country or "MX").upper() if address else "MX"
pais = "MX"
if pais_raw in ["MEXICO", "MEX", "MX"]:
pais = "MX"
elif pais_raw in ["USA", "US", "UNITED STATES"]:
pais = "US"
else:
pais = pais_raw[:2]
tax_id = cliente.rfc or ""
data = {
"nombre": (cliente.name or "")[:39],
"tax_id": tax_id[:15],
"broker": "", "calle": "", "cp": "", "ciudad": "", "estado": "", "pais": pais, "tel": "",
"manufacturer_id": "", "tipo_ext_nac": "E"
}
# TipoExtNac logic from Clarion: N (National/ManufacturerID) or E (External/TaxID)
# We'll use TaxID as default for definitive if not specified
data["manufacturer_id"] = (cliente.programs.manufacturer_id or "")[:15] if cliente.programs else ""
if cliente.programs:
data["broker"] = (cliente.programs.broker or "")[:6]
if address:
calle_base = address.streets or ""
num_base = address.exterior_number or ""
# Prevent 'None' string
calle_comp = f"{calle_base} {num_base}".strip()
# Clarion expects 20 chars for city_state: 5 CP + 11 City + 4 State
cp_formatted = (address.postal_code or "")[:5]
city_formatted = (address.city or "")[:11]
state_formatted = (address.state or "")[:4]
data["city_state"] = f"{cp_formatted:<5}{city_formatted:<11}{state_formatted:<4}"
data["calle"] = calle_comp[:35]
data["cp"] = (address.postal_code or "")[:9]
data["ciudad"] = (address.city or "").strip()[:20]
data["estado"] = (address.state or "").strip()[:2].upper()
data["tel"] = (address.phone or "")[:15]
return data
def _agregar_error(self, partida, id_err, desc, sol, tipo):
self.errores.append(ErrorValidacion(partida=partida, linea=0, descripcion=f"[ {id_err} ] {desc}", soluciones=sol, identificador=tipo))
class ScafDefProcessor(ScaiiProcessor):
"""Procesador para Importación Definitiva basado en lógica Clarion (SComprasMexID)"""
def __init__(self):
super().__init__()
def procesar_facturas(
self, db: Session, facturas_nums: List[str], empresa_dict: Dict[str, Any], request: Mainx30GenerationRequest
) -> Tuple[List[str], List[ErrorValidacion]]:
lineas = []
self.errores = []
# 1. Traer Facturas por número
facturas = db.query(InvoiceHeader).join(
InvoiceComplianceMx, InvoiceHeader.id == InvoiceComplianceMx.invoice_id
).options(
joinedload(InvoiceHeader.financials),
joinedload(InvoiceHeader.compliance_mx),
joinedload(InvoiceHeader.logistics)
).filter(
InvoiceHeader.invoice_number.in_(facturas_nums)
).all()
entry_port = request.entry_port or ""
exit_port = request.exit_port or ""
fecha_trans = datetime.now().strftime("%y%m%d")
entry_port_desc = empresa_dict.get('entry_port_desc', 'PUERTO ENTRADA')[:15]
exit_port_desc = empresa_dict.get('exit_port_desc', 'PUERTO SALIDA')[:15]
main_activity = empresa_dict.get('main_activity', 'RAW MATERIAL')[:30]
city_state = empresa_dict.get('city_state', '')[:30]
for factura in facturas:
self.cuenta_facturas += 1
f_val_total = 0.0
f_pb = 0.0
f_pn = 0.0
f_consec_partidas = 0
# MF01: Header
mod_trans = factura.logistics.transport_mode if factura.logistics else "30"
f_fecha = factura.invoice_date.strftime("%y%m%d") if factura.invoice_date else fecha_trans
lineas.append(
f"MF01{empresa_dict['broker'][:6]:<6}"
f"{exit_port[:5]:<5}"
f"{entry_port[:5]:<5}"
f"{fecha_trans}"
f" {mod_trans[:2]:<2}"
f"{f_fecha}"
f"{factura.invoice_number[:15]:<15}"
)
self.cuenta_partidas += 1
# MF03
conductor = (factura.logistics.driver_name or "")[:23] if factura.logistics else ""
carrier = (factura.logistics.carrier_id or "")[:10]
lineas.append(
f"MF03{carrier:<10}{conductor:<23}{entry_port_desc:<15}{exit_port_desc:<15}"
)
self.cuenta_partidas += 1
# MF20
num_transporte = (factura.logistics.transport_num or "")[:15] if factura.logistics else ""
# Use dynamic city/state from entry_port description if available, otherwise blank
# Clarion format: 5 digit Zip (if relevant) + City State
# We will use the entry_port_desc (e.g. CD. JUAREZ CHIH) and assume a dummy zip '00000' if not parsed
# Or better, just use the entry_port_desc fully aligned
# Using entry_port_desc directly instead of hardcoded default
cruce_desc = entry_port_desc[:20] if entry_port_desc else " "
lineas.append(
f"MF20{factura.invoice_number[:15]:<15}I{num_transporte:<15}00000{cruce_desc:<20}"
f"{exit_port[:5]:<5}{entry_port_desc:<15} "
)
self.cuenta_partidas += 1
# MF22
lineas.append(f"MF22{main_activity:<60}")
self.cuenta_partidas += 1
# IV01
flete = int(round(float(factura.financials.freight or 0))) if factura.financials else 0
self.flete_total += float(factura.financials.freight or 0) if factura.financials else 0.0
s_tax = ""; c_tax = ""
if factura.compliance_mx:
if factura.compliance_mx.sold_to_id:
c_obj = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.sold_to_id).first()
if c_obj: c_tax = (c_obj.rfc or "")[:12]
if factura.compliance_mx.provider_id:
s_obj = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if s_obj: s_tax = (s_obj.rfc or "")[:12]
# IV01 uses 11 spaces then 'C' per Clarion logic
# Adjusted validation for RFCs to avoid crashes or None
s_tax_safe = s_tax if s_tax else " "
c_tax_safe = c_tax if c_tax else " "
lineas.append(
f"IV01{factura.invoice_number[:15]:<15}{f_fecha}78{' ':<12}C"
f"{empresa_dict['broker'][:6]:<6}{flete:08d}{s_tax_safe:<13}{c_tax_safe:<13}"
)
self.cuenta_partidas += 1
# IV02: Vendor/Manufacturer Info
# Clarion: CliVen:ManufacterID, CliVen:Nombre
manufacturer_id = ""
vendor_name = ""
if factura.compliance_mx and factura.compliance_mx.provider_id:
vendor = db.query(ClientProvider).options(
joinedload(ClientProvider.programs)
).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if vendor:
manufacturer_id = (vendor.programs.manufacturer_id or "")[:16] if vendor.programs else ""
vendor_name = (vendor.name or "")[:39]
lineas.append(f"IV02{manufacturer_id:<16}{vendor_name:<39}")
self.cuenta_partidas += 1
# IV10: Responsible
lineas.append(f"IV10 {main_activity[:30]:<30}{empresa_dict['responsable'][:30]:<30}")
self.cuenta_partidas += 1
# IV11: Obs
lineas.append(f"IV11H")
self.cuenta_partidas += 1
obs_line = f"IV11F{(factura.notes or '')[:70]:<70}" if request.iv11 else "IV11F"
lineas.append(obs_line)
self.cuenta_partidas += 1
# Participants IV12-14 (S, C, T, I)
# Shipper (S)
if factura.compliance_mx and factura.compliance_mx.provider_id:
s_cliente = db.query(ClientProvider).options(
joinedload(ClientProvider.address),
joinedload(ClientProvider.programs)
).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if s_cliente:
s_data = self._obtener_datos_cliente(s_cliente)
broker_impo = s_data['broker'] or empresa_dict['broker']
lineas.append(f"IV12S{broker_impo[:6]:<6}{s_data['nombre'][:39]:<39}")
# Ensure spacing aligns with Clarion example (30 spaces + 35 address + space + CP)
lineas.append(f"IV13S{'':<30}{s_data['calle']:<35} {s_data['cp']:<9}")
# IV14: City(20)+State(2)+Country(2)+Phone(30?? No, Clarion example shows Phone then TaxID)
# Clarion example: IV14S... CITY... STMX... TAXID... TEL...
# Re-aligning based on provided example:
# IV14SAKRON NEUS 16-0919851 00000
# City (20) State (2) Country (2) Space(30) TaxID(15) Tel(5?)
lineas.append(f"IV14S{s_data['ciudad']:<20}{s_data['estado']:<2}{s_data['pais']:<2}{'':<30}{s_data['tax_id']:<15}{s_data['tel'][:5]:<5}")
self.cuenta_partidas += 3
# Consignee (C), Ship To (T), and Importer (I)
if factura.compliance_mx and factura.compliance_mx.sold_to_id:
c_cliente = db.query(ClientProvider).options(
joinedload(ClientProvider.address),
joinedload(ClientProvider.programs)
).filter(ClientProvider.id == factura.compliance_mx.sold_to_id).first()
if c_cliente:
c_data = self._obtener_datos_cliente(c_cliente)
broker_impo = c_data['broker'] or empresa_dict['broker']
# Consignee (C)
lineas.append(f"IV12C{broker_impo[:6]:<6}{c_data['nombre'][:39]:<39}")
lineas.append(f"IV13C{'':<30}{c_data['calle']:<35} {c_data['cp']:<9}")
lineas.append(f"IV14C{c_data['ciudad']:<20}{c_data['estado']:<2}{c_data['pais']:<2}{'':<30}{c_data['tax_id']:<15}{c_data['tel'][:5]:<5}")
self.cuenta_partidas += 3
# Ship To (T) - check if different
t_id = factura.compliance_mx.shipped_to_id
if t_id and t_id != factura.compliance_mx.sold_to_id:
t_cl = db.query(ClientProvider).options(
joinedload(ClientProvider.address),
joinedload(ClientProvider.programs)
).filter(ClientProvider.id == t_id).first()
if t_cl:
t_data = self._obtener_datos_cliente(t_cl)
b_t = t_data['broker'] or empresa_dict['broker']
lineas.append(f"IV12T{b_t[:6]:<6}{t_data['nombre'][:39]:<39}")
lineas.append(f"IV13T{'':<30}{t_data['calle']:<35} {t_data['cp']:<9}")
lineas.append(f"IV14T{t_data['ciudad']:<20}{t_data['estado']:<2}{t_data['pais']:<2}{'':<30}{t_data['tax_id']:<15}{t_data['tel'][:5]:<5}")
self.cuenta_partidas += 3
# Importer (I) - Usually same as Consignee in Definitive unless ShippedBy is set
i_id = factura.compliance_mx.shipped_by_id
if i_id and i_id != factura.compliance_mx.sold_to_id and i_id != t_id:
i_cl = db.query(ClientProvider).options(
joinedload(ClientProvider.address),
joinedload(ClientProvider.programs)
).filter(ClientProvider.id == i_id).first()
if i_cl:
i_data = self._obtener_datos_cliente(i_cl)
b_i = i_data['broker'] or empresa_dict['broker']
lineas.append(f"IV12I{b_i[:6]:<6}{i_data['nombre'][:39]:<39}")
lineas.append(f"IV13I{'':<30}{i_data['calle']:<35} {i_data['cp']:<9}")
lineas.append(f"IV14I{i_data['ciudad']:<20}{i_data['estado']:<2}{i_data['pais']:<2}{'':<30}{i_data['tax_id']:<15}{i_data['tel'][:5]:<5}")
self.cuenta_partidas += 3
else:
# Fallback: repeat C as I if not specified (following Clarion pattern)
lineas.append(f"IV12I{broker_impo[:6]:<6}{c_data['nombre'][:39]:<39}")
lineas.append(f"IV13I{'':<30}{c_data['calle']:<35} {c_data['cp']:<9}")
lineas.append(f"IV14I{c_data['ciudad']:<20}{c_data['estado']:<2}{c_data['pais']:<2}{'':<30}{c_data['tax_id']:<15}{c_data['tel'][:5]:<5}")
self.cuenta_partidas += 3
# Items IV20-27
items_headers = db.query(LineItem).filter(LineItem.invoice_id == factura.id).all()
item_ids = [ih.id for ih in items_headers]
if item_ids:
items_query = db.query(LineItem).filter(
LineItem.id.in_(item_ids)
).options(
joinedload(LineItem.part_info),
joinedload(LineItem.description),
joinedload(LineItem.financial),
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.unit_of_measure_info)
).all()
for line in items_query:
f_consec_partidas += 1
part_num = (line.part_info.part_number if line.part_info else "S/N")[:25]
po = (line.order or "")[:20] if line.financial else ""
# IV20
lineas.append(f"IV20{f_consec_partidas:03d} {part_num:<25}C {po:<20}")
self.cuenta_partidas += 1
# IV21
desc_ingles = (line.description.description_english or "")[:50] if line.description else ""
lineas.append(f"IV21 {desc_ingles:<50}")
self.cuenta_partidas += 1
# Data for IV22
val_me = float(line.financial.value_usd or 0) if line.financial else 0.0
qty = float(line.quantity.quantity or 0) if line.quantity else 0.0
pb = float(line.quantity.gross_weight or 0) if line.quantity else 0.0
pn = float(line.quantity.net_weight or 0) if line.quantity else 0.0
costo_u = float(line.financial.unit_cost_capture or 0) if line.financial else 0.0
um = (line.unit_of_measure_info.american_code or "PCS")[:3] if line.unit_of_measure_info else "PCS"
pais = ((line.customs.origin_country or "MX")[:2]).upper() if line.customs else "MX"
hts = (line.customs.american_fraction or "").replace(".", "")[:10] if line.customs else ""
# Formato Clarion @n...v...
val_int = int(round(val_me * 100)) # @n010v2
qty_int = int(round(qty * 100)) # @n09v2
pb_int = int(round(pb * 100)) # @n09v2
pn_int = int(round(pn * 100)) # @n09v2
costo_int = int(round(costo_u * 100000)) # @n011v5
prog_ind = "S" if line.customs and line.customs.has_origin_certificate else "N"
# IV22
lineas.append(
f"IV22{hts:<10}{prog_ind}"
f"{val_int:010d}{um:<3}{qty_int:09d}{pais:<2}0000010000000000100000 "
f"{pb_int:09d}{pn_int:09d}"
)
self.cuenta_partidas += 1
# IV24, IV26 (Zeros per logic)
lineas.append(f"IV24 0000000000 000000000 0000000000000000000000")
lineas.append(f"IV26 0000000000 000000000 0000000000000000000000")
self.cuenta_partidas += 2
# IV27
lineas.append(f"IV27{hts:<10}{costo_int:011d}0000000000000000000000000000000000000 ")
self.cuenta_partidas += 1
# IV42 (Optional)
if request.iv42:
um4 = um[:4]
lineas.append(f"IV42 0{qty_int:09d}{um4:<4}0000000000 0000000000 0000000000 0000000000 ")
self.cuenta_partidas += 1
f_val_total += val_me
f_pb += pb
f_pn += pn
# IV90: Footer per Invoice
fv_int = int(round(f_val_total * 100))
fpb_int = int(round(f_pb * 100))
fpn_int = int(round(f_pn * 100))
lineas.append(f"IV90{f_consec_partidas:05d}{fv_int:012d}{fpb_int:010d}{fpn_int:010d}")
self.cuenta_partidas += 1
self.valor_total_factura += f_val_total
self.peso_bruto_factura += f_pb
self.peso_neto_factura += f_pn
return lineas, self.errores

View File

@@ -0,0 +1,41 @@
from typing import Dict, Any
from fastapi import APIRouter, Depends, Body
from celery.result import AsyncResult
from core.celery_app import celery_app
from core.security import get_current_user
from .task import generar_transmission_definitiva_async
from .schemas import Mainx30GenerationRequest
router = APIRouter()
@router.get("/tasks/{task_id}")
async def get_task_status(
task_id: str,
current_user: Dict[str, Any] = Depends(get_current_user)
):
task_result = AsyncResult(task_id, app=celery_app)
response = {
"task_id": task_id,
"state": task_result.state,
"result": None,
"info": None
}
if task_result.state == 'FAILURE':
response["result"] = str(task_result.result)
elif task_result.state == 'SUCCESS':
response["result"] = task_result.result
elif task_result.state == 'PROCESSING':
response["info"] = task_result.info
return response
@router.post("/generate")
async def trigger_generation(
request: Mainx30GenerationRequest,
current_user: Dict[str, Any] = Depends(get_current_user)
):
tenant_id = current_user.get("tenant_id")
task = generar_transmission_definitiva_async.delay(request.model_dump(), tenant_id)
return {"task_id": task.id, "message": "Generación Definitiva iniciada"}

View File

@@ -0,0 +1,72 @@
from typing import List, Optional, Any
from pydantic import BaseModel, Field
class Mainx30GenerationRequest(BaseModel):
"""
Schema for the Mainx30 file generation request for Definitive Import
"""
manifiestos: Optional[List[str]] = Field(None, description="Lista de números de manifiesto a procesar")
facturas: Optional[List[str]] = Field(None, description="Lista de números de factura a procesar")
entry_port: Optional[str] = Field(None, description="Puerto de entrada")
exit_port: Optional[str] = Field(None, description="Puerto de salida")
regimen: Optional[str] = Field("Definitiva", description="Regimen de importación (Temporal/Definitiva)")
nomenclatura_factura: bool = Field(False, description="Usar nomenclatura basada en factura")
consolidar_rbs: bool = Field(False, description="Consolidar por fracción RB System")
emanifest_fast_blanco: bool = Field(False, description="E-Manifest y FAST en blanco")
no_enviar_emanifest: bool = Field(False, description="No enviar E-Manifest")
consolidar_partidas: bool = Field(False, description="Consolidar partidas (XML OPTIMA Y RBS2)")
main_x40_emanifest: bool = Field(False, description="Main X40 E-Manifest")
main_x30_fedex: bool = Field(False, description="Main X30 (FEDEX)")
iv11: bool = Field(False, description="IV11")
iv42: bool = Field(False, description="IV42")
class ErrorValidacion(BaseModel):
"""
Schema for validation errors during file generation
"""
partida: int
linea: int
descripcion: str
soluciones: str
identificador: str
campos: str = ""
campos2: str = ""
class Mainx30Response(BaseModel):
"""
Schema for the generation response
"""
success: bool
message: str
task_id: Optional[str] = None
archivo_generado: Optional[str] = None
ruta_archivo: Optional[str] = None
content: Optional[str] = None
# Statistics
cuenta_partidas: int = 0
valor_total: float = 0.0
flete_total: float = 0.0
peso_bruto_total: float = 0.0
peso_neto_total: float = 0.0
cuenta_facturas: int = 0
# Validation
errores: List[ErrorValidacion] = []
tiene_inconsistencias: bool = False
class EmpresaDatos(BaseModel):
broker: str
responsable: str
rfc: str
tiene_linea_express: str
nombre_empresa: str = ""
manufacturer_id: str = ""
ftp_key: str = "00SCSI"
main_activity: str = "RAW MATERIAL"
city_state: str = ""
class ConfiguracionSistema(BaseModel):
path_arch_transmision: str
utilizar_nombre_generico_mainx30: bool
utilizar_codigo_broker_cliente: bool

View File

@@ -0,0 +1,175 @@
import os
import tempfile
from datetime import date, datetime
from typing import List, Tuple, Optional, Dict, Any
from pathlib import Path
from sqlalchemy.orm import Session, joinedload
from fastapi import HTTPException
from .schemas import (
Mainx30GenerationRequest, Mainx30Response, ErrorValidacion,
EmpresaDatos, ConfiguracionSistema
)
# --- MODELOS A76 ---
from api.v1.modules.a76.general_catalogs.company.models import Company as GEmpresa
from api.v1.modules.a76.general_catalogs.ports.models import Port
# --- PROCESADORES ---
from .processors import ScafDefProcessor
class Mainx30DefinitiveService:
def __init__(self):
self.errores_validacion: List[ErrorValidacion] = []
self.cuenta_partidas = 0
self.cuenta_facturas = 0
self.valor_total = 0.0
self.flete_total = 0.0
self.peso_bruto_total = 0.0
self.peso_neto_total = 0.0
def generar_mainx30(
self,
db: Session,
request: Mainx30GenerationRequest,
task_instance=None
) -> Mainx30Response:
try:
self._inicializar_variables()
# 1. Obtener Datos de Empresa
datos_empresa = self._obtener_datos_company(db)
emp_dict = datos_empresa.model_dump()
# 2. Obtener Descripciones de Puertos
if request.entry_port:
p_ent = db.query(Port).filter(Port.port_code == request.entry_port).first()
if p_ent: emp_dict['entry_port_desc'] = p_ent.description or p_ent.location_description or ""
if request.exit_port:
p_sal = db.query(Port).filter(Port.port_code == request.exit_port).first()
if p_sal: emp_dict['exit_port_desc'] = p_sal.description or p_sal.location_description or ""
# 3. Fecha de Transmisión (YYMMDD)
fecha_transmision = datetime.now().strftime("%y%m%d")
# 4. Determinar Procesador (Always ScafDefProcessor for this service)
processor = ScafDefProcessor()
# Both 'Definitiva' and 'DEFINITIVO SCAF' use the same heavy processor
# 5. Procesar Facturas
if not request.facturas:
raise HTTPException(status_code=400, detail="No se proporcionaron facturas para procesar.")
l_facturas, e_facturas = processor.procesar_facturas(db, request.facturas, emp_dict, request)
self.errores_validacion.extend(e_facturas)
# 6. Construir Líneas del Archivo
lineas = []
# Línea A
broker = (datos_empresa.broker or "")[:6]
# Clarion uses Broker twice in 'A' record for Definitive
ftp_key = (datos_empresa.ftp_key or "00SCSI")[:6]
lineas.append(f"A {fecha_transmision}03{broker:<6}{broker:<10}{ftp_key:<6}")
self.cuenta_partidas += 1
# Agregar Líneas de Facturas
lineas.extend(l_facturas)
self.cuenta_partidas += processor.cuenta_partidas
self.cuenta_facturas = processor.cuenta_facturas
self.valor_total = processor.valor_total_factura
self.peso_bruto_total = processor.peso_bruto_factura
self.peso_neto_total = processor.peso_neto_factura
self.flete_total = processor.flete_total
# MF80 (Totales Globales)
val_int = int(round(self.valor_total * 100))
pb_int = int(round(self.peso_bruto_total * 10000))
pn_int = int(round(self.peso_neto_total * 10000))
flete_int = int(round(self.flete_total * 100))
lineas.append(
f"MF80{val_int:012d}"
f"{self.cuenta_facturas:04d}"
f"{pb_int:012d}"
f"{flete_int:08d}"
f"{pn_int:012d}"
)
self.cuenta_partidas += 1
# Línea Z (Total de líneas)
lineas.append(f"Z {self.cuenta_partidas:05d}")
self.cuenta_partidas += 1
# 7. Generar Nombre y Guardar
nombre_archivo = f"{request.facturas[0][:15]}_Mainx30.dat"
if len(request.facturas) > 1:
nombre_archivo = f"MULTIPLE_Mainx30.dat"
if request.nomenclatura_factura and len(request.facturas) == 1:
nombre_archivo = f"{request.facturas[0][:15]}_Mainx30.dat"
content = '\r\n'.join(lineas)
return Mainx30Response(
success=len(self.errores_validacion) == 0,
message="Archivo generado" if len(self.errores_validacion) == 0 else "Archivo generado con errores de validación",
archivo_generado=nombre_archivo,
ruta_archivo="",
content=content,
errores_validacion=self.errores_validacion,
cuenta_partidas=self.cuenta_partidas,
valor_total=self.valor_total,
flete_total=self.flete_total,
peso_bruto_total=self.peso_bruto_total,
peso_neto_total=self.peso_neto_total,
cuenta_facturas=self.cuenta_facturas,
tiene_inconsistencias=len(self.errores_validacion) > 0
)
except HTTPException:
raise
except Exception as e:
import traceback
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Error generando Mainx30 Definitivo: {str(e)}")
def _inicializar_variables(self):
self.errores_validacion = []
self.cuenta_partidas = 0
self.valor_total = 0.0
self.flete_total = 0.0
self.peso_bruto_total = 0.0
self.peso_neto_total = 0.0
self.cuenta_facturas = 0
def _obtener_datos_company(self, db: Session) -> EmpresaDatos:
empresa = db.query(GEmpresa).options(joinedload(GEmpresa.addresses)).first()
if not empresa:
return EmpresaDatos(broker="", responsable="", rfc="", tiene_linea_express="N", nombre_empresa="", manufacturer_id="", ftp_key="", main_activity="", city_state="")
city_state = ""
main_addr = next((a for a in (empresa.addresses or []) if a.address_type == 'main'), None)
if not main_addr and empresa.addresses:
main_addr = empresa.addresses[0]
if main_addr:
cp = (main_addr.postal_code or "")[:5]
city = (main_addr.city or "")[:11]
state = (main_addr.state or "")[:4]
city_state = f"{cp:<5}{city:<11}{state:<4}"
return EmpresaDatos(
broker=(empresa.broker_company or "")[:5],
responsable=(empresa.responsible or "")[:30],
rfc=(empresa.rfc or "")[:13],
tiene_linea_express=empresa.has_express_line or "N",
nombre_empresa=(empresa.name or "")[:40],
manufacturer_id=(empresa.manufacturer_id or "")[:10],
ftp_key=(empresa.ftp_key or "")[:10],
main_activity=(empresa.main_activity or "")[:30],
city_state=city_state[:30]
)

View File

@@ -0,0 +1,45 @@
from core.celery_app import celery_app
from core.database import get_core_db as get_db
from .service import Mainx30DefinitiveService
from .schemas import Mainx30GenerationRequest
@celery_app.task(name="generar_transmission_definitiva_async", bind=True)
def generar_transmission_definitiva_async(self, request_data: dict, tenant_id: int):
"""
Celery task to generate Mainx30 file for Definitive Import
"""
try:
# Re-create db session for task
db = next(get_db())
# Deserialize request
request = Mainx30GenerationRequest(**request_data)
service = Mainx30DefinitiveService()
response = service.generar_mainx30(db, request, task_instance=self)
# Return result as dict for Celery serialization
result = response.model_dump()
# Add base64 encoding for content to match frontend expectations
if response.content:
import base64
encoded_content = base64.b64encode(response.content.encode('utf-8')).decode('utf-8')
result['content'] = encoded_content
result['file_name'] = response.archivo_generado
result['media_type'] = "text/plain"
return result
except Exception as e:
import traceback
traceback_print = traceback.format_exc()
self.update_state(
state='FAILURE',
meta={
'exc_type': type(e).__name__,
'exc_message': str(e),
'traceback': traceback_print
}
)
raise e

View File

@@ -0,0 +1,427 @@
from typing import List, Dict, Any, Tuple
from sqlalchemy.orm import Session, joinedload
from datetime import datetime
from .schemas import Mainx30GenerationRequest, ErrorValidacion
# --- MODELOS A76 ---
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
class ScaiiProcessor:
def __init__(self):
self.cuenta_partidas = 0
self.cuenta_facturas = 0
self.valor_total_factura = 0.0
self.flete_total = 0.0
self.peso_bruto_factura = 0.0
self.peso_neto_factura = 0.0
self.errores: List[ErrorValidacion] = []
def _obtener_datos_cliente(self, cliente: ClientProvider) -> dict:
"""Extrae de manera segura los datos del cliente/dirección"""
address = cliente.address
pais_raw = (address.country or "MX").upper() if address else "MX"
pais = "MX"
if pais_raw in ["MEXICO", "MEX", "MX"]:
pais = "MX"
elif pais_raw in ["USA", "US", "UNITED STATES"]:
pais = "US"
else:
pais = pais_raw[:2]
tax_id = cliente.rfc or ""
data = {
"nombre": (cliente.name or "")[:39],
"tax_id": tax_id[:15],
"broker": "", "calle": "", "cp": "", "ciudad": "", "estado": "", "pais": pais, "tel": ""
}
if cliente.programs:
data["broker"] = (cliente.programs.broker or "")[:6]
if address:
calle_comp = f"{address.streets or ''} {address.exterior_number or ''}".strip()
# Clarion expects 20 chars for city_state: 5 CP + 11 City + 4 State
cp_formatted = (address.postal_code or "")[:5]
city_formatted = (address.city or "")[:11]
state_formatted = (address.state or "")[:4]
data["city_state"] = f"{cp_formatted:<5}{city_formatted:<11}{state_formatted:<4}"
data["calle"] = calle_comp[:35]
data["cp"] = (address.postal_code or "")[:9]
data["ciudad"] = (address.city or "")[:20]
data["estado"] = (address.state or "")[:2].upper()
data["tel"] = (address.phone or "")[:15]
return data
def procesar_facturas(
self, db: Session, manifiesto: str, empresa_dict: Dict[str, Any], request: Mainx30GenerationRequest
) -> Tuple[List[str], List[ErrorValidacion]]:
# This base method is used for Manifest-based processing (Exportacion/Legacy)
lineas = []
self.errores = []
return lineas, self.errores
def _agregar_error(self, partida, id_err, desc, sol, tipo):
self.errores.append(ErrorValidacion(partida=partida, linea=0, descripcion=f"[ {id_err} ] {desc}", soluciones=sol, identificador=tipo))
# --- PROCESADORES ESPECIFICOS ---
class ScafDefProcessor(ScaiiProcessor):
"""Procesador para Importación Definitiva"""
def __init__(self):
super().__init__()
def procesar_facturas(
self, db: Session, facturas_nums: List[str], empresa_dict: Dict[str, Any], request: Mainx30GenerationRequest
) -> Tuple[List[str], List[ErrorValidacion]]:
# Placeholder for Definitiva logic
lineas = []
self.errores = []
return lineas, self.errores
class ScafTempProcessor(ScaiiProcessor):
"""Procesador para Importación Temporal basado en lógica Clarion"""
def __init__(self):
super().__init__()
def procesar_facturas(
self, db: Session, facturas_nums: List[str], empresa_dict: Dict[str, Any], request: Mainx30GenerationRequest
) -> Tuple[List[str], List[ErrorValidacion]]:
lineas = []
self.errores = []
# 1. Traer Facturas por número (Importación Temporal trabaja por factura)
facturas = db.query(InvoiceHeader).join(
InvoiceComplianceMx, InvoiceHeader.id == InvoiceComplianceMx.invoice_id
).options(
joinedload(InvoiceHeader.financials),
joinedload(InvoiceHeader.compliance_mx),
joinedload(InvoiceHeader.logistics)
).filter(
InvoiceHeader.invoice_number.in_(facturas_nums)
).all()
# Helper for ports from request
entry_port = request.entry_port or ""
exit_port = request.exit_port or ""
# Date for transmission records (YYMMDD)
fecha_trans = datetime.now().strftime("%y%m%d")
# Port descriptions from empresa_dict (populated in service.py)
entry_port_desc = empresa_dict.get('entry_port_desc', 'PUERTO ENTRADA')[:15]
exit_port_desc = empresa_dict.get('exit_port_desc', 'PUERTO SALIDA')[:15]
main_activity = empresa_dict.get('main_activity', 'RAW MATERIAL')[:30]
city_state = empresa_dict.get('city_state', '')[:30]
for factura in facturas:
self.cuenta_facturas += 1
f_val_total = 0.0
f_pb = 0.0
f_pn = 0.0
f_consec_partidas = 0
# MF01: Header per Invoice in Importacion Temporal
mod_trans = factura.logistics.transport_mode if factura.logistics else "30"
f_fecha = factura.invoice_date.strftime("%y%m%d") if factura.invoice_date else fecha_trans
lineas.append(
f"MF01{empresa_dict['broker'][:6]:<6}"
f"{exit_port[:5]:<5}"
f"{entry_port[:5]:<5}"
f"{fecha_trans}"
f" {mod_trans[:2]:<2}"
f"{f_fecha}"
f"{factura.invoice_number[:15]:<15}"
)
self.cuenta_partidas += 1
# MF03: Dynamic Driver/Carrier info
conductor = (factura.logistics.driver_name or "") if factura.logistics else ""
# Carrier Logic from Clarion:
# IF ERRORCODE() = 35 THEN Loc:NumTransporte = '00000TRUCK'
# ELSE IF GenTra:NombreCorto = '' THEN Loc:NumTransporte = GenTra:Nombre
# ELSE Loc:NumTransporte = GenTra:NombreCorto
carrier = "00000TRUCK"
if factura.logistics:
# Logic simplified: assume carrier_id holds the correct code/name or fallback
carrier = (factura.logistics.carrier_id or "00000TRUCK")
lineas.append(
f"MF03{carrier[:10]:<10}{conductor[:23]:<23}{entry_port_desc:<15}{exit_port_desc:<15}"
)
self.cuenta_partidas += 1
# MF20
num_transporte = (factura.logistics.transport_num or "") if factura.logistics else ""
lineas.append(
f"MF20{factura.invoice_number[:15]:<15}I{num_transporte[:15]:<15}{city_state[:20]:<20}"
f"{exit_port[:5]:<5}{entry_port_desc:<15} "
)
self.cuenta_partidas += 1
# MF22
lineas.append(f"MF22{main_activity:<60}")
self.cuenta_partidas += 1
# IV01: Header
f_fecha = factura.invoice_date.strftime("%y%m%d") if factura.invoice_date else "000000"
flete = int(round(float(factura.financials.freight or 0))) if factura.financials else 0
self.flete_total += float(factura.financials.freight or 0) if factura.financials else 0.0
s_tax = ""; c_tax = ""
if factura.compliance_mx:
if factura.compliance_mx.sold_to_id:
c_obj = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.sold_to_id).first()
if c_obj: c_tax = c_obj.rfc[:12] if c_obj.rfc else ""
if factura.compliance_mx.provider_id:
s_obj = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if s_obj: s_tax = s_obj.rfc[:12] if s_obj.rfc else ""
lineas.append(
f"IV01{factura.invoice_number[:15]:<15}{f_fecha}{entry_port:<5}{' ':<11}C"
f"{empresa_dict['broker'][:6]:<6}{flete:08d}{c_tax:<13}"
)
self.cuenta_partidas += 1
# IV02: Shipper Name
lineas.append(f"IV02 {empresa_dict['nombre_empresa'][:40]:<40}")
self.cuenta_partidas += 1
# IV10: Responsible
lineas.append(f"IV10 {main_activity[:30]:<30}{empresa_dict['responsable'][:30]:<30}")
self.cuenta_partidas += 1
# IV11: Obs
lineas.append(f"IV11H")
lineas.append(f"IV11F")
self.cuenta_partidas += 2
# IV12-14 (S, C, T, I)
# Shipper (S)
if factura.compliance_mx and factura.compliance_mx.provider_id:
s_cliente = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if s_cliente:
s_data = self._obtener_datos_cliente(s_cliente)
lineas.append(f"IV12S {s_data['nombre'][:39]:<39}")
lineas.append(f"IV13S {s_data['calle'][:35]:<35}{s_data['cp']:<9}")
lineas.append(f"IV14S{s_data['ciudad'][:20]:<20}{s_data['state_full'][:2] if 'state_full' in s_data else s_data['estado'][:2]}{s_data['pais'][:2]}{s_data['tel'][:30]:<30}{s_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# Consignee (C), Ship To (T), and Intermediate (I)
if factura.compliance_mx and factura.compliance_mx.sold_to_id:
c_cliente = db.query(ClientProvider).filter(ClientProvider.id == factura.compliance_mx.sold_to_id).first()
if c_cliente:
c_data = self._obtener_datos_cliente(c_cliente)
l12c = f"IV12C {c_data['nombre'][:39]:<39}"
l13c = f"IV13C {c_data['calle'][:35]:<35}{c_data['cp']:<9}"
l14c = f"IV14C{c_data['ciudad'][:20]:<20}{c_data['estado'][:2]}{c_data['pais'][:2]}{c_data['tel'][:30]:<30}{c_data['tax_id']:<15}00000"
# Output C
lineas.extend([l12c, l13c, l14c])
self.cuenta_partidas += 3
# T (Ship To) - Only if different from C
t_id = factura.compliance_mx.shipped_to_id
if t_id and t_id != factura.compliance_mx.sold_to_id:
t_cl = db.query(ClientProvider).filter(ClientProvider.id == t_id).first()
if t_cl:
t_data = self._obtener_datos_cliente(t_cl)
lineas.append(f"IV12T {t_data['nombre'][:39]:<39}")
lineas.append(f"IV13T {t_data['calle'][:35]:<35}{t_data['cp']:<9}")
lineas.append(f"IV14T{t_data['ciudad'][:20]:<20}{t_data['estado'][:2]}{t_data['pais'][:2]}{t_data['tel'][:30]:<30}{t_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# I (Intermediate) - Only if different from C and T
i_id = factura.compliance_mx.shipped_by_id
if i_id and i_id != factura.compliance_mx.sold_to_id and i_id != t_id:
i_cl = db.query(ClientProvider).filter(ClientProvider.id == i_id).first()
if i_cl:
i_data = self._obtener_datos_cliente(i_cl)
lineas.append(f"IV12I {i_data['nombre'][:39]:<39}")
lineas.append(f"IV13I {i_data['calle'][:35]:<35}{i_data['cp']:<9}")
lineas.append(f"IV14I{i_data['ciudad'][:20]:<20}{i_data['estado'][:2]}{i_data['pais'][:2]}{i_data['tel'][:30]:<30}{i_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# Partidas IV20, IV21, IV22, IV24, IV26, IV27
# Let's get Item IDs first to ensure we find them
items_headers = db.query(LineItem).filter(LineItem.invoice_id == factura.id).all()
item_ids = [ih.id for ih in items_headers]
if item_ids:
items_query = db.query(LineItem).filter(
LineItem.id.in_(item_ids)
).options(
joinedload(LineItem.part_info),
joinedload(LineItem.description),
joinedload(LineItem.financial),
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.unit_of_measure_info)
).all()
for line in items_query:
f_consec_partidas += 1
part_num = line.part_info.part_number if line.part_info else "S/N"
# IV20
lineas.append(f"IV20{f_consec_partidas:03d} {part_num[:25]:<25}C")
self.cuenta_partidas += 1
# IV21
lineas.append(f"IV21")
self.cuenta_partidas += 1
# Data
val_me = float(line.financial.value_usd or 0) if line.financial else 0.0
qty = float(line.quantity.quantity or 0) if line.quantity else 0.0
pb = float(line.quantity.gross_weight or 0) if line.quantity else 0.0
pn = float(line.quantity.net_weight or 0) if line.quantity else 0.0
costo_u = float(line.financial.unit_cost_capture or 0) if line.financial else 0.0
um = line.unit_of_measure_info.american_code if line.unit_of_measure_info else "PCS"
pais = (line.customs.origin_country or "MX")[:2] if line.customs else "MX"
hts = (line.customs.american_fraction or "").replace(".", "")[:10] if line.customs else ""
val_int = int(round(val_me * 10000))
qty_int = int(round(qty * 10000))
pb_int = int(round(pb * 10000))
pn_int = int(round(pn * 10000))
costo_int = int(round(costo_u * 100000))
# IV22
lineas.append(
f"IV22{hts:<10} "
f"{val_int:010d}{um[:3]:<3}{qty_int:09d}{pais:<2}0000010000000000100000 "
f"{pb_int:010d}{pn_int:010d}"
)
self.cuenta_partidas += 1
# IV24, IV26 (Zeros)
lineas.append(f"IV24 0000000000 000000000 0000000000000000000000")
lineas.append(f"IV26 0000000000 000000000 0000000000000000000000")
self.cuenta_partidas += 2
# IV27
lineas.append(f"IV27{hts:<10}{costo_int:011d}0000000000000000000000000000000000000")
self.cuenta_partidas += 1
f_val_total += val_me
f_pb += pb
f_pn += pn
# IV90: Footer per Invoice
# IV90
f_val_int = int(round(f_val_total * 100))
f_pb_int = int(round(f_pb * 10000))
f_pn_int = int(round(f_pn * 10000))
lineas.append(f"IV90{f_consec_partidas:05d}{f_val_int:012d}{f_pb_int:010d}{f_pn_int:010d}")
self.cuenta_partidas += 1
self.valor_total_factura += f_val_total
self.peso_bruto_factura += f_pb
self.peso_neto_factura += f_pn
return lineas, self.errores
class ScaiiTempProcessor(ScafTempProcessor):
"""
Procesador para Importación Temporal (Versión Ligera/SCAII)
Se salta los registros MF20, MF22, IV10-14 para mayor velocidad y menor detalle.
"""
def procesar_facturas(
self, db: Session, facturas_nums: List[str], empresa_dict: Dict[str, Any], request: Mainx30GenerationRequest
) -> Tuple[List[str], List[ErrorValidacion]]:
lineas = []
self.errores = []
# 1. Traer Facturas por número
facturas = db.query(InvoiceHeader).join(
InvoiceComplianceMx, InvoiceHeader.id == InvoiceComplianceMx.invoice_id
).options(
joinedload(InvoiceHeader.financials),
joinedload(InvoiceHeader.compliance_mx),
joinedload(InvoiceHeader.logistics)
).filter(
InvoiceHeader.invoice_number.in_(facturas_nums)
).all()
entry_port = request.entry_port or ""
exit_port = request.exit_port or ""
fecha_trans = datetime.now().strftime("%y%m%d")
entry_port_desc = empresa_dict.get('entry_port_desc', 'PUERTO ENTRADA')[:15]
exit_port_desc = empresa_dict.get('exit_port_desc', 'PUERTO SALIDA')[:15]
for factura in facturas:
self.cuenta_facturas += 1
f_val_total = 0.0
f_consec_partidas = 0
# MF01
mod_trans = factura.logistics.transport_mode if factura.logistics else "30"
f_fecha = factura.invoice_date.strftime("%y%m%d") if factura.invoice_date else fecha_trans
lineas.append(
f"MF01{empresa_dict['broker'][:6]:<6}{exit_port[:5]:<5}{entry_port[:5]:<5}"
f"{fecha_trans} {mod_trans[:2]:<2}{f_fecha}{factura.invoice_number[:15]:<15}"
)
self.cuenta_partidas += 1
# MF03
conductor = (factura.logistics.driver_name or "") if factura.logistics else ""
carrier = (factura.logistics.carrier_id or "00000TRUCK")
lineas.append(f"MF03{carrier[:10]:<10}{conductor[:23]:<23}{entry_port_desc:<15}{exit_port_desc:<15}")
self.cuenta_partidas += 1
# IV01
f_fecha_iv = factura.invoice_date.strftime("%y%m%d") if factura.invoice_date else "000000"
flete = int(round(float(factura.financials.freight or 0))) if factura.financials else 0
self.flete_total += float(factura.financials.freight or 0) if factura.financials else 0.0
lineas.append(
f"IV01{factura.invoice_number[:15]:<15}{f_fecha_iv}{entry_port:<5}{' ':<11}C"
f"{empresa_dict['broker'][:6]:<6}{flete:08d}{'':<13}"
)
self.cuenta_partidas += 1
# Partidas IV20, IV21, IV22
items_headers = db.query(LineItem).filter(LineItem.invoice_id == factura.id).all()
item_ids = [ih.id for ih in items_headers]
if item_ids:
items_query = db.query(LineItem).filter(LineItem.id.in_(item_ids)).all()
for line in items_query:
f_consec_partidas += 1
part_num = line.part_info.part_number if line.part_info else "S/N"
# IV20
lineas.append(f"IV20{f_consec_partidas:03d} {part_num[:25]:<25}C")
self.cuenta_partidas += 1
# IV22
qty = float(line.quantity.quantity or 0) if line.quantity else 0.0
val_me = float(line.financial.value_usd or 0) if line.financial else 0.0
hts = (line.customs.american_fraction or "").replace(".", "")[:10] if line.customs else ""
val_int = int(round(val_me * 10000))
qty_int = int(round(qty * 10000))
lineas.append(f"IV22{hts:<10} {val_int:010d}PCS{qty_int:09d}MX0000010000000000100000 00000000000000000000")
self.cuenta_partidas += 1
f_val_total += val_me
# IV90
fv_int = int(round(f_val_total * 100))
lineas.append(f"IV90{f_consec_partidas:05d}{fv_int:012d}00000000000000000000")
self.cuenta_partidas += 1
self.valor_total_factura += f_val_total
return lineas, self.errores

View File

@@ -0,0 +1,43 @@
from typing import Dict, Any
from fastapi import APIRouter, Depends, Body
from celery.result import AsyncResult
from core.celery_app import celery_app
from core.security import get_current_user
from .task import generar_transmission_temporal_async
from .schemas import Mainx30GenerationRequest
router = APIRouter()
@router.get("/tasks/{task_id}")
async def get_task_status(
task_id: str,
current_user: Dict[str, Any] = Depends(get_current_user)
):
task_result = AsyncResult(task_id, app=celery_app)
response = {
"task_id": task_id,
"state": task_result.state,
"result": None,
"info": None
}
if task_result.state == 'FAILURE':
response["result"] = str(task_result.result)
elif task_result.state == 'SUCCESS':
response["result"] = task_result.result
elif task_result.state == 'PROCESSING':
# Ensure info is serializable
response["info"] = task_result.info
return response
@router.post("/generate")
async def trigger_generation(
request: Mainx30GenerationRequest,
current_user: Dict[str, Any] = Depends(get_current_user)
):
tenant_id = current_user.get("tenant_id")
# Pass request as dict to Celery task
task = generar_transmission_temporal_async.delay(request.model_dump(), tenant_id)
return {"task_id": task.id, "message": "Generación Temporal iniciada"}

View File

@@ -0,0 +1,77 @@
from typing import List, Optional, Any
from pydantic import BaseModel, Field
class Mainx30GenerationRequest(BaseModel):
"""
Schema for the Mainx30 file generation request
"""
manifiestos: Optional[List[str]] = Field(None, description="Lista de números de manifiesto a procesar")
facturas: Optional[List[str]] = Field(None, description="Lista de números de factura a procesar")
entry_port: Optional[str] = Field(None, description="Puerto de entrada")
exit_port: Optional[str] = Field(None, description="Puerto de salida")
regimen: Optional[str] = Field("Temporal", description="Regimen de importación (Temporal/Definitiva)")
nomenclatura_factura: bool = Field(False, description="Usar nomenclatura basada en factura")
consolidar_rbs: bool = Field(False, description="Consolidar por fracción RB System")
emanifest_fast_blanco: bool = Field(False, description="E-Manifest y FAST en blanco")
no_enviar_emanifest: bool = Field(False, description="No enviar E-Manifest")
consolidar_partidas: bool = Field(False, description="Consolidar partidas (XML OPTIMA Y RBS2)")
main_x40_emanifest: bool = Field(False, description="Main X40 E-Manifest")
main_x30_fedex: bool = Field(False, description="Main X30 (FEDEX)")
iv11: bool = Field(False, description="IV11")
iv42: bool = Field(False, description="IV42")
class ErrorValidacion(BaseModel):
"""
Schema for validation errors during file generation
"""
partida: int
linea: int
descripcion: str
soluciones: str
identificador: str
campos: str = ""
campos2: str = ""
class Mainx30Response(BaseModel):
"""
Schema for the generation response
"""
success: bool
message: str
task_id: Optional[str] = None
archivo_generado: Optional[str] = None
ruta_archivo: Optional[str] = None
content: Optional[str] = None
# Statistics
cuenta_partidas: int = 0
valor_total: float = 0.0
flete_total: float = 0.0
peso_bruto_total: float = 0.0
peso_neto_total: float = 0.0
cuenta_facturas: int = 0
# Validation
errores: List[ErrorValidacion] = []
tiene_inconsistencias: bool = False
class BrokerValidationResult(BaseModel):
es_valido: bool
mensaje_error: Optional[str] = None
broker_cliente: Optional[str] = None
class EmpresaDatos(BaseModel):
broker: str
responsable: str
rfc: str
tiene_linea_express: str
nombre_empresa: str = "AAKRON RULE CORPORATION"
manufacturer_id: str = "I10900"
ftp_key: str = "00SCSI"
main_activity: str = "RAW MATERIAL"
city_state: str = ""
class ConfiguracionSistema(BaseModel):
path_arch_transmision: str
utilizar_nombre_generico_mainx30: bool
utilizar_codigo_broker_cliente: bool

View File

@@ -0,0 +1,177 @@
import os
import tempfile
from datetime import date, datetime
from typing import List, Tuple, Optional, Dict, Any
from pathlib import Path
from sqlalchemy.orm import Session, joinedload
from fastapi import HTTPException
from .schemas import (
Mainx30GenerationRequest, Mainx30Response, ErrorValidacion,
EmpresaDatos, ConfiguracionSistema
)
# --- MODELOS A76 ---
from api.v1.modules.a76.manifests.manifest.models import Manifest
from api.v1.modules.a76.general_catalogs.company.models import Company as GEmpresa
from api.v1.modules.a76.general_catalogs.ports.models import Port
# --- PROCESADORES ---
from .processors import ScaiiProcessor, ScafDefProcessor, ScafTempProcessor, ScaiiTempProcessor
class Mainx30Service:
def __init__(self):
self.errores_validacion: List[ErrorValidacion] = []
self.cuenta_partidas = 0
self.cuenta_facturas = 0
self.valor_total = 0.0
self.flete_total = 0.0
self.peso_bruto_total = 0.0
self.peso_neto_total = 0.0
def generar_mainx30(
self,
db: Session,
request: Mainx30GenerationRequest,
task_instance=None
) -> Mainx30Response:
try:
self._inicializar_variables()
# 1. Obtener Datos de Empresa
datos_empresa = self._obtener_datos_company(db)
emp_dict = datos_empresa.model_dump()
# 2. Obtener Descripciones de Puertos
if request.entry_port:
p_ent = db.query(Port).filter(Port.port_code == request.entry_port).first()
if p_ent: emp_dict['entry_port_desc'] = p_ent.description or p_ent.location_description or ""
if request.exit_port:
p_sal = db.query(Port).filter(Port.port_code == request.exit_port).first()
if p_sal: emp_dict['exit_port_desc'] = p_sal.description or p_sal.location_description or ""
# 3. Fecha de Transmisión (Clarion @D11 = mm/dd/yy, but example uses YYMMDD)
fecha_transmision = datetime.now().strftime("%y%m%d")
# 4. Determinar Procesador
processor = ScafTempProcessor() # Now 'Temporal' defaults to Heavy (SCAF) logic per user request
if request.regimen == "TEMPORAL SCAF":
processor = ScaiiTempProcessor() # 'TEMPORAL SCAF' uses light (SCAII) logic
elif request.regimen == "Definitiva" or request.regimen == "DEFINITIVO SCAF":
processor = ScafDefProcessor()
# 5. Procesar Facturas
if not request.facturas:
raise HTTPException(status_code=400, detail="No se proporcionaron facturas para procesar.")
l_facturas, e_facturas = processor.procesar_facturas(db, request.facturas, emp_dict, request)
self.errores_validacion.extend(e_facturas)
# 6. Construir Líneas del Archivo
lineas = []
# Línea A
broker = (datos_empresa.broker or "")[:6]
ftp_key = (datos_empresa.ftp_key or "00SCSI")[:6]
lineas.append(f"A {fecha_transmision}03{broker:<6}{broker:<10}{ftp_key:<6}")
self.cuenta_partidas += 1
# Agregar Líneas de Facturas
lineas.extend(l_facturas)
self.cuenta_partidas += processor.cuenta_partidas
self.cuenta_facturas = processor.cuenta_facturas
self.valor_total = processor.valor_total_factura
self.peso_bruto_total = processor.peso_bruto_factura
self.peso_neto_total = processor.peso_neto_factura
# MF80 (Totales Globales)
val_int = int(round(self.valor_total * 100))
pb_int = int(round(self.peso_bruto_total * 10000))
pn_int = int(round(self.peso_neto_total * 10000))
flete_int = 0
lineas.append(
f"MF80{val_int:012d}"
f"{self.cuenta_facturas:04d}"
f"{pb_int:012d}"
f"{flete_int:08d}"
f"{pn_int:012d}"
)
self.cuenta_partidas += 1
# Línea Z (Total de líneas)
lineas.append(f"Z {self.cuenta_partidas:05d}")
self.cuenta_partidas += 1
# 7. Generar Nombre y Guardar
nombre_archivo = f"{request.facturas[0][:15]}_Mainx30.dat"
if len(request.facturas) > 1:
nombre_archivo = f"MULTIPLE_Mainx30.dat"
if request.nomenclatura_factura and len(request.facturas) == 1:
nombre_archivo = f"{request.facturas[0][:15]}_Mainx30.dat"
content = '\r\n'.join(lineas)
return Mainx30Response(
success=len(self.errores_validacion) == 0,
message="Archivo generado" if len(self.errores_validacion) == 0 else "Archivo generado con errores de validación",
archivo_generado=nombre_archivo,
ruta_archivo="",
content=content,
errores_validacion=self.errores_validacion,
cuenta_partidas=self.cuenta_partidas,
valor_total=self.valor_total,
peso_bruto_total=self.peso_bruto_total,
peso_neto_total=self.peso_neto_total,
cuenta_facturas=self.cuenta_facturas
)
except HTTPException:
raise
except Exception as e:
import traceback
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Error generando Mainx30: {str(e)}")
def _inicializar_variables(self):
self.errores_validacion = []
self.cuenta_partidas = 0
self.valor_total = 0.0
self.flete_total = 0.0
self.peso_bruto_total = 0.0
self.peso_neto_total = 0.0
self.cuenta_facturas = 0
def _obtener_datos_company(self, db: Session) -> EmpresaDatos:
empresa = db.query(GEmpresa).options(joinedload(GEmpresa.addresses)).first()
if not empresa:
return EmpresaDatos(broker="", responsable="", rfc="", tiene_linea_express="N", nombre_empresa="", manufacturer_id="", ftp_key="", main_activity="", city_state="")
# Get city/state from main address, or first found
city_state = ""
main_addr = next((a for a in (empresa.addresses or []) if a.address_type == 'main'), None)
if not main_addr and empresa.addresses:
main_addr = empresa.addresses[0]
if main_addr:
# Clarion expects 20 chars for city_state: 5 CP + 11 City + 4 State
cp = (main_addr.postal_code or "")[:5]
city = (main_addr.city or "")[:11]
state = (main_addr.state or "")[:4]
city_state = f"{cp:<5}{city:<11}{state:<4}"
return EmpresaDatos(
broker=(empresa.broker_company or "")[:5],
responsable=(empresa.responsible or "")[:30],
rfc=(empresa.rfc or "")[:13],
tiene_linea_express=empresa.has_express_line or "N",
nombre_empresa=(empresa.name or "")[:40],
manufacturer_id=(empresa.manufacturer_id or "")[:10],
ftp_key=(empresa.ftp_key or "")[:10],
main_activity=(empresa.main_activity or "")[:30],
city_state=city_state[:30]
)

View File

@@ -0,0 +1,46 @@
from celery import Task
from core.celery_app import celery_app
from core.celery_app import celery_app
from core.database import get_core_db as get_db
from .service import Mainx30Service
from .schemas import Mainx30GenerationRequest, Mainx30Response
@celery_app.task(name="generar_transmission_temporal_async", bind=True)
def generar_transmission_temporal_async(self, request_data: dict, tenant_id: int):
"""
Generates the transmission .dat file asynchronously using Mainx30Service
"""
try:
# Re-create db session for task
# Using next(get_db()) is a common pattern for obtaining a session in tasks
# but ensure context management
db = next(get_db())
# Deserialize request
request = Mainx30GenerationRequest(**request_data)
service = Mainx30Service()
response = service.generar_mainx30(db, request, task_instance=self)
# Return result as dict for Celery serialization
# Ensure we return valid JSON serializable dict
result = response.model_dump()
# If we returned content directly, encode it if it's bytes (it's str here)
if response.content:
import base64
# Mainx30Service returns content as string with \r\n
encoded_content = base64.b64encode(response.content.encode('utf-8')).decode('utf-8')
# Add to result to match expected format by frontend dialog
result['content'] = encoded_content
result['file_name'] = response.archivo_generado
result['media_type'] = "text/plain"
return result
except Exception as e:
import traceback
traceback.print_exc()
self.update_state(state='FAILURE', meta={'exc_type': type(e).__name__, 'exc_message': str(e)})
# Re-raise to mark task as failed in Celery
raise e