# -*- coding: utf-8 -*- """Scraper de Idealista Portugal adaptado para Hugging Face Spaces. Conserva la lógica del script original, pero: - elimina input() para que pueda ser llamado desde Gradio; - usa Playwright headless para servidor; - agrega posición por página/global para análisis; - normaliza de forma estimada la antigüedad textual del anuncio. """ from __future__ import annotations import csv import json import re import time import random from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Callable from urllib.parse import urljoin, urlsplit import pandas as pd from bs4 import BeautifulSoup from playwright._impl._errors import Error as PWError from playwright.sync_api import TimeoutError as PWTimeout from playwright.sync_api import sync_playwright BASE = "https://www.idealista.pt" DEFAULT_LANG = "es" # "es" o "pt" LISTING_SELECTORS = [ "article.item", "article[data-element-id]", ".item-info-container", "[data-element-id]", ] LISTING_SELECTOR_COMBINED = ", ".join(LISTING_SELECTORS) LISTING_COLUMNS = [ "district_slug", "source_input", "page_hint", "position_in_page", "global_position", "listing_id", "url", "title", "address_text", "location_full", "price_eur", "price_text", "price_period", "tipologia", "tipologia_text", "area_m2", "area_text", "floor_info", "listed_when", "estimated_published_at", "tag", "agency_name", "agency_url", "image_main_url", "image_main_webp", "image_count", "online_booking", "has_map_button", "scraped_at", ] COOKIE_BUTTON_TEXTS = [ "Aceptar y cerrar", "Aceitar e fechar", "Accept and close", "Aceptar todo", "Aceitar tudo", "Accept all", "Aceptar", "Aceitar", "Accept", "Estoy de acuerdo", "Estou de acordo", "OK", ] LogFn = Callable[[str], None] def _log(log_fn: LogFn | None, msg: str) -> None: if log_fn: log_fn(msg) else: print(msg) def norm_price(text: str | None) -> int | None: m = re.search(r"(\d[\d\.]*)", text or "") return int(m.group(1).replace(".", "")) if m else None def norm_area(text: str | None) -> int | None: m = re.search(r"(\d[\d\.]*)\s*m", (text or "").lower()) return int(m.group(1).replace(".", "")) if m else None def parse_tipologia(text: str | None) -> int | None: m = re.search(r"\bT(\d+)\b", (text or "").upper()) return int(m.group(1)) if m else None def parse_address_from_title(title: str | None) -> str | None: parts = (title or "").split(" en ", 1) return parts[1].strip() if len(parts) == 2 else None def build_location_full(address_text: str | None) -> str | None: addr = (address_text or "").strip() if not addr: return None addr = re.sub(r"\s*,\s*", ", ", addr) addr = re.sub(r"\s+", " ", addr).strip(" ,") if not re.search(r"\bPortugal\b", addr, flags=re.I): addr = f"{addr}, Portugal" return addr def estimate_published_at(listed_when: str | None, observed_at: datetime | None = None) -> str | None: """Convierte etiquetas tipo 'hace 3 días' en fecha estimada ISO. Es una estimación, no una fecha oficial de Idealista. Si el texto no se puede interpretar, devuelve None. """ if not listed_when: return None observed_at = observed_at or datetime.now(timezone.utc) text = listed_when.strip().lower() text = text.replace("á", "a").replace("é", "e").replace("í", "i").replace("ó", "o").replace("ú", "u") if any(x in text for x in ["ayer", "ontem"]): return (observed_at - timedelta(days=1)).date().isoformat() if any(x in text for x in ["hoy", "hoje", "ahora", "agora"]): return observed_at.date().isoformat() m = re.search(r"(\d+)\s*(minuto|minutos|min|min\.|hora|horas|h|dia|dias|semana|semanas|mes|meses)", text) if not m: return None n = int(m.group(1)) unit = m.group(2) if unit.startswith("min"): delta = timedelta(minutes=n) elif unit in {"hora", "horas", "h"}: delta = timedelta(hours=n) elif unit in {"dia", "dias"}: delta = timedelta(days=n) elif unit in {"semana", "semanas"}: delta = timedelta(weeks=n) elif unit in {"mes", "meses"}: delta = timedelta(days=30 * n) else: return None return (observed_at - delta).date().isoformat() def parse_listing_html(html: str) -> list[dict]: soup = BeautifulSoup(html, "html.parser") out: list[dict] = [] observed_at = datetime.now(timezone.utc) articles = soup.select("article.item, article[data-element-id]") for pos, art in enumerate(articles, start=1): item: dict = {} item["listing_id"] = art.get("data-element-id") a = art.select_one("a.item-link") if not a: continue item["title"] = a.get_text(strip=True) item["url"] = urljoin(BASE, a.get("href")) item["address_text"] = parse_address_from_title(item["title"]) item["location_full"] = build_location_full(item["address_text"]) price_el = art.select_one(".price-row .item-price") price_text = price_el.get_text(strip=True) if price_el else None item["price_text"] = price_text item["price_eur"] = norm_price(price_text) mper = re.search(r"/(mes|semana|d[ií]a|dia)", (price_text or "").lower()) item["price_period"] = f"/{mper.group(1)}" if mper else None details = [d.get_text(" ", strip=True) for d in art.select(".item-detail-char .item-detail")] tipologia_text = next((d for d in details if re.search(r"\bT\d+\b", d, re.I)), None) area_text = next((d for d in details if "m²" in d.lower() or "m2" in d.lower()), None) time_el = art.select_one(".item-detail-char .item-detail.txt-highlight-red") time_badge = time_el.get_text(strip=True) if time_el else None if not time_badge: for d in reversed(details): if re.search(r"\b(horas?|d[ií]as?|dias?|semanas?|meses?)\b", d, re.I): time_badge = d break floor_info = None for d in details: low = d.lower() if "planta" in low or "ascensor" in low or "elevador" in low: floor_info = d break item["tipologia_text"] = tipologia_text item["tipologia"] = parse_tipologia(tipologia_text) item["area_text"] = area_text item["area_m2"] = norm_area(area_text) item["floor_info"] = floor_info item["listed_when"] = time_badge item["estimated_published_at"] = estimate_published_at(time_badge, observed_at) item["position_in_page"] = pos tag_el = art.select_one(".listing-tags") item["tag"] = tag_el.get_text(" | ", strip=True) if tag_el else None ag_a = art.select_one("picture.logo-branding a") item["agency_name"] = ag_a.get("title") if ag_a and ag_a.get("title") else None item["agency_url"] = urljoin(BASE, ag_a.get("href")) if ag_a and ag_a.get("href") else None img = art.select_one("picture img") item["image_main_url"] = img.get("src") if img and img.get("src") else None webp = art.select_one('source[type="image/webp"]') item["image_main_webp"] = webp.get("srcset") if webp and webp.get("srcset") else None cnt = art.select_one(".item-multimedia-pictures__counter") if cnt: mcnt = re.search(r"(\d+)", cnt.get_text(strip=True)) item["image_count"] = int(mcnt.group(1)) if mcnt else None else: item["image_count"] = None item["online_booking"] = bool(art.select_one(".online-booking")) item["has_map_button"] = bool(art.select_one(".btn-show-map")) out.append(item) return out def safe_goto(page, url: str, **kwargs) -> bool: try: page.goto(url, **kwargs) return True except PWError as e: msg = str(e) if "ERR_CERT_VERIFIER_CHANGED" in msg: page.wait_for_timeout(1500) page.goto(url, **kwargs) return True raise def accept_cookies_if_needed(page, log_fn=print): try: # 1) Botones normales por rol accesible for label in COOKIE_BUTTON_TEXTS: try: btn = page.get_by_role("button", name=re.compile(label, re.I)) if btn.count() > 0: btn.first.click(timeout=3000) page.wait_for_timeout(1500) log_fn(f"[DEBUG] Cookies aceptadas con botón: {label}") return True except Exception: pass # 2) Selectores por texto visible for label in COOKIE_BUTTON_TEXTS: try: locator = page.locator( f"button:has-text('{label}'), " f"a:has-text('{label}'), " f"text='{label}'" ) if locator.count() > 0: locator.first.click(timeout=3000) page.wait_for_timeout(1500) log_fn(f"[DEBUG] Cookies aceptadas con locator/texto: {label}") return True except Exception: pass # 3) XPath exacto fuera de frames for label in COOKIE_BUTTON_TEXTS: try: xpath_locator = page.locator(f"xpath=//*[normalize-space()='{label}']") if xpath_locator.count() > 0: xpath_locator.first.click(timeout=3000) page.wait_for_timeout(1500) log_fn(f"[DEBUG] Cookies aceptadas con XPath exacto: {label}") return True except Exception: pass # 4) Búsqueda dentro de frames for frame in page.frames: for label in COOKIE_BUTTON_TEXTS: try: btn = frame.get_by_role("button", name=re.compile(label, re.I)) if btn.count() > 0: btn.first.click(timeout=3000) page.wait_for_timeout(1500) log_fn(f"[DEBUG] Cookies aceptadas en frame con botón: {label}") return True except Exception: pass try: locator = frame.locator( f"button:has-text('{label}'), " f"a:has-text('{label}'), " f"text='{label}'" ) if locator.count() > 0: locator.first.click(timeout=3000) page.wait_for_timeout(1500) log_fn(f"[DEBUG] Cookies aceptadas en frame con texto: {label}") return True except Exception: pass try: xpath_locator = frame.locator(f"xpath=//*[normalize-space()='{label}']") if xpath_locator.count() > 0: xpath_locator.first.click(timeout=3000) page.wait_for_timeout(1500) log_fn(f"[DEBUG] Cookies aceptadas en frame con XPath exacto: {label}") return True except Exception: pass except Exception as e: log_fn(f"[DEBUG] Error intentando aceptar cookies: {e}") log_fn("[DEBUG] No se detectó botón de cookies.") return False def detect_block_status(html: str, body_text: str = "") -> dict: html_l = (html or "").lower() body_l = (body_text or "").lower() datadome = ( "captcha-delivery.com" in html_l or "datadome" in html_l or "geo.captcha-delivery.com" in html_l ) idealista_blocked = ( "foi detetado um uso indevido" in body_l or "o acesso foi bloqueado" in body_l or "uso indevido" in body_l or "acesso foi bloqueado" in body_l or "access was blocked" in body_l or "access blocked" in body_l ) captcha_iframe = ( "title=\"datadome captcha\"" in html_l or "captcha-delivery.com/captcha" in html_l or ("iframe" in html_l and "captcha" in html_l) ) return { "datadome": datadome, "idealista_blocked": idealista_blocked, "captcha_iframe": captcha_iframe, "blocked": datadome or idealista_blocked or captcha_iframe, } def classify_page_state(diagnosis: dict) -> str: """Clasifica la página sin confundir bloqueo con cero resultados reales.""" flags = diagnosis.get("flags", {}) or {} selector_counts = diagnosis.get("selector_counts", {}) or {} positive_selectors = sum(1 for v in selector_counts.values() if isinstance(v, int) and v > 0) if flags.get("blocked") or flags.get("datadome") or flags.get("captcha_iframe"): return "blocked_datadome" if flags.get("access"): return "access_denied" if positive_selectors > 0: return "ok_listings" if flags.get("no_results"): return "zero_real_results" return "unknown_empty_page" def make_diagnostic_row( *, entry_index: int | None, input_url: str, page_no: int, diagnosis: dict, status: str, html_path: str | None = None, png_path: str | None = None, ) -> dict: flags = diagnosis.get("flags", {}) or {} counts = diagnosis.get("selector_counts", {}) or {} return { "ts_utc": datetime.now(timezone.utc).isoformat(), "entry_index": entry_index, "input_url": input_url, "page_no": page_no, "status": status, "title": diagnosis.get("title"), "final_url": diagnosis.get("url"), "cookies": flags.get("cookies"), "captcha": flags.get("captcha"), "access": flags.get("access"), "no_results": flags.get("no_results"), "datadome": flags.get("datadome"), "idealista_blocked": flags.get("idealista_blocked"), "captcha_iframe": flags.get("captcha_iframe"), "blocked": flags.get("blocked"), "article_item_count": counts.get("article.item"), "article_data_element_id_count": counts.get("article[data-element-id]"), "item_info_container_count": counts.get(".item-info-container"), "data_element_id_count": counts.get("[data-element-id]"), "html_path": html_path, "screenshot_path": png_path, "body_sample": (diagnosis.get("body_text") or "")[:1200].replace("\n", " "), } def diagnose_page(page, page_no, log_fn=print) -> dict: try: title = page.title() except Exception: title = None try: current_url = page.url except Exception: current_url = None try: html = page.content() except Exception: html = "" try: body_text = page.locator("body").inner_text(timeout=5000) except Exception: body_text = "" body_lower = body_text.lower() html_lower = html.lower() selector_counts = {} log_fn(f"[DEBUG] Página {page_no} · title: {title}") log_fn(f"[DEBUG] Página {page_no} · url final: {current_url}") for selector in LISTING_SELECTORS: try: count = page.locator(selector).count() selector_counts[selector] = count log_fn(f"[DEBUG] Selector '{selector}' encontrado: {count}") except Exception as e: selector_counts[selector] = -1 log_fn(f"[DEBUG] Selector '{selector}' error: {e}") block_flags = detect_block_status(html, body_text) flags = { "cookies": any(x in body_lower for x in [ "cookies", "política de cookies", "politica de cookies", "proveedores", "aceptar y cerrar", "aceitar e fechar", ]), "captcha": ( "captcha" in body_lower or "captcha" in html_lower or block_flags["captcha_iframe"] ), "access": ( "acceso denegado" in body_lower or "access denied" in body_lower or "forbidden" in body_lower or "acesso bloqueado" in body_lower or "o acesso foi bloqueado" in body_lower or block_flags["idealista_blocked"] ), "no_results": any(x in body_lower for x in [ "no hay resultados", "sin resultados", "no encontramos", "não encontrámos", "sem resultados", "no hemos encontrado", ]), "datadome": block_flags["datadome"], "idealista_blocked": block_flags["idealista_blocked"], "captcha_iframe": block_flags["captcha_iframe"], "blocked": block_flags["blocked"], } log_fn(f"[DEBUG] Flags página {page_no}: {flags}") log_fn("[DEBUG] Texto inicial body:") log_fn(body_text[:1200].replace("\n", " ")) if flags["blocked"]: log_fn("[BLOCK] Idealista/DataDome detectado. La página no entregó listados al navegador headless.") return { "title": title, "url": current_url, "html": html, "body_text": body_text, "selector_counts": selector_counts, "flags": flags, } def wait_for_list_or_dump( page, page_no, debug_dir="debug", log_fn=print, debug_prefix: str | None = None, input_url: str | None = None, entry_index: int | None = None, ) -> dict: """Espera listado o genera diagnóstico estructurado. Devuelve un dict con status. Ya no devuelve True/False/"blocked", porque eso impedía distinguir bloqueo, cero resultados reales y página vacía desconocida. """ debug_dir = Path(debug_dir) debug_dir.mkdir(parents=True, exist_ok=True) prefix = debug_prefix or "entry_unknown" input_url = input_url or getattr(page, "url", None) or "" try: page.wait_for_selector(LISTING_SELECTOR_COMBINED, timeout=25000) diagnosis = diagnose_page(page, page_no, log_fn=log_fn) status = classify_page_state(diagnosis) log_fn(f"[OK] Estado de página {page_no}: {status}.") return { "status": status, "diagnosis": diagnosis, "diagnostic_row": make_diagnostic_row( entry_index=entry_index, input_url=input_url, page_no=page_no, diagnosis=diagnosis, status=status, ), } except PWTimeout: diagnosis = diagnose_page(page, page_no, log_fn=log_fn) html = diagnosis.get("html", "") status = classify_page_state(diagnosis) html_path = debug_dir / f"debug_{prefix}_page{page_no}.html" png_path = debug_dir / f"debug_{prefix}_page{page_no}.png" json_path = debug_dir / f"debug_{prefix}_page{page_no}_diagnostic.json" try: html_path.write_text(html, encoding="utf-8", errors="replace") log_fn(f"[DEBUG] HTML guardado en: {html_path}") except Exception as e: log_fn(f"[DEBUG] No se pudo guardar HTML debug: {e}") try: page.screenshot(path=str(png_path), full_page=True) log_fn(f"[DEBUG] Screenshot guardado en: {png_path}") except Exception as e: log_fn(f"[DEBUG] No se pudo guardar screenshot: {e}") row = make_diagnostic_row( entry_index=entry_index, input_url=input_url, page_no=page_no, diagnosis=diagnosis, status=status, html_path=str(html_path), png_path=str(png_path), ) try: json_path.write_text(json.dumps(row, ensure_ascii=False, indent=2), encoding="utf-8") log_fn(f"[DEBUG] Diagnóstico JSON guardado en: {json_path}") except Exception as e: log_fn(f"[DEBUG] No se pudo guardar diagnóstico JSON: {e}") if status == "blocked_datadome": log_fn("[BLOCK] Idealista devolvió bloqueo/DataDome en lugar de listados.") log_fn("[BLOCK] Se detiene esta URL para no insistir contra el bloqueo.") elif status == "zero_real_results": log_fn("[INFO] La página parece devolver cero resultados reales, no bloqueo.") else: log_fn(f"[WARN] Sin listado detectable en página {page_no}. Estado: {status}.") return { "status": status, "diagnosis": diagnosis, "diagnostic_row": row, "html_path": str(html_path), "png_path": str(png_path), "json_path": str(json_path), } def get_next_url_from_page(page) -> str | None: a = page.query_selector('a[rel="next"]') if a and a.get_attribute("href"): return urljoin(BASE, a.get_attribute("href")) for t in ["Siguiente", "Seguinte", "Próxima", "Próximo"]: a = page.get_by_role("link", name=re.compile(t, re.I)) if a and a.count() > 0: href = a.first.get_attribute("href") if href: return urljoin(BASE, href) candidates = [] curr = page.url for el in page.query_selector_all("a[href*='pagina-']"): href = el.get_attribute("href") or "" m = re.search(r"pagina-(\d+)", href) if m: candidates.append((int(m.group(1)), urljoin(curr, href))) if candidates: candidates.sort() return candidates[-1][1] return None def get_page_number_from_url(url: str | None) -> int: m = re.search(r"pagina-(\d+)", url or "") return int(m.group(1)) if m else 1 def fetch_pages_playwright( start_url: str, max_pages: int = 80, wait_ms: int = 30000, headless: bool = True, lang: str = DEFAULT_LANG, debug_dir: Path | None = None, log_fn: LogFn | None = None, entry_index: int | None = None, ) -> dict: items: list[dict] = [] diagnostics: list[dict] = [] log = log_fn or print debug_path = debug_dir or Path("debug") final_status = "unknown_empty_page" with sync_playwright() as p: browser = p.chromium.launch(headless=headless, args=["--disable-blink-features=AutomationControlled"]) context = browser.new_context( user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ), locale="es-ES" if lang == "es" else "pt-PT", viewport={"width": 1366, "height": 900}, ignore_https_errors=True, ) page = context.new_page() current_url = start_url seen_urls: set[str] = set() seen_first_ids: set[str] = set() try: for _ in range(1, max_pages + 1): safe_goto(page, current_url, wait_until="domcontentloaded", timeout=30000) try: page.wait_for_load_state("networkidle", timeout=15000) except Exception: pass accept_cookies_if_needed(page, log_fn=log) page.wait_for_timeout(2000) try: page.evaluate("window.scrollTo(0, document.body.scrollHeight * 0.25)") page.wait_for_timeout(1000) page.evaluate("window.scrollTo(0, document.body.scrollHeight * 0.60)") page.wait_for_timeout(1000) page.evaluate("window.scrollTo(0, document.body.scrollHeight * 0.95)") page.wait_for_timeout(1500) except Exception as e: log(f"[DEBUG] No se pudo hacer scroll progresivo: {e}") curr_num = get_page_number_from_url(page.url) debug_prefix = f"entry{entry_index}" if entry_index is not None else "entry_unknown" wait_result = wait_for_list_or_dump( page, curr_num, debug_dir=debug_path, log_fn=log, debug_prefix=debug_prefix, input_url=current_url, entry_index=entry_index, ) page_status = wait_result.get("status", "unknown_empty_page") diagnostics.append(wait_result.get("diagnostic_row", {"status": page_status, "input_url": current_url})) if page_status == "blocked_datadome": log(f"[BLOCK] Se detiene la entrada {debug_prefix} por bloqueo Idealista/DataDome.") final_status = "blocked_datadome" break if page_status == "access_denied": log(f"[BLOCK] Se detiene la entrada {debug_prefix} por acceso denegado.") final_status = "access_denied" break if page_status == "zero_real_results": log(f"[INFO] Cero resultados reales detectados en página {curr_num}.") final_status = "zero_real_results" break if page_status != "ok_listings": log(f"[WARN] Sin listado detectable en página {curr_num}. Estado: {page_status}.") final_status = page_status break if page.url in seen_urls: _log(log_fn, f"[STOP] URL repetida: {page.url}") final_status = "repeated_url" break seen_urls.add(page.url) batch = parse_listing_html(page.content()) if not batch: _log(log_fn, f"[STOP] Página {curr_num} con selector, pero sin anuncios parseables.") final_status = "parse_empty_after_selector" break first_id = batch[0].get("listing_id") if first_id and first_id in seen_first_ids: _log(log_fn, f"[STOP] Primer listing repetido en p{curr_num}; posible bucle.") final_status = "repeated_first_listing" break if first_id: seen_first_ids.add(first_id) for it in batch: it["page_hint"] = curr_num pos = it.get("position_in_page") or 0 it["global_position"] = ((curr_num - 1) * 30) + int(pos) items.extend(batch) final_status = "ok_listings" _log(log_fn, f"[OK] Página {curr_num}: {len(batch)} anuncios.") next_url = get_next_url_from_page(page) if not next_url: break next_num = get_page_number_from_url(next_url) if next_num <= curr_num: _log(log_fn, f"[STOP] Paginación no avanza ({curr_num}→{next_num}).") break current_url = next_url base_wait = max(int(wait_ms), 5000) jitter = random.uniform(0.70, 1.60) real_wait_ms = int(base_wait * jitter) log( f"[WAIT] Pausa entre páginas: {real_wait_ms / 1000:.1f}s " f"(base={wait_ms}ms, jitter={jitter:.2f})." ) page.wait_for_timeout(real_wait_ms) finally: context.close() browser.close() if items and final_status == "blocked_datadome": final_status = "partial_blocked" elif items: final_status = "ok_listings" return { "items": items, "status": final_status, "diagnostics": diagnostics, "pages_seen": len(diagnostics), } def sanitize_filename(stem: str) -> str: return re.sub(r"[^a-zA-Z0-9_\-\.]+", "_", stem).strip("_") or "consulta" def normalize_input_to_url(s, lang="es"): s = (str(s or "")).strip() if not s: return None if s.startswith("http://") or s.startswith("https://"): # No modificar URLs completas. # Importante: si la URL trae ?shape=..., agregar "/" al final rompe el parámetro. return s s = s.strip("/") return f"{BASE}/{lang}/arrendar-casas/{s}/" def extract_slug_from_url(url: str | None) -> str: parts = urlsplit(url or "") path = parts.path.rstrip("/") return path.split("/")[-1] if path else "sin_slug" def load_first_column_urls_or_slugs(xlsx_path: str | Path, column_name: str | None = None) -> list[str]: df = pd.read_excel(xlsx_path, sheet_name=0) if df.empty: return [] if column_name and column_name in df.columns: col = df[column_name].astype(str) else: first_col = df.columns[0] col = df[first_col].astype(str) return [s.strip() for s in col.tolist() if str(s).strip() and str(s).strip().lower() != "nan"] def clean_dataframe(df: pd.DataFrame) -> pd.DataFrame: for col in LISTING_COLUMNS: if col not in df.columns: df[col] = None df = df[LISTING_COLUMNS].copy() return df.where(pd.notnull(df), None) def run_scrape_job( input_xlsx_path: str | Path, output_dir: str | Path, max_pages: int = 80, wait_ms: int = 30000, entry_wait_ms: int = 90000, lang: str = DEFAULT_LANG, diagnostic_mode: bool = False, stop_on_first_block: bool = True, log_fn: LogFn | None = None, ) -> dict: input_xlsx_path = Path(input_xlsx_path) output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) debug_dir = output_dir / "debug" ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") name_no_ext = sanitize_filename(input_xlsx_path.stem) entries = load_first_column_urls_or_slugs(input_xlsx_path) if not entries: raise ValueError("El Excel de entrada no contiene URLs ni slugs válidos en la primera columna.") _log(log_fn, f"[INFO] Entradas detectadas: {len(entries)}") if diagnostic_mode: _log(log_fn, "[DIAG] Modo diagnóstico activo: se generará evidencia y app.py evitará persistencia.") master: list[dict] = [] districts_queried: list[str] = [] diagnostics_rows: list[dict] = [] blocked_entries: list[str] = [] zero_real_result_entries: list[str] = [] unknown_empty_entries: list[str] = [] per_district_dir = output_dir / "out_por_distrito" per_district_dir.mkdir(parents=True, exist_ok=True) for idx, entry in enumerate(entries, start=1): base_url = normalize_input_to_url(entry, lang=lang) if not base_url: _log(log_fn, f"[WARN] Entrada vacía o inválida: {entry!r}") continue district_slug = extract_slug_from_url(base_url) safe_stem = sanitize_filename(district_slug) _log(log_fn, f"\n[RUN] {idx}/{len(entries)} · {entry} → {base_url}") fetch_result = fetch_pages_playwright( base_url, max_pages=max_pages, wait_ms=wait_ms, headless=True, lang=lang, debug_dir=debug_dir, log_fn=log_fn, entry_index=idx, ) data = fetch_result.get("items", []) entry_status = fetch_result.get("status", "unknown_empty_page") diagnostics_rows.extend(fetch_result.get("diagnostics", [])) if entry_status == "blocked_datadome": blocked_entries.append(district_slug) _log(log_fn, f"[BLOCK] El distrito {district_slug} queda como blocked_datadome; NO se marca como consultado.") elif entry_status == "zero_real_results": zero_real_result_entries.append(district_slug) districts_queried.append(district_slug) _log(log_fn, f"[INFO] El distrito {district_slug} sí se marca como consultado: cero resultados reales.") elif data: districts_queried.append(district_slug) else: unknown_empty_entries.append(district_slug) _log(log_fn, f"[WARN] El distrito {district_slug} no se marcará como consultado. Estado: {entry_status}.") scraped_at = datetime.now(timezone.utc).isoformat() for it in data: it["district_slug"] = district_slug it["source_input"] = entry it["scraped_at"] = scraped_at district_df = clean_dataframe(pd.DataFrame(data)) if data else clean_dataframe(pd.DataFrame()) (per_district_dir / f"idealista_{safe_stem}_{ts}.json").write_text( json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" ) district_df.to_csv(per_district_dir / f"idealista_{safe_stem}_{ts}.csv", index=False, encoding="utf-8") district_df.to_excel(per_district_dir / f"idealista_{safe_stem}_{ts}.xlsx", index=False) master.extend(data) if entry_status == "blocked_datadome" and stop_on_first_block: _log(log_fn, "[BLOCK] stop_on_first_block=True: se corta la corrida completa.") break if idx < len(entries): base_sleep = max(int(entry_wait_ms), 30000) / 1000 jitter = random.uniform(0.75, 1.50) sleep_seconds = base_sleep * jitter _log( log_fn, f"[WAIT] Pausa entre URLs/distritos: {sleep_seconds:.1f}s " f"(base={entry_wait_ms}ms, jitter={jitter:.2f})." ) time.sleep(sleep_seconds) master_df = clean_dataframe(pd.DataFrame(master)) if master else clean_dataframe(pd.DataFrame()) diagnostics_df = pd.DataFrame(diagnostics_rows) current_xlsx = output_dir / f"{name_no_ext}_consulta_actual_{ts}.xlsx" current_csv = output_dir / f"{name_no_ext}_consulta_actual_{ts}.csv" current_json = output_dir / f"{name_no_ext}_consulta_actual_{ts}.json" diagnostics_csv = output_dir / f"{name_no_ext}_diagnostico_{ts}.csv" diagnostics_json = output_dir / f"{name_no_ext}_diagnostico_{ts}.json" with pd.ExcelWriter(current_xlsx, engine="openpyxl") as writer: master_df.to_excel(writer, index=False, sheet_name="consulta_actual") diagnostics_df.to_excel(writer, index=False, sheet_name="diagnostico_corrida") master_df.to_csv(current_csv, index=False, encoding="utf-8") current_json.write_text(json.dumps(master, ensure_ascii=False, indent=2), encoding="utf-8") diagnostics_df.to_csv(diagnostics_csv, index=False, encoding="utf-8") diagnostics_json.write_text(json.dumps(diagnostics_rows, ensure_ascii=False, indent=2), encoding="utf-8") if blocked_entries and not master: run_status = "blocked_datadome" elif blocked_entries and master: run_status = "partial_blocked" elif zero_real_result_entries and not master: run_status = "zero_real_results" elif unknown_empty_entries and not master: run_status = "unknown_empty_page" else: run_status = "success" _log(log_fn, f"\n[OK] Consulta finalizada. Anuncios encontrados: {len(master_df)}") _log(log_fn, f"[DIAG] Estado de corrida: {run_status}") _log(log_fn, f"[DIAG] Bloqueos DataDome: {len(blocked_entries)}") _log(log_fn, f"[DIAG] Cero resultados reales: {len(zero_real_result_entries)}") return { "timestamp": ts, "entries_count": len(entries), "rows_count": len(master_df), "dataframe": master_df, "diagnostics_dataframe": diagnostics_df, "districts_queried": sorted(set(districts_queried)), "run_status": run_status, "blocked_entries": sorted(set(blocked_entries)), "blocked_entries_count": len(set(blocked_entries)), "zero_real_result_entries": sorted(set(zero_real_result_entries)), "zero_real_result_entries_count": len(set(zero_real_result_entries)), "unknown_empty_entries": sorted(set(unknown_empty_entries)), "unknown_empty_entries_count": len(set(unknown_empty_entries)), "diagnostics_csv": str(diagnostics_csv), "diagnostics_json": str(diagnostics_json), "current_xlsx": str(current_xlsx), "current_csv": str(current_csv), "current_json": str(current_json), "output_dir": str(output_dir), }