GeorgeandClaude Sonnet 5
c2503745bf
- Nuevo logger_manager.py (reemplaza logger_utilities.py): logging con formato de árbol, rotación diaria a sunat.log, y consola por stdout (antes iba por stderr y se duplicaba en logs/error.log del launchd). - Nuevo historial_manager.py: registra cada intento de consulta/descarga en historial.db para auditoría, con limpieza automática de registros de más de 180 días. - consulta_cpe.py: corrige el id de notas de crédito/débito (usa F7/F8, no 07/08) y agrega reintentos ante errores 5xx. - hana_manager.py: suma consulta a ORPC para notas de crédito (Indicator 07) y soporte de ObjType para actualizar la tabla correcta (OPCH/ORPC) al marcar PDF/XML descargado. - gestor_fallas.py / main.py: propagan objtype end-to-end y dejan constancia de cada intento en historial_manager. - Elimina login_api.py y reporte_fallas.py (código no usado / script manual sin uso reciente). - Actualiza .gitignore (sunat-venv/, historial.db*) y docs/ para reflejar el estado real del proyecto. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
460 lines
18 KiB
Python
460 lines
18 KiB
Python
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
|
|
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"
|
|
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 _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))
|
|
logger.info(f"Screenshot guardado: {path}")
|
|
|
|
# ============================================================================
|
|
def get_token(headless=False, timeout=60000):
|
|
logger.info("Iniciando navegador...")
|
|
with sync_playwright() as p:
|
|
browser = p.firefox.launch(
|
|
headless=headless,
|
|
)
|
|
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 ───────────────────────────────────
|
|
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 ──
|
|
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")
|
|
logger.info(f"Popup URL: {popup.url}")
|
|
|
|
# ─── 3. Login ─────────────────────────────────────
|
|
logger.info("3. Verificando si hay formulario de login...")
|
|
popup.wait_for_load_state("networkidle")
|
|
|
|
logger.info(f"Título popup: {popup.title()}")
|
|
has_inputs = len(popup.locator("input[type='password']").all()) > 0
|
|
logger.info(f"¿Tiene campo password?: {has_inputs}")
|
|
|
|
if has_inputs or any(k in popup.url.lower() for k in ["login", "autentica"]):
|
|
logger.info("Formulario de login detectado, rellenando credenciales...")
|
|
fill_login_form(popup)
|
|
save_screenshot(popup, "03_form_rellenado")
|
|
|
|
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")
|
|
logger.info(f"URL post-login: {popup.url}")
|
|
else:
|
|
logger.warning("No se detectó login. Imprimiendo HTML de la página...")
|
|
try:
|
|
html = popup.content()
|
|
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)
|
|
logger.info(f"'{el}' encontrado en posición {idx}: ...{html[max(0,idx-50):idx+100]}...")
|
|
except Exception as e:
|
|
logger.error(f"Error leyendo HTML: {e}")
|
|
|
|
# ─── 4. Token via consulta directa ────────────────
|
|
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:
|
|
logger.info("Token obtenido!")
|
|
save_token(token)
|
|
save_screenshot(popup, "05_token_obtenido")
|
|
browser.close()
|
|
return token
|
|
|
|
logger.error("Token no encontrado. Revisa screenshots.")
|
|
logger.info("Frames en la página:")
|
|
for i, frm in enumerate(popup.frames):
|
|
logger.info(f"[{i}] name='{frm.name}' url='{frm.url[:100]}'")
|
|
|
|
if not headless:
|
|
_esperar_enter("\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()
|
|
logger.info(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()
|
|
logger.info(f"id='{id_}' name='{name}' type='{type_}' placeholder='{placeholder}' visible={visible}")
|
|
except:
|
|
pass
|
|
|
|
logger.info(f"Rellenando RUC: {RUC}")
|
|
page.fill("#txtRuc", RUC)
|
|
|
|
logger.info(f"Rellenando Usuario: {USER}")
|
|
page.fill("#txtUsuario", USER)
|
|
|
|
logger.info(f"Rellenando Contraseña: {'*' * len(PASSWORD)}")
|
|
page.fill("#txtContrasena", PASSWORD)
|
|
|
|
captcha = page.locator("input[id='captcha']")
|
|
if captcha.is_visible():
|
|
logger.warning("CAPTCHA detectado en la página!")
|
|
save_screenshot(page, "03_captcha_detectado")
|
|
_esperar_enter("\n⏸️ Resuelve el captcha manualmente y presiona Enter...")
|
|
else:
|
|
logger.info("Captcha no detectado")
|
|
|
|
# ============================================================================
|
|
def get_token_from_direct_access(context, login_page):
|
|
logger.info("Buscando enlace con token en el menú SOL...")
|
|
token = extract_token_from_menu(login_page)
|
|
if token:
|
|
logger.info("Token encontrado en menú (href)")
|
|
return token
|
|
|
|
logger.info("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
|
|
logger.info("Frames activos:")
|
|
for frm in page.frames:
|
|
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()
|
|
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)
|
|
logger.info(f"HTML guardado en {debug_path}")
|
|
|
|
import re
|
|
|
|
tokens = re.findall(r'token[=:][^&\s"\']{50,}', html)
|
|
if tokens:
|
|
logger.info(f"Tokens encontrados (token=): {len(tokens)}")
|
|
for t in tokens[:3]:
|
|
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:
|
|
logger.info(f"JWT encontrados: {len(jwt_matches)}")
|
|
for j in jwt_matches[:3]:
|
|
logger.info(f"{j[:80]}...")
|
|
t = extract_token_from_url(j)
|
|
if t:
|
|
return t
|
|
|
|
ls = frm.evaluate("JSON.stringify(localStorage)")
|
|
logger.info(f"localStorage: {ls[:200]}")
|
|
ss = frm.evaluate("JSON.stringify(sessionStorage)")
|
|
logger.info(f"sessionStorage: {ss[:200]}")
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
|
|
return None
|
|
|
|
# ============================================================================
|
|
def click_menu_item_by_id(page, 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()
|
|
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:
|
|
logger.info("clickeando span...")
|
|
span.first.click()
|
|
else:
|
|
logger.info("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
|
|
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():
|
|
logger.info("clickeando span con texto...")
|
|
el.click()
|
|
page.wait_for_timeout(1500)
|
|
return None
|
|
except:
|
|
pass
|
|
|
|
logger.warning(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:
|
|
logger.info("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
|
|
|
|
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:
|
|
logger.info("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:
|
|
logger.info("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:
|
|
logger.info(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:
|
|
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:
|
|
logger.info("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:
|
|
logger.info("JWT encontrado en HTML")
|
|
return t
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
|
|
return None
|
|
|
|
# ============================================================================
|
|
def extract_token_from_menu(page):
|
|
logger.info("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():
|
|
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'):
|
|
logger.info(f"href: {r['href'][:120]}")
|
|
if r.get('onclick'):
|
|
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:
|
|
logger.error(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))
|
|
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__":
|
|
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/")
|
|
# ============================================================================ |