ADUANASOFT/SCAII_Sync_Client
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: initial commit del proyecto SCAII_Sync_Client

Browse Source 
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Kevin_Ramirez
2026-07-09 13:38:20 -05:00
commit 10dc063cf1
26 changed files with 2069 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored Normal file
View File

@@ -0,0 +1,6 @@
venv/
build/
dist/
__pycache__/
*.pyc
*.spec.bak

39
SCAII_Sincronizador_32bit.spec Normal file
View File

@@ -0,0 +1,39 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='SCAII_Sincronizador_32bit',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=False,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
version='C:\\Users\\PC\\AppData\\Local\\Temp\\6dd1fff1-457d-40b5-9d54-2ef8d8c29463',
)

118
cliente_api.py Normal file
View File

@@ -0,0 +1,118 @@
import requests
import os
import json
from datetime import datetime
from typing import Optional, Dict, Any
from logger_utils import log
class ClienteAPI:
def __init__(self, config: Dict[str, Any]):
self.url_api = config.get("url_api", "").rstrip('/') + '/'
self.usuario = config.get("usuario_rpa", "")
self.password = config.get("password_rpa", "")
self.sistema = config.get("sistema", "SCAII")
self.token = None
def obtener_token(self) -> bool:
"""
Equivalente a GetTokenAPI() de WinDev.
Realiza el login y guarda el token de sesión.
"""
log.info(f"Intentando login API para sistema {self.sistema} en {self.url_api}...")
# Lógica de endpoint basada en WinDev (Solo SCAII o SCAF)
endpoint = "login"
url = f"{self.url_api}{endpoint}"
payload = {
"username": self.usuario,
"password": self.password
}
try:
# En WinDev: RequestAPI.IgnoreError = httpIgnoreInvalidCertificate
response = requests.post(url, json=payload, verify=False, timeout=10)
if response.status_code == 200:
data = response.json()
# Extracción de token (Solo SCAII o SCAF)
self.token = data.get("token")
if self.token:
log.info("Login API exitoso. Token obtenido.")
return True
else:
log.error("Respuesta de login OK pero no se encontró el token.")
return False
else:
log.error(f"Error de autenticación API ({response.status_code}): {response.text}")
return False
except Exception as e:
log.error(f"Excepción en login API: {e}")
return False
def actualizar_estatus_factura(self, numero_factura: str) -> bool:
"""Actualiza el estatus de una factura en la nube"""
if not self.token:
return False
url = f"{self.url_api}registro505/descargar/{numero_factura}"
headers = {"Authorization": f"Bearer {self.token}"}
try:
# Esta es una llamada de ejemplo, ajustar según necesidad real
# response = requests.get(url, headers=headers, verify=False)
print(f"[API] Estatus de factura {numero_factura} actualizado.")
return True
except Exception as e:
self._guardar_log_error(f"Error factura {numero_factura}: {e}")
return False
def enviar_registro_api(self, endpoint: str, payload: dict) -> Optional[int]:
"""Método genérico para enviar cualquier registro (501, 505, 551, etc.)"""
if not self.token:
if not self.obtener_token():
return None
url = f"{self.url_api}{endpoint}"
headers = {
"Authorization": f"Bearer {self.token}",
"Content-Type": "application/json"
}
try:
log.debug(f"API POST -> {url}")
log.debug(f"PAYLOAD -> {json.dumps(payload)}")
response = requests.post(url, json=payload, headers=headers, verify=False, timeout=20)
log.debug(f"API STATUS <- {response.status_code}")
if response.status_code in [200, 201]:
data = response.json()
new_id = (data.get("id") or
data.get("id_pedimento") or
data.get("id_factura") or
data.get("id_partida") or
data.get("id_creado") or
data.get("id_registro"))
log.info(f"Registro enviado con éxito a {endpoint}. ID obtenido: {new_id}")
return new_id
else:
log.error(f"Error API enviando {endpoint} ({response.status_code}): {response.text}")
return None
except Exception as e:
log.error(f"Excepción enviando registro a {endpoint}: {e}")
return None
def _guardar_log_error(self, mensaje: str):
fecha = datetime.now().strftime("%Y%m%d_%H%M%S")
nombre_archivo = f"Error_API_{fecha}.txt"
try:
with open(nombre_archivo, "w") as f:
f.write(mensaje)
print(f"[ERROR] Log guardado en {nombre_archivo}")
except:
pass

117
config.py Normal file
View File

@@ -0,0 +1,117 @@
import argparse
import configparser
import os
import sys
from typing import List, Dict, Any
from logger_utils import log
class ConfiguracionApp:
def __init__(self):
self.datos: Dict[str, Any] = {
"servidor": "",
"base_datos": "",
"usuario": "sa",
"password": "",
"puerto": 0,
"sistema": "SCAF",
"pedimento": "",
"tipo_moneda": "ME",
"proceso": 1,
"url_api": "", # Se calculará dinámicamente
"progreso_actual": 0
}
self.usuario_rpa = ""
self.password_rpa = ""
def cargar(self):
log.info("Iniciando carga de configuración (Modo Parámetros)...")
log.info(f"=== ARGUMENTOS RECIBIDOS (sys.argv) ===")
log.info(f"Total de argumentos: {len(sys.argv)}")
for i, arg in enumerate(sys.argv):
log.info(f" argv[{i}] = '{arg}'")
log.info(f"=======================================")
# 2. Parseo manual de argumentos - compatible con -KEY=VALUE (WinDev) y /KEY VALUE (legacy)
params = {}
argv = sys.argv[1:]
i = 0
while i < len(argv):
arg = argv[i]
# Formato -KEY=VALUE o /KEY=VALUE
if '=' in arg:
key, value = arg.split('=', 1)
key = key.lstrip('-').lstrip('/')
params[key] = value
# Formato -KEY VALUE o /KEY VALUE (sin signo igual)
elif arg.startswith('-') or arg.startswith('/'):
key = arg.lstrip('-').lstrip('/')
if i + 1 < len(argv) and not argv[i+1].startswith('-') and not argv[i+1].startswith('/'):
params[key] = argv[i+1]
i += 1
else:
params[key] = True
i += 1
log.info(f"Parámetros interpretados: {list(params.keys())}")
self._aplicar_dict_params(params)
# 3. Validación (Sin modo test por defecto)
if not self.datos["servidor"]:
log.error("FALTA PARÁMETRO: No se recibió DBServer. El proceso no puede continuar.")
# 4. Lógica de URL Dinámica (Solo SCAII o SCAF)
self.datos["url_api"] = "https://104.192.7.152:3529/"
# 5. Valores por defecto finales
if not self.datos["puerto"] or self.datos["puerto"] == 0:
self.datos["puerto"] = 1433
# 6. Asegurar que las credenciales RPA estén en el dict de datos
self.datos["usuario_rpa"] = self.usuario_rpa
self.datos["password_rpa"] = self.password_rpa
log.info(f"Configuración cargada. Servidor: {self.datos['servidor']}, DB: {self.datos['base_datos']}, Sistema: {self.datos['sistema']}")
def _leer_archivo_ini(self):
config = configparser.ConfigParser()
# Buscamos el archivo en el mismo directorio del EXE
ruta_ini = os.path.join(os.path.dirname(os.path.abspath(sys.argv[0])), "SCAII_Sync.ini")
try:
if config.read(ruta_ini):
# Buscar sección Database de forma insensible a mayúsculas
section_db = next((s for s in config.sections() if s.lower() == "database"), None)
if section_db:
db = config[section_db]
self.datos["servidor"] = db.get("Server", self.datos["servidor"])
self.datos["base_datos"] = db.get("Database", self.datos["base_datos"])
self.datos["usuario"] = db.get("User", self.datos["usuario"])
self.datos["password"] = db.get("Pass", self.datos["password"])
self.datos["puerto"] = int(db.get("Port", "1433"))
log.info(f"Datos de DB cargados desde sección [{section_db}]")
# Buscar sección API de forma insensible a mayúsculas
section_api = next((s for s in config.sections() if s.lower() == "api"), None)
if section_api:
api = config[section_api]
self.usuario_rpa = api.get("User", self.usuario_rpa)
self.password_rpa = api.get("Pass", self.password_rpa)
log.info(f"Datos de API cargados desde sección [{section_api}]")
log.info(f"Archivo INI leído con éxito desde: {ruta_ini}")
except Exception as e:
log.error(f"Error al leer archivo INI: {e}")
pass
def _aplicar_dict_params(self, params: dict):
"""Aplica los parámetros desde un diccionario -KEY=VALUE (formato WinDev)"""
if params.get("DBServer"): self.datos["servidor"] = params["DBServer"]
if params.get("DBPort"): self.datos["puerto"] = int(params["DBPort"])
if params.get("DBUser"): self.datos["usuario"] = params["DBUser"]
if params.get("DBPass"): self.datos["password"] = params["DBPass"]
if params.get("DBDatabase"): self.datos["base_datos"] = params["DBDatabase"]
if params.get("DBSystem"): self.datos["sistema"] = params["DBSystem"]
if params.get("DBPedi"): self.datos["pedimento"] = params["DBPedi"]
if params.get("DBMoneda"): self.datos["tipo_moneda"] = params["DBMoneda"]
if params.get("DBProcess") is not None: self.datos["proceso"] = int(params["DBProcess"])
if params.get("RPAUser"): self.usuario_rpa = params["RPAUser"]
if params.get("RPAPass"): self.password_rpa = params["RPAPass"]
log.info(f"Parámetros aplicados -> Servidor: {self.datos['servidor']}, DB: {self.datos['base_datos']}, Sistema: {self.datos['sistema']}, Pedimento: {self.datos['pedimento']}")

3
e.sh Executable file
View File

@@ -0,0 +1,3 @@
#!/bin/bash
mv SCAII_Sincronizador_32bit.exe /mnt/c/Users/PC/Desktop

474
gestor_base_datos.py Normal file
View File

