ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Se integro el modulo de importacion definitiva

Browse Source
This commit is contained in:
Kevin_Ramirez
2026-02-16 11:52:27 -06:00
parent 3d691d6881
commit 5d230fc4d7
11 changed files with 822 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/api/v1/modules/a76/reports/exportacion/transmission/MAINX30/service.py
View File

@@ -283,7 +283,7 @@ class Mainx30Service:
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_nombre_archivo(self, c, r, m): return f"{m}_Mainx30.dat"
def _generar_linea_a(self, f, d):
# Sample: A 26021203AKR AKR 00SCSI

333
backend/api/v1/modules/a76/reports/importacion/transmission/definitive/MAINX30/processors.py Normal file
View File

@@ -0,0 +1,333 @@
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 Item
from api.v1.modules.a76.items.line_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.external_id or "")[:15] # Fallback for manufacturer id
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 _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 "00000TRUCK")[: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 ""
# CD. JUAREZ CHIH is often hardcoded in these legacy formats
lineas.append(
f"MF20{factura.invoice_number[:15]:<15}I{num_transporte:<15}20100CD. JUAREZ CHIH"
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
lineas.append(
f"IV01{factura.invoice_number[:15]:<15}{f_fecha}78{' ':<12}C"
f"{empresa_dict['broker'][:6]:<6}{flete:08d}{s_tax:<13}{c_tax:<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).filter(ClientProvider.id == factura.compliance_mx.provider_id).first()
if vendor:
manufacturer_id = (vendor.external_id or "")[:16]
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).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}")
lineas.append(f"IV13S {s_data['calle'][:35]:<35}{s_data['cp']:<9}")
lineas.append(f"IV14S{s_data['ciudad'][:20]:<20}{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 Importer (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)
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 {c_data['calle'][:35]:<35}{c_data['cp']:<9}")
lineas.append(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")
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).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 {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
# 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).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 {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
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 {c_data['calle'][:35]:<35}{c_data['cp']:<9}")
lineas.append(f"IV14I{c_data['ciudad'][:20]:<20}{c_data['estado'][:2]}{c_data['pais'][:2]}{c_data['tel'][:30]:<30}{c_data['tax_id']:<15}00000")
self.cuenta_partidas += 3
# Items IV20-27
items_headers = db.query(Item).filter(Item.invoice_id == factura.id).all()
item_ids = [ih.id for ih in items_headers]
if item_ids:
items_query = db.query(LineItem).filter(
LineItem.item_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.financial.purchase_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_price_usd 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

41
backend/api/v1/modules/a76/reports/importacion/transmission/definitive/MAINX30/routes.py Normal file
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"}

72
backend/api/v1/modules/a76/reports/importacion/transmission/definitive/MAINX30/schemas.py Normal file
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

174
backend/api/v1/modules/a76/reports/importacion/transmission/definitive/MAINX30/service.py Normal file
View File

@@ -0,0 +1,174 @@
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()
# 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]
)

43
backend/api/v1/modules/a76/reports/importacion/transmission/definitive/MAINX30/task.py Normal file
View File

@@ -0,0 +1,43 @@
from core.celery_app import celery_app
from core.database import CoreSessionLocal
from .service import Mainx30DefinitiveService
from .schemas import Mainx30GenerationRequest
@celery_app.task(bind=True, name="api.v1.modules.a76.reports.importacion.transmission.definitive.MAINX30.task.generar_transmission_definitiva_async")
def generar_transmission_definitiva_async(self, request_data: dict, tenant_id: int):
"""
Celery task to generate Mainx30 file for Definitive Import
"""
db = CoreSessionLocal()
try:
# Reconstruct request object
request = Mainx30GenerationRequest(**request_data)
service = Mainx30DefinitiveService()
result_obj = service.generar_mainx30(db, request, task_instance=self)
result = result_obj.model_dump()
# Add base64 encoding for content to match frontend expectations
if result_obj.content:
import base64
encoded_content = base64.b64encode(result_obj.content.encode('utf-8')).decode('utf-8')
result['content'] = encoded_content
result['file_name'] = result_obj.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
finally:
db.close()

6
backend/api/v1/modules/a76/reports/importacion/transmission/temporal/MAINX30/service.py
View File

@@ -105,12 +105,12 @@ class Mainx30Service:
self.cuenta_partidas += 1
# 7. Generar Nombre y Guardar
nombre_archivo = f"TEMPORAL_{request.facturas[0][:15]}.txt"
nombre_archivo = f"{request.facturas[0][:15]}_Mainx30.dat"
if len(request.facturas) > 1:
nombre_archivo = f"TEMPORAL_MULTIPLE.txt"
nombre_archivo = f"MULTIPLE_Mainx30.dat"
if request.nomenclatura_factura and len(request.facturas) == 1:
nombre_archivo = f"{request.facturas[0][:15]}.txt"
nombre_archivo = f"{request.facturas[0][:15]}_Mainx30.dat"
content = '\r\n'.join(lineas)

7
backend/api/v1/modules/a76/router.py
View File

@@ -61,6 +61,7 @@ from .manifests.driver.routes import router as manifest_drivers_router
from .manifests.manifiesto_anexo.routes import router as manifest_anexos_router
from .reports.exportacion.transmission.MAINX30.routes import router as transmission_router
from .reports.importacion.transmission.temporal.MAINX30.routes import router as transmission_temporal_router
from .reports.importacion.transmission.definitive.MAINX30.routes import router as transmission_definitive_router
@@ -193,6 +194,12 @@ router.include_router(
tags=["a76 / reports"]
)
router.include_router(
transmission_definitive_router,
prefix="/a76/reports/importacion/transmission/definitive",
tags=["a76 / reports"]
)
# Registrar router de bitácora
from .audit_log.router import router as audit_log_router
router.include_router(audit_log_router, prefix="/a76/audit-log", tags=["Audit Log"])