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

Se integro el modulo de importacion

Browse Source
This commit is contained in:
Kevin_Ramirez
2026-02-16 10:36:23 -06:00
parent febae9fc01
commit 3d691d6881
11 changed files with 1274 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,330 @@
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": ""
}
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"
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"{fecha_trans}"
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(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"
# 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_price_usd 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
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

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

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

175
backend/api/v1/modules/a76/reports/importacion/transmission/temporal/MAINX30/service.py Normal file
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.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
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() # Default to Temporal for this endpoint
if request.regimen == "Definitiva":
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"TEMPORAL_{request.facturas[0][:15]}.txt"
if len(request.facturas) > 1:
nombre_archivo = f"TEMPORAL_MULTIPLE.txt"
if request.nomenclatura_factura and len(request.facturas) == 1:
nombre_archivo = f"{request.facturas[0][:15]}.txt"
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]
)

46
backend/api/v1/modules/a76/reports/importacion/transmission/temporal/MAINX30/task.py Normal file
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

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

@@ -60,6 +60,7 @@ from .manifests.manifest.routes import router as manifests_router
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
@@ -186,6 +187,12 @@ router.include_router(
tags=["a76 / reports"]
)
router.include_router(
transmission_temporal_router,
prefix="/a76/reports/importacion/transmission/temporal",
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"])