119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
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
|