@@ -0,0 +1,474 @@
import pyodbc
from typing import List, Dict, Any, Optional, Tuple
from logger_utils import log
class GestorBaseDatos:
def __init__(self, config: Dict[str, Any]):
self.config = config
self.conn = None
self.cursor = None
def conectar(self) -> bool:
try:
conn_str = (
f"DRIVER={{ODBC Driver 17 for SQL Server}};"
f"SERVER={self.config['servidor']};"
f"DATABASE={self.config['base_datos']};"
f"UID={self.config['usuario']};"
f"PWD={self.config['password']};"
"Connection Timeout=30;"
)
log.info(f"Conectando a SQL Server: {self.config['servidor']} - DB: {self.config['base_datos']}...")
self.conn = pyodbc.connect(conn_str)
log.info("Conexión SQL exitosa.")
self.cursor = self.conn.cursor()
return True
except Exception as e:
log.error(f"[DB ERROR] Fallo al conectar: {e}")
return False
def obtener_datos_pedimento(self, pedimento: str, sistema: str) -> Optional[Dict[str, Any]]:
"""Busca el pedimento usando coincidencia flexible (LIKE)"""
tabla = "SPedimentos" if sistema == "SCAII" else "QPedimentos"
query = f"""
SELECT TIPO, REGIMEN, CLAVEPED, ADUANA_CRUCE, OPCIONDESTINO,
TIPOPEDIMENTOTRANSPORTEE, TIPOPEDIMENTOTRANSPORTEA, TIPOPEDIMENTOTRANSPORTES,
PEDIMENTO
FROM {tabla}
WHERE PEDIMENTO LIKE ?
"""
log.debug(f"Buscando pedimento: {pedimento} en {tabla}")
try:
self.cursor.execute(query, (f"%{pedimento.strip()}%",))
row = self.cursor.fetchone()
if row:
return {
"Tipo": row[0], "Regimen": row[1], "Clave": row[2],
"Aduana": row[3], "Destino": row[4],
"TransE": row[5], "TransA": row[6], "TransS": row[7],
"Pedimento": row[8]
}
return None
except Exception as e:
log.error(f"[DB ERROR] Error al obtener datos pedimento: {e}")
return None
def obtener_detalle_facturas(self, pedimento: str, sistema: str, regimen: str, tipo_op: str) -> List[Dict[str, Any]]:
"""Obtiene las facturas usando los querys maestros de WinDev"""
if sistema == "SCAII":
if tipo_op == "I":
if regimen == "IMD":
# Scenario: SComprasMexID (Importación Definitiva)
query = f"""
SELECT
f.PedimentoImpoDef AS Pedimento, f.Consecutivo AS Consecutivo, f.FacCMImpoDef AS NumFactura, f.FechaFactura AS Fecha,
f.Incoterm AS Incoterm, f.ValorImpoME AS ValorImpoME, f.Proveedor AS IdProveedor,
f.ValorImpoMN AS ValorImpoMN, f.ClaveMoneda AS ClaveMoneda, f.ValorImpoMC AS ValorMC,
f.MetValor AS MetodoValoracion, f.EsMixto AS EsMixto, f.Remesa AS Remesa,
f.EDocument AS EDocument, f.NumeroCertificado AS NumCertificado, f.FirmaElectronica AS FirmaElectronica,
f.NumeroNIU AS NumNIU, f.TipoDeGuiaAIdentificar AS TipoGuia, f.CantGuiasEmbarque AS CantGuias,
f.ObservacionesVU AS Observaciones, f.ContenedoresTipo AS ContenedoresTipo, f.EnviadoA AS EnviadoA,
f.FUNGECOMOCO AS FungeComoCO, f.SUBDIVISION AS Subdivision, f.TipoCambio AS TipoCambio,
f.MODOCONTINGENCIA AS ModoContingencia, f.NUMOPERACIONVU AS NumOperacionVU, f.TRANSPORTISTA AS Transportista,
f.NUMTRASPORTE AS NumTrailer, f.PESOBRUTO AS PesoBruto, f.FLETES AS Flete,
f.SEGUROS AS Seguros, f.EMBALAJES AS Embalajes, f.OTROSINCREMENTA AS OtrosIncrementa,
f.CANTBULTOS AS CantBultos, 'ME' AS TipoMoneda, p.VINCULACION AS Vinculacion,
0 AS IdCliente, '' AS CfdiUuid, '' AS CfdiPathPDF, '' AS CfdiPathXML
FROM SComprasMexID f
LEFT JOIN GClientesPro p ON f.PROVEEDOR = p.CLIENTE
WHERE f.PEDIMENTOIMPODEF LIKE ? AND f.Estatus = 'AC' ORDER BY f.Remesa
"""
else:
# Scenario: SFacImp (Importación Normal)
query = f"""
SELECT
f.PedimentoImpo AS Pedimento, f.Consecutivo AS Consecutivo, f.FacturaImpo AS NumFactura, f.FechaFactura AS Fecha,
f.Incoterm AS Incoterm, f.ValorImpoME AS ValorImpoME, f.Proveedor AS IdProveedor,
f.ValorImpoMN AS ValorImpoMN, f.ClaveMoneda AS ClaveMoneda, f.ValorImpoMC AS ValorMC,
f.MetValor AS MetodoValoracion, f.EsMixto AS EsMixto, f.Remesa AS Remesa,
f.EDocument AS EDocument, f.NumeroCertificado AS NumCertificado, f.FirmaElectronica AS FirmaElectronica,
f.NumeroNIU AS NumNIU, f.TipoDeGuiaAIdentificar AS TipoGuia, f.CantGuiasEmbarque AS CantGuias,
f.ObservacionesVU AS Observaciones, f.ContenedoresTipo AS ContenedoresTipo, f.EnviadoA AS EnviadoA,
f.FUNGECOMOCO AS FungeComoCO, f.SUBDIVISION AS Subdivision, f.TipoCambio AS TipoCambio,
f.MODOCONTINGENCIA AS ModoContingencia, f.NUMOPERACIONVU AS NumOperacionVU, f.TRANSPORTISTA AS Transportista,
f.NUMTRASPORTE AS NumTrailer, f.PESOBRUTO AS PesoBruto, f.FLETE AS Flete,
f.SEGUROS AS Seguros, f.EMBALAJES AS Embalajes, f.OTROSINCREMENTA AS OtrosIncrementa,
f.CANTBULTOS AS CantBultos, f.TIPOMONEDA AS TipoMoneda, p.VINCULACION AS Vinculacion,
0 AS IdCliente, '' AS CfdiUuid, '' AS CfdiPathPDF, '' AS CfdiPathXML
FROM SFacImp f
LEFT JOIN GClientesPro p ON f.PROVEEDOR = p.CLIENTE
WHERE f.PedimentoImpo LIKE ? AND f.Estatus = 'AC' ORDER BY f.Remesa
"""
else:
# Scenario: SFacExp (Exportación)
query = f"""
SELECT
f.PedimentoExpo AS Pedimento, f.Consecutivo AS Consecutivo, f.FacturaExpo AS NumFactura, f.FechaFactura AS Fecha,
f.Incoterm AS Incoterm,
(SELECT SUM(MatPex.ValorTotalME) FROM SPartidasExpo MatPex LEFT JOIN SPartes Partes ON Partes.NumParte = MatPex.NumParte WHERE MatPex.Consecutivo = f.Consecutivo) AS ValorImpoME,
f.Proveedor AS IdProveedor,
(SELECT SUM(MatPex.ValorTotalMN) FROM SPartidasExpo MatPex LEFT JOIN SPartes Partes ON Partes.NumParte = MatPex.NumParte WHERE MatPex.Consecutivo = f.Consecutivo) AS ValorImpoMN,
f.ClaveMoneda AS ClaveMoneda, 0 AS ValorMC,
f.MetValor AS MetodoValoracion, 0 AS EsMixto, f.Remesa AS Remesa,
f.EDocument AS EDocument, f.NumeroCertificado AS NumCertificado, f.FirmaElectronica AS FirmaElectronica,
f.NumeroNIU AS NumNIU, f.TipoDeGuiaAIdentificar AS TipoGuia, f.CantGuiasEmbarque AS CantGuias,
f.ObservacionesVU AS Observaciones, f.ContenedoresTipo AS ContenedoresTipo, f.EnviadoA AS EnviadoA,
f.FUNGECOMOCO AS FungeComoCO, f.SUBDIVISION AS Subdivision, f.TipoCambio AS TipoCambio,
f.MODOCONTINGENCIA AS ModoContingencia, f.NUMOPERACIONVU AS NumOperacionVU, f.TRANSPORTISTA AS Transportista,
f.NUMTRASPORTE AS NumTrailer, f.PESOBRUTO AS PesoBruto, f.FLETE AS Flete,
f.SEGUROS AS Seguros, f.EMBALAJES AS Embalajes, f.OTROSINCREMENTA AS OtrosIncrementa,
f.CANTBULTOS AS CantBultos, f.TIPOMONEDA AS TipoMoneda, 0 AS Vinculacion,
f.VendidoA AS IdCliente, f.CfdiUuid AS CfdiUuid, f.CfdiPathPDF AS CfdiPathPDF, f.CfdiPathXML AS CfdiPathXML
FROM SFacExp f
WHERE f.PedimentoExpo LIKE ? AND f.Estatus = 'AC' AND f.TipoFactura <> 'REPAR' ORDER BY f.Remesa
"""
else: # SCAF
if tipo_op == "E":
query = """
SELECT
f.PedimentoExpo AS Pedimento, f.Consecutivo AS Consecutivo, f.FacturaExpo AS NumFactura, f.FechaFactura AS FechaFactura,
f.Incoterm AS Incoterm, f.ValorExpoME AS ValorImpoME, f.Proveedor AS IdProveedor,
f.ValorExpoMN AS ValorImpoMN, f.ClaveMoneda AS ClaveMoneda, f.MetValor AS MetValor, f.Remesa AS Remesa,
f.EDocument AS EDocument, f.NumeroCertificado AS NumeroCertificado, f.FirmaElectronica AS FirmaElectronica,
f.NUMERONIU AS NumNIU, f.TIPODEGUIAAIDENTIFICAR AS TipoGuia, f.CANTGUIASEMBARQUE AS CantGuias,
f.OBSERVACIONESVU AS ObservacionesVU, f.CONTENEDORESTIPO AS ContenedoresTipo, f.FUNGIRCOMOCO AS FUNGECOMOCO,
f.SUBDIVISION AS SUBDIVISION, f.TIPOCAMBIO AS TipoCambio, f.MODOCONTINGENCIA AS MODOCONTINGENCIA,
f.NUMOPERACIONVU AS NUMOPERACIONVU, f.TRANSPORTISTA AS TRANSPORTISTA, f.CANTBULTOS AS CANTBULTOS,
f.TIPOMONEDA AS TipoMoneda,
(SELECT SUM(PesoBrutoKGS) FROM QEqeMaq WHERE Consecutivo = f.Consecutivo) AS PesoBruto
FROM QFacExp f
WHERE f.PedimentoExpo LIKE ? AND f.Estatus = 'AC' ORDER BY f.Remesa
"""
else: # SCAF Importación
tabla_partidas = "QEqiDef" if regimen == "IMD" else "QEqiMaq"
query = f"""
SELECT
f.PedimentoImpo AS Pedimento, f.Consecutivo AS Consecutivo, f.FacturaImpo AS NumFactura, f.FechaFactura AS FechaFactura,
f.Incoterm AS Incoterm, f.ValorImpoME AS ValorImpoME, f.Proveedor AS IdProveedor,
f.ValorImpoMN AS ValorImpoMN, f.ClaveMoneda AS ClaveMoneda, f.MetValor AS MetValor, f.Remesa AS Remesa,
f.EDocument AS EDocument, f.NumeroCertificado AS NumeroCertificado, f.FirmaElectronica AS FirmaElectronica,
f.NUMERONIU AS NumNIU, f.TIPODEGUIAAIDENTIFICAR AS TipoGuia, f.CANTGUIASEMBARQUE AS CantGuias,
f.OBSERVACIONESVU AS ObservacionesVU, f.CONTENEDORESTIPO AS ContenedoresTipo, f.FUNGIRCOMOCO AS FUNGECOMOCO,
f.SUBDIVISION AS SUBDIVISION, f.TIPOCAMBIO AS TipoCambio, f.MODOCONTINGENCIA AS MODOCONTINGENCIA,
f.NUMOPERACIONVU AS NUMOPERACIONVU, f.TRANSPORTISTA AS TRANSPORTISTA, f.CANTBULTOS AS CANTBULTOS,
f.TIPOMONEDA AS TipoMoneda,
(SELECT SUM(PesoBrutoKGS) FROM {tabla_partidas} WHERE Consecutivo = f.Consecutivo) AS PesoBruto
FROM QFacImp f
WHERE f.PedimentoImpo LIKE ? AND f.Estatus = 'AC' ORDER BY f.Remesa
"""
try:
self.cursor.execute(query, (f"%{pedimento.strip()}%",))
columns = [column[0] for column in self.cursor.description]
return [dict(zip(columns, row)) for row in self.cursor.fetchall()]
except Exception as e:
log.error(f"[DB ERROR] Error al obtener facturas: {e}")
return []
def obtener_partidas_factura(self, consecutivo_fac: int, sistema: str, tipo_op: str, regimen: str) -> List[Dict[str, Any]]:
"""Querys homologados de partidas"""
if sistema == "SCAII":
if tipo_op == "I":
tabla = "SPartidasCM" if regimen == "IMD" else "SPartidasImpo"
col_linea = "LineaCMImpDef" if regimen == "IMD" else "LineaImpo"
col_fac = "FACCMIMPODEF" if regimen == "IMD" else "FACTURAIMPO"
query = f"""
SELECT p.Consecutivo AS Consecutivo, p.{col_linea} AS Linea, p.NumParte AS NumParte,
p.CantImpo AS Cantidad, p.ValorImpoMC AS ValorComercial, p.UniMed AS UnidadMedida,
p.FraccionImpo AS Fraccion, p.PesoNetoKGS AS PesoNeto, p.TipoFraccImpo AS TipoFracc,
p.CostoUnitarioME AS CostoUnitarioME, p.ValorImpoME AS ValorDolares,
p.ValorImpoMN AS ValorPesos, p.PaisOrigen AS PaisOrigen, p.Sector AS Sector,
p.PesoBrutoKGS AS PesoBruto, p.ClaveBultos AS ClaveBultos, p.CantBultos AS CantBultos,
p.InfoAdicionEsp AS InfoAdic, p.Clase AS Clase,
(SELECT TOP 1 UMCLAVE FROM sfracciones WHERE fraccion = SUBSTRING(p.FraccionImpo, 1, 8)) AS ClaveOMA,
p.FraccionROctava AS FraccionOctava, p.DESCRIPCIONPARTE AS Descripcion,
p.{col_fac} AS NumFactura, p.CANTALTERNA AS CantidadAlterna,
p.UNIMEDALTERNA AS UnidadAlterna, p.VALORADUANASMN AS ValorAduanasMN,
p.VALORADUANASME AS ValorAduanasME,
ISNULL(sp.DESCRIPCIONI, p.INFOADICIONING) AS DescripcionIngles,
0 AS ValorAgregadoMN, p.METVALOR AS MetodoValoracion,
p.COSTOUAUXILIARME AS CostoAuxiliarME,
'' AS Marca, '' AS Modelo, 0 AS EsSubPartida
FROM {tabla} p
LEFT JOIN SPartes sp ON p.NumParte = sp.NUMPARTE
WHERE p.Consecutivo = ? ORDER BY p.{col_linea}
"""
else: # EXPO SCAII
query = f"""
SELECT p.FacturaExpo AS Consecutivo, p.Linea AS Linea, p.NumParte AS NumParte,
p.CantExpo AS Cantidad, p.ValorTotalMN AS ValorComercial, p.UniMed AS UnidadMedida,
p.FraccionExpo AS Fraccion, p.PesoNetoKGS AS PesoNeto, p.TipoFracExpo AS TipoFracc,
p.CostoUnitarioME AS CostoUnitarioME, p.ValorTotalME AS ValorDolares,
p.ValorTotalMN AS ValorPesos, p.PaisOrigen AS PaisOrigen, p.Sector AS Sector,
p.PesoBrutoKGS AS PesoBruto, p.ClaveBultos AS ClaveBultos, p.CantBultos AS CantBultos,
p.InfoAdicionEsp AS InfoAdic, p.Clase AS Clase,
(SELECT TOP 1 UMCLAVE FROM sfracciones WHERE fraccion = SUBSTRING(p.FraccionExpo, 1, 8)) AS ClaveOMA,
'' AS FraccionOctava, p.DESCRIPCIONPARTE AS Descripcion, p.FACTURAEXPO AS NumFactura,
0 AS CantidadAlterna, '' AS UnidadAlterna, p.VALORADUANASMN AS ValorAduanasMN,
p.ValorAgreME AS ValorAgregadoME, p.ValorAgreMN AS ValorAgregadoMN,
p.MetValor AS MetodoValoracion, '' AS Marca, '' AS Modelo, 0 AS EsSubPartida,
ISNULL(sp.DESCRIPCIONI, p.DESCRIPCIONPARTE) AS DescripcionIngles,
0 AS CostoAuxiliarME
FROM SPartidasExpo p
LEFT JOIN SPartes sp ON p.NumParte = sp.NUMPARTE
WHERE Consecutivo = ? ORDER BY p.LINEA
"""
else: # SCAF
if tipo_op == "I":
tabla = "QEqiDef" if regimen == "IMD" else ("QEqiMaqRep" if "REP" in regimen else "QEqiMaq")
col_linea = "LineaImpoDef" if regimen == "IMD" else "LineaImpo"
col_fac = "FACTURAIMPODEF" if regimen == "IMD" else "FACTURAIMPO"
col_cant = "CantImpoDef" if regimen == "IMD" else "CantImpo"
col_vme = "ValorME" if regimen == "IMD" else "ValorImpoME"
col_vmn = "ValorMN" if regimen == "IMD" else "ValorImpoMN"
query = f"""
SELECT p.Consecutivo AS Consecutivo, p.{col_linea} AS Linea, p.NumParte AS NumParte,
p.{col_cant} AS Cantidad, p.ValorImpoMC AS ValorComercial, p.UnidadMedida AS UnidadMedida,
p.Fraccion AS Fraccion, p.PesoNetoKGS AS PesoNeto, p.TipoFraccion AS TipoFracc,
p.CostoUnitarioDlls AS CostoUnitarioME, p.{col_vme} AS ValorDolares,
p.{col_vmn} AS ValorPesos, p.PaisOrigen AS PaisOrigen, p.Sector AS Sector,
p.PesoBrutoKGS AS PesoBruto, p.ClaveBultos AS ClaveBultos, p.CantBultos AS CantBultos,
p.DescripcionE AS InfoAdic, p.Clase AS Clase,
(SELECT TOP 1 UMCLAVE FROM sfracciones WHERE fraccion = SUBSTRING(p.Fraccion, 1, 8)) AS ClaveOMA,
p.FRACCIONROCTAVA AS FraccionOctava, ISNULL(c.DESCRIPCIONE, p.DESCRIPCIONE) AS Descripcion,
p.{col_fac} AS NumFactura, 0 AS CantidadAlterna, ' ' AS UnidadAlterna,
p.ValorAduanasMN AS ValorAduanasMN, p.ValorAduanasME AS ValorAduanasME,
0 AS ValorAgregadoMN, p.METVALOR AS MetodoValoracion,
p.Marca AS Marca, p.Modelo AS Modelo, p.EsSubPartida AS EsSubPartida,
ISNULL(c.DESCRIPCIONI, p.DESCRIPCIONE) AS DescripcionIngles,
0 AS CostoAuxiliarME
FROM {tabla} p
LEFT JOIN QClaAct c ON c.CLASE = p.NumParte
WHERE p.Consecutivo = ? ORDER BY p.{col_linea}
"""
else: # EXPO SCAF
query = """
SELECT p.Consecutivo AS Consecutivo, p.LineaExpo AS Linea, p.FacturaImpo AS NumParte,
p.CantExpo AS Cantidad, 0 AS ValorComercial, p.UnidadMedida AS UnidadMedida,
p.FraccionExpo AS Fraccion, p.PesoNetoKGS AS PesoNeto, p.TipoFraccion AS TipoFracc,
p.CostoUnitarioDlls AS CostoUnitarioME, p.ValorExpoME AS ValorDolares,
p.ValorExpoMN AS ValorPesos, p.PaisOrigen AS PaisOrigen, p.Sector AS Sector,
p.PesoBrutoKGS AS PesoBruto, p.ClaveBultos AS ClaveBultos, p.CantBultos AS CantBultos,
p.DescripcionE AS InfoAdic, p.Clase AS Clase,
(SELECT TOP 1 UMCLAVE FROM sfracciones WHERE fraccion = SUBSTRING(p.FraccionExpo, 1, 8)) AS ClaveOMA,
' ' AS FraccionOctava, ISNULL(c.DESCRIPCIONE, p.DESCRIPCIONE) AS Descripcion,
p.FACTURAEXPO AS NumFactura,
0 AS CantidadAlterna, ' ' AS UnidadAlterna, p.VALORADUANASMN AS ValorAduanasMN,
0 AS ValorAgregadoME, 0 AS ValorAgregadoMN, p.MetValor AS MetodoValoracion,
p.Marca AS Marca, p.Modelo AS Modelo, p.ESSUBPARTIDA AS EsSubPartida,
ISNULL(c.DESCRIPCIONI, p.DESCRIPCIONE) AS DescripcionIngles,
0 AS CostoAuxiliarME
FROM QEqeMaq p
LEFT JOIN QClaAct c ON c.CLASE = p.FacturaImpo
WHERE p.Consecutivo = ? ORDER BY p.LineaExpo
"""
try:
self.cursor.execute(query, (consecutivo_fac,))
columns = [column[0] for column in self.cursor.description]
return [dict(zip(columns, row)) for row in self.cursor.fetchall()]
except Exception as e:
print(f"[DB ERROR] Error al obtener partidas: {e}")
return []
def obtener_total_vehiculos(self, pedimento: str, sistema: str) -> int:
"""Calcula el total de vehículos (Trailers)"""
if sistema == "SCAII":
query = """
SELECT DISTINCT MatFex.NUMTRAILER
FROM SPartidasExpo MatPex
LEFT OUTER JOIN SFacExp MatFex ON MatFex.Consecutivo = MatPex.Consecutivo
WHERE (MatFex.PedimentoExpo LIKE ?) AND MatFex.Estatus = 'AC'
"""
try:
self.cursor.execute(query, (f"%{pedimento.strip()}%",))
rows = self.cursor.fetchall()
return len(rows)
except:
return 0
else: # SCAF
# En SCAF buscamos en las tablas de partidas (QEqiMaq/QEqeMaq/etc)
# Primero intentamos obtener el tipo de pedimento para saber qué tabla usar
datos_pedi = self.obtener_datos_pedimento(pedimento, sistema)
if not datos_pedi: return 0
tipo_op = datos_pedi.get("Tipo")
regimen = datos_pedi.get("Regimen", "")
if tipo_op == "E":
tabla = "QEqeMaq"
else:
tabla = "QEqiDef" if regimen == "IMD" else "QEqiMaq"
query = f"SELECT COUNT(DISTINCT NUMTRAILER) FROM {tabla} WHERE PEDIMENTO LIKE ?"
try:
self.cursor.execute(query, (f"%{pedimento.strip()}%",))
row = self.cursor.fetchone()
return int(row[0]) if row else 0
except:
return 0
return 0
def obtener_datos_empresa(self, sistema: str = "SCAII") -> Optional[Dict[str, Any]]:
# Casi todos los sistemas (SCAII/SCAF) usan GEmpresa como tabla global
tabla = "GEmpresa"
try:
self.cursor.execute(f"SELECT TOP 1 NOMBRE, RFC, NUMDEEXPORTADORCONFIABLE FROM {tabla}")
row = self.cursor.fetchone()
if row:
return {"NOMBRE": row[0], "RFC": row[1], "NUMDEEXPORTADORCONFIABLE": row[2] if len(row)>2 else ""}
return None
except Exception as e:
print(f"[DB ERROR] Error al obtener datos de empresa: {e}")
return None
def obtener_credenciales_rpa(self) -> tuple:
"""Lee usuario/password del API desde GEmpresa, igual que WinDev en produccion.
Usa los campos NOMBREBDINTER y CLAVEPREVALIDADOR."""
try:
self.cursor.execute("SELECT TOP 1 NOMBREBDINTER, CLAVEPREVALIDADOR FROM GEmpresa")
row = self.cursor.fetchone()
if row and row[0] and row[1]:
log.info(f"Credenciales RPA obtenidas de GEmpresa (usuario: {row[0]})")
return str(row[0]).strip(), str(row[1]).strip()
else:
log.warning("GEmpresa no tiene NOMBREBDINTER/CLAVEPREVALIDADOR configurados.")
return "", ""
except Exception as e:
log.error(f"[DB ERROR] Error al obtener credenciales RPA de GEmpresa: {e}")
return "", ""
def obtener_datos_proveedor(self, id_proveedor: int, sistema: str) -> Optional[Dict[str, Any]]:
if sistema == "SCAII":
tabla = "GClientesPro"
query = f"SELECT NOMBRE, PAIS, TAXID, VINCULACION, CLAVETRANSFER FROM {tabla} WHERE CLIENTE = ?"
else:
tabla = "GClientesPro"
query = f"SELECT NOMBRE, PAIS, TAXID, VINCULACION, CLAVETRANSFER FROM {tabla} WHERE CLIENTE = ?"
try:
self.cursor.execute(query, (id_proveedor,))
row = self.cursor.fetchone()
if row:
return {
"NOMBRE": row[0], "PAIS": row[1], "TAXID": row[2],
"VINCULACION": row[3], "CLAVETRANSFER": row[4] if len(row) > 4 else ""
}
return None
except Exception as e:
print(f"[DB ERROR] Error al obtener proveedor: {e}")
return None
def obtener_datos_transporte(self, clave_transporte: str, sistema: str = "SCAII") -> Optional[Dict[str, Any]]:
if not clave_transporte:
return None
if sistema == "SCAII":
tabla = "GTransportista"
query = f"SELECT RFC, CODIGOCAAT, PAIS FROM {tabla} WHERE CLAVETRANS = ?"
else:
tabla = "GTransportista" # Confirmado por WinDev
query = f"SELECT RFC, CODIGOCAAT, PAIS FROM {tabla} WHERE CLAVETRANS = ?"
try:
self.cursor.execute(query, (clave_transporte,))
row = self.cursor.fetchone()
if row:
return {"RFC": row[0], "CODIGOCAAT": row[1], "PAIS": row[2]}
return None
except Exception as e:
print(f"[DB ERROR] Error al obtener transporte: {e}")
return None
def obtener_datos_trailer(self, num_trailer: str, sistema: str = "SCAII") -> Optional[Dict[str, Any]]:
if not num_trailer:
return None
if sistema == "SCAII":
tabla = "GTrailers"
query = f"SELECT NUMEROPLACAS, PAIS FROM {tabla} WHERE NUMTRAILER = ?"
else:
tabla = "GTrailers" # Confirmado por WinDev
query = f"SELECT NUMEROPLACAS, PAIS FROM {tabla} WHERE NUMTRAILER = ?"
try:
self.cursor.execute(query, (num_trailer,))
row = self.cursor.fetchone()
if row:
return {"NUMEROPLACAS": row[0], "PAIS": row[1]}
return None
except Exception as e:
print(f"[DB ERROR] Error al obtener trailer: {e}")
return None
def obtener_um_tarifa_fraccion(self, fraccion: str, sistema: str = "SCAII") -> Optional[str]:
if not fraccion:
return None
fraccion_corta = str(fraccion)[:8].replace(".", "")
tabla = "SFracciones" if sistema == "SCAII" else "QFracciones"
try:
self.cursor.execute(f"SELECT TOP 1 UMCLAVE FROM {tabla} WHERE FRACCION LIKE ?", (fraccion_corta + '%',))
row = self.cursor.fetchone()
return str(row[0]).strip() if row else None
except Exception as e:
print(f"[DB ERROR] Error UM Tarifa: {e}")
return None
def obtener_unidad_aduana_scaii(self, clave_aduana: str) -> Optional[str]:
if not clave_aduana:
return None
try:
self.cursor.execute("SELECT TOP 1 CLAVEUNI FROM GUniMedida WHERE CLAVEOMA = ?", (clave_aduana,))
row = self.cursor.fetchone()
return str(row[0]).strip() if row else None
except Exception as e:
return None
def obtener_conversion_unidades(self, um_origen: str, um_destino: str) -> Tuple[str, float]:
if not um_origen or not um_destino or um_origen == um_destino:
return "M", 1.0
try:
self.cursor.execute("SELECT FACTORCONV FROM GConversiones WHERE CLAVEUNI1 = ? AND CLAVEUNI2 = ?", (um_origen, um_destino))
row = self.cursor.fetchone()
if row: return "M", float(row[0])
self.cursor.execute("SELECT FACTORCONV FROM GConversiones WHERE CLAVEUNI1 = ? AND CLAVEUNI2 = ?", (um_destino, um_origen))
row = self.cursor.fetchone()
if row: return "D", float(row[0])
return "M", 1.0
except:
return "M", 1.0
def obtener_identificadores_partida(self, consecutivo: int, linea: int, modulo: str) -> List[Dict[str, Any]]:
if not consecutivo or not linea or not modulo:
return []
query = """
SELECT ID, COMPLEMENTO1, COMPLEMENTO2, COMPLEMENTO3
FROM GIdentificadoresPartidas
WHERE CONSECUTIVOFACTURA = ? AND LINEAPARTIDA = ? AND MODULO = ?
"""
try:
self.cursor.execute(query, (consecutivo, linea, modulo))
cols = [column[0] for column in self.cursor.description]
return [dict(zip(cols, row)) for row in self.cursor.fetchall()]
except Exception as e:
print(f"[DB ERROR] Error al obtener identificadores de partida: {e}")
return []
def obtener_datos_unidad_medida(self, clave_raw: str, sistema: str) -> Optional[Dict[str, Any]]:
tabla = "Sunidades" if sistema == "SCAII" else "Qunidades"
query = f"SELECT CLAVE, DESCRIPCION, CLAVEOMA FROM {tabla} WHERE CLAVE = ? OR DESCRIPCION = ?"
try:
self.cursor.execute(query, (clave_raw, clave_raw))
row = self.cursor.fetchone()
if row:
return {"CLAVE": row[0], "DESCRIPCION": row[1], "CLAVEOMA": row[2] if len(row) > 2 else " "}
return None
except Exception as e:
print(f"[DB ERROR] Error al obtener unidad: {e}")
return None
def cerrar(self):
if self.conn:
self.conn.close()

