import os import time from pathlib import Path from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout from hana_manager import marcar_pdf_descargado, marcar_xml_descargado from logger_manager import get_logger logger = get_logger("descargador") # ============================================================================ URL_SOVOS = "https://public-einvoicing.sovos.com/pe" # ============================================================================ def descargar_comprobante(page, ruc, tipo, serie, numero, correlativo, monto, fecha, download_dir): page.goto(URL_SOVOS, wait_until="domcontentloaded", timeout=30000) try: page.locator('input[name="senderTaxId"]').wait_for(state="visible", timeout=30000) except Exception as e: try: snippet = (page.inner_text("body") or "")[:600].replace("\n", " ") except Exception: snippet = "" logger.error(f" ❌ No se encontró el campo RUC. URL={page.url}") logger.error(f" 🔎 Contenido visible: {snippet}") raise page.fill('input[name="senderTaxId"]', ruc) page.click('#mui-component-select-invoiceTypeCode') page.wait_for_selector(f'li[role="option"][data-value="{tipo}"]', timeout=5000) page.click(f'li[role="option"][data-value="{tipo}"]') time.sleep(0.5) page.fill('input[name="correlative"]', correlativo) page.fill('input[name="payableAmount"]', str(monto)) date_input = page.locator('//input[contains(@placeholder, "Ej: 202")]') date_input.click() date_input.fill(fecha) time.sleep(0.5) submit_btn = page.locator('//button[contains(text(), "Ver documento")]') if submit_btn.is_disabled(): raise ValueError("Botón 'Ver documento' deshabilitado. Verifique los campos.") submit_btn.click() download_btn = page.locator("[data-testid='btn-download-document']") errorLocator = page.locator(".sovosSnackbar--error, :text('Document not found')") try: download_btn.wait_for(state="visible", timeout=15000) except PlaywrightTimeout: if errorLocator.count() > 0 and errorLocator.first.is_visible(): raise FileNotFoundError("El comprobante no existe en el portal de Sovos (Document not found).") raise TimeoutError("Tiempo de espera agotado sin encontrar el comprobante ni error.") nombre_base = f"{ruc}-{tipo}-{serie}-{numero}" with page.expect_download(timeout=30000) as download_info: download_btn.click() download_pdf = download_info.value pdf_path = Path(download_dir) / f"{nombre_base}.pdf" download_pdf.save_as(str(pdf_path)) page.click('button:has-text(".pdf"), button:has-text(".xml")') time.sleep(1) page.click('li[role="menuitem"]:has-text(".xml")') page.wait_for_selector('button:has-text(".xml")', timeout=5000) with page.expect_download(timeout=30000) as download_info: download_btn.click() download_xml = download_info.value xml_path = Path(download_dir) / f"{nombre_base}.xml" download_xml.save_as(str(xml_path)) return [pdf_path.name, xml_path.name] # ============================================================================ def descargar_documentos_pendientes(docs, download_dir): Path(download_dir).mkdir(parents=True, exist_ok=True) docs_descargar = [ d for d in docs if (d.get("tienePDF") or "N").strip() == "N" or (d.get("tieneXML") or "N").strip() == "N" ] if not docs_descargar: logger.info(" ✅ No hay documentos para descargar") return {"exitosos": 0, "fallidos": 0, "archivos": []} logger.info(f"\n📥 Descargando {len(docs_descargar)} comprobantes desde Sovos...") exitosos = 0 fallidos = 0 archivos_descargados = [] with sync_playwright() as p: browser = p.firefox.launch(headless=True) context = browser.new_context(accept_downloads=True, ignore_https_errors=True) page = context.new_page() for idx, doc in enumerate(docs_descargar): ruc = (doc.get("LicTradNum") or "").strip() tipo = (doc.get("Indicator") or "").strip() serie = (doc.get("FolioPref") or "").strip() numero_raw = doc.get("FolioNum") or 0 numero = str(int(numero_raw)) monto = doc.get("DocTotal") or 0 fecha = (doc.get("TaxDate") or "").strip() correlativo = f"{serie}-{numero}" schema = (doc.get("schema") or "").strip() docentry = doc.get("DocEntry") logger.info(f" [{idx+1}/{len(docs_descargar)}] DocEntry={docentry} DocNum={doc.get('DocNum')} " f"{ruc}-{tipo}-{correlativo}") try: archivos = descargar_comprobante( page, ruc, tipo, serie, numero, correlativo, monto, fecha, download_dir ) exitosos += 1 archivos_descargados.extend(archivos) logger.info(f" ✅ Descargado: PDF={archivos[0] if len(archivos) > 0 else ''} " f"| XML={archivos[1] if len(archivos) > 1 else ''}") if schema and docentry is not None: if (doc.get("tienePDF") or "N").strip() == "N": marcar_pdf_descargado(schema, docentry, doc.get("urlPDF"), archivos[0] if len(archivos) > 0 else "") if (doc.get("tieneXML") or "N").strip() == "N": marcar_xml_descargado(schema, docentry, doc.get("urlXML"), archivos[1] if len(archivos) > 1 else "") time.sleep(1.5) except Exception as e: fallidos += 1 logger.error(f" ❌ Error: {e}") browser.close() logger.info(f"\n📊 Resultado: ✅ {exitosos} éxitos | ❌ {fallidos} fallidos") return {"exitosos": exitosos, "fallidos": fallidos, "archivos": archivos_descargados} # ============================================================================