diff --git a/.gitignore b/.gitignore index f859bc5..5a016d9 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Entorno virtual sunat-env/ +sunat-venv/ venv/ .env @@ -17,6 +18,9 @@ logs/ # Base de datos SQLite fallas_descarga.db +historial.db +historial.db-shm +historial.db-wal # Reportes generados reportes/ diff --git a/consulta_cpe.py b/consulta_cpe.py index 8da2389..0018e2a 100644 --- a/consulta_cpe.py +++ b/consulta_cpe.py @@ -9,8 +9,10 @@ import requests import urllib3 from pathlib import Path from datetime import datetime +from logger_manager import get_logger urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +logger = get_logger(__name__) # ============================================================================ class TokenExpiredError(Exception): @@ -25,7 +27,7 @@ BASE_PATH = "/v1/contribuyente/consultacpe/comprobantes" TIPO_COMPROBANTE = { "01": "Factura", "02": "Recibo por Honorarios", "03": "Boleta de Venta", - "04": "Liquidación de Compra", "07": "Nota de Crédito", "08": "Nota de Débito", + "04": "Liquidación de Compra", "07": "Factura - Nota de Crédito", "08": "Factura - Nota de Débito", } TIPO_DESCARGA = {"01": "PDF", "02": "XML", "03": "CDR"} @@ -33,8 +35,7 @@ TIPO_DESCARGA = {"01": "PDF", "02": "XML", "03": "CDR"} # ============================================================================ def load_token(): if not TOKEN_FILE.exists(): - print(f"❌ No se encontró {TOKEN_FILE}") - print(" Ejecuta primero: python get_token.py") + logger.error(f"No se encontró {TOKEN_FILE}. Ejecuta primero: python get_token.py") return None data = json.loads(TOKEN_FILE.read_text()) @@ -43,16 +44,22 @@ def load_token(): now = time.time() if expires and now > expires: - print(f"⚠️ Token expiró. Ejecuta de nuevo: python get_token.py") + logger.warning("Token expiró. Ejecuta de nuevo: python get_token.py") return None remaining = expires - now - print(f"🔑 Token válido por {int(remaining // 60)} min {int(remaining % 60)} seg") + logger.info(f"Token válido por {int(remaining // 60)} min {int(remaining % 60)} seg") return token # ============================================================================ +# La API de consultacpe no acepta el código de Indicator de HANA tal cual para +# notas de crédito/débito: espera "F7"/"F8" en vez de "07"/"08" (confirmado +# contra la URL real que usa el formulario de SUNAT), sin importar la serie. +TIPO_URL = {"07": "F7", "08": "F8"} + def cpe_id(ruc, tipo, serie, numero): - return f"{ruc}-{tipo}-{serie}-{numero}-2" + tipo_url = TIPO_URL.get(tipo, tipo) + return f"{ruc}-{tipo_url}-{serie}-{numero}-2" # ============================================================================ def get_headers(token): @@ -70,8 +77,7 @@ def consultar_cpe(token, ruc_emisor, tipo, serie, numero, descargar=True, verbos url = f"{API_HOST}{BASE_PATH}/{cid}" if verbose: - print(f"\n🌐 Consultando: {ruc_emisor}-{tipo}-{serie}-{numero}") - print(f" {url}") + logger.info(f"Consultando: {ruc_emisor}-{tipo}-{serie}-{numero} ({url})") session = requests.Session() session.verify = False @@ -82,20 +88,27 @@ def consultar_cpe(token, ruc_emisor, tipo, serie, numero, descargar=True, verbos try: r = session.get(url, timeout=120) if verbose: - print(f"📡 {r.status_code}") - break + logger.info(f"Respuesta: {r.status_code}") + if r.status_code < 500: + break + if intento < 2: + if verbose: + logger.warning(f"Error {r.status_code}, reintentando...") + time.sleep(2 * (intento + 1)) + elif verbose: + logger.error(f"Error {r.status_code} tras 3 intentos") except Exception as e: if verbose: - print(f" ⚠️ Intento {intento+1} falló: {e}") + logger.warning(f"Intento {intento+1} falló: {e}") if intento < 2: time.sleep(3) else: if verbose: - print(f" ❌ Error después de 3 intentos") - return None, 0 + logger.error("Error después de 3 intentos") + return None, 0, None if r is None: - return None, 0, 0 + return None, 0, None if r.status_code == 200: data = r.json() @@ -105,26 +118,26 @@ def consultar_cpe(token, ruc_emisor, tipo, serie, numero, descargar=True, verbos json_file = DOWNLOAD_DIR / f"{safe}_{ts}.json" json_file.write_text(json.dumps(data, indent=2, ensure_ascii=False)) if verbose: - print(f"💾 JSON: {json_file}") + logger.info(f"JSON: {json_file}") if descargar: for tipo_dl in ["02", "01", "03"]: descargar_archivo(token, cid, tipo_dl, ts, verbose=verbose) - return data, 200 + return data, 200, str(json_file) elif r.status_code == 401: if verbose: - print(" ❌ Token inválido. Ejecuta: python get_token.py") + logger.error("Token inválido. Ejecuta: python get_token.py") raise TokenExpiredError("Token expirado o inválido (401)") elif r.status_code == 404: if verbose: - print(" ❌ No encontrado (404)") + logger.error("No encontrado (404)") else: if verbose: - print(f" ❌ Error {r.status_code}: {r.text[:300]}") + logger.error(f"Error {r.status_code}: {r.text[:300]}") - return None, r.status_code + return None, r.status_code, None # ============================================================================ def descargar_archivo(token, cid, tipo, ts, verbose=True): @@ -132,8 +145,7 @@ def descargar_archivo(token, cid, tipo, ts, verbose=True): url = f"{API_HOST}{BASE_PATH}/{cid}/{tipo}" if verbose: - print(f"\n📥 Descargando {label}...") - print(f" {url}") + logger.info(f"Descargando {label}... ({url})") session = requests.Session() session.verify = False @@ -144,16 +156,23 @@ def descargar_archivo(token, cid, tipo, ts, verbose=True): try: r = session.get(url, timeout=120) if verbose: - print(f"📡 {r.status_code}") - break + logger.info(f"Respuesta: {r.status_code}") + if r.status_code < 500: + break + if intento < 2: + if verbose: + logger.warning(f"Error {r.status_code}, reintentando...") + time.sleep(2 * (intento + 1)) + elif verbose: + logger.error(f"Error {r.status_code} tras 3 intentos") except Exception as e: if verbose: - print(f" ⚠️ Intento {intento+1} falló: {e}") + logger.warning(f"Intento {intento+1} falló: {e}") if intento < 2: time.sleep(3) else: if verbose: - print(f" ❌ Error después de 3 intentos") + logger.error("Error después de 3 intentos") return False if r is None: @@ -165,7 +184,7 @@ def descargar_archivo(token, cid, tipo, ts, verbose=True): if not val_archivo: if verbose: - print(f" ⚠️ Respuesta sin valArchivo") + logger.warning("Respuesta sin valArchivo") return False raw = base64.b64decode(val_archivo) @@ -175,7 +194,7 @@ def descargar_archivo(token, cid, tipo, ts, verbose=True): zip_path = DOWNLOAD_DIR / zip_name zip_path.write_bytes(raw) if verbose: - print(f" 💾 ZIP: {zip_path} ({len(raw)} bytes)") + logger.info(f"ZIP: {zip_path} ({len(raw)} bytes)") try: with zipfile.ZipFile(io.BytesIO(raw)) as zf: @@ -200,23 +219,23 @@ def descargar_archivo(token, cid, tipo, ts, verbose=True): out_path = DOWNLOAD_DIR / out_name out_path.write_bytes(content) if verbose: - print(f" ✅ {out_name} ({len(content)} bytes)") + logger.info(f"{out_name} ({len(content)} bytes)") except Exception as e: if verbose: - print(f" ⚠️ Error extrayendo ZIP: {e}") + logger.warning(f"Error extrayendo ZIP: {e}") return True elif r.status_code == 401: if verbose: - print(" ❌ Token inválido (401)") + logger.error("Token inválido (401)") raise TokenExpiredError("Token expirado o inválido (401)") elif r.status_code == 400: if verbose: - print(f" ⚠️ No disponible (400)") + logger.warning("No disponible (400)") else: if verbose: - print(f" ❌ Error {r.status_code}") + logger.error(f"Error {r.status_code}") return False diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index f93ab4a..69dcdf1 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -26,6 +26,7 @@ WantedBy=multi-user.target ``` ```bash +# Guarda el bloque [Unit]/[Service]/[Install] de arriba en un archivo, ej. sunat-cpe.service sudo cp sunat-cpe.service /etc/systemd/system/ sudo systemctl enable sunat-cpe sudo systemctl start sunat-cpe @@ -63,7 +64,7 @@ NSSM redirige stdout/stderr a `C:\ruta\descarga_xml_sunat\logs\` por defecto (el Como servicio, el daemon usa `get_token_headless()` (headless, Firefox real) para renovar el token automáticamente sin intervención. Si Firefox no está instalado, el daemon lo detecta y usa `get_token.py headless=False` como respaldo (requiere escritorio). Para renovar manualmente: ```cmd -C:\ruta\descarga_xml_sunat\sunat-env\Scripts\python get_token.py +C:\ruta\descarga_xml_sunat\sunat-venv\Scripts\python get_token.py ``` ### Actualización @@ -83,4 +84,4 @@ COPY . . CMD ["python", "main.py"] ``` -Asegúrate de que `requirements.txt` incluya: `playwright`, `python-dotenv`, `requests`, `hdbcli`, `matplotlib`. +Asegúrate de que `requirements.txt` incluya: `playwright`, `python-dotenv`, `requests`, `hdbcli`. diff --git a/docs/MODULES.md b/docs/MODULES.md index 67069c7..a4fb632 100644 --- a/docs/MODULES.md +++ b/docs/MODULES.md @@ -2,22 +2,22 @@ | Archivo | Función | |---|---| -| `main.py` | Daemon orquestador. Bucle infinito con horarios configurables, obtiene token, consulta HANA, procesa pendientes (nuevos + reintentos). | +| `main.py` | Daemon orquestador. Bucle infinito con horarios configurables (`scheduler.json`), obtiene token, consulta HANA, procesa pendientes (nuevos + reintentos). | | `get_token_headless.py` | Login headless via Firefox real. Sin ventana visible. | | `get_token.py` | Login SOL via Playwright (headless=False). Navega menú → extrae JWT de `sessionStorage['SUNAT.token']`. | | `consulta_cpe.py` | Consulta API REST `api-cpe.sunat.gob.pe`. Retorna `(data, status_code)`. Lanza `TokenExpiredError` en 401. | | `hana_manager.py` | Conexión HANA vía `hdbcli`. Consulta pendientes en `OPCH`, actualiza flags de descarga. | | `gestor_fallas.py` | SQLite `fallas_descarga.db`. Registro de fallas con conteo de intentos y backoff progresivo (0min → 5min → 15min → ... → 12h, máx 20). | +| `historial_manager.py` | SQLite `historial.db`. Registra cada intento de consulta/descarga (éxito o falla) para auditoría; limpia registros de más de 180 días. Incluye CLI (`python historial_manager.py --ruc/--serie/--numero/--tipo`) para buscar intentos. | +| `logger_manager.py` | Logging con formato de árbol (indentado por tareas anidadas) a consola + archivo. Rotación diaria en `logs/sunat_YYYY-MM-DD.log`, limpieza automática de logs con más de 7 días. | | `smb_manager.py` | Monta recurso SMB (`mount_smbfs`/`mount.cifs`/`net use`) y sincroniza PDF/XML al NAS. Limpia JSON/ZIP temporales. | -| `login_api.py` | Intento de `password_grant()` / `authorization_code` directo a SUNAT (sin navegador, generalmente ya no funciona). Se mantiene como referencia. | -| `logger_utilities.py` | Redirige stdout/stderr a `logs/YYYY-MM-DD.log` + terminal. Rotación diaria. | -| `reporte_fallas.py` | Genera resumen textual + 3 gráficos PNG desde `fallas_descarga.db`. | +| `scheduler.json` | Horarios activos (`HORARIOS`) y días de la semana en que corre el daemon (`DIAS_ACTIVOS`). Editable sin tocar código. | | `token.json` | Token JWT auto-generado con expiración. | | `.env` | Credenciales HANA y SUNAT. | ## Flujo -1. `main.py` verifica día/horario → renueva token si expiró (get_token_headless → get_token.py headless=False) +1. `main.py` verifica día/horario (`scheduler.json`) → renueva token si expiró (get_token_headless → get_token.py headless=False) 2. `obtener_pendientes()` trae todos los documentos de OPCH donde `tienePDF='N'` o `tieneXML='N'` 3. Separa en **nuevos** (no están en fallas DB) y **reintentos** (fallas pendientes con backoff cumplido) 4. Por cada doc: @@ -26,5 +26,8 @@ - Si OK → `descargar_archivo()` (XML + PDF + CDR) - `marcar_pdf_descargado()` / `marcar_xml_descargado()` actualizan HANA - Si era reintento exitoso → `eliminar_falla()` + - `registrar_intento()` deja constancia del resultado en `historial.db` 5. `sincronizar_a_smb()` monta NAS (si no está montado), copia PDF/XML via rsync (Mac/Linux) o shutil.move (Windows), y elimina JSON/ZIP temporales de `descargas/` -6. Espera intervalo configurado (5 min modo rápido, 1 h modo full) y repite +6. Espera intervalo configurado (según `scheduler.json`, ej. 5 min modo rápido, 30 min/1 h modo full) y repite + +Al arrancar, `main.py` también corre mantenimiento: `limpiar_logs_antiguos(dias=7)` y `limpiar_historial_antiguo(dias=180)`. diff --git a/docs/SETUP.md b/docs/SETUP.md index bc46ca0..bafb6dd 100644 --- a/docs/SETUP.md +++ b/docs/SETUP.md @@ -12,8 +12,8 @@ ```bash # Crear y activar entorno virtual (recomendado) -python3 -m venv sunat-env -source sunat-env/bin/activate +python3 -m venv sunat-venv +source sunat-venv/bin/activate # Instalar dependencias pip install -r requirements.txt @@ -50,7 +50,7 @@ SMB_SUBFOLDER=COMPRAS/TAGUMEDICA/ADJUNTOS_CORREOS ```bash # Activar entorno virtual -source sunat-env/bin/activate +source sunat-venv/bin/activate # Obtener token manualmente (con navegador visible) python get_token.py @@ -61,10 +61,10 @@ python get_token_headless.py # Daemon completo (automático) python main.py -# Reporte de fallas (resumen texto + gráficos PNG) -python reporte_fallas.py +# Buscar intentos de descarga en el historial +python historial_manager.py --ruc 20503662869 --serie F001 --numero 123 ``` ## Logs -Todos los logs se guardan en `logs/YYYY-MM-DD.log`. Rotación automática diaria, los archivos mayores a 7 días se eliminan. +Todos los logs se guardan en `logs/sunat_YYYY-MM-DD.log`. Rotación automática diaria, los archivos mayores a 7 días se eliminan. diff --git a/gestor_fallas.py b/gestor_fallas.py index c7b1aa4..5f8985b 100644 --- a/gestor_fallas.py +++ b/gestor_fallas.py @@ -16,12 +16,18 @@ def init_db(): tipo TEXT NOT NULL, serie TEXT NOT NULL, numero TEXT NOT NULL, + objtype INTEGER NOT NULL DEFAULT 18, error TEXT NOT NULL, intentos INTEGER NOT NULL DEFAULT 1, proximo_intento TEXT, fecha_ultimo TEXT NOT NULL ) """) + columnas = {row[1] for row in conn.execute("PRAGMA table_info(fallas)")} + if "obj_type" in columnas and "objtype" not in columnas: + conn.execute("ALTER TABLE fallas RENAME COLUMN obj_type TO objtype") + elif "objtype" not in columnas: + conn.execute("ALTER TABLE fallas ADD COLUMN objtype INTEGER NOT NULL DEFAULT 18") # ============================================================================ def _calcular_proximo_intento(intentos): if intentos >= 20: @@ -44,7 +50,7 @@ def _calcular_proximo_intento(intentos): delta = timedelta(hours=12) return (datetime.now() + delta).isoformat() # ============================================================================ -def registrar_falla(docentry, ruc, tipo_cpe, serie, numero, error, intentos=None): +def registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, error, intentos=None): ahora = datetime.now().isoformat() if intentos is None: with _conn() as conn: @@ -56,9 +62,9 @@ def registrar_falla(docentry, ruc, tipo_cpe, serie, numero, error, intentos=None with _conn() as conn: conn.execute(""" INSERT OR REPLACE INTO fallas - (docentry, ruc, tipo, serie, numero, error, intentos, proximo_intento, fecha_ultimo) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, (docentry, ruc, tipo_cpe, serie, numero, str(error)[:200], intentos, proximo, ahora)) + (docentry, ruc, tipo, serie, numero, objtype, error, intentos, proximo_intento, fecha_ultimo) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, (docentry, ruc, tipo_cpe, serie, numero, obj_type, str(error)[:200], intentos, proximo, ahora)) # ============================================================================ def eliminar_falla(docentry): with _conn() as conn: diff --git a/get_token.py b/get_token.py index 6845242..2c3b162 100644 --- a/get_token.py +++ b/get_token.py @@ -4,9 +4,11 @@ import time from pathlib import Path from dotenv import load_dotenv from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout +from logger_manager import get_logger # ============================================================================ load_dotenv() +logger = get_logger(__name__) SOL_URL = "https://www.sunat.gob.pe/sol.html" MENU_URL = "https://e-menu.sunat.gob.pe/cl-ti-itmenucabina/MenuInternet.htm" @@ -18,15 +20,23 @@ RUC = os.getenv("SUNAT_RUC", "").strip() USER = os.getenv("SUNAT_USER", "").strip() PASSWORD = os.getenv("SUNAT_PASSWORD", "").strip() +# ============================================================================ +def _esperar_enter(mensaje): + """input() que no revienta si no hay terminal interactiva (ej. corriendo como LaunchAgent).""" + try: + input(mensaje) + except EOFError: + logger.warning("Sin terminal interactiva disponible, continuando sin pausa.") + # ============================================================================ def save_screenshot(page, name): path = SCREENSHOT_DIR / f"{name}.png" page.screenshot(path=str(path)) - print(f" 📸 Screenshot guardado: {path}") + logger.info(f"Screenshot guardado: {path}") # ============================================================================ def get_token(headless=False, timeout=60000): - print("🚀 Iniciando navegador...") + logger.info("Iniciando navegador...") with sync_playwright() as p: browser = p.firefox.launch( headless=headless, @@ -41,71 +51,71 @@ def get_token(headless=False, timeout=60000): page.set_default_timeout(timeout) # ─── 1. Ir a SOL ─────────────────────────────────── - print("1. Navegando a sol.html...") + logger.info("1. Navegando a sol.html...") page.goto(SOL_URL, wait_until="domcontentloaded") save_screenshot(page, "01_sol_home") # ─── 2. Click "Ingresar" en Mis Tramites y Consultas ── - print("2. Click en 'Ingresar' de Mis Tramites y Consultas...") + logger.info("2. Click en 'Ingresar' de Mis Tramites y Consultas...") with page.expect_popup() as popup_info: page.evaluate("tramiteConsulta()") popup = popup_info.value popup.wait_for_load_state("domcontentloaded") save_screenshot(popup, "02_popup_login") - print(f" Popup URL: {popup.url}") + logger.info(f"Popup URL: {popup.url}") # ─── 3. Login ───────────────────────────────────── - print("3. Verificando si hay formulario de login...") + logger.info("3. Verificando si hay formulario de login...") popup.wait_for_load_state("networkidle") - print(f" Título popup: {popup.title()}") + logger.info(f"Título popup: {popup.title()}") has_inputs = len(popup.locator("input[type='password']").all()) > 0 - print(f" ¿Tiene campo password?: {has_inputs}") + logger.info(f"¿Tiene campo password?: {has_inputs}") if has_inputs or any(k in popup.url.lower() for k in ["login", "autentica"]): - print(" Formulario de login detectado, rellenando credenciales...") + logger.info("Formulario de login detectado, rellenando credenciales...") fill_login_form(popup) save_screenshot(popup, "03_form_rellenado") - print(" Enviando formulario...") + logger.info("Enviando formulario...") popup.keyboard.press("Enter") popup.wait_for_load_state("networkidle", timeout=30000) popup.wait_for_timeout(3000) save_screenshot(popup, "04_post_login") - print(f" URL post-login: {popup.url}") + logger.info(f"URL post-login: {popup.url}") else: - print(" No se detectó login. Imprimiendo HTML de la página...") + logger.warning("No se detectó login. Imprimiendo HTML de la página...") try: html = popup.content() - print(f" HTML length: {len(html)}") + logger.info(f"HTML length: {len(html)}") # find form-related elements for el in ["form", "input", "login", "password", "ruc", "usuario"]: if el in html.lower(): idx = html.lower().index(el) - print(f" '{el}' encontrado en posición {idx}: ...{html[max(0,idx-50):idx+100]}...") + logger.info(f"'{el}' encontrado en posición {idx}: ...{html[max(0,idx-50):idx+100]}...") except Exception as e: - print(f" Error leyendo HTML: {e}") + logger.error(f"Error leyendo HTML: {e}") # ─── 4. Token via consulta directa ──────────────── - print("4. Accediendo directamente a consulta CPE con sesión activa...") + logger.info("4. Accediendo directamente a consulta CPE con sesión activa...") token = get_token_from_direct_access(context, popup) # ─── 5. Guardar o debug ─────────────────────────── if token: - print(f" ✅ Token obtenido!") + logger.info("Token obtenido!") save_token(token) save_screenshot(popup, "05_token_obtenido") browser.close() return token - print(" ❌ Token no encontrado. Revisa screenshots.") - print("\n📋 Frames en la página:") + logger.error("Token no encontrado. Revisa screenshots.") + logger.info("Frames en la página:") for i, frm in enumerate(popup.frames): - print(f" [{i}] name='{frm.name}' url='{frm.url[:100]}'") + logger.info(f"[{i}] name='{frm.name}' url='{frm.url[:100]}'") if not headless: - input("\n⏸️ Presiona Enter para cerrar el navegador...") + _esperar_enter("\n⏸️ Presiona Enter para cerrar el navegador...") browser.close() return None @@ -114,7 +124,7 @@ def fill_login_form(page): page.wait_for_timeout(1000) inputs = page.locator("input").all() - print(f" Inputs encontrados: {len(inputs)}") + logger.info(f"Inputs encontrados: {len(inputs)}") for inp in inputs: try: name = inp.get_attribute("name") or "" @@ -122,36 +132,36 @@ def fill_login_form(page): type_ = inp.get_attribute("type") or "" placeholder = inp.get_attribute("placeholder") or "" visible = inp.is_visible() - print(f" id='{id_}' name='{name}' type='{type_}' placeholder='{placeholder}' visible={visible}") + logger.info(f"id='{id_}' name='{name}' type='{type_}' placeholder='{placeholder}' visible={visible}") except: pass - print(f" Rellenando RUC: {RUC}") + logger.info(f"Rellenando RUC: {RUC}") page.fill("#txtRuc", RUC) - print(f" Rellenando Usuario: {USER}") + logger.info(f"Rellenando Usuario: {USER}") page.fill("#txtUsuario", USER) - print(f" Rellenando Contraseña: {'*' * len(PASSWORD)}") + logger.info(f"Rellenando Contraseña: {'*' * len(PASSWORD)}") page.fill("#txtContrasena", PASSWORD) captcha = page.locator("input[id='captcha']") if captcha.is_visible(): - print(" ⚠️ CAPTCHA detectado en la página!") + logger.warning("CAPTCHA detectado en la página!") save_screenshot(page, "03_captcha_detectado") - input("\n⏸️ Resuelve el captcha manualmente y presiona Enter...") + _esperar_enter("\n⏸️ Resuelve el captcha manualmente y presiona Enter...") else: - print(" ✅ Captcha no detectado") + logger.info("Captcha no detectado") # ============================================================================ def get_token_from_direct_access(context, login_page): - print(" Buscando enlace con token en el menú SOL...") + logger.info("Buscando enlace con token en el menú SOL...") token = extract_token_from_menu(login_page) if token: - print(f" ✅ Token encontrado en menú (href)") + logger.info("Token encontrado en menú (href)") return token - print(" Navegando árbol de menú SOL...") + logger.info("Navegando árbol de menú SOL...") return navigate_sol_menu(login_page) # ============================================================================ @@ -188,63 +198,63 @@ def navigate_sol_menu(page): page.wait_for_timeout(2000) # Debug: imprimir info de todos los frames - print(" Frames activos:") + logger.info("Frames activos:") for frm in page.frames: - print(f" [{frm.name}] {frm.url[:120]}") + logger.info(f"[{frm.name}] {frm.url[:120]}") if "consultacpe" in frm.url or (frm.name and "consulta" in frm.name.lower()): try: html = frm.content() - print(f" Content length: {len(html)}") + logger.info(f"Content length: {len(html)}") debug_path = SCREENSHOT_DIR / "iframe_content.html" with open(debug_path, "w", encoding="utf-8") as f: f.write(html) - print(f" HTML guardado en {debug_path}") + logger.info(f"HTML guardado en {debug_path}") import re tokens = re.findall(r'token[=:][^&\s"\']{50,}', html) if tokens: - print(f" Tokens encontrados (token=): {len(tokens)}") + logger.info(f"Tokens encontrados (token=): {len(tokens)}") for t in tokens[:3]: - print(f" {t[:80]}...") + logger.info(f"{t[:80]}...") jwt_matches = re.findall(r'eyJ[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+', html) if jwt_matches: - print(f" JWT encontrados: {len(jwt_matches)}") + logger.info(f"JWT encontrados: {len(jwt_matches)}") for j in jwt_matches[:3]: - print(f" {j[:80]}...") + logger.info(f"{j[:80]}...") t = extract_token_from_url(j) if t: return t ls = frm.evaluate("JSON.stringify(localStorage)") - print(f" localStorage: {ls[:200]}") + logger.info(f"localStorage: {ls[:200]}") ss = frm.evaluate("JSON.stringify(sessionStorage)") - print(f" sessionStorage: {ss[:200]}") + logger.info(f"sessionStorage: {ss[:200]}") except Exception as e: - print(f" Error: {e}") + logger.error(f"Error: {e}") return None # ============================================================================ def click_menu_item_by_id(page, element_id, expected_text): - print(f" Buscando elemento: #{element_id} ('{expected_text}')...") + logger.info(f"Buscando elemento: #{element_id} ('{expected_text}')...") for frm in page.frames: try: el = frm.locator(f"#{element_id}").first if el.count() > 0: visible = el.is_visible() - print(f" encontrado en frame '{frm.name}' visible={visible}") + logger.info(f"encontrado en frame '{frm.name}' visible={visible}") # Intentar click en el span interior span = frm.locator(f"#{element_id} span.spanNivelDescripcion") if span.count() > 0: - print(f" clickeando span...") + logger.info("clickeando span...") span.first.click() else: - print(f" clickeando li directamente...") + logger.info("clickeando li directamente...") el.click() page.wait_for_timeout(1500) @@ -256,19 +266,19 @@ def click_menu_item_by_id(page, element_id, expected_text): pass # Fallback: buscar por texto - print(f" No encontrado por ID. Buscando por texto...") + logger.info("No encontrado por ID. Buscando por texto...") for frm in page.frames: try: el = frm.locator(f"span:has-text('{expected_text}')").first if el.count() > 0 and el.is_visible(): - print(f" clickeando span con texto...") + logger.info("clickeando span con texto...") el.click() page.wait_for_timeout(1500) return None except: pass - print(f" ❌ No se pudo encontrar: {expected_text}") + logger.warning(f"No se pudo encontrar: {expected_text}") return None # ============================================================================ @@ -277,7 +287,7 @@ def check_for_token(page): for url in urls: t = extract_token_from_url(url) if t: - print(f" ✅ Token encontrado en URL") + logger.info("Token encontrado en URL") return t import re @@ -288,13 +298,13 @@ def check_for_token(page): if "consultacpe" not in frm_url and "nuevaconsulta" not in (frm.name or ""): continue - print(f" Buscando token en frame '{frm.name}' ({frm_url[:100]}...)") + logger.info(f"Buscando token en frame '{frm.name}' ({frm_url[:100]}...)") # Intentar sessionStorage try: ss_token = frm.evaluate("sessionStorage.getItem('SUNAT.token')") if ss_token and len(ss_token) > 50: - print(f" ✅ Token encontrado en sessionStorage") + logger.info("Token encontrado en sessionStorage") return ss_token except: pass @@ -303,7 +313,7 @@ def check_for_token(page): try: ls_token = frm.evaluate("localStorage.getItem('SUNAT.token')") if ls_token and len(ls_token) > 50: - print(f" ✅ Token encontrado en localStorage") + logger.info("Token encontrado en localStorage") return ls_token except: pass @@ -316,7 +326,7 @@ def check_for_token(page): ss_data = json.loads(all_ss) for key, value in ss_data.items(): if "token" in key.lower() and len(str(value)) > 50: - print(f" ✅ Token encontrado en sessionStorage key='{key}'") + logger.info(f"Token encontrado en sessionStorage key='{key}'") return value except: pass @@ -326,13 +336,13 @@ def check_for_token(page): matches = re.findall(r'token=([a-zA-Z0-9_\-\.]+)', content) for m in matches: if len(m) > 50: - print(f" ✅ Token encontrado en HTML") + logger.info("Token encontrado en HTML") return m matches = re.findall(r'access_token["\']?\s*[:=]\s*["\']([a-zA-Z0-9_\-\.]+)["\']', content) for m in matches: if len(m) > 50: - print(f" ✅ Access Token encontrado en HTML") + logger.info("Access Token encontrado en HTML") return m # Buscar JWT en todo el HTML @@ -340,17 +350,17 @@ def check_for_token(page): for j in jwt_matches: t = extract_token_from_url(j) if t: - print(f" ✅ JWT encontrado en HTML") + logger.info("JWT encontrado en HTML") return t except Exception as e: - print(f" Error: {e}") + logger.error(f"Error: {e}") return None # ============================================================================ def extract_token_from_menu(page): - print(" Analizando estructura del menú SOL...") + logger.info("Analizando estructura del menú SOL...") for frm in page.frames: try: @@ -384,18 +394,18 @@ def extract_token_from_menu(page): if result: for r in result: if r.get('text') and r['text'].strip(): - print(f" [{r['tag']}] vis={r['visible']} disp={r['display']} id='{r['id']}' cls='{r['cls']}'") - print(f" text: {r['text'][:100]}") + logger.info(f"[{r['tag']}] vis={r['visible']} disp={r['display']} id='{r['id']}' cls='{r['cls']}'") + logger.info(f"text: {r['text'][:100]}") if r.get('href'): - print(f" href: {r['href'][:120]}") + logger.info(f"href: {r['href'][:120]}") if r.get('onclick'): - print(f" onclick: {r['onclick'][:120]}") + logger.info(f"onclick: {r['onclick'][:120]}") if 'token=' in r.get('href', ''): token = extract_token_from_url(r['href']) if token: return token except Exception as e: - print(f" Error analizando frame: {e}") + logger.error(f"Error analizando frame: {e}") return None @@ -431,8 +441,8 @@ def save_token(token): data = {"token": token, "expires_at": 0} TOKEN_FILE.write_text(json.dumps(data, indent=2)) - print(f" 💾 Token guardado en: {TOKEN_FILE}") - print(f" ⏰ Expira: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(data['expires_at']))}") + logger.info(f"Token guardado en: {TOKEN_FILE}") + logger.info(f"Expira: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(data['expires_at']))}") # ============================================================================ if __name__ == "__main__": diff --git a/get_token_headless.py b/get_token_headless.py index 2a4ad96..a9ca700 100644 --- a/get_token_headless.py +++ b/get_token_headless.py @@ -10,6 +10,9 @@ from get_token import ( fill_login_form, save_token, save_screenshot, check_for_token, navigate_sol_menu, extract_token_from_menu, ) +from logger_manager import get_logger + +logger = get_logger(__name__) # ============================================================================ USER_AGENT = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) " @@ -47,16 +50,16 @@ def get_token_headless(timeout=90000): page = context.new_page() page.set_default_timeout(timeout) - print("1. Sesión desde SOL...") + logger.info("1. Sesión desde SOL...") page.goto("https://www.sunat.gob.pe/sol.html", wait_until="domcontentloaded") - print("2. Navegando a menú directo...") + logger.info("2. Navegando a menú directo...") page.goto(MENU_URL, wait_until="domcontentloaded") page.wait_for_load_state("networkidle") has_inputs = len(page.locator("input[type='password']").all()) > 0 if has_inputs or any(k in page.url.lower() for k in ["login", "autentica", "auth"]): - print("3. Login detectado, rellenando credenciales...") + logger.info("3. Login detectado, rellenando credenciales...") fill_login_form(page) page.keyboard.press("Enter") try: @@ -65,9 +68,9 @@ def get_token_headless(timeout=90000): pass page.wait_for_timeout(3000) else: - print("3. Sin login (sesión activa)") + logger.info("3. Sin login (sesión activa)") - print("4. Buscando token en menú...") + logger.info("4. Buscando token en menú...") page.wait_for_timeout(2000) token = check_for_token(page) if not token: @@ -79,11 +82,11 @@ def get_token_headless(timeout=90000): browser.close() if token: - print(" ✅ Token obtenido (headless)") + logger.info("Token obtenido (headless)") save_token(token) return token - print(" ❌ No se encontró token") + logger.error("No se encontró token") return None # ============================================================================ diff --git a/hana_manager.py b/hana_manager.py index 9af2911..1fd8654 100644 --- a/hana_manager.py +++ b/hana_manager.py @@ -2,9 +2,11 @@ import os import json from pathlib import Path from dotenv import load_dotenv +from logger_manager import get_logger # ============================================================================ load_dotenv() +logger = get_logger(__name__) HANA_HOST = os.getenv("HANA_HOST", "") HANA_PORT = int(os.getenv("HANA_PORT", 30015)) @@ -27,7 +29,7 @@ def get_connection(): password=HANA_PASSWORD ) except Exception as e: - print(f" ❌ Error conectando a HANA: {e}") + logger.error(f"Error conectando a HANA: {e}") return None return _conn @@ -49,6 +51,7 @@ def obtener_pendientes() -> list[dict]: "LicTradNum", "FolioPref", "FolioNum", + "ObjType", "U_PGX_REPIMP" "urlPDF", "U_PGX_PDF" "tienePDF", "U_PGX_REPIMPXML" "urlXML", @@ -56,9 +59,35 @@ def obtener_pendientes() -> list[dict]: FROM "{HANA_SCHEMA}".OPCH WHERE ("U_PGX_PDF" = 'N' OR "U_PGX_XML" = 'N') AND EXTRACT(YEAR FROM "DocDate") = 2026 - AND "Indicator" IN ('01', '02','08') + AND "Indicator" IN ('01','02','03') AND "CANCELED" = 'N' - ORDER BY EXTRACT(MONTH FROM "DocDate") DESC + AND "LicTradNum" NOT IN ('20100047218','20100130204','20100053455') + AND SUBSTRING("FolioPref", 1, 1) NOT BETWEEN '0' AND '9' + + UNION ALL + + SELECT + "DocEntry", + "DocNum", + "DocDate", + EXTRACT(MONTH FROM "DocDate") "Mes", + "Indicator", + "LicTradNum", + "FolioPref", + "FolioNum", + "ObjType", + "U_PGX_REPIMP" "urlPDF", + "U_PGX_PDF" "tienePDF", + "U_PGX_REPIMPXML" "urlXML", + "U_PGX_XML" "tieneXML" + FROM "{HANA_SCHEMA}".ORPC + WHERE ("U_PGX_PDF" = 'N' OR "U_PGX_XML" = 'N') + AND EXTRACT(YEAR FROM "DocDate") = 2026 + AND "Indicator" IN ('07') + AND "CANCELED" = 'N' + AND "LicTradNum" NOT IN ('20100047218','20100130204','20100053455') + AND SUBSTRING("FolioPref", 1, 1) NOT BETWEEN '0' AND '9' + """ cursor = conn.cursor() cursor.execute(query) @@ -68,45 +97,47 @@ def obtener_pendientes() -> list[dict]: result = [dict(zip(columns, row)) for row in rows] return result except Exception as e: - print(f" ❌ Error en query HANA: {e}") + logger.error(f"Error en query HANA: {e}") return [] # ============================================================================ -def marcar_pdf_descargado(docentry: int, url: str, ruta: str): +def marcar_pdf_descargado(docentry: int, url: str, ruta: str, obj_type: int = 18): conn = get_connection() if not conn: return False + tabla = "ORPC" if int(obj_type) == 19 else "OPCH" try: cursor = conn.cursor() cursor.execute(f""" - UPDATE "{HANA_SCHEMA}".OPCH + UPDATE "{HANA_SCHEMA}".{tabla} SET "U_PGX_PDF" = 'Y' WHERE "DocEntry" = ? - """, (docentry)) + """, (docentry,)) conn.commit() cursor.close() return True except Exception as e: - print(f" ❌ Error actualizando PDF HANA: {e}") + logger.error(f"Error actualizando PDF HANA: {e}") return False # ============================================================================ -def marcar_xml_descargado(docentry: int, url: str, ruta: str): +def marcar_xml_descargado(docentry: int, url: str, ruta: str, obj_type: int = 18): conn = get_connection() if not conn: return False + tabla = "ORPC" if int(obj_type) == 19 else "OPCH" try: cursor = conn.cursor() cursor.execute(f""" - UPDATE "{HANA_SCHEMA}".OPCH + UPDATE "{HANA_SCHEMA}".{tabla} SET "U_PGX_XML" = 'Y' WHERE "DocEntry" = ? - """, (docentry)) + """, (docentry,)) conn.commit() cursor.close() return True except Exception as e: - print(f" ❌ Error actualizando XML HANA: {e}") + logger.error(f"Error actualizando XML HANA: {e}") return False # ============================================================================ diff --git a/historial_manager.py b/historial_manager.py new file mode 100644 index 0000000..36ff34c --- /dev/null +++ b/historial_manager.py @@ -0,0 +1,121 @@ +import sqlite3 +from pathlib import Path +from datetime import datetime, timedelta +from logger_manager import get_logger + +DB_PATH = Path(__file__).parent / "historial.db" +logger = get_logger(__name__) +# ============================================================================ +def _conn(): + return sqlite3.connect(str(DB_PATH)) +# ============================================================================ +def init_db(): + with _conn() as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute(""" + CREATE TABLE IF NOT EXISTS historial ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + fecha TEXT NOT NULL, + docentry INTEGER NOT NULL, + docnum TEXT, + ruc TEXT NOT NULL, + tipo TEXT NOT NULL, + serie TEXT NOT NULL, + numero TEXT NOT NULL, + objtype INTEGER NOT NULL DEFAULT 18, + es_retry INTEGER NOT NULL DEFAULT 0, + intento_num INTEGER, + consulta_status INTEGER, + pdf_resultado TEXT, + xml_resultado TEXT, + resultado TEXT NOT NULL, + detalle TEXT, + json_path TEXT + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_historial_docentry ON historial(docentry)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_historial_fecha ON historial(fecha)") +# ============================================================================ +def registrar_intento(docentry, docnum, ruc, tipo, serie, numero, objtype, es_retry, + intento_num, consulta_status, pdf_resultado, xml_resultado, + resultado, detalle="", json_path=None): + ahora = datetime.now().isoformat() + with _conn() as conn: + conn.execute(""" + INSERT INTO historial + (fecha, docentry, docnum, ruc, tipo, serie, numero, objtype, es_retry, + intento_num, consulta_status, pdf_resultado, xml_resultado, resultado, + detalle, json_path) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, (ahora, docentry, docnum, ruc, tipo, serie, numero, objtype, int(es_retry), + intento_num, consulta_status, pdf_resultado, xml_resultado, resultado, + str(detalle)[:500] if detalle else detalle, json_path)) +# ============================================================================ +def limpiar_historial_antiguo(dias=180): + limite = (datetime.now() - timedelta(days=dias)).isoformat() + with _conn() as conn: + cursor = conn.execute("DELETE FROM historial WHERE fecha < ?", (limite,)) + count = cursor.rowcount + if count > 0: + logger.info(f"Se eliminaron {count} registros de historial de más de {dias} días.") + return count +# ============================================================================ +def historial_docentry(docentry) -> list[dict]: + with _conn() as conn: + conn.row_factory = sqlite3.Row + cursor = conn.execute( + "SELECT * FROM historial WHERE docentry = ? ORDER BY id ASC", (docentry,) + ) + return [dict(row) for row in cursor.fetchall()] +# ============================================================================ +def buscar_intentos(ruc=None, serie=None, numero=None, tipo=None) -> list[dict]: + """Busca intentos registrados filtrando por cualquier combinación de ruc/serie/numero/tipo.""" + condiciones = [] + params = [] + if ruc: + condiciones.append("ruc = ?") + params.append(ruc) + if serie: + condiciones.append("serie = ?") + params.append(serie) + if numero: + condiciones.append("numero = ?") + params.append(str(numero)) + if tipo: + condiciones.append("tipo = ?") + params.append(tipo) + + where = " AND ".join(condiciones) if condiciones else "1=1" + with _conn() as conn: + conn.row_factory = sqlite3.Row + cursor = conn.execute(f"SELECT * FROM historial WHERE {where} ORDER BY fecha ASC", params) + return [dict(row) for row in cursor.fetchall()] +# ============================================================================ +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Buscar intentos de descarga en historial.db") + parser.add_argument("--ruc", help="RUC del emisor") + parser.add_argument("--serie", help="Serie del comprobante (ej: F001)") + parser.add_argument("--numero", help="Número del comprobante") + parser.add_argument("--tipo", help="Tipo de comprobante (01, 03, 07, 08...)") + args = parser.parse_args() + + if not any([args.ruc, args.serie, args.numero, args.tipo]): + parser.error("Indica al menos un filtro: --ruc, --serie, --numero o --tipo") + + filas = buscar_intentos(ruc=args.ruc, serie=args.serie, numero=args.numero, tipo=args.tipo) + + if not filas: + print("No se encontraron intentos con esos filtros.") + else: + print(f"{len(filas)} intento(s) encontrado(s):\n") + for f in filas: + print( + f"{f['fecha']} DocEntry={f['docentry']} " + f"{f['ruc']}-{f['tipo']}-{f['serie']}-{f['numero']} " + f"intento={f['intento_num']} retry={'sí' if f['es_retry'] else 'no'} " + f"status={f['consulta_status']} resultado={f['resultado']} " + f"pdf={f['pdf_resultado']} xml={f['xml_resultado']} detalle={f['detalle']}" + ) +# ============================================================================ diff --git a/logger_manager.py b/logger_manager.py new file mode 100644 index 0000000..ac31fb9 --- /dev/null +++ b/logger_manager.py @@ -0,0 +1,146 @@ +import logging +import re +import contextvars +import os +import sys +import time +from contextlib import contextmanager +from logging.handlers import TimedRotatingFileHandler +from pathlib import Path + +LOG_DIR = Path(__file__).parent / "logs" +LOG_FILE = LOG_DIR / "sunat.log" +_LOG_CONFIGURED = False + +_ANSI_RE = re.compile(r"\033\[[0-9;]*m") + +_stack: contextvars.ContextVar[list] = contextvars.ContextVar("log_stack", default=[]) +# ============================================================================ + +class TreeFormatter(logging.Formatter): + def format(self, record): + return f"{self.formatTime(record)} ─── {record.getMessage()}" + + def formatTime(self, record, datefmt=None): + import datetime + return datetime.datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S") + +# ============================================================================ +class PlainFileFormatter(TreeFormatter): + """Igual que TreeFormatter pero sin códigos ANSI de color, para el archivo.""" + def format(self, record): + return _ANSI_RE.sub("", super().format(record)) + +# ============================================================================ +def _build_prefix(stack, is_last): + if not stack: + return "" + parts = [] + for sib_is_last in stack[:-1]: + parts.append(" " if sib_is_last else "│ ") + parts.append("└── " if is_last else "├── ") + return "".join(parts) + +# ============================================================================ +class _TaskContext: + def __init__(self, logger, msg, emoji, is_last): + self._logger = logger + self._msg = msg + self._emoji = emoji + self._is_last = is_last + + def __enter__(self): + stack = _stack.get() + prefix = _build_prefix(stack, self._is_last) + emoji_str = f"{self._emoji} " if self._emoji else "" + self._logger._logger.info(f"{prefix}{emoji_str}{self._msg}") + _stack.set(stack + [self._is_last]) + return TaskLogger(self._logger._logger) + + def __exit__(self, exc_type, exc_val, exc_tb): + _stack.set(_stack.get()[:-1]) + +# ============================================================================ +class TaskLogger: + def __init__(self, logger): + self._logger = logger + + @contextmanager + def task(self, msg, emoji=""): + with _TaskContext(self, msg, emoji, is_last=False) as sub_logger: + yield sub_logger + + @contextmanager + def last_task(self, msg, emoji=""): + with _TaskContext(self, msg, emoji, is_last=True) as sub_logger: + yield sub_logger + + def info(self, msg, *args, _last=False, **kwargs): + self._log(logging.INFO, msg, args, kwargs, _last) + + def warning(self, msg, *args, _last=False, **kwargs): + self._log(logging.WARNING, msg, args, kwargs, _last) + + def error(self, msg, *args, _last=False, **kwargs): + self._log(logging.ERROR, msg, args, kwargs, _last) + + def _log(self, level, msg, args, kwargs, is_last): + stack = _stack.get() + prefix = _build_prefix(stack, is_last) + self._logger.log(level, f"{prefix}{msg}", *args, **kwargs) + +def _rotated_name(default_name): + """TimedRotatingFileHandler entrega 'sunat.log.2026-08-30'; lo convertimos + a 'sunat_2026-08-30.log' para que sea más legible como nombre de archivo.""" + date_part = default_name.rsplit(".", 1)[-1] + return str(LOG_FILE.with_name(f"{LOG_FILE.stem}_{date_part}{LOG_FILE.suffix}")) + +# ============================================================================ +def limpiar_logs_antiguos(dias=7): + """Borra los logs rotados (sunat_YYYY-MM-DD.log) con más de 'dias' de antigüedad.""" + limite = time.time() - dias * 86400 + count = 0 + for p in LOG_DIR.glob(f"{LOG_FILE.stem}_*{LOG_FILE.suffix}"): + try: + if p.stat().st_mtime < limite: + p.unlink() + count += 1 + except OSError: + pass + return count + +# ============================================================================ +def setup_logging(level=None, console=True): + global _LOG_CONFIGURED + if _LOG_CONFIGURED: + return + + if level is None: + level_name = os.getenv("LOG_LEVEL", "INFO").upper() + level = getattr(logging, level_name, logging.INFO) + + LOG_DIR.mkdir(exist_ok=True) + + file_handler = TimedRotatingFileHandler( + LOG_FILE, + when="midnight", + interval=1, + backupCount=0, # la retención la maneja limpiar_logs_antiguos() con el namer de abajo + encoding="utf-8", + ) + file_handler.namer = _rotated_name + file_handler.setFormatter(PlainFileFormatter()) + + handlers = [file_handler] + if console: + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setFormatter(TreeFormatter()) + handlers.append(console_handler) + + logging.basicConfig(level=level, handlers=handlers, force=True) + _LOG_CONFIGURED = True + +# ============================================================================ +def get_logger(name="sunat"): + setup_logging() + return TaskLogger(logging.getLogger(name)) \ No newline at end of file diff --git a/logger_utilities.py b/logger_utilities.py deleted file mode 100644 index 65f40cf..0000000 --- a/logger_utilities.py +++ /dev/null @@ -1,93 +0,0 @@ -import sys -import os -from datetime import datetime, timedelta - -_stdout_original = sys.stdout -_stderr_original = sys.stderr -_file_stream = None -_log_dir = "logs" - -# ============================================================================ -class MultiStream: - def __init__(self, *streams): - self.streams = list(streams) - - def write(self, data): - for stream in self.streams: - stream.write(data) - stream.flush() - - def flush(self): - for stream in self.streams: - stream.flush() - - def reemplazar_file(self, old_fh, new_fh): - for i, s in enumerate(self.streams): - if s is old_fh: - self.streams[i] = new_fh - return - -# ============================================================================ -def _ruta_log(hoy=None): - if hoy is None: - hoy = datetime.now().strftime("%Y-%m-%d") - if not os.path.exists(_log_dir): - os.makedirs(_log_dir) - return os.path.join(_log_dir, f"{hoy}.log") - -# ============================================================================ -def setup_logger(): - global _file_stream - log_path = _ruta_log() - _file_stream = open(log_path, "a", encoding="utf-8") - sys.stdout = MultiStream(_stdout_original, _file_stream) - sys.stderr = MultiStream(_stderr_original, _file_stream) - return log_path - -# ============================================================================ -def rotar_log(): - global _file_stream - hoy = datetime.now().strftime("%Y-%m-%d") - log_path = _ruta_log(hoy) - - if _file_stream and os.path.abspath(_file_stream.name) == os.path.abspath(log_path): - return - - nuevo_fh = open(log_path, "a", encoding="utf-8") - if isinstance(sys.stdout, MultiStream): - sys.stdout.reemplazar_file(_file_stream, nuevo_fh) - if isinstance(sys.stderr, MultiStream): - sys.stderr.reemplazar_file(_file_stream, nuevo_fh) - try: - _file_stream.close() - except Exception: - pass - _file_stream = nuevo_fh - print(f"📄 Log rotado a: {log_path}") - -# ============================================================================ -def limpiar_logs_antiguos(dias=7): - """ - Elimina archivos de log con más de 'n' días de antigüedad. - """ - log_dir = "logs" - if not os.path.exists(log_dir): - return - - limite = datetime.now() - timedelta(days=dias) - count = 0 - - for archivo in os.listdir(log_dir): - ruta_archivo = os.path.join(log_dir, archivo) - if os.path.isfile(ruta_archivo) and archivo.endswith(".log"): - fecha_archivo = datetime.fromtimestamp(os.path.getmtime(ruta_archivo)) - if fecha_archivo < limite: - try: - os.remove(ruta_archivo) - count += 1 - except Exception as e: - print(f"⚠️ Error al eliminar log {archivo}: {e}") - - if count > 0: - print(f"🧹 Se eliminaron {count} archivos de log antiguos (más de {dias} días).") -# ============================================================================ \ No newline at end of file diff --git a/login_api.py b/login_api.py deleted file mode 100644 index 30204d3..0000000 --- a/login_api.py +++ /dev/null @@ -1,309 +0,0 @@ -import os -import json -import re -import time -import base64 -from pathlib import Path -from dotenv import load_dotenv -import requests -import urllib3 - -# ============================================================================ -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - -load_dotenv() - -TOKEN_FILE = Path(__file__).parent / "token.json" -CLIENT_ID = "cd8e7afb-a0e2-4217-9187-88206e4ba7af" -TOKEN_URL = f"https://api-seguridad.sunat.gob.pe/v1/clientessol/{CLIENT_ID}/oauth2/token/" - -RUC = os.getenv("SUNAT_RUC", "").strip() -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] - padding = 4 - len(payload_b64) % 4 - if padding != 4: - payload_b64 += "=" * padding - return json.loads(base64.b64decode(payload_b64)) - except: - return None - -# ============================================================================ -def save_token(token): - payload = decode_jwt(token) or {} - expires_at = payload.get("exp", 0) - data = { - "token": token, - "expires_at": expires_at, - "client_id": payload.get("clientId", ""), - "ruc": payload.get("sub", ""), - "usuario_sol": payload.get("userdata", {}).get("usuarioSOL", ""), - "_source": "login_api.py", - "_grant_type": payload.get("grantType", ""), - } - TOKEN_FILE.write_text(json.dumps(data, indent=2)) - print(f"\n✅ Token guardado en: {TOKEN_FILE}") - print(f" Expira: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(expires_at))}") - print(f" RUC: {data['ruc']}") - print(f" Usuario SOL: {data['usuario_sol']}") - return True - -# ============================================================================ -def try_password_grant(): - print("\n🔑 Intentando grant_type=password...") - data = { - "grant_type": "password", - "username": J_USERNAME, - "password": PASSWORD, - "client_id": CLIENT_ID, - } - headers = { - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0", - "Content-Type": "application/x-www-form-urlencoded", - "Accept": "application/json", - } - - try: - r = requests.post(TOKEN_URL, data=data, headers=headers, timeout=30, verify=False) - print(f" {r.status_code}") - if r.status_code == 200: - j = r.json() - token = j.get("access_token") - if token: - print(f" ✅ Password grant funcionó!") - save_token(token) - return True - print(f" ⚠️ Sin access_token en respuesta: {str(j)[:200]}") - else: - print(f" ❌ Response: {r.text[:300]}") - except Exception as e: - 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...") - - session = requests.Session() - session.verify = False - session.headers.update({ - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0", - "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - }) - - menu_url = "https://e-menu.sunat.gob.pe/cl-ti-itmenu/MenuInternet.htm" - - try: - r = session.get(menu_url, allow_redirects=True, timeout=30) - log(f" GET MenuInternet: {r.status_code}") - log(f" URL final: {r.url}") - - html = r.text - (Path(__file__).parent / "descargas" / "menu_response.html").write_text(html) - - # Extraer URL del redirect() en el JavaScript - redirect_match = re.search(r'redirect\s*\(\s*["\']([^"\']+)["\']', html) - if not redirect_match: - print(" ❌ No se encontró URL de login en MenuInternet.htm") - return False - - authen_url = redirect_match.group(1) - log(f"\n Login URL: {authen_url}") - - # Extraer state y redirect_uri de la authen URL - state_match = re.search(r'state=([^&\s"\']+)', authen_url) - state_from_url = state_match.group(1) if state_match else None - if state_from_url: - log(f" state: {state_from_url[:60]}...") - state_decoded = state_from_url # No decodear: el + de base64 se pierde con unquote - - redirect_uri_match = re.search(r'redirect_uri=([^&\s"\']+)', authen_url) - redirect_uri = requests.utils.unquote(redirect_uri_match.group(1)) if redirect_uri_match else None - - # 2. Acceder a la URL de authen para obtener el formulario de login - log("\n 2. Accediendo a authen...") - r2 = session.get(authen_url, timeout=30) - log(f" GET authen: {r2.status_code}") - log(f" URL final: {r2.url[:200]}") - - html2 = r2.text - (Path(__file__).parent / "descargas" / "login_page.html").write_text(html2) - - # Extraer state y originalUrl de la URL final (authen los pasa a loginMenuSol) - final_state_match = re.search(r'[?&]state=([^&\s]+)', r2.url) - final_orig_match = re.search(r'[?&]originalUrl=([^&\s]+)', r2.url) - final_state = requests.utils.unquote(final_state_match.group(1)) if final_state_match else state_decoded - final_orig = requests.utils.unquote(final_orig_match.group(1)) if final_orig_match else redirect_uri - - # Verificar CAPTCHA real: buscar - # (el captcha solo se renderiza via JS tras 3 fallos; con requests el JS no corre) - html_no_script = re.sub(r']*>.*?', '', html2, flags=re.DOTALL | re.IGNORECASE) - if re.search(r'id=[\'"]imgCaptcha[\'"]', html_no_script): - print(" ⚠️ CAPTCHA visible detectado (imgCaptcha)") - return False - print(" ✅ Sin CAPTCHA visible") - - # Extraer form action - action_match = re.search(r']*action=[\'"]([^\'"]+)[\'"]', html2) - form_action = action_match.group(1) if action_match else "j_security_check" - log(f" form action: {form_action}") - - # Construir POST URL (relativo a la URL actual) - base_path = r2.url.split("?")[0].rsplit("/", 1)[0] + "/" - if form_action.startswith("http"): - post_url = form_action - elif form_action.startswith("/"): - base = re.match(r'(https?://[^/]+)', r2.url) - post_url = f"{base.group(1)}{form_action}" if base else form_action - else: - post_url = base_path + form_action - log(f" POST URL: {post_url}") - - # Extraer lang de la URL actual - lang_match = re.search(r'lang=([^&\s]+)', r2.url) - lang = lang_match.group(1) if lang_match else "es-PE" - - # 3. POST credenciales - log("\n 3. Enviando credenciales...") - form_data = { - "tipo": "2", - "dni": "", - "custom_ruc": RUC, - "j_username": J_USERNAME, - "j_password": PASSWORD, - "captcha": "", - "originalUrl": redirect_uri or "", - "lang": lang, - "state": state_decoded or "", - } - - r3 = session.post(post_url, data=form_data, allow_redirects=False, timeout=30) - log(f" POST login: {r3.status_code}") - location = r3.headers.get("Location", "") - log(f" Location: {location[:200]}") - - # 4. Seguir redirects buscando authorization code - auth_code = None - next_url = location if r3.status_code in (301, 302, 303, 307, 308) else None - - for i in range(15): - if not next_url: - break - - if "code=" in next_url: - code_match = re.search(r'code=([^&]+)', next_url) - if code_match: - auth_code = code_match.group(1) - log(f"\n ✅ Authorization code obtenido en redirect {i+1}!") - break - - r4 = session.get(next_url, allow_redirects=False, timeout=30) - log(f" Redirect {i+1}: {r4.status_code} -> {r4.url[:100]}") - next_url = r4.headers.get("Location") if r4.status_code in (301, 302, 303, 307, 308) else None - - if auth_code: - log(f"\n 4. Siguiendo redirect a AutenticaMenuInternet.htm con session activa...") - # El SOL portal intercambia el code internamente y genera el menú - # Seguir la cadena de redirects hasta llegar al menú SOL - next_url = location - for i in range(10): - if not next_url: - break - log(f" GET redirect {i+1}: {next_url[:120]}") - r5 = session.get(next_url, allow_redirects=(i == 0), timeout=30) - log(f" Status: {r5.status_code}") - log(f" URL final: {r5.url[:150]}") - - # Buscar token en la respuesta - jwt_matches = re.findall(r'eyJ[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+', r5.text) - for j in jwt_matches: - if len(j) > 100: - print(f"\n ✅ JWT encontrado en HTML!") - save_token(j) - return True - - # Buscar en headers - for h, v in r5.headers.items(): - if 'token' in h.lower() or 'auth' in h.lower(): - log(f" Header {h}: {v[:100]}") - - # Buscar en cookies - for c in session.cookies: - if 'token' in c.name.lower(): - log(f" Cookie {c.name}: {c.value[:80]}") - - next_url = r5.headers.get("Location") if r5.status_code in (301, 302, 303, 307, 308) else None - - # Si no encontramos token en las respuestas, extraerlo del contenido - # Buscar sessionStorage o localStorage hints - log(f"\n Buscando token en última respuesta...") - patterns = [ - r'access_token["\']?\s*[=:]\s*["\']([a-zA-Z0-9_\-\.]{100,})["\']', - r'token["\']?\s*[=:]\s*["\']([a-zA-Z0-9_\-\.]{100,})["\']', - r'["\']SUNAT\.token["\']?\s*[=:]\s*["\']([a-zA-Z0-9_\-\.]+)["\']', - ] - for p in patterns: - m = re.search(p, r5.text if 'r5' in dir() else '') - if m: - print(f"\n ✅ Token encontrado via patrón!") - save_token(m.group(1)) - return True - - print(f"\n ❌ Token no encontrado en respuestas HTTP") - print(f" El flujo completo es: login SOL -> menú -> click item -> token en sessionStorage") - print(f" Esto requiere JavaScript. Usa: python get_token.py") - else: - if location: - print(f" ❌ No se obtuvo code. Última location: {location[:150]}") - else: - print(f" ❌ No hubo redirect después del POST") - print(f" Respuesta: {r3.text[:500]}") - - except Exception as e: - print(f" ❌ Error: {e}") - import traceback - traceback.print_exc() - - return False - -# ============================================================================ -def main(): - print("=" * 60) - print("🔐 LOGIN SUNAT API (sin Playwright)") - print("=" * 60) - print(f"\n📋 Config:") - print(f" RUC: {RUC}") - print(f" Usuario: {USER}") - print(f" Client ID: {CLIENT_ID}") - print(f" Token URL: {TOKEN_URL}") - - # Intentar password grant primero - if try_password_grant(): - return - - # Si falla, intentar authorization_code flow - print("\n⚠️ Password grant falló. Intentando authorization_code flow...") - if try_auth_code_flow(): - return - - print("\n❌ No se pudo obtener token automáticamente.") - print(" Posibles causas:") - print(" - CAPTCHA requerido (usa: python get_token.py)") - print(" - UUID de cliente incorrecto") - 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 f5f8bef..95fdb78 100644 --- a/main.py +++ b/main.py @@ -14,9 +14,10 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) load_dotenv() +from logger_manager import get_logger, setup_logging, limpiar_logs_antiguos, LOG_FILE from hana_manager import obtener_pendientes, marcar_pdf_descargado, marcar_xml_descargado, cerrar_conexion -from logger_utilities import setup_logger, rotar_log, limpiar_logs_antiguos from gestor_fallas import init_db, registrar_falla, eliminar_falla, obtener_fallas_pendientes, obtener_docentries_falla, total_fallas +from historial_manager import init_db as init_historial_db, registrar_intento, limpiar_historial_antiguo from smb_manager import sincronizar_a_smb from consulta_cpe import ( TokenExpiredError, load_token, get_headers, consultar_cpe, @@ -24,6 +25,8 @@ from consulta_cpe import ( API_HOST, BASE_PATH, DOWNLOAD_DIR ) +logger = get_logger(__name__) + # ============================================================================ # CONFIGURACIÓN DE HORARIOS (desde scheduler.json) # ============================================================================ @@ -40,7 +43,7 @@ def _cargar_scheduler(): try: DIAS_ACTIVOS, HORARIOS = _cargar_scheduler() except Exception as e: - print(f"⚠️ Error cargando {SCHEDULER_FILE}: {e}. Usando defaults.") + logger.warning(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 @@ -97,38 +100,38 @@ def obtener_token(): """Obtiene o renueva el token si es necesario.""" token = load_token() if token: - print(" 🔑 Token existente válido, reutilizando") + logger.info("Token existente válido, reutilizando") return token - print("\n🔑 Token no disponible o expirado. Obteniendo nuevo...") + logger.info("Token no disponible o expirado. Obteniendo nuevo...") from get_token_headless import get_token_headless as obtener_headless, firefox_instalado if not firefox_instalado(): - print(" ⚠️ Firefox no instalado. Saltando headless...") + logger.warning("Firefox no instalado. Saltando headless...") else: try: if obtener_headless(): - print(" ✅ Token obtenido vía get_token_headless (headless)") + logger.info("Token obtenido vía get_token_headless (headless)") return load_token() except Exception as e: - print(f" ⚠️ Headless falló: {e}") + logger.warning(f"Headless falló: {e}") - print(" ⚠️ Headless no disponible. Intentando get_token.py con navegador visible...") + logger.warning("Headless no disponible. Intentando get_token.py con navegador visible...") from get_token import get_token as obtener_token_playwright obtener_token_playwright(headless=False, timeout=120000) for _ in range(30): tk = load_token() if tk: - print(" ✅ Token obtenido vía Playwright (get_token.py)") + logger.info("Token obtenido vía Playwright (get_token.py)") return tk time.sleep(1) - print(" ⚠️ No se detectó token tras 30s. Continuando...") + logger.warning("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...") + logger.info("Buscando documentos pendientes en HANA...") docs_hana = obtener_pendientes() fallas_docentries = obtener_docentries_falla() docs_nuevos = [d for d in docs_hana if d.get("DocEntry") not in fallas_docentries] @@ -142,16 +145,18 @@ def procesar_lote(token): "Indicator": d["tipo"], "FolioPref": d["serie"], "FolioNum": d["numero"], + "ObjType": d.get("objtype", 18), + "Intentos": d.get("intentos", 1), "tienePDF": "N", "tieneXML": "N", }) total_fallas_count = total_fallas() if not docs_nuevos and not docs_retry: - print(" ✅ No hay documentos pendientes") + logger.info("No hay documentos pendientes") return - print(f" 📄 {len(docs_hana)} en HANA | 🆕 {len(docs_nuevos)} nuevos | 🔄 {len(docs_retry)} reintentos | 🗂️ {total_fallas_count} en fallas\n") + logger.info(f"{len(docs_hana)} en HANA | {len(docs_nuevos)} nuevos | {len(docs_retry)} reintentos | {total_fallas_count} en fallas") DOWNLOAD_DIR.mkdir(exist_ok=True) # ======================================================================== @@ -166,123 +171,136 @@ def procesar_lote(token): numero = doc.get("FolioNum") or 0 docentry = doc.get("DocEntry") or 0 docnum = doc.get("DocNum") or 0 + obj_type = doc.get("ObjType") or 18 + intento_num = doc.get("Intentos") or 1 tiene_pdf = (doc.get("tienePDF") or "N").strip() tiene_xml = (doc.get("tieneXML") or "N").strip() if not ruc or not tipo_cpe or not serie or not numero: - print(f" ⚠️ DocEntry={docentry} DocNum={docnum}: datos incompletos, saltando") + logger.warning(f"DocEntry={docentry} DocNum={docnum}: datos incompletos, saltando") return ultimo = idx == total - 1 - raiz = "└──" if ultimo else "├──" tag = "🔄" if es_retry else "📄" - print(f"{raiz} {tag} [{idx+1}/{total}] {ruc}-{tipo_cpe}-{serie}-{numero} [DocEntry={docentry} | DocNum={docnum}]") - - prefijo = " " if ultimo else "│ " - prefijo_hijo = prefijo cid = cpe_id(ruc, tipo_cpe, serie, numero) - for intento_token in range(2): - ok_pdf = True - ok_xml = True - hubo_error = False + task_cm = logger.last_task if ultimo else logger.task + with task_cm(f"[{idx+1}/{total}] {ruc}-{tipo_cpe}-{serie}-{numero} [DocEntry={docentry} | DocNum={docnum}]", emoji=tag) as doc_log: + for intento_token in range(2): + ok_pdf = True + ok_xml = True + hubo_error = False - try: - time.sleep(random.uniform(1.0, 2.0)) - data, status_code = consultar_cpe(token, ruc, tipo_cpe, serie, numero, descargar=False, verbose=False) - - if not data: - print(f"{prefijo_hijo}└── Consulta: {color_rojo}Error {status_code}{color_fin}") - registrar_falla(docentry, ruc, tipo_cpe, serie, numero, f"Error HTTP {status_code}") - return - - print(f"{prefijo_hijo}├── Consulta: {color_verde}✅ OK{color_fin}") - - # ─── PDF ─── - if tiene_pdf == "Y": - pdf_msg = f"{color_verde}✅ ya tenía{color_fin}" - else: + try: time.sleep(random.uniform(1.0, 2.0)) - ok_pdf = descargar_archivo(token, cid, "01", - datetime.now().strftime("%Y%m%d_%H%M%S"), - verbose=False) - if ok_pdf: - dl_url = f"{API_HOST}{BASE_PATH}/{cid}/01" - local_path = str(DOWNLOAD_DIR / f"{cid[:-2] if cid.endswith('-2') else cid}.pdf") - marcar_pdf_descargado(docentry, dl_url, local_path) - pdf_msg = f"{color_verde}📥 descargado{color_fin}" + data, status_code, json_path = consultar_cpe(token, ruc, tipo_cpe, serie, numero, descargar=False, verbose=False) + + if not data: + doc_log.error(f"Consulta: {color_rojo}Error {status_code}{color_fin}", _last=True) + registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, f"Error HTTP {status_code}") + registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry, + intento_num, status_code, "no_aplica", "no_aplica", "error", + f"Error HTTP {status_code}", None) + return + + doc_log.info(f"Consulta: {color_verde}✅ OK{color_fin}") + + # ─── PDF ─── + if tiene_pdf == "Y": + pdf_msg = f"{color_verde}✅ ya tenía{color_fin}" else: - pdf_msg = f"{color_amarillo}⚠️ error{color_fin}" - hubo_error = True + time.sleep(random.uniform(1.0, 2.0)) + ok_pdf = descargar_archivo(token, cid, "01", + datetime.now().strftime("%Y%m%d_%H%M%S"), + verbose=False) + if ok_pdf: + dl_url = f"{API_HOST}{BASE_PATH}/{cid}/01" + local_path = str(DOWNLOAD_DIR / f"{cid[:-2] if cid.endswith('-2') else cid}.pdf") + marcar_pdf_descargado(docentry, dl_url, local_path, obj_type) + pdf_msg = f"{color_verde}📥 descargado{color_fin}" + else: + pdf_msg = f"{color_amarillo}⚠️ error{color_fin}" + hubo_error = True - print(f"{prefijo_hijo}├── PDF: {pdf_msg}") + doc_log.info(f"PDF: {pdf_msg}") - # ─── XML ─── - if tiene_xml == "Y": - xml_msg = f"{color_verde}✅ ya tenía{color_fin}" - else: - time.sleep(random.uniform(1.0, 2.0)) - ok_xml = descargar_archivo(token, cid, "02", - datetime.now().strftime("%Y%m%d_%H%M%S"), - verbose=False) - if ok_xml: - dl_url = f"{API_HOST}{BASE_PATH}/{cid}/02" - local_path = str(DOWNLOAD_DIR / f"{cid[:-2] if cid.endswith('-2') else cid}.xml") - marcar_xml_descargado(docentry, dl_url, local_path) - xml_msg = f"{color_verde}📥 descargado{color_fin}" + # ─── XML ─── + if tiene_xml == "Y": + xml_msg = f"{color_verde}✅ ya tenía{color_fin}" else: - xml_msg = f"{color_amarillo}⚠️ error{color_fin}" - hubo_error = True + time.sleep(random.uniform(1.0, 2.0)) + ok_xml = descargar_archivo(token, cid, "02", + datetime.now().strftime("%Y%m%d_%H%M%S"), + verbose=False) + if ok_xml: + dl_url = f"{API_HOST}{BASE_PATH}/{cid}/02" + local_path = str(DOWNLOAD_DIR / f"{cid[:-2] if cid.endswith('-2') else cid}.xml") + marcar_xml_descargado(docentry, dl_url, local_path, obj_type) + xml_msg = f"{color_verde}📥 descargado{color_fin}" + else: + xml_msg = f"{color_amarillo}⚠️ error{color_fin}" + hubo_error = True - print(f"{prefijo_hijo}├── XML: {xml_msg}") - print(f"{prefijo_hijo}├── HANA: {ic(ok_pdf and ok_xml, '✅ actualizado', '⚠️ pendiente')}") + doc_log.info(f"XML: {xml_msg}") + doc_log.info(f"HANA: {ic(ok_pdf and ok_xml, '✅ actualizado', '⚠️ pendiente')}") - if ok_pdf and ok_xml: - if es_retry: - print(f"{prefijo_hijo}├── SQLite: {color_amarillo}🗑️ quitando registro{color_fin}") - eliminar_falla(docentry) - print(f"{prefijo_hijo}└── Resultado: {color_verde}✅ OK{color_fin}") - else: - fallos = [s for s, ok in [("PDF", ok_pdf), ("XML", ok_xml)] if not ok] - print(f"{prefijo_hijo}└── Resultado: {color_amarillo}⚠️ parcial{color_fin} ({', '.join(fallos)})") - if hubo_error: - registrar_falla(docentry, ruc, tipo_cpe, serie, numero, f"falló: {', '.join(fallos)}") + if ok_pdf and ok_xml: + if es_retry: + doc_log.info(f"SQLite: {color_amarillo}🗑️ quitando registro{color_fin}") + eliminar_falla(docentry) + doc_log.info(f"Resultado: {color_verde}✅ OK{color_fin}", _last=True) + detalle_historial = "OK" + else: + fallos = [s for s, ok in [("PDF", ok_pdf), ("XML", ok_xml)] if not ok] + doc_log.warning(f"Resultado: {color_amarillo}⚠️ parcial{color_fin} ({', '.join(fallos)})", _last=True) + detalle_historial = f"falló: {', '.join(fallos)}" + if hubo_error: + registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, detalle_historial) - break + pdf_resultado = "ya_tenia" if tiene_pdf == "Y" else ("descargado" if ok_pdf else "error") + xml_resultado = "ya_tenia" if tiene_xml == "Y" else ("descargado" if ok_xml else "error") + registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry, + intento_num, status_code, pdf_resultado, xml_resultado, + "ok" if (ok_pdf and ok_xml) else "parcial", + detalle_historial, json_path) - except TokenExpiredError: - if intento_token == 0: - print(f"{prefijo_hijo}├── Token expirado, renovando...") - nuevo_token = obtener_token() - if nuevo_token: - token = nuevo_token - print(f"{prefijo_hijo}├── Token renovado, reintentando...") - continue - print(f"{prefijo_hijo}└── Resultado: {color_rojo}Token no renovado{color_fin}") - else: - print(f"{prefijo_hijo}└── Resultado: {color_rojo}Token expiró{color_fin}") - registrar_falla(docentry, ruc, tipo_cpe, serie, numero, "Token expiró") - break + break - except Exception as e: - print(f"{prefijo_hijo}└── Resultado: {color_rojo}error{color_fin} ({e})") - registrar_falla(docentry, ruc, tipo_cpe, serie, numero, str(e)) - break + except TokenExpiredError: + if intento_token == 0: + doc_log.info("Token expirado, renovando...") + nuevo_token = obtener_token() + if nuevo_token: + token = nuevo_token + doc_log.info("Token renovado, reintentando...") + continue + doc_log.error(f"Resultado: {color_rojo}Token no renovado{color_fin}", _last=True) + else: + doc_log.error(f"Resultado: {color_rojo}Token expiró{color_fin}", _last=True) + registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, "Token expiró") + registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry, + intento_num, None, "no_aplica", "no_aplica", "error", "Token expiró", None) + break + + except Exception as e: + doc_log.error(f"Resultado: {color_rojo}error{color_fin} ({e})", _last=True) + registrar_falla(docentry, ruc, tipo_cpe, serie, numero, obj_type, str(e)) + registrar_intento(docentry, docnum, ruc, tipo_cpe, serie, numero, obj_type, es_retry, + intento_num, None, "no_aplica", "no_aplica", "error", str(e), None) + break # ======================================================================== for idx, doc in enumerate(docs_nuevos): _procesar(doc, idx, len(docs_nuevos), es_retry=False) - if docs_retry: - print() 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...") + logger.warning("Señal de parada recibida. Deteniendo...") _running = False # ============================================================================ @@ -291,30 +309,30 @@ def main(): signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) - log_path = setup_logger() + setup_logging() limpiar_logs_antiguos(dias=7) init_db() - print(f"📄 Log: {log_path}") + init_historial_db() + limpiar_historial_antiguo(dias=180) + logger.info(f"Log: {LOG_FILE}") - print("=" * 60) - print("🚀 DAEMON DESCARGA XML/PDF SUNAT") - print("=" * 60) - print(f"\n📅 Días activos: {DIAS_ACTIVOS} (0=lun, 6=dom)") - print(f"⏰ Horarios:") + logger.info("=" * 60) + logger.info("DAEMON DESCARGA XML/PDF SUNAT") + logger.info("=" * 60) + logger.info(f"Días activos: {DIAS_ACTIVOS} (0=lun, 6=dom)") + logger.info("Horarios:") for hi, mi, hf, mf, intervalo, modo in HORARIOS: - print(f" {hi:02d}:{mi:02d} - {hf:02d}:{mf:02d} | cada {intervalo}s | {modo}") - print() + logger.info(f"{hi:02d}:{mi:02d} - {hf:02d}:{mf:02d} | cada {intervalo}s | {modo}") token = None while _running: ahora = datetime.now() - rotar_log() if ahora.weekday() not in DIAS_ACTIVOS: espera = segundos_hasta_proximo_dia_activo(ahora) prox = ahora + timedelta(seconds=espera) - print(f"⏸️ Día no activo. Siguiente inicio: {prox.strftime('%a %Y-%m-%d %H:%M:%S')}") + logger.info(f"Día no activo. Siguiente inicio: {prox.strftime('%a %Y-%m-%d %H:%M:%S')}") for _ in range(int(espera)): if not _running: break @@ -325,7 +343,7 @@ def main(): if not intervalo_info: espera = segundos_hasta_proximo_inicio(ahora) prox = ahora + timedelta(seconds=espera) - print(f"⏸️ Fuera de horario. Siguiente inicio: {prox.strftime('%H:%M:%S')}") + logger.info(f"Fuera de horario. Siguiente inicio: {prox.strftime('%H:%M:%S')}") for _ in range(int(espera)): if not _running: break @@ -334,14 +352,14 @@ def main(): intervalo, modo = intervalo_info - print(f"\n{'#'*60}") - print(f"# 🔄 CICLO: {ahora.strftime('%Y-%m-%d %H:%M:%S')} | modo={modo} | intervalo={intervalo}s") - print(f"{'#'*60}") + logger.info("#" * 60) + logger.info(f"CICLO: {ahora.strftime('%Y-%m-%d %H:%M:%S')} | modo={modo} | intervalo={intervalo}s") + logger.info("#" * 60) if not token: token = obtener_token() if not token: - print(" ❌ No se pudo obtener token. Esperando 60s...") + logger.error("No se pudo obtener token. Esperando 60s...") time.sleep(60) continue @@ -350,7 +368,7 @@ def main(): try: sincronizar_a_smb(str(DOWNLOAD_DIR)) except Exception as e: - print(f" ⚠️ Error en sincronización SMB: {e}") + logger.warning(f"Error en sincronización SMB: {e}") # Verificar si el token sigue válido para el próximo ciclo from consulta_cpe import TOKEN_FILE as tf @@ -360,21 +378,21 @@ def main(): expires = data.get("expires_at", 0) remaining = expires - time.time() if remaining < 300: - print(f"\n⚠️ Token por expirar ({int(remaining)}s). Se renovará en próximo ciclo.") + logger.warning(f"Token por expirar ({int(remaining)}s). Se renovará en próximo ciclo.") token = None except: pass - print(f"\n⏳ Esperando {intervalo}s hasta próximo ciclo...") + logger.info(f"Esperando {intervalo}s hasta próximo ciclo...") for _ in range(intervalo): if not _running: break time.sleep(1) cerrar_conexion() - print("\n👋 Daemon detenido.") + logger.info("Daemon detenido.") # ============================================================================ if __name__ == "__main__": main() -# ============================================================================ \ No newline at end of file +# ============================================================================ diff --git a/reporte_fallas.py b/reporte_fallas.py deleted file mode 100644 index 0297505..0000000 --- a/reporte_fallas.py +++ /dev/null @@ -1,168 +0,0 @@ -import sqlite3 -from pathlib import Path -from datetime import datetime -from collections import Counter - -# ============================================================================ -try: - import matplotlib - matplotlib.use("Agg") - import matplotlib.pyplot as plt - import matplotlib.ticker as mticker - HAS_MPL = True -except ImportError: - HAS_MPL = False - -DB_PATH = Path(__file__).parent / "fallas_descarga.db" -REPORTES_DIR = Path(__file__).parent / "reportes" -REPORTES_DIR.mkdir(exist_ok=True) - -COLORS = { - "Error HTTP 500": "#e74c3c", - "falló: PDF": "#e67e22", - "falló: XML": "#3498db", - "falló: PDF, XML": "#9b59b6", -} -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") - print("=" * 60) - - total = _query("SELECT COUNT(*) AS c FROM fallas")[0]["c"] - print(f"\n📄 Total fallas registradas: {total}") - - if total == 0: - print("\n✅ No hay fallas registradas.") - return - - print("\n── Distribución por tipo de error ──") - for r in _query("SELECT COUNT(*) AS c, error FROM fallas GROUP BY error ORDER BY c DESC"): - pct = r["c"] / total * 100 - print(f" {r['error']:25s} {r['c']:>5d} ({pct:5.1f}%)") - - print("\n── Top 10 RUC con más fallas ──") - for i, r in enumerate(_query("SELECT COUNT(*) AS c, ruc FROM fallas GROUP BY ruc ORDER BY c DESC LIMIT 10"), 1): - print(f" {i:2d}. {r['ruc']} ({r['c']} fallas)") - - print("\n── Distribución por intentos ──") - for r in _query("SELECT intentos, COUNT(*) AS c FROM fallas GROUP BY intentos ORDER BY intentos"): - print(f" {r['intentos']:2d} intento(s) → {r['c']:>5d} documento(s)") - - print("\n── Fallas por día ──") - 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: - return None - - labels = [r["error"] for r in data] - values = [r["c"] for r in data] - colors = [_color_error(l) for l in labels] - - fig, ax = plt.subplots(figsize=(10, 5)) - bars = ax.barh(labels, values, color=colors, edgecolor="white", height=0.6) - for b, v in zip(bars, values): - ax.text(b.get_width() + max(values) * 0.01, b.get_y() + b.get_height() / 2, - str(v), va="center", fontsize=11, fontweight="bold") - ax.set_xlabel("Cantidad de fallas") - ax.set_title("Distribución de fallas por tipo de error", fontsize=14, fontweight="bold") - ax.margins(x=0.15) - ax.spines[["top", "right"]].set_visible(False) - fig.tight_layout() - path = REPORTES_DIR / "barras_errores.png" - fig.savefig(path, dpi=150) - 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: - return None - - data.reverse() - labels = [r["ruc"] for r in data] - values = [r["c"] for r in data] - - fig, ax = plt.subplots(figsize=(10, 5)) - bars = ax.barh(labels, values, color="#e67e22", edgecolor="white", height=0.6) - for b, v in zip(bars, values): - ax.text(b.get_width() + 0.3, b.get_y() + b.get_height() / 2, - str(v), va="center", fontsize=10, fontweight="bold") - ax.set_xlabel("Cantidad de fallas") - ax.set_title("Top 10 RUC con más fallas", fontsize=14, fontweight="bold") - ax.margins(x=0.15) - ax.spines[["top", "right"]].set_visible(False) - fig.tight_layout() - path = REPORTES_DIR / "barras_ruc.png" - fig.savefig(path, dpi=150) - 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: - return None - - labels = [r["error"] for r in data] - values = [r["c"] for r in data] - colors = [_color_error(l) for l in labels] - - fig, ax = plt.subplots(figsize=(7, 7)) - wedges, texts, autotexts = ax.pie( - values, labels=labels, autopct="%1.1f%%", - colors=colors, startangle=90, textprops={"fontsize": 11} - ) - for at in autotexts: - at.set_fontweight("bold") - ax.set_title("Proporción de fallas por tipo", fontsize=14, fontweight="bold") - fig.tight_layout() - path = REPORTES_DIR / "torta_errores.png" - fig.savefig(path, dpi=150) - plt.close(fig) - return path - -# ============================================================================ -def main(): - reporte_texto() - - if not HAS_MPL: - print("\n⚠️ matplotlib no instalado. Ejecuta: pip3 install matplotlib") - return - - paths = [] - for name, fn in [("barras_errores", grafico_barras_errores), - ("barras_ruc", grafico_barras_ruc), - ("torta_errores", grafico_torta)]: - p = fn() - if p: - paths.append(p) - - if paths: - print(f"\n📁 Gráficos guardados en: {REPORTES_DIR}") - for p in paths: - 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/requirements.txt b/requirements.txt index 22baac4..c0eef80 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,22 +1,13 @@ -certifi==2026.6.17 -charset-normalizer==3.4.9 -contourpy==1.3.3 -cycler==0.12.1 -fonttools==4.63.0 -greenlet==3.5.3 -hdbcli==2.29.25 -idna==3.18 -kiwisolver==1.5.0 -matplotlib==3.11.0 -numpy==2.5.1 -packaging==26.2 -pillow==12.3.0 -playwright==1.61.0 -pyee==13.0.1 -pyparsing==3.3.2 -python-dateutil==2.9.0.post0 -python-dotenv==1.2.2 -requests==2.34.2 -six==1.17.0 -typing_extensions==4.16.0 -urllib3==2.7.0 +certifi +charset-normalizer +greenlet +hdbcli +idna +packaging +playwright==1.49.1 +pyee +python-dotenv +requests +six +typing_extensions +urllib3 diff --git a/scheduler.json b/scheduler.json index cb93168..3b1d794 100644 --- a/scheduler.json +++ b/scheduler.json @@ -1,9 +1,9 @@ { "HORARIOS": [ - [6, 30, 13, 15, 10, "rapido"], + [6, 30, 13, 15, 300, "rapido"], [13, 15, 14, 0, 1800, "full"], - [14, 0, 17, 30, 10, "rapido"], + [14, 0, 17, 30, 300, "rapido"], [17, 30, 21, 30, 1800, "full"] ], - "DIAS_ACTIVOS": [0, 1, 2, 3, 4] + "DIAS_ACTIVOS": [0, 1, 2, 3, 4, 5] } diff --git a/smb_manager.py b/smb_manager.py index 252b4cb..3dc7474 100644 --- a/smb_manager.py +++ b/smb_manager.py @@ -4,9 +4,11 @@ import platform import subprocess from pathlib import Path from dotenv import load_dotenv +from logger_manager import get_logger # ============================================================================ load_dotenv() +logger = get_logger(__name__) SMB_SERVER = os.getenv("SMB_SERVER", "").strip() SMB_SHARE = os.getenv("SMB_SHARE", "").strip() @@ -50,7 +52,7 @@ def esta_montado(mount_point): def montar_smb(): mount_point = get_mount_point() if not mount_point: - print(" ⚠️ No hay punto de montaje configurado para este SO.") + logger.warning("No hay punto de montaje configurado para este SO.") return False if esta_montado(mount_point): @@ -82,20 +84,20 @@ def montar_smb(): ] subprocess.run(cmd, check=True, capture_output=True) - print(f" ✅ SMB montado en: {mount_point}") + logger.info(f"SMB montado en: {mount_point}") return True except Exception as e: - print(f" ⚠️ Error montando SMB: {e}") + logger.error(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}") + logger.warning(f"La carpeta local no existe: {ruta_local}") return False if not montar_smb(): - print(" ❌ No se pudo sincronizar: SMB no montado.") + logger.error("No se pudo sincronizar: SMB no montado.") return False mount_point = get_mount_point() @@ -112,12 +114,12 @@ def sincronizar_a_smb(ruta_local): if not os.path.exists(ruta_destino): os.makedirs(ruta_destino) - print(f" 📁 Carpeta creada: {ruta_destino}") + logger.info(f"Carpeta creada: {ruta_destino}") source = os.path.abspath(ruta_local) destination = os.path.abspath(ruta_destino) - print(f" 🚀 Sincronizando a {current_os.upper()}: {destination}") + logger.info(f"Sincronizando a {current_os.upper()}: {destination}") try: if current_os in ("mac", "linux"): @@ -126,12 +128,12 @@ def sincronizar_a_smb(ruta_local): f"-exec rsync -av --remove-source-files {{}} {destination}/ \\;" ) subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True) - print(f" ✅ Archivos copiados a NAS") + logger.info("Archivos copiados a NAS") subprocess.run( f"find {source} -type f \\( -name '*.json' -o -name '*.zip' \\) -delete", shell=True, check=True, capture_output=True, text=True, ) - print(f" 🧹 Limpieza de temporales completada") + logger.info("Limpieza de temporales completada") elif current_os == "windows": import shutil for archivo in os.listdir(source): @@ -143,14 +145,14 @@ def sincronizar_a_smb(ruta_local): if os.path.exists(dst_path): os.remove(dst_path) shutil.move(src_path, dst_path) - print(f" ✅ Archivos copiados a NAS") + logger.info("Archivos copiados a NAS") for archivo in os.listdir(source): if archivo.lower().endswith((".json", ".zip")): os.remove(os.path.join(source, archivo)) - print(f" 🧹 Limpieza de temporales completada") + logger.info("Limpieza de temporales completada") return True except Exception as e: - print(f" ❌ Error en sincronización: {e}") + logger.error(f"Error en sincronización: {e}") return False # ============================================================================ \ No newline at end of file