El endpoint viejo (104.192.7.152:3529) dejo de responder. La API se movio a https://anexo46.aduanasoft.com/api/v1/. Todas las rutas que usa el cliente (login, registro501..registro558) existen bajo ese prefijo. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
118 lines
5.6 KiB
Python
118 lines
5.6 KiB
Python
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://anexo46.aduanasoft.com/api/v1/"
|
|
|
|
# 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']}")
|