import asyncio import pandas as pd from playwright.async_api import async_playwright import logging from datetime import datetime, timedelta import re import numpy as np import requests # Configuración de logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") # Fechas de check-in y check-out checkin_date = datetime.now().strftime("%Y-%m-%d") checkout_date = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d") # checkin_date = '2025-12-06' # checkout_date = '2025-12-07' # Lista de ciudades y sus URLs base ciudades = [ {"nombre": "Acapulco", "url": "https://www.pricetravel.com/es/hoteles-en-caleta-acapulco-acapulco-guerrero-mexico_d635825"}, {"nombre": "Bahía Huatulco", "url": "https://www.pricetravel.com/es/hoteles-en-bahia-de-santa-cruz-huatulco-oaxaca-mexico_d87770"}, {"nombre": "Cancún", "url": "https://www.pricetravel.com/es/hoteles-en-cancun_d54455"}, {"nombre": "Corredor Los Cabos", "url": "https://www.pricetravel.com/es/hoteles-en-corredor-turistico-baja-california-sur-mexico_d87097"}, {"nombre": "Ensenada", "url": "https://www.pricetravel.com/es/hoteles-en-ensenada_d60255"}, {"nombre": "Isla Mujeres", "url": "https://www.pricetravel.com/es/hoteles-en-isla-mujeres_d647435"}, {"nombre": "La Paz", "url": "https://www.pricetravel.com/es/hoteles-en-la-paz_d55417"}, {"nombre": "Los Cabos", "url": "https://www.pricetravel.com/es/hoteles-en-los-cabos_d69398"}, {"nombre": "Mazatlán", "url": "https://www.pricetravel.com/es/hoteles-en-mazatlan_d55873"}, {"nombre": "Mérida", "url": "https://www.pricetravel.com/es/hoteles-en-merida_d55688"}, {"nombre": "Playa del Carmen", "url": "https://www.pricetravel.com/es/hoteles-en-playa-del-carmen_d76001"}, {"nombre": "Puerto Escondido", "url": "https://www.pricetravel.com/es/hoteles-en-puerto-escondido_d56255"}, {"nombre": "Puerto Vallarta", "url": "https://www.pricetravel.com/es/hoteles-en-puerto-vallarta_d56252"}, {"nombre": "Riviera Nayarit", "url": "https://www.pricetravel.com/es/hoteles-en-riviera-nayarit_d119425"}, {"nombre": "San José del Cabo", "url": "https://www.pricetravel.com/es/hoteles-en-san-jose-del-cabo_d61763"}, {"nombre": "San Miguel de Allende", "url": "https://www.pricetravel.com/es/hoteles-en-san-miguel-de-allende_d62865"}, {"nombre": "Tulum", "url": "https://www.pricetravel.com/es/hoteles-en-tulum_d71281"}, {"nombre": "Veracruz", "url": "https://www.pricetravel.com/es/hoteles-en-puerto-de-veracruz_d88051"} ] async def scrape_ciudad(playwright, ciudad, url_base, paginas): """ Scrapea información de hoteles para una ciudad específica. """ browser = await playwright.chromium.launch(headless=True) try: context = await browser.new_context(user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36") page = await context.new_page() resultados = [] for page_num in range(1, paginas + 1): url = f"{url_base}?adults=1&checkin={checkin_date}&checkout={checkout_date}&page={page_num}" try: logging.info(f"🔄 Cargando URL: {url}") await page.goto(url, timeout=90000) await page.wait_for_load_state("domcontentloaded") # dispara lazy-load for _ in range(4): await page.mouse.wheel(0, 1400) await page.wait_for_timeout(400) # tarjetas await page.wait_for_selector('a.HotelCard', state="attached", timeout=15000) hotel_cards = await page.locator('a.HotelCard').all() if not hotel_cards: logging.warning(f"⚠️ No se encontraron hoteles en {ciudad}, página {page_num}") continue for hotel in hotel_cards: data = {"ciudad": ciudad} data["hoteles"] = await safe_extract(hotel, '//h2[@class="ng-binding"]', "N/A") data["precios"] = await extract_price(hotel)#, '//div[@class="footer__pricePerNight"]', "N/A") data["estrellas"] = await extract_stars(hotel) # Verifica si el precio está disponible if data["precios"] == "N/A": logging.warning(f"⚠️ No se encontró precio para el hotel: {data['hoteles']} en {ciudad}") resultados.append(data) except Exception as e: logging.error(f"❌ Error en {ciudad}, página {page_num}: {e}") continue if resultados and len(resultados) < 10: for r in resultados[:5]: logging.info(f"[TEST] {r['hoteles']} -> {r['estrellas']}") if resultados: return pd.DataFrame(resultados) else: logging.warning(f"⚠️ No se encontraron resultados para {ciudad}") return pd.DataFrame() finally: await browser.close() async def safe_extract(element, selector, default_value): """ Extrae texto de un elemento de forma segura. """ try: return await element.locator(selector).inner_text() except: return default_value STAR_RE = re.compile(r'icon-([0-9](?:[-\.]5)?)\-star', re.I) async def extract_stars(hotel) -> float | str: """ Lee las estrellas del elemento . Soporta 3, 4, 5 y medios (3-5 => 3.5 / 3.5). Devuelve float o "N/A". """ try: # anclado al bloque de contenido de la card el = hotel.locator('css=i.content__stars') if await el.count() == 0: # fallback por si cambian BEM: cualquier con content__stars en la clase el = hotel.locator('css=i[class*="content__stars"]') if await el.count() > 0: cls = await el.first.get_attribute("class") or "" m = STAR_RE.search(cls) if m: raw = m.group(1).replace('-', '.') try: return float(raw) except: pass except: pass return "N/A" async def extract_price(hotel) -> float | str: """ Extrae el precio por noche del elemento dentro de footer__pricePerNight. Devuelve el valor numérico como float (sin símbolo ni moneda) o "N/A". """ try: # Anclado al div footer__pricePerNight que contiene el currency-display el = hotel.locator('css=div.footer__pricePerNight currency-display') if await el.count() == 0: # Fallback por si cambian las clases: busca cualquier currency-display dentro de footer__pricePerNight el = hotel.locator('css=div.footer__pricePerNight currency-display') if await el.count() > 0: # Obtener el texto completo del elemento currency-display text = await el.first.text_content() if text: # Extraer solo el número, eliminando el símbolo $ y la moneda (MXN) import re price_match = re.search(r'\$\s*([\d,]+)', text) if price_match: raw_price = price_match.group(1).replace(',', '') # Elimina comas try: return float(raw_price) except ValueError: pass except Exception: pass return "N/A" async def main(): """ Función principal que coordina el scraping de todas las ciudades. """ async with async_playwright() as playwright: tareas = [ scrape_ciudad(playwright, c["nombre"], c["url"], paginas=5) # Se agregó el argumento 'paginas' for c in ciudades ] try: dfs = await asyncio.gather(*tareas, return_exceptions=True) for i, result in enumerate(dfs): if isinstance(result, Exception): logging.error(f"❌ Error en la tarea {ciudades[i]['nombre']}: {result}") dfs = [df for df in dfs if isinstance(df, pd.DataFrame)] # Filtra errores df_final = pd.concat(dfs, ignore_index=True) archivo = f"Hoteles_PriceTravel/Hoteles_PriceTravel_{checkin_date}.csv" df_final.to_csv(archivo, index=False, encoding='utf-8-sig') logging.info(f"✅ Datos guardados en: {archivo}") except Exception as e: logging.error(f"❌ Error global: {e}") # if __name__ == "__main__": # asyncio.run(main()) def procesar_resumen(checkin_date: str) -> pd.DataFrame: ruta = f"Hoteles_PriceTravel/Hoteles_PriceTravel_{checkin_date}.csv" df = pd.read_csv(ruta) df['estrellas_redondeadas'] = df['estrellas'].apply( lambda x: np.floor(x + 0.5) if pd.notna(x) else np.nan ) df_total = ( df.groupby('ciudad')['precios'] .mean() .reset_index(name='prom_total') .round() ) df_4 = ( df[df['estrellas_redondeadas'] == 4] .groupby('ciudad')['precios'] .mean() .reset_index(name='prom_4_estrellas') .round() ) df_5 = ( df[df['estrellas_redondeadas'] == 5] .groupby('ciudad')['precios'] .mean() .reset_index(name='prom_5_estrellas') .round() ) df_final = ( df_total .merge(df_4, on='ciudad', how='left') .merge(df_5, on='ciudad', how='left') ) df_final.to_csv( f"Hoteles_resumen_mxn_dls/Hoteles_resumen_{checkin_date}.csv", index=False, encoding='utf-8-sig' ) return df_final def obtener_precio_dolar() -> float | None: url = "https://api.exchangerate-api.com/v4/latest/USD" r = requests.get(url, timeout=15) if r.status_code == 200: data = r.json() return data["rates"].get("MXN") return None def convertir_a_dolares(df: pd.DataFrame, tipo_cambio: float, checkin_date: str): df_usd = df.copy() cols = ['prom_total', 'prom_4_estrellas', 'prom_5_estrellas'] for c in cols: if c in df_usd.columns: df_usd[c] = (df_usd[c] / tipo_cambio).round(2) df_usd.to_csv( f"Hoteles_resumen_mxn_dls/Hoteles_resumen_{checkin_date}_dolares.csv", index=False, encoding='utf-8-sig' ) return df_usd def ejecutar_postproceso(): checkin_date = datetime.now().strftime("%Y-%m-%d") df_resumen = procesar_resumen(checkin_date) tipo_cambio = obtener_precio_dolar() if tipo_cambio: convertir_a_dolares(df_resumen, tipo_cambio, checkin_date) logging.info(f"💵 Tipo de cambio usado: {tipo_cambio:.2f}") else: logging.warning("⚠️ No se pudo obtener tipo de cambio") if __name__ == "__main__": asyncio.run(main()) ejecutar_postproceso()