scraper.py

# -*- 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),
    }