import requests import json import urllib3 import time from logger_manager import get_logger logger = get_logger() # Desactivar advertencias de SSL no verificado urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # ============================================================================ def token(datos): CLIENT_ID=datos["Client_ID"] CLIENT_SECRET=datos["Client_Secret"] url = f"https://api-seguridad.sunat.gob.pe/v1/clientesextranet/{CLIENT_ID}/oauth2/token/" headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'application/json, text/plain, */*', 'Origin': 'https://api-seguridad.sunat.gob.pe' } payload = 'grant_type=client_credentials' + '&' payload += 'scope=https://api.sunat.gob.pe/v1/contribuyente/contribuyentes' + '&' payload += f'client_id={CLIENT_ID}' + '&' payload += f'client_secret={CLIENT_SECRET}' response = requests.request("POST", url, headers=headers, data=payload, verify=False, timeout=30) logger.info(f" 📤 Token request enviado") logger.info(f" 📥 Token response: {response.status_code}") if response.status_code == 200: result = response.json() if "access_token" in result: token = result["access_token"] return token else: return "" else: return "" # ============================================================================ def validar_comprobante(token,RUC,payload:dict): url = f"https://api.sunat.gob.pe/v1/contribuyente/contribuyentes/{RUC}/validarcomprobante" # payload = { # "numRuc": "20538298485", # "codComp": "08", # "numeroSerie": "E001", # "numero": "3773", # "fechaEmision": "15/11/2023", # "monto": "1328.52" # } headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {token}', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'application/json, text/plain, */*' } max_retries = 3 for attempt in range(max_retries): try: response = requests.request("POST", url, headers=headers, json=payload, verify=False, timeout=30) logger.info(f" 📥 Status code: {response.status_code}") return response.json() except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: logger.warning(f" ⚠️ Error de conexión (intento {attempt + 1}/{max_retries}): {e}") if attempt < max_retries - 1: time_to_wait = 2 ** (attempt + 1) logger.info(f" 🔄 Reintentando en {time_to_wait} segundos...") time.sleep(time_to_wait) else: raise e # ============================================================================