import asyncio import base64 import logging import re from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeout, Frame from app.scrapers.base import BaseScraper, ScraperError from app.utils.captcha import CaptchaSolver logger = logging.getLogger(__name__) class PoderJudicialScraper(BaseScraper): uses_playwright = True source_name = "Poder Judicial" base_url = "https://scw.pjn.gov.ar/scw/home.seam" SAFE_FETCH_TIMEOUT = 60 max_retries = 1 # Timeouts (milisegundos) DEFAULT_NAVIGATION_TIMEOUT = 45000 DEFAULT_ELEMENT_TIMEOUT = 12000 FORM_WAIT_TIMEOUT = 10000 RESULTS_WAIT_TIMEOUT = 15000 CAPTCHA_FRAME_WAIT = 2000 # Delays (segundos) INITIAL_PAGE_DELAY = 3 AFTER_TAB_CLICK_DELAY = 2 AFTER_VER_DESAFIO_DELAY = 2.5 AFTER_CAPTCHA_TYPE_DELAY = 0.3 AFTER_CAPTCHA_ACCEPT_DELAY = 3 AFTER_FILL_DELAY = 0.3 AFTER_SUBMIT_DELAY = 10 RESULTS_CHECK_DELAY = 3 DOUBLE_CLICK_DELAY = 0.5 # Selectores CSS SELECTOR_TAB_POR_PARTE = [ "td#formPublica\\:porParte\\:header\\:inactive", "td[id='formPublica:porParte:header:inactive']", ] SELECTOR_INPUT_NOMBRE = [ "input[id*='nomIntervParte']", "input[name*='nomIntervParte']", ] SELECTOR_BUTTON_BUSCAR = [ "input[id*='buscarPorParteButton']", "button[id*='buscarPorParteButton']", ] SELECTOR_JURISDICCION = "select#formPublica\\:camaraPartes" SELECTOR_CAPTCHA_TOKEN = [ "input#captcha-response", "input[name='captcha-response']", ] SELECTOR_TABLA_RESULTADOS = "table[id*='tablaPartes']" SELECTOR_NO_RESULTS = "td:has-text('No se encontraron')" async def fetch(self, cuit_or_name: str, **kwargs) -> dict: cuit_clean = self.clean_cuit(cuit_or_name) if cuit_or_name.replace('-', '').isdigit() else None identifier = self.format_cuit(cuit_clean) if cuit_clean else cuit_or_name apellido = kwargs.get("apellido", "").lower().strip() result = await self._fetch_web(identifier) if apellido and result.get("causas"): filtered = [] for c in result["causas"]: caratula = (c.get("caratula", "") or c.get("caratula_completa", "")).lower() if caratula and apellido in caratula: filtered.append(c) result["causas"] = filtered return result async def _fetch_web(self, identifier: str) -> dict: proxy_url = self.get_proxy() try: async with async_playwright() as p: browser, context, page = await self.get_stealth_context(p, proxy_url) try: return await self._pjn_flow(page, identifier) finally: await browser.close() except ScraperError: raise except Exception as e: self.logger.warning(f"[PJN] Error tecnico: {e}") raise ScraperError(self.source_name, f"Error tecnico: {e}") async def _pjn_flow(self, page, identifier: str) -> dict: self.logger.info(f"[PJN] Navegando a {self.base_url}") try: await page.goto(self.base_url, wait_until="commit", timeout=self.DEFAULT_NAVIGATION_TIMEOUT) except Exception as e: self.logger.debug(f"[PJN] goto warning: {e}") await asyncio.sleep(self.INITIAL_PAGE_DELAY) captcha_frame = await self._wait_for_captcha_frame(page) if not captcha_frame: self.logger.warning("[PJN] CAPTCHA frame no aparecio") return {"causas": [], "nota": "CAPTCHA frame no disponible"} tab_ok = await self._click_por_parte_tab(page) if tab_ok: await asyncio.sleep(self.AFTER_TAB_CLICK_DELAY) form_ok = await self._wait_for_por_parte_form(page) if not form_ok: self.logger.warning("[PJN] Formulario Por parte no disponible") return {"causas": [], "nota": "Formulario no disponible"} captcha_ok = await self._solve_captcha(page, captcha_frame) if not captcha_ok: return {"causas": [], "nota": "CAPTCHA no resuelto"} await self._fill_and_submit(page, identifier) causas = await self._parse_results(page) self.logger.info(f"[PJN] {len(causas)} causas para '{identifier}'") return {"causas": causas} async def _click_por_parte_tab(self, page) -> bool: self.logger.debug("[PJN] Click pestaña Por parte") for sel in self.SELECTOR_TAB_POR_PARTE: try: el = await page.query_selector(sel) if el and await el.is_visible(): await el.click() return True except Exception: continue try: await page.evaluate( "document.getElementById('formPublica:porParte:header:inactive')?.click()" ) return True except Exception as e: self.logger.warning(f"[PJN] No se pudo clickear pestana: {e}") return False async def _wait_for_por_parte_form(self, page) -> bool: for sel in self.SELECTOR_INPUT_NOMBRE: try: await page.wait_for_selector(sel, timeout=self.FORM_WAIT_TIMEOUT) return True except PlaywrightTimeout: continue return "nomIntervParte" in (await page.content()) async def _get_captcha_frame(self, page) -> Frame | None: for frame in page.frames: if frame.name == "captcha-frame": return frame for frame in page.frames: if "captcha.pjn.gov.ar" in frame.url: return frame return None async def _wait_for_captcha_frame(self, page, max_attempts=8) -> Frame | None: for i in range(max_attempts): for frame in page.frames: furl = frame.url or "" fname = frame.name or "" if "captcha" in furl.lower() or "captcha" in fname.lower(): return frame if i < 3: self.logger.debug(f"[PJN] Esperando CAPTCHA frame ({i+1}/{max_attempts})") await asyncio.sleep(self.CAPTCHA_FRAME_WAIT / 1000) # Convertir ms a segundos return None async def _solve_captcha(self, page, captcha_frame: Frame) -> bool: self.logger.info("[PJN] Resolviendo CAPTCHA...") clicked = await self._click_ver_desafio(captcha_frame) if not clicked: self.logger.warning("[PJN] No se pudo clickear VER DESAFIO") return False await asyncio.sleep(self.AFTER_VER_DESAFIO_DELAY) image_bytes = await self._extract_captcha_image(captcha_frame) if not image_bytes: self.logger.warning("[PJN] No se pudo extraer imagen CAPTCHA") return False # Guardar imagen para debugging (ruta relativa al proyecto) try: from pathlib import Path project_root = Path(__file__).parent.parent.parent debug_path = project_root / "scripts" / "pjn_last_captcha.png" debug_path.parent.mkdir(parents=True, exist_ok=True) with open(debug_path, "wb") as f: f.write(image_bytes) self.logger.debug(f"[PJN] CAPTCHA guardado en {debug_path}") except Exception as e: self.logger.debug(f"[PJN] No se pudo guardar CAPTCHA debug: {e}") text = await self._ocr_captcha(image_bytes) if not text: self.logger.warning("[PJN] OCR no pudo leer CAPTCHA") return False self.logger.info(f"[PJN] CAPTCHA OCR: '{text}'") await self._type_captcha_answer(captcha_frame, text) await asyncio.sleep(self.AFTER_CAPTCHA_TYPE_DELAY) await self._click_accept(captcha_frame) await asyncio.sleep(self.AFTER_CAPTCHA_ACCEPT_DELAY) token = await self._get_captcha_token(page) if token: self.logger.info(f"[PJN] CAPTCHA resuelto, token: {token[:20]}...") return True self.logger.warning("[PJN] CAPTCHA fallo (sin token tras ACEPTAR)") return False async def _click_ver_desafio(self, frame: Frame) -> bool: try: btn = await frame.query_selector("button.terminos-button") if btn and await btn.is_visible(): await btn.click() return True except Exception: pass try: await frame.evaluate("document.querySelector('button.terminos-button')?.click()") return True except Exception: return False async def _extract_captcha_image(self, frame: Frame) -> bytes | None: try: img = await frame.query_selector(".text-challenge-image img") if not img: img = await frame.query_selector("img[src*='data:image']") if not img: return None src = await img.get_attribute("src") if not src: return None if src.startswith("data:image"): b64_data = src.split(",", 1)[1] elif re.match(r"^[A-Za-z0-9+/=]{100,}$", src): b64_data = src else: return None padding = len(b64_data) % 4 if padding: b64_data += "=" * (4 - padding) return base64.b64decode(b64_data) except Exception as e: self.logger.debug(f"[PJN] Error extrayendo imagen: {e}") return None async def _ocr_captcha(self, image_bytes: bytes) -> str | None: solver = CaptchaSolver() b64 = base64.b64encode(image_bytes).decode() text = await solver.solve_image_captcha_local(b64) if text and len(text) >= 3: self.logger.info(f"[PJN] OCR ddddocr: '{text}'") return text text = await solver.solve_image_captcha_preprocessed(image_bytes) if text and len(text) >= 3: self.logger.info(f"[PJN] OCR preprocessed: '{text}'") return text text = await solver.solve_image_captcha_groq(image_bytes) if text and len(text) >= 3: self.logger.info(f"[PJN] OCR Groq: '{text}'") return text return None async def _type_captcha_answer(self, frame: Frame, text: str): try: inp = await frame.query_selector("input.text-challenge-input") if inp: await inp.fill("") await inp.type(text, delay=50) return except Exception: pass try: await frame.evaluate( f"document.querySelector('input.text-challenge-input').value = '{text}'" ) except Exception: pass async def _click_accept(self, frame: Frame): try: btn = await frame.query_selector("button.accept-challenge-button") if btn and await btn.is_visible(): await btn.click() await asyncio.sleep(self.DOUBLE_CLICK_DELAY) # Double click: el sitio requiere dos clicks if await btn.is_visible(): await btn.click() return except Exception: pass try: await frame.evaluate( f"document.querySelector('button.accept-challenge-button')?.click();" f"setTimeout(() => document.querySelector('button.accept-challenge-button')?.click(), {int(self.DOUBLE_CLICK_DELAY * 1000)});" ) except Exception: pass async def _get_captcha_token(self, page) -> str: # Intentar ambos selectores selectors_js = [ "document.getElementById('captcha-response')?.value || ''", "document.querySelector('input[name=\"captcha-response\"]')?.value || ''", ] for js in selectors_js: try: token = await page.evaluate(js) if token: return token except Exception: pass return "" async def _fill_and_submit(self, page, identifier: str): # Seleccionar jurisdicción "Todos/Todas" selected_jurisdiccion = await page.evaluate(""" (() => { const sel = document.getElementById('formPublica:camaraPartes'); if (!sel) return 'no_select'; for (const o of sel.options) { if (o.text.toLowerCase().includes('todos') || o.text.toLowerCase().includes('todas')) { o.selected = true; sel.dispatchEvent(new Event('change', {bubbles: true})); return 'selected: ' + o.text; } } if (sel.options.length > 1) { sel.options[1].selected = true; sel.dispatchEvent(new Event('change', {bubbles: true})); return 'selected first: ' + sel.options[1].text; } return 'no_options'; })() """) self.logger.info(f"[PJN] Jurisdiccion: {selected_jurisdiccion}") # Llenar campo de nombre filled = False for sel in self.SELECTOR_INPUT_NOMBRE: try: el = await page.query_selector(sel) if el: await el.fill(identifier) filled = True break except Exception: continue if not filled: self.logger.warning("[PJN] No se pudo llenar campo nombre") await asyncio.sleep(self.AFTER_FILL_DELAY) # Click botón buscar for sel in self.SELECTOR_BUTTON_BUSCAR: try: el = await page.query_selector(sel) if el and await el.is_visible(): await el.click(no_wait_after=True) self.logger.debug("[PJN] Submit clicked") await asyncio.sleep(self.AFTER_SUBMIT_DELAY) return except Exception: continue # Fallback: Enter try: await page.keyboard.press("Enter") except Exception: pass async def _parse_results(self, page) -> list[dict]: # Esperar resultados con múltiples intentos resultados_aparecieron = False for attempt in range(5): try: await page.wait_for_selector( f"{self.SELECTOR_TABLA_RESULTADOS}, {self.SELECTOR_NO_RESULTS}, .ui-messages-error", timeout=self.RESULTS_WAIT_TIMEOUT, ) resultados_aparecieron = True break except PlaywrightTimeout: self.logger.debug(f"[PJN] Esperando resultados intento {attempt+1}") if attempt < 4: # No esperar después del último intento await asyncio.sleep(self.RESULTS_CHECK_DELAY) # Si después de 5 intentos no aparecieron resultados, retornar vacío if not resultados_aparecieron: self.logger.warning("[PJN] Timeout esperando resultados, retornando vacío") return [] # Verificar si no hay resultados try: no_results = await page.query_selector(self.SELECTOR_NO_RESULTS) if no_results: self.logger.info("[PJN] Sin resultados para esta busqueda") return [] except Exception: pass # Buscar tabla de resultados causas = [] tabla = await page.query_selector(self.SELECTOR_TABLA_RESULTADOS) if not tabla: self.logger.debug("[PJN] No se encontro tabla de resultados") return [] filas = await tabla.query_selector_all("tr") self.logger.debug(f"[PJN] Tabla tiene {len(filas)} filas") # Parsear headers para detectar orden de columnas col_map = await self._parse_table_headers(filas[0] if filas else None) # Parsear filas de datos for fila in filas[1:]: celdas = await fila.query_selector_all("td") if len(celdas) < 2: continue try: causa = { "expediente": await self._get_cell_text(celdas, col_map.get("expediente", 0)), "fuero": "Nacional", "caratula": await self._get_cell_text(celdas, col_map.get("caratula", 1)), "estado": await self._get_cell_text(celdas, col_map.get("estado", 2)) or "Activa", "juzgado": await self._get_cell_text(celdas, col_map.get("juzgado", 3)), "jurisdiccion": "Nacional", "fecha": await self._get_cell_text(celdas, col_map.get("fecha", 4)), "caratula_completa": await self._get_cell_text(celdas, col_map.get("caratula_completa", 5)), } causas.append(causa) except Exception as e: self.logger.debug(f"[PJN] Error parseando fila: {e}") continue return causas async def _parse_table_headers(self, header_row) -> dict: """ Parsea headers de tabla para detectar orden de columnas. Retorna mapeo de campo → índice de columna. """ if not header_row: # Fallback: orden por defecto return { "expediente": 0, "caratula": 1, "estado": 2, "juzgado": 3, "fecha": 4, "caratula_completa": 5, } try: header_cells = await header_row.query_selector_all("th, td") headers = [] for cell in header_cells: text = (await cell.inner_text()).strip().lower() headers.append(text) # Mapear columnas por nombre col_map = {} for i, h in enumerate(headers): if "expediente" in h or "número" in h or "nro" in h: col_map["expediente"] = i elif "carátula" in h and "completa" not in h: col_map["caratula"] = i elif "estado" in h or "situación" in h: col_map["estado"] = i elif "juzgado" in h or "tribunal" in h or "órgano" in h: col_map["juzgado"] = i elif "fecha" in h: col_map["fecha"] = i elif "completa" in h: col_map["caratula_completa"] = i # Asegurar que al menos tenemos expediente y carátula if "expediente" not in col_map: col_map["expediente"] = 0 if "caratula" not in col_map: col_map["caratula"] = 1 if 1 < len(headers) else 0 self.logger.debug(f"[PJN] Mapeo columnas: {col_map}") return col_map except Exception as e: self.logger.debug(f"[PJN] Error parseando headers: {e}, usando fallback") return { "expediente": 0, "caratula": 1, "estado": 2, "juzgado": 3, "fecha": 4, "caratula_completa": 5, } async def _get_cell_text(self, cells: list, index: int) -> str: """Helper para extraer texto de celda con validación.""" try: if 0 <= index < len(cells): return (await cells[index].inner_text()).strip() except Exception: pass return ""