commit 2be35d7e55bb7093ea617da1494760daed849de8 Author: George Date: Fri Jul 17 12:21:31 2026 -0500 primer commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f859bc5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,34 @@ +# Entorno virtual +sunat-env/ +venv/ +.env + +# Credenciales y configuración sensible +token.json + +# Descargas +descargas/ + +# Screenshots (debug) +screenshots/ + +# Logs +logs/ + +# Base de datos SQLite +fallas_descarga.db + +# Reportes generados +reportes/ + +# Python cache +__pycache__/ +*.pyc +*.pyo +*.egg-info/ + +# macOS +.DS_Store + +# VS Code (opcional, comentar si quieres mantener config) +.vscode/ diff --git a/consulta_cpe.py b/consulta_cpe.py new file mode 100644 index 0000000..d2cb16b --- /dev/null +++ b/consulta_cpe.py @@ -0,0 +1,244 @@ +import os +import json +import time +import argparse +import base64 +import zipfile +import io +import requests +import urllib3 +from pathlib import Path +from datetime import datetime + +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +class TokenExpiredError(Exception): + pass + +TOKEN_FILE = Path(__file__).parent / "token.json" +DOWNLOAD_DIR = Path(__file__).parent / "descargas" +DOWNLOAD_DIR.mkdir(exist_ok=True) + +API_HOST = "https://api-cpe.sunat.gob.pe" +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", +} + +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") + return None + + data = json.loads(TOKEN_FILE.read_text()) + token = data.get("token") + expires = data.get("expires_at", 0) + now = time.time() + + if expires and now > expires: + print(f"⚠️ 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") + 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}", + "Accept": "application/json", + "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", + "Origin": "https://e-factura.sunat.gob.pe", + "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}" + + if verbose: + print(f"\n🌐 Consultando: {ruc_emisor}-{tipo}-{serie}-{numero}") + print(f" {url}") + + session = requests.Session() + session.verify = False + session.headers.update(get_headers(token)) + + r = None + for intento in range(3): + try: + r = session.get(url, timeout=120) + if verbose: + print(f"📡 {r.status_code}") + break + except Exception as e: + if verbose: + print(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 + + if r is None: + return None, 0, 0 + + if r.status_code == 200: + data = r.json() + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + safe = cid.replace("/", "_") + + 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}") + + if descargar: + for tipo_dl in ["02", "01", "03"]: + descargar_archivo(token, cid, tipo_dl, ts, verbose=verbose) + + return data, 200 + + elif r.status_code == 401: + if verbose: + print(" ❌ 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)") + else: + if verbose: + print(f" ❌ Error {r.status_code}: {r.text[:300]}") + + 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}" + + if verbose: + print(f"\n📥 Descargando {label}...") + print(f" {url}") + + session = requests.Session() + session.verify = False + session.headers.update(get_headers(token)) + + r = None + for intento in range(3): + try: + r = session.get(url, timeout=120) + if verbose: + print(f"📡 {r.status_code}") + break + except Exception as e: + if verbose: + print(f" ⚠️ Intento {intento+1} falló: {e}") + if intento < 2: + time.sleep(3) + else: + if verbose: + print(f" ❌ Error después de 3 intentos") + return False + + if r is None: + return False + + if r.status_code == 200: + data = r.json() + val_archivo = data.get("valArchivo", "") + + if not val_archivo: + if verbose: + print(f" ⚠️ Respuesta sin valArchivo") + return False + + raw = base64.b64decode(val_archivo) + + file_base = cid[:-2] if cid.endswith("-2") else cid + zip_name = f"{file_base}-{label}.zip" + zip_path = DOWNLOAD_DIR / zip_name + zip_path.write_bytes(raw) + if verbose: + print(f" 💾 ZIP: {zip_path} ({len(raw)} bytes)") + + try: + with zipfile.ZipFile(io.BytesIO(raw)) as zf: + for member in zf.namelist(): + content = zf.read(member) + member_lower = member.lower() + + if label == "PDF" and member_lower.endswith(".pdf"): + out_name = f"{file_base}.pdf" + elif label == "XML": + if not member_lower.endswith(".xml"): + continue + if member.startswith("R-") or member.startswith("r-"): + out_name = f"CDR-{file_base}.xml" + else: + out_name = f"{file_base}.xml" + elif label == "CDR" and member_lower.endswith(".xml"): + out_name = f"CDR-{file_base}.xml" + else: + continue + + out_path = DOWNLOAD_DIR / out_name + out_path.write_bytes(content) + if verbose: + print(f" ✅ {out_name} ({len(content)} bytes)") + except Exception as e: + if verbose: + print(f" ⚠️ Error extrayendo ZIP: {e}") + + return True + + elif r.status_code == 401: + if verbose: + print(" ❌ Token inválido (401)") + raise TokenExpiredError("Token expirado o inválido (401)") + elif r.status_code == 400: + if verbose: + print(f" ⚠️ No disponible (400)") + else: + if verbose: + print(f" ❌ Error {r.status_code}") + + return False + + +def main(): + print("=" * 60) + print("🔍 CONSULTA CPE - SUNAT API") + print("=" * 60) + + parser = argparse.ArgumentParser(description="Consultar CPE en SUNAT") + parser.add_argument("--ruc", required=True, help="RUC del emisor") + parser.add_argument("--tipo", required=True, help="Tipo (01=Factura, 03=Boleta, 07=NC, 08=ND)") + parser.add_argument("--serie", required=True, help="Serie (ej: F007)") + parser.add_argument("--numero", required=True, help="Número del comprobante") + parser.add_argument("--no-download", action="store_true", help="No descargar archivos") + + args = parser.parse_args() + token = load_token() + if not token: + return + + consultar_cpe(token, args.ruc, args.tipo, args.serie, args.numero, descargar=not args.no_download) + + +if __name__ == "__main__": + main() diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md new file mode 100644 index 0000000..7f912c1 --- /dev/null +++ b/docs/DEPLOYMENT.md @@ -0,0 +1,86 @@ +# Despliegue + +## systemd (Linux) + +### Requisitos adicionales +- **Google Chrome** instalado (para `get_token_headless.py`) +- **rsync** y **cifs-utils** (para montar SMB y sincronizar) + +### Servicio + +```ini +[Unit] +Description=SUNAT CPE Downloader Daemon +After=network.target + +[Service] +Type=simple +User=tu_usuario +WorkingDirectory=/ruta/a/descarga_xml_sunat +ExecStart=/usr/bin/python3 main.py +Restart=always +RestartSec=30 + +[Install] +WantedBy=multi-user.target +``` + +```bash +sudo cp sunat-cpe.service /etc/systemd/system/ +sudo systemctl enable sunat-cpe +sudo systemctl start sunat-cpe +``` + +## NSSM (Windows) + +### Requisitos +- **Google Chrome** instalado (cualquier versión) — necesario para `get_token_headless.py` +- Python 3.12+ con dependencias instaladas +- **Unidad de red mapeable** (SMB) para sincronizar descargas + +### Instalación como servicio + +```cmd +:: Instalar servicio +nssm install SUNAT-CPE "C:\Python312\python.exe" "C:\ruta\descarga_xml_sunat\main.py" + +:: Configurar directorio de trabajo +nssm set SUNAT-CPE AppDirectory "C:\ruta\descarga_xml_sunat" + +:: Configurar reinicio automático +nssm set SUNAT-CPE AppExit Default Exit +nssm set SUNAT-CPE AppThrottle 30000 + +:: Iniciar servicio +nssm set SUNAT-CPE Start SERVICE_AUTO_START +nssm start SUNAT-CPE +``` + +### Logs +NSSM redirige stdout/stderr a `C:\ruta\descarga_xml_sunat\logs\` por defecto (el logger del daemon también escribe allí). + +### Renovación de token +Como servicio, el daemon usa `get_token_headless()` (headless, Chrome real) para renovar el token automáticamente sin intervención. Si Chrome 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 +``` + +### Actualización +```cmd +nssm stop SUNAT-CPE +:: actualizar archivos +nssm start SUNAT-CPE +``` + +## Docker (opcional) + +```dockerfile +FROM python:3.12-slim +RUN pip install -r requirements.txt && playwright install chromium +WORKDIR /app +COPY . . +CMD ["python", "main.py"] +``` + +Asegúrate de que `requirements.txt` incluya: `playwright`, `python-dotenv`, `requests`, `hdbcli`, `matplotlib`. diff --git a/docs/FLOWCHART.md b/docs/FLOWCHART.md new file mode 100644 index 0000000..61a166e --- /dev/null +++ b/docs/FLOWCHART.md @@ -0,0 +1,62 @@ +```mermaid +flowchart TD + subgraph Daemon["🚀 Daemon (main.py)"] + A["Inicio"] --> B{"Día activo y
en horario?"} + B -->|No| C["Esperar"] + C --> B + B -->|Sí| D["Token válido?"] + D -->|No| E["get_token_headless()
channel=chrome headless=True
→ get_token.py headless=False"] + E --> F["token.json"] + D -->|Sí| F + end + + subgraph HANA["📊 Obtener pendientes"] + F --> G["Consultar HANA
obtener_pendientes()"] + G --> H["542 documentos
en OPCH"] + H --> I["Filtrar contra
fallas_descarga.db"] + I --> J["🆕 Nuevos
(sin falla previa)"] + I --> K["🔄 Reintentos
(fallas pendientes)"] + end + + subgraph Nuevos["📄 Procesar nuevos"] + L["Por cada nuevo:"] --> M["consultar_cpe()
GET api-cpe.sunat.gob.pe
/v1/.../{ruc}-{tipo}-{serie}-{num}-2"] + M --> N{"Status?"} + N -->|200| O["descargar_archivo()
XML (02) / PDF (01) / CDR (03)"] + O --> P["ZIP → extraer en
descargas/"] + P --> Q["marcar_pdf/xml_descargado()
UPDATE HANA OPCH"] + Q --> R["✅ OK"] + N -->|401| S["⚠️ TokenExpiredError"] + S --> T["Renovar token"] + T --> M + N -->|500/404/otros| U["registrar_falla()
→ SQLite con backoff"] + end + + subgraph Reintentos["🔄 Procesar reintentos"] + V["Por cada reintento
(backoff cumplido):"] --> W["consultar_cpe()"] + W --> X{"Status?"} + X -->|200| Y["descargar_archivo()"] + Y --> Z["marcar_pdf/xml_descargado()"] + Z --> AA["eliminar_falla()
DELETE FROM fallas"] + AA --> AB["✅ OK + 🗑️ quitado de SQLite"] + X -->|401| AC["Renovar token → reintentar"] + X -->|500/404/otros| AD["actualizar_falla()
incrementar intento
→ próximo backoff"] + end + + AD --> AF["sincronizar_a_smb()
montar SMB
rsync PDF/XML → NAS
limpiar JSON/ZIP"] + + AF --> AE["Esperar intervalo
→ próximo ciclo"] + R --> AF + U --> AD + AB --> AF + + style A fill:#1a1a2e,stroke:#e94560,color:#fff + style E fill:#16213e,stroke:#0f3460,color:#fff + style G fill:#0f3460,stroke:#e94560,color:#fff + style M fill:#533483,stroke:#e94560,color:#fff + style O fill:#1a1a2e,stroke:#e94560,color:#fff + style P fill:#16213e,stroke:#0f3460,color:#fff + style Q fill:#0f3460,stroke:#e94560,color:#fff + style U fill:#3d0000,stroke:#ff4444,color:#fff + style AA fill:#003300,stroke:#44ff44,color:#fff + style AD fill:#3d0000,stroke:#ff4444,color:#fff +``` diff --git a/docs/MODULES.md b/docs/MODULES.md new file mode 100644 index 0000000..f28e117 --- /dev/null +++ b/docs/MODULES.md @@ -0,0 +1,30 @@ +# Módulos + +| Archivo | Función | +|---|---| +| `main.py` | Daemon orquestador. Bucle infinito con horarios configurables, obtiene token, consulta HANA, procesa pendientes (nuevos + reintentos). | +| `get_token_headless.py` | Login headless via Chrome real (`channel="chrome"`). 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). | +| `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`. | +| `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) +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: + - `consultar_cpe()` → si 401, renueva token y reintenta + - Si falla (500/404/etc) → `registrar_falla()` con backoff progresivo + - Si OK → `descargar_archivo()` (XML + PDF + CDR) + - `marcar_pdf_descargado()` / `marcar_xml_descargado()` actualizan HANA + - Si era reintento exitoso → `eliminar_falla()` +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 diff --git a/docs/SETUP.md b/docs/SETUP.md new file mode 100644 index 0000000..2be7c48 --- /dev/null +++ b/docs/SETUP.md @@ -0,0 +1,70 @@ +# Configuración + +## Requisitos + +- Python 3.10+ +- Playwright (chromium) +- Acceso a HANA (hdbcli) +- **Google Chrome** (para `get_token_headless.py`) +- **rsync** (Mac/Linux) o **net use** (Windows) para SMB + +## Instalación + +```bash +# Crear y activar entorno virtual (recomendado) +python3 -m venv sunat-env +source sunat-env/bin/activate + +# Instalar dependencias +pip install -r requirements.txt +playwright install chromium +``` + +## Variables de Entorno (`.env`) + +```ini +# SUNAT +SUNAT_RUC=20503662869 +SUNAT_USER=TAGUMED3 +SUNAT_PASSWORD=xxxxx + +# HANA +HANA_HOST=192.168.1.175 +HANA_PORT=30015 +HANA_USER=SYSTEM +HANA_PASSWORD=SAPB1Admin.25 +HANA_SCHEMA=SBO_TAGUMEDICA_21 + +# SMB / NAS +SMB_SERVER=192.168.1.175 +SMB_SHARE=FileSAP +SMB_USER=tu_usuario +SMB_PASSWORD=tu_contraseña +SMB_MOUNT_MAC=/Volumes/FileSAP +SMB_MOUNT_LINUX=/mnt/filesap +SMB_MOUNT_WINDOWS=S: +SMB_SUBFOLDER=COMPRAS/TAGUMEDICA/ADJUNTOS_CORREOS +``` + +## Ejecución + +```bash +# Activar entorno virtual +source sunat-env/bin/activate + +# Obtener token manualmente (con navegador visible) +python get_token.py + +# Obtener token headless (sin ventana, requiere Chrome real) +python get_token_headless.py + +# Daemon completo (automático) +python main.py + +# Reporte de fallas (resumen texto + gráficos PNG) +python reporte_fallas.py +``` + +## 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. diff --git a/gestor_fallas.py b/gestor_fallas.py new file mode 100644 index 0000000..bab1b95 --- /dev/null +++ b/gestor_fallas.py @@ -0,0 +1,86 @@ +import sqlite3 +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(""" + CREATE TABLE IF NOT EXISTS fallas ( + docentry INTEGER PRIMARY KEY, + ruc TEXT NOT NULL, + tipo TEXT NOT NULL, + serie TEXT NOT NULL, + numero TEXT NOT NULL, + error TEXT NOT NULL, + intentos INTEGER NOT NULL DEFAULT 1, + proximo_intento TEXT, + fecha_ultimo TEXT NOT NULL + ) + """) + +def _calcular_proximo_intento(intentos): + if intentos >= 20: + return None + if intentos == 1: + delta = timedelta(minutes=0) + elif intentos == 2: + delta = timedelta(minutes=5) + elif intentos == 3: + delta = timedelta(minutes=15) + elif intentos == 4: + delta = timedelta(minutes=30) + elif intentos == 5: + delta = timedelta(hours=1) + elif intentos <= 10: + delta = timedelta(hours=2) + elif intentos <= 15: + delta = timedelta(hours=6) + 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: + with _conn() as conn: + row = conn.execute( + "SELECT intentos FROM fallas WHERE docentry = ?", (docentry,) + ).fetchone() + intentos = (row[0] + 1) if row else 1 + proximo = _calcular_proximo_intento(intentos) + 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)) + +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: + conn.row_factory = sqlite3.Row + cursor = conn.execute(""" + SELECT * FROM fallas + WHERE (proximo_intento IS NULL OR proximo_intento <= ?) + AND intentos < 20 + 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] diff --git a/get_token.py b/get_token.py new file mode 100644 index 0000000..e9b2864 --- /dev/null +++ b/get_token.py @@ -0,0 +1,449 @@ +import os +import json +import time +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" +MENU_URL = "https://e-menu.sunat.gob.pe/cl-ti-itmenucabina/MenuInternet.htm" +TOKEN_FILE = Path(__file__).parent / "token.json" +SCREENSHOT_DIR = Path(__file__).parent / "screenshots" +SCREENSHOT_DIR.mkdir(exist_ok=True) + +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: + browser = p.chromium.launch( + headless=headless, + args=["--disable-blink-features=AutomationControlled"], + ) + context = browser.new_context( + viewport={"width": 1280, "height": 800}, + locale="es-PE", + timezone_id="America/Lima", + ) + + page = context.new_page() + page.set_default_timeout(timeout) + + # ─── 1. Ir a SOL ─────────────────────────────────── + print("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...") + 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}") + + # ─── 3. Login ───────────────────────────────────── + print("3. Verificando si hay formulario de login...") + popup.wait_for_load_state("networkidle") + + print(f" Título popup: {popup.title()}") + has_inputs = len(popup.locator("input[type='password']").all()) > 0 + print(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...") + fill_login_form(popup) + save_screenshot(popup, "03_form_rellenado") + + print(" 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}") + else: + print(" No se detectó login. Imprimiendo HTML de la página...") + try: + html = popup.content() + print(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]}...") + except Exception as e: + print(f" Error leyendo HTML: {e}") + + # ─── 4. Token via consulta directa ──────────────── + print("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!") + 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:") + for i, frm in enumerate(popup.frames): + print(f" [{i}] name='{frm.name}' url='{frm.url[:100]}'") + + if not headless: + input("\n⏸️ Presiona Enter para cerrar el navegador...") + browser.close() + return None + + +def fill_login_form(page): + page.wait_for_timeout(1000) + + inputs = page.locator("input").all() + print(f" Inputs encontrados: {len(inputs)}") + for inp in inputs: + try: + name = inp.get_attribute("name") or "" + id_ = inp.get_attribute("id") or "" + 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}") + except: + pass + + print(f" Rellenando RUC: {RUC}") + page.fill("#txtRuc", RUC) + + print(f" Rellenando Usuario: {USER}") + page.fill("#txtUsuario", USER) + + print(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!") + save_screenshot(page, "03_captcha_detectado") + input("\n⏸️ Resuelve el captcha manualmente y presiona Enter...") + 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) + if token: + print(f" ✅ Token encontrado en menú (href)") + return token + + print(" Navegando árbol de menú SOL...") + return navigate_sol_menu(login_page) + + +def navigate_sol_menu(page): + page.wait_for_timeout(2000) + + token = check_for_token(page) + if token: + return token + + # Navegar el árbol: nivel1 → nivel2 → nivel3 → nivel4 + # IDs conocidos del menú: + menu_ruta = [ + ("nivel1_11", "Comprobantes de pago"), + ("nivel2_11_38", "Comprobantes de Pago"), + ("nivel3_11_38_1", "Consulta de Comprobantes de Pago"), + ("nivel4_11_38_1_1_1", "Nueva Consulta de comprobantes de pago"), + ] + + for item_id, item_text in menu_ruta: + token = click_menu_item_by_id(page, item_id, item_text) + if token: + return token + page.wait_for_timeout(2000) + save_screenshot(page, f"menu_{item_id}") + + if "nivel4" in item_id: + save_screenshot(page, f"menu_final") + page.wait_for_timeout(5000) + for _ in range(15): + token = check_for_token(page) + if token: + return token + page.wait_for_timeout(2000) + + # Debug: imprimir info de todos los frames + print(" Frames activos:") + for frm in page.frames: + print(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)}") + + 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}") + + import re + + tokens = re.findall(r'token[=:][^&\s"\']{50,}', html) + if tokens: + print(f" Tokens encontrados (token=): {len(tokens)}") + for t in tokens[:3]: + print(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)}") + for j in jwt_matches[:3]: + print(f" {j[:80]}...") + t = extract_token_from_url(j) + if t: + return t + + ls = frm.evaluate("JSON.stringify(localStorage)") + print(f" localStorage: {ls[:200]}") + ss = frm.evaluate("JSON.stringify(sessionStorage)") + print(f" sessionStorage: {ss[:200]}") + except Exception as e: + print(f" Error: {e}") + + return None + + +def click_menu_item_by_id(page, element_id, expected_text): + print(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}") + + # Intentar click en el span interior + span = frm.locator(f"#{element_id} span.spanNivelDescripcion") + if span.count() > 0: + print(f" clickeando span...") + span.first.click() + else: + print(f" clickeando li directamente...") + el.click() + + page.wait_for_timeout(1500) + token = check_for_token(page) + if token: + return token + return None + except Exception as e: + pass + + # Fallback: buscar por texto + print(f" 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...") + el.click() + page.wait_for_timeout(1500) + return None + except: + pass + + 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: + t = extract_token_from_url(url) + if t: + print(f" ✅ Token encontrado en URL") + return t + + import re + + for frm in page.frames: + try: + frm_url = frm.url + 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]}...)") + + # 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") + return ss_token + except: + pass + + # Intentar localStorage + try: + ls_token = frm.evaluate("localStorage.getItem('SUNAT.token')") + if ls_token and len(ls_token) > 50: + print(f" ✅ Token encontrado en localStorage") + return ls_token + except: + pass + + # Buscar en todo el sessionStorage/localStorage + try: + all_ss = frm.evaluate("JSON.stringify(sessionStorage)") + if "token" in all_ss.lower(): + import json + 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}'") + return value + except: + pass + + # Buscar en el HTML + content = frm.content() + matches = re.findall(r'token=([a-zA-Z0-9_\-\.]+)', content) + for m in matches: + if len(m) > 50: + print(f" ✅ 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") + return m + + # Buscar JWT en todo el HTML + jwt_matches = re.findall(r'eyJ[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+\.[a-zA-Z0-9_\-]+', content) + for j in jwt_matches: + t = extract_token_from_url(j) + if t: + print(f" ✅ JWT encontrado en HTML") + return t + + except Exception as e: + print(f" Error: {e}") + + return None + + +def extract_token_from_menu(page): + print(" Analizando estructura del menú SOL...") + + for frm in page.frames: + try: + result = frm.evaluate("""() => { + const items = []; + const all = document.querySelectorAll('*'); + for (const el of all) { + const text = (el.textContent || '').trim().replace(/\\u00a0/g, ' '); + if (!text) continue; + // Encontrar items del menú (Comprobantes, consulta CPE, etc.) + if (text.includes('Comprobantes') || text.includes('consulta') || text.includes('Nueva Consulta')) { + if (text.length < 200) { + const rect = el.getBoundingClientRect(); + const style = window.getComputedStyle(el); + items.push({ + tag: el.tagName, + text: text.substring(0, 100), + visible: rect.width > 0 && rect.height > 0, + display: style.display, + href: el.href || el.getAttribute('href') || '', + onclick: (el.getAttribute('onclick') || '').substring(0, 200), + id: el.id || '', + cls: el.className || '' + }); + } + } + } + return items; + }""") + + 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]}") + if r.get('href'): + print(f" href: {r['href'][:120]}") + if r.get('onclick'): + print(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}") + + return None + + +def extract_token_from_url(url): + if "token=" in url: + token = url.split("token=")[-1] + token = token.split("&")[0] + token = token.split("#")[0] + if len(token) > 50: + return token + return None + + +def save_token(token): + try: + payload_b64 = token.split(".")[1] + padding = 4 - len(payload_b64) % 4 + if padding != 4: + payload_b64 += "=" * padding + import base64 + payload = json.loads(base64.b64decode(payload_b64)) + + 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", ""), + } + except: + 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']))}") + + +if __name__ == "__main__": + print("=" * 60) + print("🔐 OBTENCIÓN DE TOKEN SUNAT") + print("=" * 60) + + token = get_token(headless=False) + if token: + print(f"\n✅ Token obtenido exitosamente!") + print(f" Token (primeros 50 chars): {token[:50]}...") + else: + print(f"\n❌ No se pudo obtener el token.") + print(" Revisa las capturas en la carpeta screenshots/") diff --git a/get_token_headless.py b/get_token_headless.py new file mode 100644 index 0000000..4203e4c --- /dev/null +++ b/get_token_headless.py @@ -0,0 +1,99 @@ +import os +import time +import shutil +import platform +from pathlib import Path +from playwright.sync_api import sync_playwright + +from get_token import ( + MENU_URL, TOKEN_FILE, SCREENSHOT_DIR, + 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": + return bool(shutil.which("chrome")) or any( + os.path.exists(p) for p in [ + "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe", + "C:\\Program Files (x86)\\Google\\Chrome\\Application\\chrome.exe", + ] + ) + elif sistema == "darwin": + return os.path.exists("/Applications/Google Chrome.app") + elif sistema == "linux": + 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( + channel="chrome", + headless=True, + args=[ + "--disable-blink-features=AutomationControlled", + "--disable-popup-blocking", + "--no-sandbox", + "--window-size=1280,800", + ], + ) + context = browser.new_context( + viewport={"width": 1280, "height": 800}, + locale="es-PE", + timezone_id="America/Lima", + user_agent=USER_AGENT, + ) + page = context.new_page() + page.set_default_timeout(timeout) + + print("1. Sesión desde SOL...") + page.goto("https://www.sunat.gob.pe/sol.html", wait_until="domcontentloaded") + + print("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...") + fill_login_form(page) + page.keyboard.press("Enter") + try: + page.wait_for_load_state("networkidle", timeout=30000) + except Exception: + pass + page.wait_for_timeout(3000) + else: + print("3. Sin login (sesión activa)") + + print("4. Buscando token en menú...") + page.wait_for_timeout(2000) + token = check_for_token(page) + if not token: + token = navigate_sol_menu(page) + if not token: + token = extract_token_from_menu(page) + + context.close() + browser.close() + + if token: + print(" ✅ Token obtenido (headless)") + save_token(token) + return token + + print(" ❌ No se encontró token") + return None + + +if __name__ == "__main__": + get_token_headless() diff --git a/hana_manager.py b/hana_manager.py new file mode 100644 index 0000000..ab8fd17 --- /dev/null +++ b/hana_manager.py @@ -0,0 +1,134 @@ +import os +import json +from pathlib import Path +from dotenv import load_dotenv + +load_dotenv() + +HANA_HOST = os.getenv("HANA_HOST", "") +HANA_PORT = int(os.getenv("HANA_PORT", 30015)) +HANA_USER = os.getenv("HANA_USER", "") +HANA_PASSWORD = os.getenv("HANA_PASSWORD", "") +HANA_SCHEMA = os.getenv("HANA_SCHEMA", "") + +_conn = None + + +def get_connection(): + global _conn + if _conn is None: + try: + from hdbcli import dbapi + _conn = dbapi.connect( + address=HANA_HOST, + port=HANA_PORT, + user=HANA_USER, + password=HANA_PASSWORD + ) + except Exception as e: + print(f" ❌ Error conectando a HANA: {e}") + return None + return _conn + + +def obtener_pendientes() -> list[dict]: + + conn = get_connection() + if not conn: + return [] + + try: + query = f""" + SELECT + "DocEntry", + "DocNum", + "DocDate", + EXTRACT(MONTH FROM "DocDate") "Mes", + "Indicator", + "LicTradNum", + "FolioPref", + "FolioNum", + "U_PGX_REPIMP" "urlPDF", + "U_PGX_PDF" "tienePDF", + "U_PGX_REPIMPXML" "urlXML", + "U_PGX_XML" "tieneXML" + 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 "CANCELED" = 'N' + ORDER BY EXTRACT(MONTH FROM "DocDate") DESC + """ + cursor = conn.cursor() + cursor.execute(query) + columns = [desc[0] for desc in cursor.description] + rows = cursor.fetchall() + cursor.close() + result = [dict(zip(columns, row)) for row in rows] + return result + except Exception as e: + print(f" ❌ Error en query HANA: {e}") + return [] + + +def marcar_pdf_descargado(docentry: int, url: str, ruta: str): + conn = get_connection() + if not conn: + return False + try: + cursor = conn.cursor() + cursor.execute(f""" + UPDATE "{HANA_SCHEMA}".OPCH + SET "U_PGX_PDF" = 'Y' + WHERE "DocEntry" = ? + """, (docentry)) + conn.commit() + cursor.close() + return True + except Exception as e: + 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: + return False + try: + cursor = conn.cursor() + cursor.execute(f""" + UPDATE "{HANA_SCHEMA}".OPCH + SET "U_PGX_XML" = 'Y' + WHERE "DocEntry" = ? + """, (docentry)) + conn.commit() + cursor.close() + return True + except Exception as e: + print(f" ❌ Error actualizando XML HANA: {e}") + return False + + +def cerrar_conexion(): + global _conn + if _conn: + try: + _conn.close() + except: + pass + _conn = None + + +if __name__ == "__main__": + print("=" * 60) + print("📋 HANA MANAGER - Prueba") + print("=" * 60) + docs = obtener_pendientes() + print(f"\n📄 Documentos pendientes: {len(docs)}") + for d in docs[:5]: + print(f" DocEntry={d['DocEntry']} DocNum={d['DocNum']} " + f"DocDate={d['DocDate']} Indicator={d['Indicator']} " + f"Folio={d['FolioPref']}-{d['FolioNum']} " + f"PDF={d['tienePDF']} XML={d['tieneXML']}") + if len(docs) > 5: + print(f" ... y {len(docs) - 5} más") diff --git a/logger_utilities.py b/logger_utilities.py new file mode 100644 index 0000000..65f40cf --- /dev/null +++ b/logger_utilities.py @@ -0,0 +1,93 @@ +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 new file mode 100644 index 0000000..625d9a3 --- /dev/null +++ b/login_api.py @@ -0,0 +1,307 @@ +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_7) AppleWebKit/537.36", + "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_7) AppleWebKit/537.36", + "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() diff --git a/main.py b/main.py new file mode 100644 index 0000000..46e9448 --- /dev/null +++ b/main.py @@ -0,0 +1,365 @@ +import os +import sys +import json +import time +import signal +import random +from datetime import datetime, timedelta +from pathlib import Path +from dotenv import load_dotenv +import urllib3 + +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +load_dotenv() + +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 smb_manager import sincronizar_a_smb +from consulta_cpe import ( + TokenExpiredError, load_token, get_headers, consultar_cpe, + descargar_archivo, cpe_id, + API_HOST, BASE_PATH, DOWNLOAD_DIR +) + +# ============================================================================ +# CONFIGURACIÓN DE HORARIOS +# ============================================================================ +# (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 + +# ─── Colores ──────────────────────────────────────────── +color_verde = "\033[92m" +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: + inicio = hi * 60 + mi + fin = hf * 60 + mf + if inicio <= minutos < fin: + 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) + if prox.weekday() in DIAS_ACTIVOS: + target = prox.replace(hour=6, minute=30, second=0, microsecond=0) + return int((target - dt).total_seconds()) + return 86400 + + +def obtener_token(): + """Obtiene o renueva el token si es necesario.""" + token = load_token() + if token: + print(" 🔑 Token existente válido, reutilizando") + return token + + print("\n🔑 Token no disponible o expirado. Obteniendo nuevo...") + + from get_token_headless import get_token_headless as obtener_headless, chrome_instalado + if not chrome_instalado(): + print(" ⚠️ Google Chrome no instalado. Saltando headless...") + else: + try: + if obtener_headless(): + print(" ✅ Token obtenido vía get_token_headless (headless)") + return load_token() + except Exception as e: + print(f" ⚠️ Headless falló: {e}") + + print(" ⚠️ 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)") + return tk + time.sleep(1) + 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...") + 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] + docs_retry_raw = obtener_fallas_pendientes() + docs_retry = [] + for d in docs_retry_raw: + docs_retry.append({ + "DocEntry": d["docentry"], + "DocNum": d.get("numero", ""), + "LicTradNum": d["ruc"], + "Indicator": d["tipo"], + "FolioPref": d["serie"], + "FolioNum": d["numero"], + "tienePDF": "N", + "tieneXML": "N", + }) + total_fallas_count = total_fallas() + + if not docs_nuevos and not docs_retry: + print(" ✅ 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") + DOWNLOAD_DIR.mkdir(exist_ok=True) + + # ======================================================================== + def _procesar(doc, idx, total, es_retry=False): + nonlocal token + if not _running: + return + + ruc = doc.get("LicTradNum", "").strip() + tipo_cpe = doc.get("Indicator", "").strip() + serie = doc.get("FolioPref", "").strip() + numero = str(doc.get("FolioNum", "")).strip() + docentry = doc.get("DocEntry") + docnum = doc.get("DocNum", "") + tiene_pdf = doc.get("tienePDF", "N") + tiene_xml = doc.get("tieneXML", "N") + + if not ruc or not tipo_cpe or not serie or not numero: + print(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 + + 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: + 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}" + else: + pdf_msg = f"{color_amarillo}⚠️ error{color_fin}" + hubo_error = True + + print(f"{prefijo_hijo}├── 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}" + 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')}") + + 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)}") + + break + + 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 + + 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 + + # ======================================================================== + + 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...") + _running = False + + +def main(): + global _running + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + log_path = setup_logger() + limpiar_logs_antiguos(dias=7) + init_db() + print(f"📄 Log: {log_path}") + + 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:") + for hi, mi, hf, mf, intervalo, modo in HORARIOS: + print(f" {hi:02d}:{mi:02d} - {hf:02d}:{mf:02d} | cada {intervalo}s | {modo}") + print() + + 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')}") + for _ in range(int(espera)): + if not _running: + break + time.sleep(1) + continue + + intervalo_info = obtener_intervalo(ahora) + 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')}") + for _ in range(int(espera)): + if not _running: + break + time.sleep(1) + continue + + 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}") + + if not token: + token = obtener_token() + if not token: + print(" ❌ No se pudo obtener token. Esperando 60s...") + time.sleep(60) + continue + + procesar_lote(token) + + try: + sincronizar_a_smb(str(DOWNLOAD_DIR)) + except Exception as e: + print(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 + if tf.exists(): + try: + data = json.loads(tf.read_text()) + 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.") + token = None + except: + pass + + print(f"\n⏳ Esperando {intervalo}s hasta próximo ciclo...") + for _ in range(intervalo): + if not _running: + break + time.sleep(1) + + cerrar_conexion() + print("\n👋 Daemon detenido.") + + +if __name__ == "__main__": + main() diff --git a/reporte_fallas.py b/reporte_fallas.py new file mode 100644 index 0000000..779d338 --- /dev/null +++ b/reporte_fallas.py @@ -0,0 +1,166 @@ +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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..c8e486c --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +playwright +python-dotenv +requests +hdbcli +matplotlib diff --git a/smb_manager.py b/smb_manager.py new file mode 100644 index 0000000..6c6b4bb --- /dev/null +++ b/smb_manager.py @@ -0,0 +1,154 @@ +import os +import re +import platform +import subprocess +from pathlib import Path +from dotenv import load_dotenv + +load_dotenv() + +SMB_SERVER = os.getenv("SMB_SERVER", "").strip() +SMB_SHARE = os.getenv("SMB_SHARE", "").strip() +SMB_USER = os.getenv("SMB_USER", "").strip() +SMB_PASSWORD = os.getenv("SMB_PASSWORD", "").strip() +SMB_SUBFOLDER = os.getenv("SMB_SUBFOLDER", "").strip() +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": + return "mac" + if system == "windows": + return "windows" + return "linux" + + +def get_mount_point(): + current_os = get_current_os() + mapping = { + "mac": SMB_MOUNT_MAC, + "linux": SMB_MOUNT_LINUX, + "windows": SMB_MOUNT_WINDOWS, + } + return mapping.get(current_os, "") + + +def esta_montado(mount_point): + if not mount_point: + return False + if get_current_os() == "windows": + return os.path.exists(mount_point) + return os.path.ismount(mount_point) or os.path.exists( + os.path.join(mount_point, ".mounted_check") + ) + + +def montar_smb(): + mount_point = get_mount_point() + if not mount_point: + print(" ⚠️ No hay punto de montaje configurado para este SO.") + return False + + if esta_montado(mount_point): + return True + + current_os = get_current_os() + + if current_os != "windows" and not os.path.exists(mount_point): + os.makedirs(mount_point) + + try: + if current_os == "mac": + cmd = ( + f"mount_smbfs //'{SMB_USER}':'{SMB_PASSWORD}'" + f"@{SMB_SERVER}/{SMB_SHARE} {mount_point}" + ) + subprocess.run(cmd, shell=True, check=True, capture_output=True) + elif current_os == "linux": + cmd = ( + f"sudo mount -t cifs //{SMB_SERVER}/{SMB_SHARE} {mount_point}" + f" -o username={SMB_USER},password={SMB_PASSWORD}" + ) + subprocess.run(cmd, shell=True, check=True, capture_output=True) + elif current_os == "windows": + ruta_smb = f"\\\\{SMB_SERVER}\\{SMB_SHARE}" + cmd = [ + "net", "use", mount_point, ruta_smb, + SMB_PASSWORD, f"/user:{SMB_USER}", "/persistent:no", + ] + subprocess.run(cmd, check=True, capture_output=True) + + print(f" ✅ SMB montado en: {mount_point}") + return True + except Exception as e: + 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}") + return False + + if not montar_smb(): + print(" ❌ No se pudo sincronizar: SMB no montado.") + return False + + mount_point = get_mount_point() + current_os = get_current_os() + + if current_os == "windows" and mount_point.endswith(":") and len(mount_point) == 2: + mount_point += "\\" + + subfolder = SMB_SUBFOLDER.lstrip("/") + if subfolder: + ruta_destino = os.path.normpath(os.path.join(mount_point, subfolder)) + else: + ruta_destino = mount_point + + if not os.path.exists(ruta_destino): + os.makedirs(ruta_destino) + print(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}") + + try: + if current_os in ("mac", "linux"): + cmd = ( + f"find {source} -maxdepth 1 \\( -name '*.pdf' -o -name '*.xml' \\) " + 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") + 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") + elif current_os == "windows": + import shutil + for archivo in os.listdir(source): + if not archivo.lower().endswith((".pdf", ".xml")): + continue + src_path = os.path.join(source, archivo) + dst_path = os.path.join(destination, archivo) + if os.path.isfile(src_path): + if os.path.exists(dst_path): + os.remove(dst_path) + shutil.move(src_path, dst_path) + print(f" ✅ 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") + + return True + except Exception as e: + print(f" ❌ Error en sincronización: {e}") + return False diff --git a/test_headless_token.py b/test_headless_token.py new file mode 100644 index 0000000..a383faa --- /dev/null +++ b/test_headless_token.py @@ -0,0 +1,325 @@ +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()