197
interfaz.py Normal file
View File

@@ -0,0 +1,197 @@
import flet as ft
import threading
from typing import Optional
class InterfazFlet:
def __init__(self, cliente_sinc=None):
self.cliente = cliente_sinc
self.page: Optional[ft.Page] = None
self.COLOR_TEXTO_TITULO = "#0066CC"
self.COLOR_FRANJA_CELESTE = "#7CE0F6"
self.lista_resultados = ft.Column(
expand=True, spacing=2, scroll=ft.ScrollMode.ALWAYS, auto_scroll=True
)
self.cabecera_lista = ft.Container(
content=ft.Row([
ft.Text("Pedimento", width=150, color=ft.Colors.WHITE, weight="bold", size=12),
ft.Text("Factura", width=120, color=ft.Colors.WHITE, weight="bold", size=12),
ft.Text("Remesa", width=80, color=ft.Colors.WHITE, weight="bold", size=12),
ft.Text("Estatus de Sincronización", expand=True, color=ft.Colors.WHITE, weight="bold", size=12),
]),
bgcolor=ft.Colors.BLUE_GREY_900,
padding=ft.padding.only(left=20, top=10, right=20, bottom=10),
border_radius=ft.border_radius.only(top_left=5, top_right=5)
)
self.barra_progreso = ft.ProgressBar(value=0, color=ft.Colors.BLUE_600, bgcolor=ft.Colors.GREY_300)
self.status_label = ft.Text("Listo", color=ft.Colors.BLACK, size=13, weight="w500")
self.porcentaje_label = ft.Text("0%", color=ft.Colors.BLUE_700, weight="bold", size=14)
def main(self, page: ft.Page):
self.page = page
page.title = "Sistema de Control de Aduanas e Inventarios"
page.theme_mode = ft.ThemeMode.LIGHT
page.window_width = 1000
page.window_height = 650
page.padding = 20
page.bgcolor = ft.Colors.GREY_50
titulo = ft.Text("Comunicación SCAII - WINSAAI.", size=24, color=self.COLOR_TEXTO_TITULO, weight="bold")
franjas_decorativas = ft.Column([
ft.Container(height=3, bgcolor=ft.Colors.BLACK),
ft.Container(height=15, bgcolor=self.COLOR_FRANJA_CELESTE)
], spacing=0)
lista_container = ft.Container(
content=ft.Column([self.cabecera_lista, self.lista_resultados], spacing=0),
expand=True, bgcolor=ft.Colors.WHITE,
border=ft.border.all(1, ft.Colors.GREY_300), border_radius=8
)
footer = ft.Container(
content=ft.Row([
ft.Column([
ft.Row([self.status_label, ft.Container(expand=True), self.porcentaje_label]),
self.barra_progreso
], expand=True, spacing=5)
], alignment=ft.MainAxisAlignment.START, vertical_alignment=ft.CrossAxisAlignment.CENTER),
padding=15, bgcolor=ft.Colors.WHITE, border_radius=8,
border=ft.border.all(1, ft.Colors.GREY_200),
shadow=ft.BoxShadow(blur_radius=5, color=ft.Colors.GREY_200, offset=ft.Offset(0, 2))
)
page.add(titulo, franjas_decorativas, lista_container,
ft.Divider(height=10, color=ft.Colors.TRANSPARENT), footer)
if self.cliente:
threading.Thread(target=self._hilo_sincronizacion, daemon=True).start()
# ------------------------------------------------------------------ #
def agregar_fila(self, pedimento, factura, remesa, datos_completos, estatus):
async def _do():
# Panel de detalles (oculto inicialmente)
campos_relevantes = {
"NumFactura": "Factura",
"Remesa": "Remesa",
"IdProveedor": "ID Proveedor",
"ValorDolares": "Valor Dólares",
"PesoBruto": "Peso Bruto",
"Transportista": "Transportista",
"NumTrailer": "Caja/Trailer",
"Incoterm": "Incoterm",
"Consecutivo": "Folio Interno"
}
filas_detalles = [
ft.Row([
ft.Text(f"{etiqueta}:", weight="bold", width=120, size=12, color=ft.Colors.GREY_700),
ft.Text(str(datos_completos.get(clave, "N/A")), size=12, color=ft.Colors.BLUE_700),
], spacing=10)
for clave, etiqueta in campos_relevantes.items()
]
panel_detalles = ft.Container(
content=ft.Column(filas_detalles, spacing=6),
visible=False, # <-- oculto por defecto
bgcolor=ft.Colors.BLUE_50,
padding=ft.padding.only(left=30, top=10, right=20, bottom=10),
border=ft.border.only(
bottom=ft.BorderSide(1, ft.Colors.BLUE_100),
left=ft.BorderSide(3, ft.Colors.BLUE_400), # acento visual izquierdo
),
)
icono_expand = ft.Icon(ft.Icons.KEYBOARD_ARROW_DOWN, size=16, color=ft.Colors.GREY_400)
def toggle_detalles(e, p=panel_detalles, i=icono_expand):
p.visible = not p.visible
i.name = ft.Icons.KEYBOARD_ARROW_UP if p.visible else ft.Icons.KEYBOARD_ARROW_DOWN
self.page.update()
fila_header = ft.Container(
content=ft.Row([
ft.Text(pedimento, width=150, size=12, color=ft.Colors.BLACK),
ft.Text(factura, width=120, size=12, color=ft.Colors.BLACK),
ft.Text(remesa, width=80, size=12, color=ft.Colors.BLACK),
ft.Text(estatus, expand=True, size=12, color=ft.Colors.BLUE_700, weight="bold"),
icono_expand,
]),
padding=ft.padding.symmetric(horizontal=20, vertical=8),
border=ft.border.only(bottom=ft.BorderSide(1, ft.Colors.GREY_100)),
on_click=toggle_detalles,
ink=True,
tooltip="Clic para ver detalles"
)
# Agrupamos header + panel en una Column
fila_completa = ft.Column([fila_header, panel_detalles], spacing=0)
self.lista_resultados.controls.append(fila_completa)
self.page.update()
if self.page:
self.page.run_task(_do)
# ------------------------------------------------------------------ #
def actualizar_estado_fila(self, indice, estatus):
async def _do():
if indice < len(self.lista_resultados.controls):
# controls[indice] = Column → controls[0] = header Container → Row → controls[3] = Text estatus
fila_header = self.lista_resultados.controls[indice].controls[0]
fila_header.content.controls[3].value = estatus
self.page.update()
if self.page:
self.page.run_task(_do)
def actualizar_progreso(self, valor):
async def _do():
self.barra_progreso.value = valor
self.porcentaje_label.value = f"{int(valor * 100)}%"
self.page.update()
if self.page:
self.page.run_task(_do)
def actualizar_estado_texto(self, texto):
async def _do():
self.status_label.value = texto
self.page.update()
if self.page:
self.page.run_task(_do)
# ------------------------------------------------------------------ #
def _hilo_sincronizacion(self):
try:
if self.cliente:
self.cliente.proceso_principal()
async def _done():
self.status_label.value = "Listo"
self.barra_progreso.value = 1.0
self.porcentaje_label.value = "100%"
self.page.update()
if self.page:
self.page.run_task(_done)
except Exception as ex:
print(f"[UI ERROR] {ex}")
async def _err():
self.status_label.value = f"Error: {ex}"
self.page.update()
if self.page:
self.page.run_task(_err)
def lanzar_interfaz(cliente=None):
app = InterfazFlet(cliente)
if cliente:
cliente.gui = app
ft.app(target=app.main)
return app
if __name__ == "__main__":
ft.app(target=InterfazFlet().main)

