diff --git a/consulta_cpe.py b/consulta_cpe.py index d2cb16b..66aa0f7 100644 --- a/consulta_cpe.py +++ b/consulta_cpe.py @@ -12,6 +12,7 @@ from datetime import datetime urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +# ============================================================================ class TokenExpiredError(Exception): pass @@ -29,7 +30,7 @@ TIPO_COMPROBANTE = { TIPO_DESCARGA = {"01": "PDF", "02": "XML", "03": "CDR"} - +# ============================================================================ def load_token(): if not TOKEN_FILE.exists(): print(f"❌ No se encontró {TOKEN_FILE}") @@ -49,11 +50,11 @@ def load_token(): print(f"🔑 Token válido por {int(remaining // 60)} min {int(remaining % 60)} seg") return token - +# ============================================================================ def cpe_id(ruc, tipo, serie, numero): return f"{ruc}-{tipo}-{serie}-{numero}-2" - +# ============================================================================ def get_headers(token): return { "Authorization": f"Bearer {token}", @@ -63,7 +64,7 @@ def get_headers(token): "Referer": "https://e-factura.sunat.gob.pe/", } - +# ============================================================================ def consultar_cpe(token, ruc_emisor, tipo, serie, numero, descargar=True, verbose=True): cid = cpe_id(ruc_emisor, tipo, serie, numero) url = f"{API_HOST}{BASE_PATH}/{cid}" @@ -125,7 +126,7 @@ def consultar_cpe(token, ruc_emisor, tipo, serie, numero, descargar=True, verbos return None, r.status_code - +# ============================================================================ def descargar_archivo(token, cid, tipo, ts, verbose=True): label = TIPO_DESCARGA.get(tipo, tipo) url = f"{API_HOST}{BASE_PATH}/{cid}/{tipo}" @@ -219,7 +220,7 @@ def descargar_archivo(token, cid, tipo, ts, verbose=True): return False - +# ============================================================================ def main(): print("=" * 60) print("🔍 CONSULTA CPE - SUNAT API") @@ -239,6 +240,6 @@ def main(): consultar_cpe(token, args.ruc, args.tipo, args.serie, args.numero, descargar=not args.no_download) - +# ============================================================================ if __name__ == "__main__": main() diff --git a/gestor_fallas.py b/gestor_fallas.py index bab1b95..c7b1aa4 100644 --- a/gestor_fallas.py +++ b/gestor_fallas.py @@ -3,10 +3,10 @@ from pathlib import Path from datetime import datetime, timedelta DB_PATH = Path(__file__).parent / "fallas_descarga.db" - +# ============================================================================ def _conn(): return sqlite3.connect(str(DB_PATH)) - +# ============================================================================ def init_db(): with _conn() as conn: conn.execute(""" @@ -22,7 +22,7 @@ def init_db(): fecha_ultimo TEXT NOT NULL ) """) - +# ============================================================================ def _calcular_proximo_intento(intentos): if intentos >= 20: return None @@ -43,7 +43,7 @@ def _calcular_proximo_intento(intentos): else: delta = timedelta(hours=12) return (datetime.now() + delta).isoformat() - +# ============================================================================ def registrar_falla(docentry, ruc, tipo_cpe, serie, numero, error, intentos=None): ahora = datetime.now().isoformat() if intentos is None: @@ -59,11 +59,11 @@ def registrar_falla(docentry, ruc, tipo_cpe, serie, numero, error, intentos=None (docentry, ruc, tipo, serie, numero, error, intentos, proximo_intento, fecha_ultimo) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (docentry, ruc, tipo_cpe, serie, numero, str(error)[:200], intentos, proximo, ahora)) - +# ============================================================================ def eliminar_falla(docentry): with _conn() as conn: conn.execute("DELETE FROM fallas WHERE docentry = ?", (docentry,)) - +# ============================================================================ def obtener_fallas_pendientes() -> list[dict]: ahora = datetime.now().isoformat() with _conn() as conn: @@ -75,12 +75,13 @@ def obtener_fallas_pendientes() -> list[dict]: ORDER BY intentos ASC, fecha_ultimo ASC """, (ahora,)) return [dict(row) for row in cursor.fetchall()] - +# ============================================================================ def obtener_docentries_falla() -> set[int]: with _conn() as conn: cursor = conn.execute("SELECT docentry FROM fallas") return {row[0] for row in cursor.fetchall()} - +# ============================================================================ def total_fallas() -> int: with _conn() as conn: return conn.execute("SELECT COUNT(*) FROM fallas").fetchone()[0] +# ============================================================================ \ No newline at end of file diff --git a/get_token.py b/get_token.py index e9b2864..d675953 100644 --- a/get_token.py +++ b/get_token.py @@ -5,6 +5,7 @@ from pathlib import Path from dotenv import load_dotenv from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout +# ============================================================================ load_dotenv() SOL_URL = "https://www.sunat.gob.pe/sol.html" @@ -17,13 +18,13 @@ RUC = os.getenv("SUNAT_RUC", "").strip() USER = os.getenv("SUNAT_USER", "").strip() PASSWORD = os.getenv("SUNAT_PASSWORD", "").strip() - +# ============================================================================ def save_screenshot(page, name): path = SCREENSHOT_DIR / f"{name}.png" page.screenshot(path=str(path)) print(f" 📸 Screenshot guardado: {path}") - +# ============================================================================ def get_token(headless=False, timeout=60000): print("🚀 Iniciando navegador...") with sync_playwright() as p: @@ -109,7 +110,7 @@ def get_token(headless=False, timeout=60000): browser.close() return None - +# ============================================================================ def fill_login_form(page): page.wait_for_timeout(1000) @@ -143,7 +144,7 @@ def fill_login_form(page): else: print(" ✅ Captcha no detectado") - +# ============================================================================ def get_token_from_direct_access(context, login_page): print(" Buscando enlace con token en el menú SOL...") token = extract_token_from_menu(login_page) @@ -154,7 +155,7 @@ def get_token_from_direct_access(context, login_page): print(" Navegando árbol de menú SOL...") return navigate_sol_menu(login_page) - +# ============================================================================ def navigate_sol_menu(page): page.wait_for_timeout(2000) @@ -227,7 +228,7 @@ def navigate_sol_menu(page): return None - +# ============================================================================ def click_menu_item_by_id(page, element_id, expected_text): print(f" Buscando elemento: #{element_id} ('{expected_text}')...") @@ -271,7 +272,7 @@ def click_menu_item_by_id(page, element_id, expected_text): print(f" ❌ No se pudo encontrar: {expected_text}") return None - +# ============================================================================ def check_for_token(page): urls = [page.url] + [f.url for f in page.frames] for url in urls: @@ -348,7 +349,7 @@ def check_for_token(page): return None - +# ============================================================================ def extract_token_from_menu(page): print(" Analizando estructura del menú SOL...") @@ -399,7 +400,7 @@ def extract_token_from_menu(page): return None - +# ============================================================================ def extract_token_from_url(url): if "token=" in url: token = url.split("token=")[-1] @@ -409,7 +410,7 @@ def extract_token_from_url(url): return token return None - +# ============================================================================ def save_token(token): try: payload_b64 = token.split(".")[1] @@ -434,7 +435,7 @@ def save_token(token): print(f" 💾 Token guardado en: {TOKEN_FILE}") print(f" ⏰ Expira: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(data['expires_at']))}") - +# ============================================================================ if __name__ == "__main__": print("=" * 60) print("🔐 OBTENCIÓN DE TOKEN SUNAT") @@ -447,3 +448,4 @@ if __name__ == "__main__": else: print(f"\n❌ No se pudo obtener el token.") print(" Revisa las capturas en la carpeta screenshots/") +# ============================================================================ \ No newline at end of file diff --git a/get_token_headless.py b/get_token_headless.py index 4203e4c..f74d40c 100644 --- a/get_token_headless.py +++ b/get_token_headless.py @@ -10,14 +10,14 @@ from get_token import ( fill_login_form, save_token, save_screenshot, check_for_token, navigate_sol_menu, extract_token_from_menu, ) - +# ============================================================================ 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" ) - +# ============================================================================ def chrome_instalado(): sistema = platform.system().lower() if sistema == "windows": @@ -33,7 +33,7 @@ def chrome_instalado(): return bool(shutil.which("google-chrome") or shutil.which("google-chrome-stable")) return False - +# ============================================================================ def get_token_headless(timeout=90000): with sync_playwright() as p: browser = p.chromium.launch( @@ -94,6 +94,7 @@ def get_token_headless(timeout=90000): print(" ❌ No se encontró token") return None - +# ============================================================================ if __name__ == "__main__": get_token_headless() +# ============================================================================ \ No newline at end of file diff --git a/hana_manager.py b/hana_manager.py index ab8fd17..9af2911 100644 --- a/hana_manager.py +++ b/hana_manager.py @@ -3,6 +3,7 @@ import json from pathlib import Path from dotenv import load_dotenv +# ============================================================================ load_dotenv() HANA_HOST = os.getenv("HANA_HOST", "") @@ -13,7 +14,7 @@ HANA_SCHEMA = os.getenv("HANA_SCHEMA", "") _conn = None - +# ============================================================================ def get_connection(): global _conn if _conn is None: @@ -30,7 +31,7 @@ def get_connection(): return None return _conn - +# ============================================================================ def obtener_pendientes() -> list[dict]: conn = get_connection() @@ -70,7 +71,7 @@ def obtener_pendientes() -> list[dict]: print(f" ❌ Error en query HANA: {e}") return [] - +# ============================================================================ def marcar_pdf_descargado(docentry: int, url: str, ruta: str): conn = get_connection() if not conn: @@ -89,7 +90,7 @@ def marcar_pdf_descargado(docentry: int, url: str, ruta: str): print(f" ❌ Error actualizando PDF HANA: {e}") return False - +# ============================================================================ def marcar_xml_descargado(docentry: int, url: str, ruta: str): conn = get_connection() if not conn: @@ -108,7 +109,7 @@ def marcar_xml_descargado(docentry: int, url: str, ruta: str): print(f" ❌ Error actualizando XML HANA: {e}") return False - +# ============================================================================ def cerrar_conexion(): global _conn if _conn: @@ -118,7 +119,7 @@ def cerrar_conexion(): pass _conn = None - +# ============================================================================ if __name__ == "__main__": print("=" * 60) print("📋 HANA MANAGER - Prueba") @@ -132,3 +133,4 @@ if __name__ == "__main__": f"PDF={d['tienePDF']} XML={d['tieneXML']}") if len(docs) > 5: print(f" ... y {len(docs) - 5} más") +# ============================================================================ \ No newline at end of file diff --git a/login_api.py b/login_api.py index 625d9a3..917d2e0 100644 --- a/login_api.py +++ b/login_api.py @@ -8,6 +8,7 @@ from dotenv import load_dotenv import requests import urllib3 +# ============================================================================ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) load_dotenv() @@ -21,11 +22,11 @@ USER = os.getenv("SUNAT_USER", "").strip() PASSWORD = os.getenv("SUNAT_PASSWORD", "").strip() J_USERNAME = USER - +# ============================================================================ def log(msg): print(f" {msg}") - +# ============================================================================ def decode_jwt(token): try: payload_b64 = token.split(".")[1] @@ -36,7 +37,7 @@ def decode_jwt(token): except: return None - +# ============================================================================ def save_token(token): payload = decode_jwt(token) or {} expires_at = payload.get("exp", 0) @@ -56,7 +57,7 @@ def save_token(token): print(f" Usuario SOL: {data['usuario_sol']}") return True - +# ============================================================================ def try_password_grant(): print("\n🔑 Intentando grant_type=password...") data = { @@ -88,7 +89,7 @@ def try_password_grant(): print(f" ❌ Error: {e}") return False - +# ============================================================================ def try_auth_code_flow(): print("\n🔑 Intentando flujo authorization_code...") log("1. Accediendo a MenuInternet.htm para obtener redirect...") @@ -275,7 +276,7 @@ def try_auth_code_flow(): return False - +# ============================================================================ def main(): print("=" * 60) print("🔐 LOGIN SUNAT API (sin Playwright)") @@ -302,6 +303,7 @@ def main(): print(" - Cambios en el flujo OAuth2 de SUNAT") print("\n Usa get_token.py como respaldo: python get_token.py") - +# ============================================================================ if __name__ == "__main__": main() +# ============================================================================ \ No newline at end of file diff --git a/main.py b/main.py index 398f613..644aa79 100644 --- a/main.py +++ b/main.py @@ -9,6 +9,7 @@ from pathlib import Path from dotenv import load_dotenv import urllib3 +# ============================================================================ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) load_dotenv() @@ -24,18 +25,31 @@ from consulta_cpe import ( ) # ============================================================================ -# CONFIGURACIÓN DE HORARIOS +# CONFIGURACIÓN DE HORARIOS (desde scheduler.json) # ============================================================================ -# (hora_inicio, min_inicio, hora_fin, min_fin, intervalo_segundos, modo) -HORARIOS = [ - (6, 30, 13, 15, 300, "rapido"), # 06:30 - 13:15 -> cada 5 minutos - (13, 15, 14, 0, 3600, "full"), # 13:15 - 14:00 -> cada 1 hora - (14, 0, 17, 30, 300, "rapido"), # 14:00 - 17:30 -> cada 5 minutos - (17, 30, 21, 30, 3600, "full"), # 17:30 - 21:30 -> cada 1 hora -] +SCHEDULER_FILE = Path(__file__).parent / "scheduler.json" -# 0=lun, 1=mar, 2=mié, 3=jue, 4=vie, 5=sáb, 6=dom -DIAS_ACTIVOS = [0, 1, 2, 3, 4] + +def _cargar_scheduler(): + with open(SCHEDULER_FILE, encoding="utf-8") as f: + data = json.load(f) + horarios = [tuple(h) for h in data["HORARIOS"]] + return data["DIAS_ACTIVOS"], horarios + + +try: + DIAS_ACTIVOS, HORARIOS = _cargar_scheduler() +except Exception as e: + print(f"⚠️ Error cargando {SCHEDULER_FILE}: {e}. Usando defaults.") + # (hora_inicio, min_inicio, hora_fin, min_fin, intervalo_segundos, modo) + HORARIOS = [ + (6, 30, 13, 15, 300, "rapido"), # 06:30 - 13:15 -> cada 5 minutos + (13, 15, 14, 0, 3600, "full"), # 13:15 - 14:00 -> cada 1 hora + (14, 0, 17, 30, 300, "rapido"), # 14:00 - 17:30 -> cada 5 minutos + (17, 30, 21, 30, 3600, "full"), # 17:30 - 21:30 -> cada 1 hora + ] + # 0=lun, 1=mar, 2=mié, 3=jue, 4=vie, 5=sáb, 6=dom + DIAS_ACTIVOS = [0, 1, 2, 3, 4] _running = True @@ -45,14 +59,14 @@ color_amarillo = "\033[93m" color_rojo = "\033[91m" color_fin = "\033[0m" - +# ============================================================================ def ic(ok, texto_ok, texto_fail): """Icono + texto según estado.""" icono = "✅" if ok else "⚠️" color = color_verde if ok else color_amarillo return f"{icono} {color}{texto_ok if ok else texto_fail}{color_fin}" - +# ============================================================================ def obtener_intervalo(dt: datetime) -> tuple[int, str] | None: minutos = dt.hour * 60 + dt.minute for hi, mi, hf, mf, intervalo, modo in HORARIOS: @@ -62,14 +76,14 @@ def obtener_intervalo(dt: datetime) -> tuple[int, str] | None: return (intervalo, modo) return None - +# ============================================================================ def segundos_hasta_proximo_inicio(dt: datetime) -> int: target = dt.replace(hour=6, minute=30, second=0, microsecond=0) if dt >= target: target += timedelta(days=1) return int((target - dt).total_seconds()) - +# ============================================================================ def segundos_hasta_proximo_dia_activo(dt: datetime) -> int: for dias in range(1, 8): prox = dt + timedelta(days=dias) @@ -78,7 +92,7 @@ def segundos_hasta_proximo_dia_activo(dt: datetime) -> int: return int((target - dt).total_seconds()) return 86400 - +# ============================================================================ def obtener_token(): """Obtiene o renueva el token si es necesario.""" token = load_token() @@ -111,7 +125,7 @@ def obtener_token(): print(" ⚠️ No se detectó token tras 30s. Continuando...") return load_token() - +# ============================================================================ def procesar_lote(token): """Procesa documentos pendientes desde HANA.""" print(f"\n📋 Buscando documentos pendientes en HANA...") @@ -265,13 +279,13 @@ def procesar_lote(token): for idx, doc in enumerate(docs_retry): _procesar(doc, idx, len(docs_retry), es_retry=True) - +# ============================================================================ def signal_handler(signum, frame): global _running print("\n\n⏹️ Señal de parada recibida. Deteniendo...") _running = False - +# ============================================================================ def main(): global _running signal.signal(signal.SIGINT, signal_handler) @@ -360,6 +374,7 @@ def main(): cerrar_conexion() print("\n👋 Daemon detenido.") - +# ============================================================================ if __name__ == "__main__": main() +# ============================================================================ \ No newline at end of file diff --git a/reporte_fallas.py b/reporte_fallas.py index 779d338..0297505 100644 --- a/reporte_fallas.py +++ b/reporte_fallas.py @@ -3,6 +3,7 @@ from pathlib import Path from datetime import datetime from collections import Counter +# ============================================================================ try: import matplotlib matplotlib.use("Agg") @@ -24,14 +25,14 @@ COLORS = { } CMAP = list(COLORS.values()) - +# ============================================================================ def _query(sql, params=()): with sqlite3.connect(str(DB_PATH)) as conn: conn.row_factory = sqlite3.Row rows = conn.execute(sql, params).fetchall() return [dict(r) for r in rows] - +# ============================================================================ def reporte_texto(): print("\n" + "=" * 60) print("📊 REPORTE DE FALLAS - SUNAT") @@ -61,11 +62,11 @@ def reporte_texto(): for r in _query("SELECT DATE(fecha_ultimo) AS dia, COUNT(*) AS c FROM fallas GROUP BY dia ORDER BY dia"): print(f" {r['dia']} → {r['c']} falla(s)") - +# ============================================================================ def _color_error(error): return COLORS.get(error, "#95a5a6") - +# ============================================================================ def grafico_barras_errores(): data = _query("SELECT COUNT(*) AS c, error FROM fallas GROUP BY error ORDER BY c DESC") if not data: @@ -90,7 +91,7 @@ def grafico_barras_errores(): plt.close(fig) return path - +# ============================================================================ def grafico_barras_ruc(): data = _query("SELECT COUNT(*) AS c, ruc FROM fallas GROUP BY ruc ORDER BY c DESC LIMIT 10") if not data: @@ -115,7 +116,7 @@ def grafico_barras_ruc(): plt.close(fig) return path - +# ============================================================================ def grafico_torta(): data = _query("SELECT COUNT(*) AS c, error FROM fallas GROUP BY error ORDER BY c DESC") if not data: @@ -139,7 +140,7 @@ def grafico_torta(): plt.close(fig) return path - +# ============================================================================ def main(): reporte_texto() @@ -161,6 +162,7 @@ def main(): print(f" {p.name}") print(f"\n💡 Abre con: open {REPORTES_DIR}") - +# ============================================================================ if __name__ == "__main__": main() +# ============================================================================ \ No newline at end of file diff --git a/scheduler.json b/scheduler.json new file mode 100644 index 0000000..cb93168 --- /dev/null +++ b/scheduler.json @@ -0,0 +1,9 @@ +{ + "HORARIOS": [ + [6, 30, 13, 15, 10, "rapido"], + [13, 15, 14, 0, 1800, "full"], + [14, 0, 17, 30, 10, "rapido"], + [17, 30, 21, 30, 1800, "full"] + ], + "DIAS_ACTIVOS": [0, 1, 2, 3, 4] +} diff --git a/smb_manager.py b/smb_manager.py index 6c6b4bb..252b4cb 100644 --- a/smb_manager.py +++ b/smb_manager.py @@ -5,6 +5,7 @@ import subprocess from pathlib import Path from dotenv import load_dotenv +# ============================================================================ load_dotenv() SMB_SERVER = os.getenv("SMB_SERVER", "").strip() @@ -16,7 +17,7 @@ SMB_MOUNT_MAC = os.getenv("SMB_MOUNT_MAC", "").strip() SMB_MOUNT_LINUX = os.getenv("SMB_MOUNT_LINUX", "").strip() SMB_MOUNT_WINDOWS = os.getenv("SMB_MOUNT_WINDOWS", "").strip() - +# ============================================================================ def get_current_os(): system = platform.system().lower() if system == "darwin": @@ -25,7 +26,7 @@ def get_current_os(): return "windows" return "linux" - +# ============================================================================ def get_mount_point(): current_os = get_current_os() mapping = { @@ -35,7 +36,7 @@ def get_mount_point(): } return mapping.get(current_os, "") - +# ============================================================================ def esta_montado(mount_point): if not mount_point: return False @@ -45,7 +46,7 @@ def esta_montado(mount_point): os.path.join(mount_point, ".mounted_check") ) - +# ============================================================================ def montar_smb(): mount_point = get_mount_point() if not mount_point: @@ -87,7 +88,7 @@ def montar_smb(): print(f" ⚠️ Error montando SMB: {e}") return False - +# ============================================================================ def sincronizar_a_smb(ruta_local): if not os.path.isdir(ruta_local): print(f" ⚠️ La carpeta local no existe: {ruta_local}") @@ -152,3 +153,4 @@ def sincronizar_a_smb(ruta_local): except Exception as e: print(f" ❌ Error en sincronización: {e}") return False +# ============================================================================ \ No newline at end of file diff --git a/test_headless_token.py b/test_headless_token.py deleted file mode 100644 index a383faa..0000000 --- a/test_headless_token.py +++ /dev/null @@ -1,325 +0,0 @@ -import os -import re -import json -import time -import base64 -from pathlib import Path -from dotenv import load_dotenv -from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout - -load_dotenv() - -from get_token import ( - SOL_URL, MENU_URL, TOKEN_FILE, SCREENSHOT_DIR, - RUC, USER, PASSWORD, - fill_login_form, save_screenshot, save_token, - check_for_token, navigate_sol_menu, - click_menu_item_by_id, extract_token_from_menu, -) - -TEST_DIR = SCREENSHOT_DIR -TEST_DIR.mkdir(exist_ok=True) - -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" -) - -BROWSER_ARGS = [ - "--disable-blink-features=AutomationControlled", - "--disable-popup-blocking", - "--no-sandbox", - "--disable-web-security", - "--disable-features=IsolateOrigins,site-per-process", - "--window-size=1280,800", - "--start-maximized", -] - - -def captura(page, tag, nombre): - safe = f"{tag}_{nombre}" - page.screenshot(path=str(TEST_DIR / f"{safe}.png")) - - -def _salto_linea(texto, ancho=100): - for i in range(0, len(texto), ancho): - print(f" {texto[i:i+ancho]}") - - -def _login_si_necesario(page, tag): - page.wait_for_load_state("networkidle") - has_inputs = len(page.locator("input[type='password']").all()) > 0 - necesita_login = has_inputs or any( - k in page.url.lower() for k in ["login", "autentica", "auth"] - ) - if not necesita_login: - print(" No se detectó login") - return True - - print(" Login detectado. Rellenando credenciales...") - fill_login_form(page) - captura(page, tag, "form_rellenado") - - page.keyboard.press("Enter") - try: - page.wait_for_load_state("networkidle", timeout=30000) - except Exception as e: - print(f" ⚠️ Timeout post-login: {e}") - page.wait_for_timeout(3000) - captura(page, tag, "post_login") - print(f" URL post-login: {page.url}") - return True - - -def _buscar_token_en_pagina(page, tag): - time.sleep(2) - token = check_for_token(page) - if token: - return token - - token = navigate_sol_menu(page) - if token: - return token - - token = extract_token_from_menu(page) - return token - - -def _navegar_con_cookies(page, url, tag): - page.goto(url, wait_until="domcontentloaded") - captura(page, tag, "navegacion") - print(f" URL tras navegar: {page.url}") - - -# ─── ESTRATEGIAS ───────────────────────────────────────────── - - -def estrategia_1_fixed_popup(context, page): - tag = "e1" - print("\n" + "=" * 60) - print("ESTRATEGIA 1: Popup con --disable-popup-blocking + event listener") - print("=" * 60) - - page.goto(SOL_URL, wait_until="domcontentloaded") - captura(page, tag, "sol_home") - - popup_found = [None] - got_popup = [False] - - def on_page(new_page): - if not got_popup[0]: - print(f" 📌 Página detectada: {new_page.url[:100]}") - popup_found[0] = new_page - got_popup[0] = True - - page.context().on("page", on_page) - - with page.expect_popup() as popup_info: - page.evaluate("tramiteConsulta()") - - try: - pup = popup_info.value - pup.wait_for_load_state("domcontentloaded", timeout=20000) - print(f" Popup URL: {pup.url}") - captura(pup, tag, "popup") - - if "chrome-error" in pup.url: - print(" ⚠️ Popup en chrome-error. Reintentando recarga...") - try: - pup.wait_for_load_state("domcontentloaded", timeout=20000) - except: - pass - captura(pup, tag, "popup_retry") - if "chrome-error" in pup.url: - print(" ❌ Sigue en chrome-error tras recarga") - print(f" URL final: {pup.url}") - return None - - _login_si_necesario(pup, tag) - token = _buscar_token_en_pagina(pup, tag) - return token - - except Exception as e: - print(f" ❌ Error con popup: {e}") - return None - - -def estrategia_2_direct_menu(context, page): - tag = "e2" - print("\n" + "=" * 60) - print("ESTRATEGIA 2: Menú directo con sesión desde SOL") - print("=" * 60) - - page.goto(SOL_URL, wait_until="domcontentloaded") - captura(page, tag, "sol_home") - - _navegar_con_cookies(page, MENU_URL, tag) - - if not _login_si_necesario(page, tag): - return None - - token = _buscar_token_en_pagina(page, tag) - return token - - -def estrategia_3_direct_login(context, page): - tag = "e3" - print("\n" + "=" * 60) - print("ESTRATEGIA 3: Login directo al menú (sin SOL)") - print("=" * 60) - - _navegar_con_cookies(page, MENU_URL, tag) - - if not _login_si_necesario(page, tag): - return None - - token = _buscar_token_en_pagina(page, tag) - return token - - -def estrategia_4_popup_url_directo(context, page): - tag = "e4" - print("\n" + "=" * 60) - print("ESTRATEGIA 4: Extraer URL del popup y navegar directo") - print("=" * 60) - - page.goto(SOL_URL, wait_until="domcontentloaded") - captura(page, tag, "sol_home") - - try: - popup_url = page.evaluate("""() => { - const fn = tramiteConsulta.toString(); - const match = fn.match(/window\\.open\\(['"]([^'"]+)['"]/); - return match ? match[1] : null; - }""") - if popup_url: - if popup_url.startswith("/"): - from urllib.parse import urljoin - popup_url = urljoin(SOL_URL, popup_url) - print(f" URL del popup: {popup_url}") - else: - print(" No se pudo extraer URL") - return None - except Exception as e: - print(f" Error extrayendo URL: {e}") - return None - - _navegar_con_cookies(page, popup_url, tag) - - if not _login_si_necesario(page, tag): - return None - - token = _buscar_token_en_pagina(page, tag) - return token - - -# ─── CONFIGURACIONES DE LANZAMIENTO ────────────────────────── - - -def lanzar_chrome(p, channel=None, extra_args=None): - """Lanza navegador con la configuracion dada.""" - args = list(BROWSER_ARGS) - if extra_args: - args.extend(extra_args) - - kwargs = { - "headless": True, - "args": args, - } - if channel: - kwargs["channel"] = channel - print(f" Lanzando: {channel} (headless)") - else: - print(f" Lanzando: chromium empaquetado (headless)") - - browser = p.chromium.launch(**kwargs) - - context = browser.new_context( - viewport={"width": 1280, "height": 800}, - locale="es-PE", - timezone_id="America/Lima", - user_agent=USER_AGENT, - no_viewport=False, - ignore_https_errors=True, - ) - - page = context.new_page() - page.set_default_timeout(90000) - - return browser, context, page - - -def ejecutar_escenario(p, nombre_browser, channel, extra_args): - """Ejecuta todas las estrategias con un navegador/configuracion.""" - print(f"\n{'#'*60}") - print(f"# ESCENARIO: {nombre_browser}") - print(f"{'#'*60}") - - browser, context, page = lanzar_chrome(p, channel, extra_args) - - token = None - estrategias = [ - ("1_fixed_popup", estrategia_1_fixed_popup), - ("2_direct_menu", estrategia_2_direct_menu), - ("3_direct_login", estrategia_3_direct_login), - ("4_popup_url_directo", estrategia_4_popup_url_directo), - ] - - for nombre, fn in estrategias: - print() - try: - token = fn(context, page) - except Exception as e: - print(f" ⚠️ Excepción en {nombre}: {e}") - import traceback - traceback.print_exc() - token = None - - if token: - print(f"\n✅ Token obtenido con '{nombre}' en escenario '{nombre_browser}'!") - return token - else: - print(f"\n ❌ '{nombre}' falló") - - context.close() - browser.close() - return None - - -def main(): - print("=" * 60) - print("🔐 TEST HEADLESS TOKEN - SUNAT") - print("=" * 60) - print(f" RUC: {RUC}") - print(f" USER: {USER}") - - if not RUC or not USER or not PASSWORD: - print("\n❌ Faltan credenciales en .env (SUNAT_RUC, SUNAT_USER, SUNAT_PASSWORD)") - return - - escenarios = [ - ("Chrome real (channel=chrome)", "chrome", None), - ("Chromium empaquetado", None, None), - ("Chromium + extra stealth", None, ["--disable-gpu", "--disable-accelerated-2d-canvas"]), - ] - - with sync_playwright() as p: - for nombre_browser, channel, extra_args in escenarios: - token = ejecutar_escenario(p, nombre_browser, channel, extra_args) - if token: - save_token(token) - print(f"\n✅ Token guardado en {TOKEN_FILE}") - print(f" Primeros 50 chars: {token[:50]}...") - - # Quitar token de prueba, no guardarlo permanentemente - if TOKEN_FILE.exists(): - TOKEN_FILE.unlink() - return - - print(f"\n❌ Todas las estrategias y escenarios fallaron.") - print(" Revisa screenshots en:", TEST_DIR) - - -if __name__ == "__main__": - main()