# -*- coding: utf-8 -*- """Aplicación Gradio para Hugging Face Spaces. Flujo: 1) Subir Excel con URLs o slugs. 2) Ejecutar scraping en modo headless. 3) Guardar consulta actual en Excel. 4) Persistir en Neon de forma incremental. 5) Detectar desapariciones como posible desplazamiento/arrendamiento. 6) Descargar Excel de consulta actual y workbook incremental analítico. 7) Enviar archivos por Telegram si las variables están configuradas. """ from __future__ import annotations import os import traceback from datetime import datetime, timezone from pathlib import Path import shutil import gradio as gr from db import ( DEFAULT_NOT_SEEN_THRESHOLD, DEFAULT_PROBABLY_RENTED_THRESHOLD, connect, create_run, export_incremental_excel, finish_run, get_database_url, upsert_listings, ) from scraper import run_scrape_job from telegram_utils import send_result_files, telegram_configured APP_TITLE = "Idealista Portugal · Scraper incremental Neon + Telegram" WORKDIR = Path(os.getenv("WORKDIR", "/tmp/idealista_outputs")) WORKDIR.mkdir(parents=True, exist_ok=True) def zip_debug_folder(output_dir: str | Path) -> str | None: output_dir = Path(output_dir) debug_dir = output_dir / "debug" if not debug_dir.exists(): return None zip_base = output_dir / "debug_files" zip_path = shutil.make_archive(str(zip_base), "zip", debug_dir) return zip_path def _append_log(lines: list[str], message: str) -> None: lines.append(message) def _safe_filename(file_path: str | None) -> str: return Path(file_path).name if file_path else "archivo_entrada.xlsx" def run_pipeline( excel_file: str | None, max_pages: int, wait_ms: int, entry_wait_ms: int, lang: str, send_telegram: bool, diagnostic_mode: bool, stop_on_first_block: bool, probably_rented_threshold: int, not_seen_threshold: int, ) -> tuple[str, str | None, str | None, str | None, str]: logs: list[str] = [] run_id: int | None = None incremental_xlsx: str | None = None current_xlsx: str | None = None debug_zip: str | None = None try: if not excel_file: raise ValueError("Debes subir un archivo Excel con URLs o slugs en la primera columna.") if int(not_seen_threshold) >= int(probably_rented_threshold): raise ValueError("El umbral 'no visto recientemente' debe ser menor que el umbral 'probablemente arrendado'.") ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") output_dir = WORKDIR / f"run_{ts}" output_dir.mkdir(parents=True, exist_ok=True) _append_log(logs, "[1/6] Iniciando scraping en Hugging Face Space...") scrape_result = run_scrape_job( input_xlsx_path=excel_file, output_dir=output_dir, max_pages=int(max_pages), wait_ms=int(wait_ms), entry_wait_ms=int(entry_wait_ms), lang=lang, diagnostic_mode=bool(diagnostic_mode), stop_on_first_block=bool(stop_on_first_block), log_fn=lambda msg: _append_log(logs, msg), ) current_xlsx = scrape_result["current_xlsx"] df = scrape_result["dataframe"] districts_queried = scrape_result.get("districts_queried", []) run_status = scrape_result.get("run_status", "unknown") blocked_entries_count = int(scrape_result.get("blocked_entries_count", 0)) zero_real_result_entries_count = int(scrape_result.get("zero_real_result_entries_count", 0)) unknown_empty_entries_count = int(scrape_result.get("unknown_empty_entries_count", 0)) skip_neon = bool(diagnostic_mode) or (blocked_entries_count > 0 and len(df) == 0) debug_zip = zip_debug_folder(output_dir) if debug_zip: _append_log(logs, f"[DEBUG] ZIP de archivos debug generado: {debug_zip}") else: _append_log(logs, "[DEBUG] No se generó ZIP debug porque no existe carpeta debug.") _append_log(logs, "[2/6] Preparando persistencia incremental en Neon...") if skip_neon: stats = { "inserted": 0, "updated": 0, "unchanged": 0, "reactivated": 0, "missing_updated": 0, "probably_rented": 0, "snapshots": 0, } if diagnostic_mode: _append_log(logs, "[DIAG] Modo diagnóstico activo: se omite Neon.") else: _append_log(logs, "[BLOCK] Se omite Neon porque hubo bloqueo y no se obtuvieron anuncios reales.") elif get_database_url(): with connect() as conn: run_id = create_run( conn, source_filename=_safe_filename(excel_file), entries_count=int(scrape_result["entries_count"]), districts_queried=districts_queried, metadata={ "max_pages": max_pages, "wait_ms": wait_ms, "entry_wait_ms": entry_wait_ms, "lang": lang, "space_runtime": "huggingface-docker", "probably_rented_threshold": probably_rented_threshold, "not_seen_threshold": not_seen_threshold, "diagnostic_mode": diagnostic_mode, "stop_on_first_block": stop_on_first_block, "run_status": run_status, "blocked_entries_count": blocked_entries_count, "zero_real_result_entries_count": zero_real_result_entries_count, "unknown_empty_entries_count": unknown_empty_entries_count, }, ) stats = upsert_listings( conn, df, run_id=run_id, districts_queried=districts_queried, probably_rented_threshold=int(probably_rented_threshold), not_seen_threshold=int(not_seen_threshold), ) finish_run(conn, run_id, stats, status="partial_blocked" if blocked_entries_count else "success") incremental_xlsx = str(output_dir / f"base_incremental_idealista_neon_{ts}.xlsx") export_incremental_excel(conn, incremental_xlsx) _append_log( logs, "[OK] Neon actualizado: " f"nuevos={stats.get('inserted', 0)}, " f"actualizados={stats.get('updated', 0)}, " f"reactivados={stats.get('reactivated', 0)}, " f"sin_cambio={stats.get('unchanged', 0)}, " f"missing_actualizados={stats.get('missing_updated', 0)}, " f"probablemente_arrendados={stats.get('probably_rented', 0)}, " f"snapshots={stats.get('snapshots', 0)}.", ) else: stats = { "inserted": 0, "updated": 0, "unchanged": 0, "reactivated": 0, "missing_updated": 0, "probably_rented": 0, "snapshots": 0, } _append_log(logs, "[WARN] Neon no configurado. Se generó solo el Excel de consulta actual.") _append_log(logs, "[3/6] Archivos Excel generados.") summary = ( "Consulta Idealista finalizada.\n" f"Estado técnico: {run_status}\n" f"Anuncios encontrados: {len(df)}\n" f"Distritos consultados: {len(districts_queried)}\n" f"Bloqueos DataDome: {blocked_entries_count}\n" f"Cero resultados reales: {zero_real_result_entries_count}\n" f"Páginas vacías desconocidas: {unknown_empty_entries_count}\n" f"Nuevos insertados: {stats.get('inserted', 0)}\n" f"Actualizados: {stats.get('updated', 0)}\n" f"Reactivados: {stats.get('reactivated', 0)}\n" f"Sin cambio: {stats.get('unchanged', 0)}\n" f"No vistos en distritos consultados: {stats.get('missing_updated', 0)}\n" f"Probablemente arrendados/desplazados: {stats.get('probably_rented', 0)}\n" f"Run ID Neon: {run_id if run_id is not None else 'sin Neon'}" ) if send_telegram: _append_log(logs, "[4/6] Enviando resultados por Telegram...") tg_statuses = send_result_files(current_xlsx, incremental_xlsx, summary) for status in tg_statuses: _append_log(logs, f"[TELEGRAM] {status}") else: _append_log(logs, "[4/6] Envío por Telegram desactivado para esta corrida.") _append_log(logs, "[5/6] Reporte de desplazamiento y categorías de velocidad incluido en el workbook incremental.") _append_log(logs, "[6/6] Proceso terminado.") status_md = ( "### Resultado\n" f"- **Estado técnico:** {run_status}\n" f"- **Anuncios encontrados:** {len(df)}\n" f"- **Distritos consultados:** {len(districts_queried)}\n" f"- **Bloqueos DataDome:** {blocked_entries_count}\n" f"- **Cero resultados reales:** {zero_real_result_entries_count}\n" f"- **Páginas vacías desconocidas:** {unknown_empty_entries_count}\n" f"- **Nuevos insertados en Neon:** {stats.get('inserted', 0)}\n" f"- **Actualizados en Neon:** {stats.get('updated', 0)}\n" f"- **Reactivados:** {stats.get('reactivated', 0)}\n" f"- **Sin cambio:** {stats.get('unchanged', 0)}\n" f"- **No vistos en distritos consultados:** {stats.get('missing_updated', 0)}\n" f"- **Probablemente arrendados/desplazados:** {stats.get('probably_rented', 0)}\n" f"- **Snapshots históricos:** {stats.get('snapshots', 0)}\n" f"- **Run ID Neon:** {run_id if run_id is not None else 'No disponible'}\n" f"- **Telegram:** {'configurado' if telegram_configured() else 'no configurado'}\n\n" "El Excel incremental incluye las pestañas: Base incremental, Mayor desplazamiento, Mayor antigüedad, " "Menor antigüedad, Cambios precio y Corridas." ) return status_md, current_xlsx, incremental_xlsx, debug_zip, "\n".join(logs) except Exception as exc: error = f"{type(exc).__name__}: {exc}" _append_log(logs, f"[ERROR] {error}") _append_log(logs, traceback.format_exc()) if run_id is not None and get_database_url(): try: with connect() as conn: finish_run( conn, run_id, { "scraped": 0, "inserted": 0, "updated": 0, "unchanged": 0, "reactivated": 0, "missing_updated": 0, "probably_rented": 0, "snapshots": 0, }, status="error", error_message=error, ) except Exception: pass status_md = "### Error\n" + error return status_md, current_xlsx, incremental_xlsx, debug_zip, "\n".join(logs) def build_app() -> gr.Blocks: db_status = "Configurado" if get_database_url() else "No configurado" tg_status = "Configurado" if telegram_configured() else "No configurado" with gr.Blocks(title=APP_TITLE) as demo: gr.Markdown( f"# {APP_TITLE}\n" "Sube un Excel con URLs o slugs de Idealista Portugal en la primera columna. " "La app genera el Excel de la consulta, actualiza Neon de forma incremental, " "detecta anuncios posiblemente arrendados por desaparición en distritos consultados, clasifica la velocidad de desplazamiento por corridas visibles —pensado para ejecución cada tercer día— y puede enviar archivos por Telegram.\n\n" f"**Neon:** {db_status} · **Telegram:** {tg_status}" ) with gr.Row(): excel_file = gr.File(label="Excel de entrada", file_types=[".xlsx", ".xls"], type="filepath") with gr.Column(): max_pages = gr.Slider( label="Máximo de páginas por distrito", minimum=1, maximum=20, value=3, step=1, ) wait_ms = gr.Slider( label="Espera base entre páginas (ms)", minimum=5000, maximum=120000, value=30000, step=5000, ) entry_wait_ms = gr.Slider( label="Espera entre URLs/distritos (ms)", minimum=30000, maximum=300000, value=90000, step=10000, ) lang = gr.Dropdown(label="Idioma URL", choices=["es", "pt"], value="es") send_telegram = gr.Checkbox(label="Enviar resultados por Telegram", value=True) diagnostic_mode = gr.Checkbox( label="Modo diagnóstico: no persistir en Neon aunque haya base configurada", value=True, ) stop_on_first_block = gr.Checkbox( label="Cortar corrida completa al primer bloqueo DataDome", value=True, ) with gr.Accordion("Parámetros de desplazamiento", open=False): gr.Markdown("La categoría de velocidad se calcula cuando el anuncio llega a `probably_rented`: 1 corrida visible = muy rápido; 2-3 = rápido; 4-6 = normal; 7-10 = lento; 11+ = muy lento. Este criterio está calibrado para ejecución cada tercer día.") not_seen_threshold = gr.Slider( label="Corridas consecutivas sin aparecer para 'no visto recientemente'", minimum=1, maximum=10, value=DEFAULT_NOT_SEEN_THRESHOLD, step=1, ) probably_rented_threshold = gr.Slider( label="Corridas consecutivas sin aparecer para 'probablemente arrendado/desplazado'", minimum=2, maximum=15, value=DEFAULT_PROBABLY_RENTED_THRESHOLD, step=1, ) run_btn = gr.Button("Ejecutar consulta", variant="primary") status = gr.Markdown(label="Resultado") with gr.Row(): current_file = gr.File(label="Descargar Excel de consulta actual") incremental_file = gr.File(label="Descargar Excel de base incremental Neon") debug_file = gr.File(label="Descargar debug HTML/PNG") logs = gr.Textbox(label="Log de ejecución", lines=20) run_btn.click( fn=run_pipeline, inputs=[ excel_file, max_pages, wait_ms, entry_wait_ms, lang, send_telegram, diagnostic_mode, stop_on_first_block, probably_rented_threshold, not_seen_threshold, ], outputs=[status, current_file, incremental_file, debug_file, logs], ) return demo if __name__ == "__main__": port = int(os.getenv("PORT", "7860")) build_app().queue(default_concurrency_limit=1).launch(server_name="0.0.0.0", server_port=port)