475 lines
28 KiB
Python
475 lines
28 KiB
Python
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()
|