38
logger_utils.py Normal file
View File

@@ -0,0 +1,38 @@
import logging
import os
import sys
def setup_logger():
# Obtener la ruta del ejecutable o del script
if getattr(sys, 'frozen', False):
# Si es un ejecutable generado por PyInstaller
application_path = os.path.dirname(sys.executable)
else:
# Si es el script de Python normal
application_path = os.path.dirname(os.path.abspath(__file__))
log_file = os.path.join(application_path, "debug_sync.log")
# Crear el logger
logger = logging.getLogger("SCAII_Sync")
logger.setLevel(logging.DEBUG)
# Evitar duplicados si se llama varias veces
if not logger.handlers:
# Formato: [2024-05-06 10:00:00] [INFO] Mensaje
formatter = logging.Formatter('[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
# Handler para archivo (modo 'a' para anexar, 'w' para sobrescribir en cada inicio)
file_handler = logging.FileHandler(log_file, mode='w', encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# Handler para consola
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return logger
# Instancia global para importar en otros archivos
log = setup_logger()

59
main.py Executable file
View File

@@ -0,0 +1,59 @@
#!/usr/bin/env python3
import sys
from config import ConfiguracionApp
from gestor_base_datos import GestorBaseDatos
from motor_sincronizacion import MotorSincronizacion
from cliente_api import ClienteAPI
from interfaz import lanzar_interfaz
from logger_utils import log
class ClienteSincronizacionSCAII:
def __init__(self):
self.config = ConfiguracionApp()
self.config.cargar()
self.db = GestorBaseDatos(self.config.datos)
# Cliente API con sistema correcto para Login
self.api = ClienteAPI(self.config.datos)
# Motor vinculado al API
self.motor = MotorSincronizacion(self.db, self.api)
self.gui = None
def iniciar(self):
# 1. Conexión a la DB
if not self.db.conectar():
log.error("No se pudo conectar a la base de datos al iniciar.")
return
# 2. Lanzar la Interfaz (Flet)
log.info("Lanzando Interfaz gráfica (Flet)...")
self.gui = lanzar_interfaz(self)
def proceso_principal(self):
"""Orquestación delegada al Motor"""
# Vincular callbacks a la UI
if self.gui:
self.motor.callback_progreso = self.gui.actualizar_progreso
self.motor.callback_fila = self.gui.agregar_fila
self.motor.callback_actualizar_fila = self.gui.actualizar_estado_fila
self.motor.callback_estado_texto = self.gui.actualizar_estado_texto
# Ejecutar la lógica robusta del motor (que ya tiene el debug)
exito = self.motor.proceso_principal()
if exito:
if self.gui:
self.gui.status_label.value = "¡Sincronización Exitosa!"
self.gui.page.update()
else:
log.error("El proceso de sincronización falló.")
if self.gui:
self.gui.status_label.value = "Error en Proceso (Ver debug_sync.log)"
self.gui.page.update()
if __name__ == "__main__":
app = ClienteSincronizacionSCAII()
app.iniciar()

218
motor_sincronizacion.py Normal file
View File

@@ -0,0 +1,218 @@
import time
import os
from typing import List, Dict, Any, Tuple, Optional
from registros.scaii.reg_501 import Generador501 as Gen501SCAII
from registros.scaf.reg_501 import Generador501SCAF as Gen501SCAF
from registros.scaii.reg_502 import Generador502SCAII as Gen502SCAII
from registros.scaf.reg_502 import Generador502SCAF as Gen502SCAF
from registros.scaii.reg_503 import Generador503SCAII as Gen503SCAII
from registros.scaf.reg_503 import Generador503SCAF as Gen503SCAF
from registros.scaii.reg_504 import Generador504SCAII as Gen504SCAII
from registros.scaf.reg_504 import Generador504SCAF as Gen504SCAF
from registros.scaii.reg_505 import Generador505SCAII as Gen505SCAII
from registros.scaf.reg_505 import Generador505SCAF as Gen505SCAF
from registros.scaii.reg_551 import Generador551SCAII as Gen551SCAII
from registros.scaf.reg_551 import Generador551SCAF as Gen551SCAF
from registros.scaii.reg_554 import Generador554SCAII as Gen554SCAII
from registros.scaf.reg_554 import Generador554SCAF as Gen554SCAF
from registros.scaii.reg_558 import Generador558 as Gen558SCAII
from registros.scaf.reg_558 import Generador558SCAF as Gen558SCAF
from logger_utils import log
class MotorSincronizacion:
def __init__(self, gestor_db, cliente_api):
self.db = gestor_db
self.api = cliente_api
self.tipo_moneda_global = "ME"
self.callback_progreso = None
self.callback_fila = None
self.callback_estado_texto = None
self.callback_actualizar_fila = None
def proceso_principal(self):
"""Metodo de entrada robusto para la Interfaz"""
config_datos = self.db.config
pedimento = config_datos.get("pedimento")
sistema = config_datos.get("sistema", "SCAII")
log.info(f"Iniciando proceso principal para Pedimento: {pedimento}, Sistema: {sistema}")
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto("Iniciando conexiones...")
# 1. Conectar a la Base de Datos
if not self.db.conectar():
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto("Error: Fallo conexión SQL Server.")
return False
# 1b. Leer credenciales RPA de GEmpresa (igual que WinDev en producción)
# Si vienen por parámetro (/RPAUser), se usan esas. Si no, se leen de la BD.
if not self.api.usuario or not self.api.password:
rpa_user, rpa_pass = self.db.obtener_credenciales_rpa()
if rpa_user and rpa_pass:
self.api.usuario = rpa_user
self.api.password = rpa_pass
else:
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto("Error: No hay credenciales RPA en GEmpresa.")
log.error("No se encontraron credenciales RPA en GEmpresa. Configura NOMBREBDINTER y CLAVEPREVALIDADOR.")
return False
# 2. Obtener Token de la API
if not self.api.obtener_token():
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto("Error: Fallo de autenticación API.")
return False
if not pedimento: return False
return self.ejecutar_logica_generacion(pedimento, sistema)
def ejecutar_logica_generacion(self, pedimento: str, sistema: str) -> bool:
"""Orquestador Gen_Regs"""
# 1. Buscar pedimento (Intento 1: Tal cual viene)
datos_pedi = self.db.obtener_datos_pedimento(pedimento, sistema)
# 2. Buscar pedimento (Intento 2: Sin guiones)
if not datos_pedi and "-" in pedimento:
log.info("Intento 2: Buscando pedimento sin guiones...")
datos_pedi = self.db.obtener_datos_pedimento(pedimento.replace("-", ""), sistema)
# 3. Buscar pedimento (Intento 3: Solo los últimos 7 dígitos - El número consecutivo)
if not datos_pedi and len(pedimento) >= 7:
solo_consecutivo = pedimento.replace("-", "").replace(" ", "")[-7:]
log.info(f"Intento 3: Buscando solo por consecutivo final: {solo_consecutivo}...")
datos_pedi = self.db.obtener_datos_pedimento(solo_consecutivo, sistema)
if not datos_pedi:
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto(f"Error: Pedimento {pedimento} no encontrado.")
return False
tipo_op = datos_pedi.get("Tipo")
regimen = datos_pedi.get("Regimen", "")
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto(f"Pedimento {tipo_op}/{regimen} encontrado. Buscando facturas...")
log.info(f"Pedimento encontrado: {tipo_op}/{regimen}. Buscando facturas...")
datos_empresa = self.db.obtener_datos_empresa(sistema)
facturas = self.db.obtener_detalle_facturas(pedimento, sistema, regimen, tipo_op)
if not facturas:
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto(f"Aviso: 0 facturas para {pedimento}")
return False
self.facturas_cache = facturas
totales = self.calcular_totales(facturas, self.tipo_moneda_global)
id_proveedor = facturas[0].get("IdProveedor")
datos_cliente = self.db.obtener_datos_proveedor(id_proveedor, sistema)
total_trailers = self.db.obtener_total_vehiculos(pedimento, sistema)
gen_501 = Gen501SCAII if sistema == "SCAII" else Gen501SCAF
json_501 = gen_501.preparar_json(datos_pedi, totales, datos_cliente, total_trailers)
id_pedimento_api = self.api.enviar_registro_api("registro501", json_501)
if not id_pedimento_api: return False
num_trailer = facturas[0].get("NumTrailer")
datos_trailer = self.db.obtener_datos_trailer(num_trailer, sistema)
gen_502 = Gen502SCAII if sistema == "SCAII" else Gen502SCAF
self.api.enviar_registro_api("registro502", gen_502.preparar_json(id_pedimento_api, datos_trailer))
gen_503 = Gen503SCAII if sistema == "SCAII" else Gen503SCAF
self.api.enviar_registro_api("registro503", gen_503.preparar_json(id_pedimento_api, facturas[0]))
log.info(f"Procesando {len(facturas)} facturas...")
for idx, fac in enumerate(facturas):
self.idx_actual = idx
if self.callback_fila:
self.callback_fila(pedimento, str(fac.get("NumFactura")), str(fac.get("Remesa")), fac, "Procesando...")
time.sleep(0.02)
gen_504 = Gen504SCAII if sistema == "SCAII" else Gen504SCAF
for j504 in gen_504.preparar_listado_json(id_pedimento_api, fac.get("ContenedoresTipo")):
self.api.enviar_registro_api("registro504", j504)
id_prov = fac.get("IdProveedor")
prov_data = self.db.obtener_datos_proveedor(id_prov, sistema)
trans_data = self.db.obtener_datos_transporte(fac.get("Transportista"), sistema)
gen_505 = Gen505SCAII if sistema == "SCAII" else Gen505SCAF
json_505 = gen_505.preparar_json(fac, id_pedimento_api, datos_empresa, prov_data, trans_data, self.tipo_moneda_global, 1.0)
json_505["tipo_operacion_505"] = tipo_op
id_factura_api = self.api.enviar_registro_api("registro505", json_505)
if id_factura_api:
self.ejecutar_generacion_partidas(fac, id_factura_api, sistema, tipo_op, regimen, json_505)
if getattr(self, "callback_actualizar_fila", None):
self.callback_actualizar_fila(idx, "Enviada correctamente")
else:
if getattr(self, "callback_actualizar_fila", None):
self.callback_actualizar_fila(idx, "Error enviando")
if getattr(self, "callback_progreso", None):
self.callback_progreso((idx + 1) / len(facturas))
time.sleep(0.05)
log.info("Sincronización finalizada con éxito.")
return True
def calcular_totales(self, facturas: List[Dict[str, Any]], moneda_objetivo: str) -> Dict[str, float]:
totales = {"peso_bruto": 0.0, "bultos": 0.0, "flete": 0.0, "seguros": 0.0, "embalajes": 0.0, "otros_incrementa": 0.0}
for fac in facturas:
totales["peso_bruto"] += float(fac.get("PesoBruto") or 0)
totales["bultos"] += float(fac.get("CANTBULTOS") or 0)
tc = float(fac.get("TipoCambio") or 1.0)
m_fac = fac.get("ClaveMoneda")
def conv(v):
val = float(v or 0)
if moneda_objetivo == "ME": return val if m_fac == "ME" else val / tc
return val * tc if m_fac == "MN" else val
totales["flete"] += conv(fac.get("Flete"))
totales["seguros"] += conv(fac.get("Seguros"))
totales["embalajes"] += conv(fac.get("Embalajes"))
totales["otros_incrementa"] += conv(fac.get("OtrosIncrementa"))
return totales
def realizar_conversion_tarifa(self, cant_c: float, um_c: str, frac: str, peso: float, sist: str) -> Tuple[float, str]:
cl_um = self.db.obtener_um_tarifa_fraccion(frac, sist)
if not cl_um: return cant_c, ""
if str(cl_um) == "1": return peso, "1"
um_s = self.db.obtener_unidad_aduana_scaii(clave_aduana=cl_um)
if not um_s or um_s == um_c: return cant_c, str(cl_um)
t, f = self.db.obtener_conversion_unidades(um_c, um_s)
cant_t = cant_c * f if t == "M" else (cant_c / f if f != 0 else cant_c)
return cant_t, str(cl_um)
def ejecutar_generacion_partidas(self, fac: Dict[str, Any], id_f: int, sist: str, t_op: str, reg: str, j505: Dict[str, Any]):
partidas = self.db.obtener_partidas_factura(fac.get("Consecutivo"), sist, t_op, reg)
if not partidas: return
for i, par in enumerate(partidas, start=1):
if getattr(self, "callback_estado_texto", None):
self.callback_estado_texto(f"Enviando Partida {i}/{len(partidas)}...")
# Obtener datos de la unidad de medida para el mapeo numérico
um_raw = par.get("UnidadMedida", "")
um_data = self.db.obtener_datos_unidad_medida(um_raw, sist) if um_raw else None
ct, ut = self.realizar_conversion_tarifa(float(par.get("Cantidad") or 0), par.get("UnidadMedida", ""), par.get("Fraccion", ""), float(par.get("PesoNeto") or 0), sist)
if sist == "SCAII":
j551 = Gen551SCAII.preparar_json(par, id_f, {}, um_data, str(fac.get("MetodoValoracion") or ""), j505["moneda_facturacion_505"], fac.get("Remesa"), i, fac.get("TipoMoneda", "ME"))
else:
j551 = Gen551SCAF.preparar_json(par, id_f, {}, um_data, str(fac.get("MetodoValoracion") or ""), j505["moneda_facturacion_505"], fac.get("Remesa"), i)
j551["cantidad_tarifa_551"] = ct
if ut: j551["unidad_medida_tarifa_551"] = ut
j551["numero_partida_remesa_551"] = str(par.get("Linea", i))
id_p = self.api.enviar_registro_api("registro551", j551)
if id_p:
mod = ("SCAII" if sist == "SCAII" else "SCAF") + ("-IT" if t_op == "I" and reg != "IMD" else ("-ID" if t_op == "I" else ("-E" if sist == "SCAII" else "-ER")))
for iden in self.db.obtener_identificadores_partida(fac.get("Consecutivo"), par.get("Linea"), mod):
self.api.enviar_registro_api("registro554", (Gen554SCAII if sist == "SCAII" else Gen554SCAF).preparar_json(iden, id_p))
if par.get("InfoAdic", "").strip():
self.api.enviar_registro_api("registro558", (Gen558SCAII if sist == "SCAII" else Gen558SCAF).preparar_json(id_p, par.get("InfoAdic")))
time.sleep(0.01)

61
registros/scaf/reg_501.py Normal file
View File

@@ -0,0 +1,61 @@
from typing import Dict, Any
class Generador501SCAF:
@staticmethod
def preparar_json(datos_pedi: Dict[str, Any], totales: Dict[str, float],
datos_cliente: Dict[str, Any], total_trailers: int) -> Dict[str, Any]:
"""
Lógica específica para SCAF (Registro 501).
"""
# Lógica de Destino/Zona (SCAF a veces usa códigos distintos, pero el if parece igual)
destino = datos_pedi.get("OPCIONDESTINO", "")
codigo_destino = "9" if "Interior" in destino else "7" if any(x in destino for x in ["Región", "Franja"]) else ""
clave_transfer = datos_cliente.get("CLAVETRANSFER", "").strip() if datos_cliente else ""
# Construcción del Registro 501 SCAF
registro = {
"tipo_operacion_501": datos_pedi.get("Tipo"),
"clave_documento_501": datos_pedi.get("Clave"),
"numero_pedimento_501": (datos_pedi.get("Pedimento") or "")[8:15],
"codigo_importador_exportador_501": clave_transfer if clave_transfer else " ",
"monto_fletes_501": totales["flete"],
"monto_seguros_501": totales["seguros"],
"costo_embalajes_501": totales["embalajes"],
"cargos_incrementables_501": totales["otros_incrementa"],
"cargos_deducibles_501": 0,
"peso_total_bruto_501": totales["peso_bruto"],
"total_bultos_501": totales["bultos"],
"metodo_transporte_entrada_501": datos_pedi.get("TransE"),
"metodo_transporte_arribo_501": datos_pedi.get("TransA"),
"metodo_transporte_salida_501": datos_pedi.get("TransS"),
"codigo_destino_zona_501": codigo_destino,
"numero_referencia_501": "",
"moneda_costos_incrementables_501": "ME",
"codigo_aduana_despacho_501": datos_pedi.get("Aduana"),
"numero_patente_501": (datos_pedi.get("Pedimento") or "")[3:7],
"origenPedimento": 1,
"total_vehiculos_501": total_trailers,
"rfc_importador_exportador_501": " ",
"curp_importador_exportador_501": " ",
"nombre_completo_importador_exportador_501": " ",
"calle_direccion_importador_exportador_501": " ",
"numero_exterior_direccion_importador_exportador_501": " ",
"numero_interior_direccion_importador_exportador_501": " ",
"colonia_direccion_importador_exportador_501": " ",
"localidad_direccion_importador_exportador_501": " ",
"municipio_direccion_importador_exportador_501": " ",
"estado_direccion_importador_exportador_501": " ",
"pais_direccion_importador_exportador_501": " ",
"codigo_postal_direccion_importador_exportador_501": " ",
"monto_fletes_decrementables_501": 0,
"monto_seguros_decrementables_501": 0,
"monto_carga_decrementables_501": 0,
"monto_descarga_decrementables_501": 0,
"monto_otros_decrementables_501": 0,
"MarcaBulto1_501": "S/M", # SCAF usa S/M mayormente
"MarcaBulto2_501": "S/N"
}
return registro

18
registros/scaf/reg_502.py Normal file
View File

@@ -0,0 +1,18 @@
from typing import Dict, Any
class Generador502SCAF:
@staticmethod
def preparar_json(id_pedimento: int, datos_transporte: Dict[str, Any]) -> Dict[str, Any]:
"""
Lógica para Registro 502 (Transporte) - SCAF.
"""
placas = datos_transporte.get("NUMEROPLACAS", " ") if datos_transporte else " "
pais = datos_transporte.get("PAIS", " ") if datos_transporte else " "
registro = {
"id_pedimento": id_pedimento,
"numero_placas_502": placas,
"codigo_pais_transporte_502": pais
}
return registro

21
registros/scaf/reg_503.py Normal file
View File

@@ -0,0 +1,21 @@
from typing import Dict, Any
class Generador503SCAF:
@staticmethod
def preparar_json(id_pedimento: int, factura_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Lógica para Registro 503 (Guías) - SCAF.
"""
num_niu = factura_data.get("NumNIU", " ")
tipo_guia = factura_data.get("TipoGuia", " ")
total_guias = factura_data.get("CantGuias", 0)
registro = {
"id_pedimento": id_pedimento,
"numero_guia_manifiesto_embarque_503": num_niu if num_niu else " ",
"identificador_guia_503": tipo_guia if tipo_guia else " ",
"total_guias_503": total_guias if total_guias else 0
}
return registro

33
registros/scaf/reg_504.py Normal file
View File

@@ -0,0 +1,33 @@
from typing import List, Dict, Any
class Generador504SCAF:
@staticmethod
def preparar_listado_json(id_pedimento: int, contenedores_raw: str) -> List[Dict[str, Any]]:
"""
Procesa el string de contenedores (CONT1|TIPO1,CONT2|TIPO2)
y genera una lista de registros para la API - Versión SCAF.
"""
registros = []
if not contenedores_raw or contenedores_raw.strip() == "":
return registros
bloques = contenedores_raw.split(",")
for bloque in bloques:
if not bloque.strip():
continue
partes = bloque.split("|")
num_contenedor = partes[0].strip() if len(partes) > 0 else ""
tipo_contenedor = partes[1].strip() if len(partes) > 1 else " "
if num_contenedor:
registros.append({
"id_pedimento_501": id_pedimento,
"numero_contenedor_504": num_contenedor,
"tipo_contenedor_504": tipo_contenedor,
"status_504": "1"
})
return registros

104
registros/scaf/reg_505.py Normal file
View File

@@ -0,0 +1,104 @@
from typing import Dict, Any, Optional
class Generador505SCAF:
@staticmethod
def preparar_json(fac: Dict[str, Any], id_pedimento: int, datos_empresa: Dict[str, Any],
datos_prov: Dict[str, Any], datos_trans: Optional[Dict[str, Any]],
tipo_moneda_global: str, tc_global: float) -> Dict[str, Any]:
"""
Lógica para Registro 505 (Facturas) - SCAF.
"""
clave_moneda = fac.get("ClaveMoneda", "")
valor_mc = fac.get("ValorImpoMC", 0)
valor_me = fac.get("ValorImpoME", 0)
if tipo_moneda_global == "ME":
clave_moneda = "USD"
valor_mc = valor_me
elif tipo_moneda_global == "MN":
clave_moneda = "MXP"
valor_mc = fac.get("ValorImpoMN", 0)
# SCAF usa el mismo mapeo de Valor Aduana que SCAII
met_val_raw = fac.get("MetValor", 0)
met_val_map = {1: "VTM", 2: "VMI", 3: "VMS", 4: "VPU", 5: "VR", 6: "A78"}
met_val_str = f"VALADU.{met_val_map.get(met_val_raw, met_val_raw)}"
# 3. Incoterm
incoterm = fac.get("Incoterm", "")
incoterm_full = incoterm if incoterm else ""
# 4. Otros mapeos
obs = fac.get("ObservacionesVU", "").replace("\r", "").replace("\n", "").replace("\t", "")
contenedores = fac.get("ContenedoresTipo", "").replace(",", "|")
registro = {
"numero_factura_505": fac.get("NumFactura"),
"fecha_facturacion_505": str(fac.get("FechaFactura"))[:10],
"termino_facturacion_505": incoterm if incoterm else "",
"moneda_facturacion_505": clave_moneda if clave_moneda else "",
"valor_moneda_facturacion_505": str(float(valor_mc or 0)),
"valor_dolares_505": str(float(valor_me or 0)),
"codigo_proveedor_comprador_505": datos_prov.get("CLAVETRANSFER", ""),
"e_document_505": fac.get("EDocument", ""),
"serie_certificado_cove_505": fac.get("NumeroCertificado", ""),
"firma_electronica_cove_505": fac.get("FirmaElectronica", ""),
"observaciones_factura_505": obs if obs else "",
"numero_economico_505": contenedores if contenedores else "",
"tipo_contenedor_numero_economico_505": str(fac.get("Remesa") or ""),
"numero_contenedor_505": contenedores if contenedores else "",
"tipo_contenedor_505": str(fac.get("Remesa") or ""),
"numero_remesa_505": str(fac.get("Remesa") or ""),
"tipo_operacion_505": "",
"certificado_origen_505": "S" if fac.get("FUNGECOMOCO") else "N",
"numero_certificado_origen_505": "",
"numero_exportador_confiable_505": datos_empresa.get("NUMDEEXPORTADORCONFIABLE") or "",
"incoterm_505": incoterm_full,
"metodo_valoracion_505": met_val_str,
"vinculacion_505": str(datos_prov.get("VINCULACION", 0)),
"subdivision_505": "S" if fac.get("SUBDIVISION") else "N",
"tipo_cambio_usd_505": str(float(fac.get("TipoCambio") or 0)),
"se_genero_en_contingencia_505": "S" if fac.get("MODOCONTINGENCIA") else "N",
"operacion_cove_505": fac.get("NUMOPERACIONVU", ""),
"transportista_clave_505": fac.get("TRANSPORTISTA", ""),
"rfc_505": datos_trans.get("RFC", "") if datos_trans else "",
"caat_505": datos_trans.get("CODIGOCAAT", "") if datos_trans else "",
"id_pedimento": id_pedimento,
"PaisFactura_505": datos_prov.get("PAIS", ""),
"Bultos_cantidad_505": str(fac.get("CANTBULTOS", 0)),
"provcomp_numregidtrib_505": " ",
"provcomp_rfc_505": " ",
"provcomp_curp_505": " ",
"provcomp_nombre_505": " ",
"provcomp_calle_505": " ",
"provcomp_numero_exterior_505": " ",
"provcomp_numero_interior_505": " ",
"provcomp_colonia_505": " ",
"provcomp_localidad_505": " ",
"provcomp_municipio_505": " ",
"provcomp_estado_505": " ",
"provcomp_pais_505": " ",
"provcomp_codigo_postal_505": " ",
"destinatario_clave_505": " ",
"destinatario_numregidtrib_505": " ",
"destinatario_rfc_505": " ",
"destinatario_curp_505": " ",
"destinatario_nombre_505": " ",
"destinatario_calle_505": " ",
"destinatario_numero_exterior_505": " ",
"destinatario_numero_interior_505": " ",
"destinatario_colonia_505": " ",
"destinatario_localidad_505": " ",
"destinatario_referencia_505": " ",
"destinatario_municipio_505": " ",
"destinatario_estado_505": " ",
"destinatario_pais_505": " ",
"destinatario_codigo_postal_505": " ",
"uuid_505": " ",
"path_pdf_cfdi_505": " ",
"path_xml_cfdi_505": " ",
"estatus_cove_505": " "
}
return registro

101
registros/scaf/reg_551.py Normal file
View File

@@ -0,0 +1,101 @@
from typing import Dict, Any, Optional
class Generador551SCAF:
@staticmethod
def preparar_json(par: Dict[str, Any], id_factura: int, datos_parte: Dict[str, Any],
um_data: Optional[Dict[str, Any]], met_val: str, cla_mon: str,
num_remesa: int, num_partida_remesa: int) -> Dict[str, Any]:
"""
Lógica para Registro 551 (Partidas) - SCAF.
Homologado con la lógica de pesos y valores de WinDev.
"""
# 1. Lógica de Fracción (Regla Octava)
frac_raw = par.get("Fraccion", "")
frac_oct = par.get("FraccionOctava", "")
frac_final = frac_oct if frac_oct and frac_oct.strip() else frac_raw
# 2. Descripciones
desc_esp = par.get("Descripcion", " ").replace("\r", "").replace("\n", "").replace("\t", "")
# 3. Arancel (Mapping)
tipo_frac = par.get("TipoFracc", "")
arancel_map = {"TLCS": 3, "GENERAL": 0, "PROSEC": 1}
id_arancel = str(arancel_map.get(tipo_frac, 0))
# 4. Unidad de Medida OMA y Comercial
oma_final = str(par.get("ClaveOMA") or " ").strip()
oma_raw = str(par.get("UnidadMedida", "")).strip().upper()
if not oma_final or oma_final == "None":
oma_final = "C62_1" if oma_raw in ["6", "PZA"] else oma_raw
map_um_num = {
"PZA": "6", "PIEZA": "6", "PZ": "6", "PC": "6", "PIEZAS": "6",
"KGS": "1", "KGM": "1", "KG": "1", "KILOS": "1", "KILO": "1",
"LTS": "5", "L": "5", "LITROS": "5", "LITRO": "5",
"MTR": "2", "M": "2", "METROS": "2", "METRO": "2",
"PAR": "7", "PARES": "7",
"CJ": "3", "CAJA": "3", "CAJAS": "3",
"JGO": "12", "JUEGO": "12", "SET": "12", "KIT": "12"
}
# Priorizar la clave que venga de la base de datos (CLAVEOMA)
if um_data and um_data.get("CLAVEOMA"):
um_comercial_num = str(um_data["CLAVEOMA"]).strip()
else:
um_comercial_num = map_um_num.get(oma_raw, oma_raw)
registro = {
"id_factura": id_factura,
"fraccion_551": str(frac_final),
"descripcion_mercancia_551": desc_esp[:250],
"numero_parte_551": str(par.get("NumParte") or " "),
"valor_mercancia_551": float(par.get("ValorDolares") or 0), # SCAF suele usar Dolares para MC
"cantidad_comercial_551": float(par.get("Cantidad") or 0),
"unidad_medida_comercial_551": um_comercial_num,
"cantidad_tarifa_551": float(par.get("Cantidad") or 0),
"umt_551": " ",
"valor_agregado_551": "0",
"vinculacion_551": "0",
"metodo_valoracion_551": str(met_val),
"marca_551": str(par.get("Marca") or " ").strip() or " ",
"modelo_551": str(par.get("Modelo") or " ").strip() or " ",
"pais_origen_destino_551": str(par.get("PaisOrigen") or " "),
"pais_comprador_vendedor_551": "MEX",
"entidad_federativa_origen_551": " ",
"entidad_federativa_destino_551": " ",
"entidad_federativa_comprador_551": " ",
"entidad_federativa_vendedor_551": " ",
"identificador_arancel_aplicar_551": id_arancel,
"clave_industria_551": str(par.get("Sector") or " "),
"peso_bruto_551": float(par.get("PesoBruto") or 0),
"uso_mercancia_551": " ",
"estado_mercancia_551": " ",
"moneda_partida_551": str(cla_mon),
"numero_factura_551": str(par.get("NumFactura") or " "),
"tipo_bulto_551": " ",
"cantidad_bultos_551": float(par.get("CantBultos") or 0),
# --- PARIDAD WINDEV ---
"peso_bruto_kg_551": str(float(par.get("PesoNeto") or 0)),
"peso_neto_kg_551": str(float(par.get("ValorAduanasMN") or 0)),
"numero_remesa_551": str(num_remesa or " "),
"numero_partida_remesa_551": str(num_partida_remesa),
"cantidad_alterna_551": float(par.get("CantidadAlterna") or 0),
"unidad_medida_alterna_551": str(par.get("UnidadAlterna", " ")),
"descripcion_ingles_551": " ",
"identificador_sistema_scaii_scaf_551": "SCAF",
"vu_submodelo_551": " ",
"vu_serie_551": " ",
"valor_unitario_aduana_551": str(float(par.get("CostoUnitarioME") or 0)),
"valor_dolares_551": str(float(par.get("ValorDolares") or 0)),
"unidad_medida_oma_551": oma_final,
"valor_dolares_cove_551": "0",
"subdivision_fraccion_551": str(frac_final)[-2:] if len(str(frac_final)) >= 2 else "00",
"codigo_carta_porte_551": " ",
"nodo_551": "85858585",
"CantidadComercialOMA_551": str(float(par.get("Cantidad") or 0))
}
return registro

16
registros/scaf/reg_554.py Normal file
View File

@@ -0,0 +1,16 @@
from typing import Dict, Any
class Generador554SCAF:
@staticmethod
def preparar_json(iden: Dict[str, Any], id_partida: int) -> Dict[str, Any]:
"""
Lógica para Registro 554 (Identificadores de Partidas) - SCAF.
"""
registro = {
"tipo_caso_554": iden.get("identificador", " "),
"complemento_caso_1_554": iden.get("complemento_1", " "),
"complemento_caso_2_554": iden.get("complemento_2", " "),
"complemento_caso_3_554": iden.get("complemento_3", " "),
"id_partida": id_partida
}
return registro

17
registros/scaf/reg_558.py Normal file
View File

@@ -0,0 +1,17 @@
from typing import Dict, Any
class Generador558SCAF:
@staticmethod
def preparar_json(id_partida: int, observaciones: str) -> Dict[str, Any]:
"""
Lógica para Registro 558 (Observaciones de Partidas) - SCAF.
"""
if not observaciones:
observaciones = " "
obs_limpia = observaciones.replace("\r", "").replace("\n", "").replace("\t", "").strip()
return {
"observaciones_558": obs_limpia[:1000],
"id_partida": id_partida
}

62
registros/scaii/reg_501.py Normal file
View File

@@ -0,0 +1,62 @@
from typing import Dict, Any
class Generador501:
@staticmethod
def preparar_json(datos_pedi: Dict[str, Any], totales: Dict[str, float],
datos_cliente: Dict[str, Any], total_trailers: int) -> Dict[str, Any]:
"""
Unifica la lógica de Get_Regs501, _Def y _Expo en una sola estructura.
"""
# Lógica de Destino/Zona (Opcional)
destino = datos_pedi.get("Destino", "")
codigo_destino = "9" if "Interior" in destino else "7" if any(x in destino for x in ["Región", "Franja"]) else ""
# Clave del Importador/Exportador
clave_transfer = datos_cliente.get("CLAVETRANSFER", "").strip() if datos_cliente else ""
# Construcción del Registro 501
registro = {
"tipo_operacion_501": datos_pedi.get("Tipo") or "",
"clave_documento_501": datos_pedi.get("Clave") or "",
"numero_pedimento_501": (datos_pedi.get("Pedimento") or "")[8:15],
"codigo_importador_exportador_501": clave_transfer if clave_transfer else " ",
"monto_fletes_501": totales["flete"],
"monto_seguros_501": totales["seguros"],
"costo_embalajes_501": totales["embalajes"],
"cargos_incrementables_501": totales["otros_incrementa"],
"cargos_deducibles_501": 0,
"peso_total_bruto_501": totales["peso_bruto"],
"total_bultos_501": totales["bultos"],
"metodo_transporte_entrada_501": str(datos_pedi.get("TransE") or ""),
"metodo_transporte_arribo_501": str(datos_pedi.get("TransA") or ""),
"metodo_transporte_salida_501": str(datos_pedi.get("TransS") or ""),
"codigo_destino_zona_501": codigo_destino,
"numero_referencia_501": "",
"moneda_costos_incrementables_501": "ME", # O tomar de la primera factura
"codigo_aduana_despacho_501": str(datos_pedi.get("Aduana") or ""),
"numero_patente_501": (datos_pedi.get("Pedimento") or "")[3:7],
"origenPedimento": 1,
"total_vehiculos_501": total_trailers,
"rfc_importador_exportador_501": " ",
"curp_importador_exportador_501": " ",
"nombre_completo_importador_exportador_501": " ",
"calle_direccion_importador_exportador_501": " ",
"numero_exterior_direccion_importador_exportador_501": " ",
"numero_interior_direccion_importador_exportador_501": " ",
"colonia_direccion_importador_exportador_501": " ",
"localidad_direccion_importador_exportador_501": " ",
"municipio_direccion_importador_exportador_501": " ",
"estado_direccion_importador_exportador_501": " ",
"pais_direccion_importador_exportador_501": " ",
"codigo_postal_direccion_importador_exportador_501": " ",
"monto_fletes_decrementables_501": 0,
"monto_seguros_decrementables_501": 0,
"monto_carga_decrementables_501": 0,
"monto_descarga_decrementables_501": 0,
"monto_otros_decrementables_501": 0,
"MarcaBulto1_501": "S/N",
"MarcaBulto2_501": "S/N"
}
return registro

20
registros/scaii/reg_502.py Normal file
View File

@@ -0,0 +1,20 @@
from typing import Dict, Any
class Generador502SCAII:
@staticmethod
def preparar_json(id_pedimento: int, datos_transporte: Dict[str, Any]) -> Dict[str, Any]:
"""
Lógica para Registro 502 (Transporte) - SCAII.
"""
# En WinDev: GTrailers.NUMEROPLACAS si existe, si no " "
placas = datos_transporte.get("NUMEROPLACAS", " ") if datos_transporte else " "
pais = datos_transporte.get("PAIS", " ") if datos_transporte else " "
registro = {
"id_pedimento": id_pedimento,
"numero_placas_502": placas,
"codigo_pais_transporte_502": pais
}
return registro

21
registros/scaii/reg_503.py Normal file
View File

@@ -0,0 +1,21 @@
from typing import Dict, Any
class Generador503SCAII:
@staticmethod
def preparar_json(id_pedimento: int, factura_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Lógica para Registro 503 (Guías) - SCAII.
"""
num_niu = factura_data.get("num_niu", " ")
tipo_guia = factura_data.get("tipo_guia", " ")
total_guias = factura_data.get("total_guias", 0)
registro = {
"id_pedimento": id_pedimento,
"numero_guia_manifiesto_embarque_503": num_niu if num_niu else " ",
"identificador_guia_503": tipo_guia if tipo_guia else " ",
"total_guias_503": total_guias if total_guias else 0
}
return registro

35
registros/scaii/reg_504.py Normal file
View File

@@ -0,0 +1,35 @@
from typing import List, Dict, Any
class Generador504SCAII:
@staticmethod
def preparar_listado_json(id_pedimento: int, contenedores_raw: str) -> List[Dict[str, Any]]:
"""
Procesa el string de contenedores (CONT1|TIPO1,CONT2|TIPO2)
y genera una lista de registros para la API.
"""
registros = []
if not contenedores_raw or contenedores_raw.strip() == "":
return registros
# En WinDev: FOR EACH STRING sContenedores OF DS_Facturas.c1 SEPARATED BY ","
bloques = contenedores_raw.split(",")
for bloque in bloques:
if not bloque.strip():
continue
# En WinDev: (sNumeroContenedor,sTipoContenedor) = StringSplit(sContenedores,"|")
partes = bloque.split("|")
num_contenedor = partes[0].strip() if len(partes) > 0 else ""
tipo_contenedor = partes[1].strip() if len(partes) > 1 else " "
if num_contenedor:
registros.append({
"id_pedimento_501": id_pedimento,
"numero_contenedor_504": num_contenedor,
"tipo_contenedor_504": tipo_contenedor,
"status_504": "1"
})
return registros

106
registros/scaii/reg_505.py Normal file
View File

@@ -0,0 +1,106 @@
from typing import Dict, Any, Optional
class Generador505SCAII:
@staticmethod
def preparar_json(fac: Dict[str, Any], id_pedimento: int, datos_empresa: Dict[str, Any],
datos_prov: Dict[str, Any], datos_trans: Optional[Dict[str, Any]],
tipo_moneda_global: str, tc_global: float) -> Dict[str, Any]:
"""
Lógica para Registro 505 (Facturas) - SCAII.
"""
# 1. Lógica de Moneda y Valores
clave_moneda = fac.get("ClaveMoneda", "")
valor_mc = fac.get("ValorMC", 0)
valor_me = fac.get("ValorImpoME", 0)
if tipo_moneda_global == "ME":
clave_moneda = "USD"
valor_mc = valor_me
elif tipo_moneda_global == "MN":
clave_moneda = "MXP"
# Asumiendo lógica estándar si no es FP
valor_mc = fac.get("ValorImpoMN", 0)
# 2. Lógica de Método de Valoración (Mapping)
met_val_raw = fac.get("MetodoValoracion", 0)
met_val_map = {1: "VTM", 2: "VMI", 3: "VMS", 4: "VPU", 5: "VR", 6: "A78"}
met_val_str = f"VALADU.{met_val_map.get(met_val_raw, met_val_raw)}"
# 3. Incoterm
incoterm = fac.get("Incoterm", "")
incoterm_full = incoterm if incoterm else ""
# 4. Otros mapeos
obs = fac.get("Observaciones", "").replace("\r", "").replace("\n", "").replace("\t", "")
contenedores = fac.get("ContenedoresTipo", "").replace(",", "|")
registro = {
"numero_factura_505": fac.get("NumFactura"),
"fecha_facturacion_505": str(fac.get("Fecha"))[:10], # YYYY-MM-DD
"termino_facturacion_505": incoterm if incoterm else "",
"moneda_facturacion_505": clave_moneda if clave_moneda else "",
"valor_moneda_facturacion_505": str(float(valor_mc or 0)),
"valor_dolares_505": str(float(valor_me or 0)),
"codigo_proveedor_comprador_505": datos_prov.get("CLAVETRANSFER", ""),
"e_document_505": fac.get("EDocument", ""),
"serie_certificado_cove_505": fac.get("NumCertificado", ""),
"firma_electronica_cove_505": fac.get("FirmaElectronica", ""),
"observaciones_factura_505": obs if obs else "",
"numero_economico_505": contenedores if contenedores else "",
"tipo_contenedor_numero_economico_505": str(fac.get("Remesa") or ""),
"numero_contenedor_505": contenedores if contenedores else "",
"tipo_contenedor_505": str(fac.get("Remesa") or ""),
"numero_remesa_505": str(fac.get("Remesa") or ""),
"tipo_operacion_505": "", # Se llenará en el motor
"certificado_origen_505": "S" if fac.get("FungeComoCO") else "N",
"numero_certificado_origen_505": "",
"numero_exportador_confiable_505": (datos_empresa.get("NUMDEEXPORTADORCONFIABLE") or "") if datos_empresa else "",
"incoterm_505": incoterm_full,
"metodo_valoracion_505": met_val_str,
"vinculacion_505": str(datos_prov.get("VINCULACION") or "0"),
"subdivision_505": "S" if fac.get("Subdivision") else "N",
"tipo_cambio_usd_505": str(float(fac.get("TipoCambio") or 0)),
"se_genero_en_contingencia_505": "S" if fac.get("ModoContingencia") else "N",
"operacion_cove_505": fac.get("NumOperacionVU") or "",
"transportista_clave_505": fac.get("Transportista") or "",
"rfc_505": (datos_trans.get("RFC") or "") if datos_trans else "",
"caat_505": (datos_trans.get("CODIGOCAAT") or "") if datos_trans else "",
"id_pedimento": id_pedimento,
"PaisFactura_505": datos_prov.get("PAIS", ""),
"Bultos_cantidad_505": str(fac.get("CantBultos", 0)),
"provcomp_numregidtrib_505": " ",
"provcomp_rfc_505": " ",
"provcomp_curp_505": " ",
"provcomp_nombre_505": " ",
"provcomp_calle_505": " ",
"provcomp_numero_exterior_505": " ",
"provcomp_numero_interior_505": " ",
"provcomp_colonia_505": " ",
"provcomp_localidad_505": " ",
"provcomp_municipio_505": " ",
"provcomp_estado_505": " ",
"provcomp_pais_505": " ",
"provcomp_codigo_postal_505": " ",
"destinatario_clave_505": " ",
"destinatario_numregidtrib_505": " ",
"destinatario_rfc_505": " ",
"destinatario_curp_505": " ",
"destinatario_nombre_505": " ",
"destinatario_calle_505": " ",
"destinatario_numero_exterior_505": " ",
"destinatario_numero_interior_505": " ",
"destinatario_colonia_505": " ",
"destinatario_localidad_505": " ",
"destinatario_referencia_505": " ",
"destinatario_municipio_505": " ",
"destinatario_estado_505": " ",
"destinatario_pais_505": " ",
"destinatario_codigo_postal_505": " ",
"uuid_505": " ",
"path_pdf_cfdi_505": " ",
"path_xml_cfdi_505": " ",
"estatus_cove_505": " "
}
return registro

144
registros/scaii/reg_551.py Normal file
View File

@@ -0,0 +1,144 @@
from typing import Dict, Any, Optional
class Generador551SCAII:
@staticmethod
def preparar_json(par: Dict[str, Any], id_factura: int, datos_parte: Dict[str, Any],
um_data: Optional[Dict[str, Any]], met_val: str, cla_mon: str,
num_remesa: int, num_partida_remesa: int, tipo_moneda_fac: str) -> Dict[str, Any]:
"""
Lógica para Registro 551 (Partidas) - SCAII.
Implementa paridad total con la lógica de Clarion/WinDev.
"""
# 1. Lógica de Fracción (Regla Octava)
frac_raw = par.get("Fraccion", "")
frac_oct = par.get("FraccionOctava", "")
frac_final = frac_oct if frac_oct else frac_raw
# 2. Descripciones (Limpieza de CR/LF como en WinDev)
desc_esp = par.get("Descripcion", " ").replace("\r", "").replace("\n", "").replace("\t", "")
desc_ing = " " # Por ahora
# 3. Arancel (Mapping de Clarion)
tipo_frac = par.get("TipoFracc", "")
arancel_map = {"TLCS": 3, "GENERAL": 0, "PROSEC": 1}
id_arancel = str(arancel_map.get(tipo_frac, 0))
# 4. Lógica de Valor Comercial (Loc_ValorExpoMC)
# ME = ValorDolares, MN = ValorPesos, else ValorComercial
if tipo_moneda_fac == "ME":
valor_comercial = float(par.get("ValorDolares") or 0)
elif tipo_moneda_fac == "MN":
valor_comercial = float(par.get("ValorPesos") or 0)
else:
valor_comercial = float(par.get("ValorComercial") or 0)
# 5. Unidad de Medida OMA y Comercial (Mapeo reverso para WinDev)
oma_final = str(par.get("ClaveOMA") or " ").strip()
oma_raw = str(par.get("UnidadMedida", "")).strip().upper()
if not oma_final or oma_final == "None":
oma_final = "C62_1" if oma_raw in ["6", "PZA"] else oma_raw
map_um_num = {
"PZA": "6", "PIEZA": "6", "PZ": "6", "PC": "6", "PIEZAS": "6",
"KGS": "1", "KGM": "1", "KG": "1", "KILOS": "1", "KILO": "1",
"LTS": "5", "L": "5", "LITROS": "5", "LITRO": "5",
"MTR": "2", "M": "2", "METROS": "2", "METRO": "2",
"PAR": "7", "PARES": "7",
"CJ": "3", "CAJA": "3", "CAJAS": "3",
"JGO": "12", "JUEGO": "12", "SET": "12", "KIT": "12"
}
# Priorizar la clave que venga de la base de datos (CLAVEOMA)
if um_data and um_data.get("CLAVEOMA"):
um_comercial_num = str(um_data["CLAVEOMA"]).strip()
else:
um_comercial_num = map_um_num.get(oma_raw, oma_raw)
# 6. Cálculo del Valor Agregado
valor_agregado_me = float(par.get("ValorAgregadoME") or 0)
valor_aduanas_me = float(par.get("ValorAduanasME") or 0)
valor_dolares = float(par.get("ValorDolares") or 0)
valor_aduanas_mn = float(par.get("ValorAduanasMN") or 0)
valor_pesos = float(par.get("ValorPesos") or 0)
costo_auxiliar_me = float(par.get("CostoAuxiliarME") or 0)
cantidad = float(par.get("Cantidad") or 0)
# 1ra Opción (EXPO): Si es Exportación, el valor ya viene calculado en ValorAgregadoME
if valor_agregado_me > 0:
valor_agregado = valor_agregado_me
# 2da Opción (IMPO - Exactitud WinDev): Costo Unitario Auxiliar * Cantidad
elif costo_auxiliar_me > 0:
valor_agregado = costo_auxiliar_me * cantidad
# 3ra Opción: Si Valor Aduanas ME existe
elif valor_aduanas_me > 0 and valor_aduanas_me > valor_dolares:
valor_agregado = valor_aduanas_me - valor_dolares
# 4ta Opción: Calcular mediante el tipo de cambio implícito
elif valor_aduanas_mn > 0 and valor_aduanas_mn > valor_pesos and valor_dolares > 0:
tc_implicito = valor_pesos / valor_dolares if valor_dolares else 1
valor_agregado = (valor_aduanas_mn - valor_pesos) / tc_implicito
else:
valor_agregado = 0
valor_agregado = round(max(0, valor_agregado), 2)
# 7. Método de Valoración específico por partida
metodo_val = str(par.get("MetodoValoracion") or "").strip()
if not metodo_val:
metodo_val = str(met_val) if met_val else ""
registro = {
"id_factura": id_factura,
"fraccion_551": str(frac_final),
"descripcion_mercancia_551": desc_esp[:250],
"numero_parte_551": str(par.get("NumParte") or " "),
"valor_mercancia_551": valor_comercial,
"cantidad_comercial_551": float(par.get("Cantidad") or 0),
"unidad_medida_comercial_551": um_comercial_num, # Ej. "6"
"cantidad_tarifa_551": float(par.get("Cantidad") or 0), # Se ajusta en el motor
"umt_551": " ", # WinDev envía un espacio vacío aquí
"valor_agregado_551": str(valor_agregado),
"vinculacion_551": "0",
"metodo_valoracion_551": metodo_val,
"marca_551": str(par.get("Marca") or " ").strip() or " ",
"modelo_551": str(par.get("Modelo") or " ").strip() or " ",
"pais_origen_destino_551": str(par.get("PaisOrigen") or " "),
"pais_comprador_vendedor_551": "MEX", # Generalmente MEX en Impo
"entidad_federativa_origen_551": " ",
"entidad_federativa_destino_551": " ",
"entidad_federativa_comprador_551": " ",
"entidad_federativa_vendedor_551": " ",
"identificador_arancel_aplicar_551": id_arancel,
"clave_industria_551": str(par.get("Sector") or " "),
"peso_bruto_551": float(par.get("PesoBruto") or 0),
"uso_mercancia_551": " ",
"estado_mercancia_551": " ",
"moneda_partida_551": str(cla_mon),
"numero_factura_551": str(par.get("NumFactura") or " "),
"tipo_bulto_551": " ", # Se llena en motor si aplica
"cantidad_bultos_551": float(par.get("CantBultos") or 0),
# --- PARIDAD WINDEV ---
"peso_bruto_kg_551": float(par.get("PesoNeto") or 0), # WinDev envía Peso Neto aquí
"peso_neto_kg_551": float(par.get("ValorAduanasMN") or 0), # HACK WinDev: Envía ValorAduanasMN aquí
"numero_remesa_551": str(num_remesa or " "),
"numero_partida_remesa_551": str(par.get("Linea") or num_partida_remesa),
"cantidad_alterna_551": float(par.get("CantidadAlterna") or 0),
"unidad_medida_alterna_551": str(par.get("UnidadAlterna") or " "),
"descripcion_ingles_551": str(par.get("DescripcionIngles") or " ").strip()[:80],
"identificador_sistema_scaii_scaf_551": "SCAII",
"vu_submodelo_551": " ",
"vu_serie_551": " ",
"valor_unitario_aduana_551": float(par.get("CostoUnitarioME") or 0),
"valor_dolares_551": float(par.get("ValorDolares") or 0),
"unidad_medida_oma_551": oma_final, # Código de la fracción (ej. '6')
"valor_dolares_cove_551": "0",
"subdivision_fraccion_551": str(frac_final)[-2:] if len(str(frac_final)) >= 2 else "00",
"codigo_carta_porte_551": " ",
"nodo_551": "85858585",
"CantidadComercialOMA_551": str(float(par.get("Cantidad") or 0))
}
return registro

23
registros/scaii/reg_554.py Normal file
View File

@@ -0,0 +1,23 @@
from typing import Dict, Any
class Generador554SCAII:
@staticmethod
def preparar_json(iden: Dict[str, Any], id_partida: int) -> Dict[str, Any]:
"""
Lógica para Registro 554 (Identificadores de Partidas) - SCAII.
"""
registro = {
"tipo_caso_554": iden.get("identificador", " "),
"complemento_caso_1_554": iden.get("complemento_1", " "),
"complemento_caso_2_554": iden.get("complemento_2", " "),
"complemento_caso_3_554": iden.get("complemento_2", " "), # WinDev repite C3 con C3, pero aquí mapeamos C1, C2, C3
"id_partida": id_partida
}
# En el código WinDev: FormURL_Reg554.complemento_caso_3_554 = DS_IntPar.C3
# Pero DS_IntPar.C3 es COMPLEMENTO2. DS_IntPar.C4 es COMPLEMENTO3.
# Ajustaré para que use el complemento_3 real si existe.
if iden.get("complemento_3"):
registro["complemento_caso_3_554"] = iden.get("complemento_3")
return registro

18
registros/scaii/reg_558.py Normal file
View File

@@ -0,0 +1,18 @@
from typing import Dict, Any
class Generador558:
@staticmethod
def preparar_json(id_partida: int, observaciones: str) -> Dict[str, Any]:
"""
Lógica para Registro 558 (Observaciones de Partidas).
Limpia caracteres especiales como saltos de línea y tabuladores.
"""
if not observaciones:
observaciones = " "
obs_limpia = observaciones.replace("\r", "").replace("\n", "").replace("\t", "").strip()
return {
"observaciones_558": obs_limpia[:1000],
"id_partida": id_partida
}