GHSsda/HotelManagement
0
1"""2Flora HTF Engine v6.13Gestione date, inserimento DB e ciclo principale delegando l'estrazione ad antiban.py4 5FIX v6.1:6 - GIORNI_DA_ANALIZZARE: warning esplicito se rimasto a 1 (valore da test)7 - Commento "Save ONLY successful scrapes" corretto: anche i sold-out vengono8 salvati in DB (is_sold_out=1) per tracciare la disponibilità storica.9"""10import asyncio11import random12import gc13import json14import logging15import os16import re17import sys18from datetime import datetime, timedelta19from pathlib import Path20from postgrest import SyncPostgrestClient21from antiban import scrape_with_antiban, retry_soldouts22from playwright.async_api import async_playwright23 24logging.basicConfig(25 level=logging.INFO,26 format="%(asctime)s [%(levelname)s] %(message)s",27 handlers=[28 logging.FileHandler("scraper.log", encoding="utf-8"),29 logging.StreamHandler()30 ]31)32log = logging.getLogger(__name__)33 34DEFAULT_SUPABASE_URL = "https://fsnhksmnzcvltsdnfhmn.supabase.co"35SUPABASE_URL = os.getenv("SUPABASE_URL", DEFAULT_SUPABASE_URL).rstrip("/")36SUPABASE_REST_URL = os.getenv("SUPABASE_REST_URL", f"{SUPABASE_URL}/rest/v1").rstrip("/")37SUPABASE_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImZzbmhrc21uemN2bHRzZG5maG1uIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImlhdCI6MTc3MzY1NjMwMiwiZXhwIjoyMDg5MjMyMzAyfQ.kpWNMOaIO4xcCUYbENVZm1Wcy-lrDPl9UpN4CtMpW0c"38if not SUPABASE_KEY:39 raise RuntimeError(40 "❌ Nessuna chiave Supabase trovata. "41 "Imposta SUPABASE_SERVICE_ROLE_KEY come variabile d'ambiente."42 )43 44FAILED_PAYLOADS_FILE = Path(__file__).with_name("failed_supabase_payloads.jsonl")45 46client = SyncPostgrestClient(47 SUPABASE_REST_URL,48 headers={49 "apikey": SUPABASE_KEY,50 "Authorization": f"Bearer {SUPABASE_KEY}"51 }52)53 54# ----------------------------------------------------------------------55# Ensure the `market_data` table exists and has the expected schema.56# ----------------------------------------------------------------------57def _ensure_market_data_table():58 try:59 client.from_('market_data').select('id').limit(1).execute()60 log.info("✅ Connessione a Supabase OK")61 except Exception as e:62 log.error("❌ Table `market_data` does not exist. "63 "Run `market_data.sql` in the Supabase SQL editor "64 "to create it, then restart the scraper.")65 raise SystemExit(1)66 67_ensure_market_data_table()68 69HOTEL_MAP = {70 "Hotel Flora": "https://www.booking.com/hotel/it/flora-milano.it.html",71 "Hotel Ibis Centro": "https://www.booking.com/hotel/it/ibismilanocentromilano.it.html",72 "Hotel Garda": "https://www.booking.com/hotel/it/garda.it.html",73 "Hotel Mennini": "https://www.booking.com/hotel/it/mennini.it.html",74 "Hotel Bernina": "https://www.booking.com/hotel/it/hotelbernina.it.html",75 "Hotel Canova": "https://www.booking.com/hotel/it/canovahotel.it.html",76 "Hotel Mythos": "https://www.booking.com/hotel/it/hotel-mythos.it.html",77 "Hotel Delle Nazioni": "https://www.booking.com/hotel/it/tidellenazioni.it.html",78 "Spice Hotel": "https://www.booking.com/hotel/it/Spice.it.html",79 "Hotel Folen": "https://www.booking.com/hotel/it/folen.it.html",80}81 82# FIX: warning esplicito se GIORNI_DA_ANALIZZARE è rimasto al valore di test.83# Imposta 365 per la produzione completa (365 giorni avanti in chunk da CHUNK_GIORNI).84GIORNI_DA_ANALIZZARE = int(os.getenv("GIORNI_DA_ANALIZZARE", "365"))85CHUNK_GIORNI = int(os.getenv("CHUNK_GIORNI", "5"))86CONCURRENCY = int(os.getenv("CONCURRENCY", "2"))87 88if GIORNI_DA_ANALIZZARE <= 3:89 log.warning(90 "⚠️ GIORNI_DA_ANALIZZARE=%d — valore molto basso, probabilmente rimasto da un test. "91 "Imposta la variabile d'ambiente GIORNI_DA_ANALIZZARE=365 per la produzione.",92 GIORNI_DA_ANALIZZARE,93 )94 95# ── Feature flags ──────────────────────────────────────────────────────────────96SKIP_IF_SCRAPED_TODAY = True97RETRY_SOLDOUTS = True98SOLDOUT_RETRY_DELAY = 60 # secondi prima del secondo tentativo sui sold-out99 100 101def salva_su_db(dati: list[dict]):102 """Upsert a chunk of rows into `market_data`."""103 if not dati:104 return105 106 unsupported_columns: set[str] = set()107 last_error = None108 109 for _ in range(8):110 payload = [_market_data_payload(row, unsupported_columns) for row in dati]111 try:112 client.from_('market_data').upsert(113 payload,114 on_conflict="scrape_date,stay_date,hotel_name,room_type"115 ).execute()116 117 omitted = f" Colonne omesse: {sorted(unsupported_columns)}." if unsupported_columns else ""118 log.info(f"✅ Salvato chunk di {len(payload)} record su Supabase.{omitted}")119 return120 except Exception as e:121 last_error = e122 missing = _missing_column_from_error(e)123 if missing:124 unsupported_columns.add(missing)125 log.warning(126 "⚠️ Colonna `%s` non presente nella schema cache Supabase: "127 "ritento il salvataggio senza quella colonna.", missing,128 )129 continue130 break131 132 log.error(f"❌ Errore caricamento DB: {last_error}")133 if dati:134 log.error(f"Primo record: {dati[0]}")135 _backup_failed_payload(dati, str(last_error))136 137 138def _market_data_payload(row: dict, unsupported_columns: set[str]) -> dict:139 payload = {140 "scrape_timestamp": row.get("scrape_timestamp"),141 "scrape_date": row.get("scrape_date"),142 "scrape_hour": row.get("scrape_hour"),143 "stay_date": row.get("stay_date"),144 "hotel_name": row.get("hotel_name"),145 "room_type": row.get("room_type"),146 "price": row.get("price"),147 "price_verified": bool(row.get("price_verified", False)),148 "is_sold_out": bool(row.get("is_sold_out", False)),149 "currency": row.get("currency", "EUR"),150 "load_timestamp": row.get("scrape_timestamp"),151 }152 return {k: v for k, v in payload.items() if k not in unsupported_columns}153 154 155def _missing_column_from_error(exc: Exception) -> str | None:156 match = re.search(r"Could not find the '([^']+)' column", str(exc))157 return match.group(1) if match else None158 159 160def _backup_failed_payload(dati: list[dict], error: str) -> None:161 try:162 with FAILED_PAYLOADS_FILE.open("a", encoding="utf-8") as f:163 f.write(json.dumps({164 "created_at": datetime.now().isoformat(),165 "error": error,166 "rows": dati,167 }, ensure_ascii=False) + "\n")168 log.warning(f"💾 Payload non salvato archiviato in {FAILED_PAYLOADS_FILE.name}")169 except Exception as backup_error:170 log.error(f"Impossibile creare backup payload Supabase: {backup_error}")171 172 173async def esegui_ciclo():174 now = datetime.now()175 scrape_date = now.strftime("%Y-%m-%d")176 scrape_hour = now.hour177 ts = now.isoformat()178 179 start_date = now + timedelta(days=1)180 date_list = [181 (start_date + timedelta(days=i)).strftime("%Y-%m-%d")182 for i in range(GIORNI_DA_ANALIZZARE)183 ]184 185 # ── Skip-set: combinazioni già scrappate oggi ──────────────────────────────186 already_scraped: set[tuple] = set()187 if SKIP_IF_SCRAPED_TODAY:188 try:189 rows = (190 client.from_('market_data')191 .select("hotel_name,stay_date,room_type")192 .eq("scrape_date", scrape_date)193 .execute()194 )195 for row in (rows.data or []):196 already_scraped.add((row["hotel_name"], row["stay_date"], row["room_type"]))197 log.info(198 f"ℹ️ SKIP_IF_SCRAPED_TODAY: {len(already_scraped)} combinazioni "199 f"già presenti oggi — verranno saltate."200 )201 except Exception as e:202 log.warning(f"⚠️ Impossibile caricare dati già scrapati: {e}")203 204 # ── Accumulatori statistiche per hotel ────────────────────────────────────205 hotel_stats: dict[str, dict] = {206 h: {"total": 0, "found": 0, "verified": 0, "soldout": 0}207 for h in HOTEL_MAP208 }209 global_stats = {"total": 0, "found": 0, "verified": 0, "soldout": 0}210 211 async with async_playwright() as p:212 browser = await p.chromium.launch(213 headless=True,214 args=["--no-sandbox", "--disable-blink-features=AutomationControlled"]215 )216 217 for chunk_start in range(0, len(date_list), CHUNK_GIORNI):218 chunk = date_list[chunk_start:chunk_start + CHUNK_GIORNI]219 results = []220 semaphore = asyncio.Semaphore(CONCURRENCY)221 222 async def scrape_task(d: str, pax: int, h_name: str, h_url: str):223 tipo = "Singola" if pax == 1 else "Doppia"224 225 if (h_name, d, tipo) in already_scraped:226 log.debug(f"⏭ Skip (già scrapato oggi): {h_name} | {d} | {tipo}")227 return228 229 async with semaphore:230 hs = hotel_stats[h_name]231 hs["total"] += 1232 global_stats["total"] += 1233 234 p_val, verified = await scrape_with_antiban(browser, h_name, h_url, d, pax)235 236 if p_val is not None:237 hs["found"] += 1238 global_stats["found"] += 1239 if verified:240 hs["verified"] += 1241 global_stats["verified"] += 1242 243 else:244 hs["soldout"] += 1245 global_stats["soldout"] += 1246 247 # FIX: sia i successi che i sold-out vengono salvati in DB.248 # I sold-out (price=None, is_sold_out=1) tracciano la disponibilità249 # storica e vengono ritentati dal secondo passaggio retry_soldouts().250 results.append({251 "scrape_timestamp": ts,252 "scrape_date": scrape_date,253 "scrape_hour": scrape_hour,254 "stay_date": d,255 "hotel_name": h_name,256 "room_type": tipo,257 "price": p_val,258 "price_verified": verified,259 "is_sold_out": 0 if p_val is not None else 1,260 })261 262 tasks = [263 scrape_task(d, pax, h_name, h_url)264 for d in chunk265 for pax in (1, 2)266 for h_name, h_url in HOTEL_MAP.items()267 ]268 269 await asyncio.gather(*tasks, return_exceptions=True)270 271 if RETRY_SOLDOUTS:272 await retry_soldouts(browser, results, semaphore, HOTEL_MAP, delay=SOLDOUT_RETRY_DELAY)273 274 salva_su_db(results)275 gc.collect()276 277 if chunk_start + CHUNK_GIORNI < len(date_list):278 pausa = random.uniform(8.0, 15.0)279 log.info(f"Pausa inter-chunk di {pausa:.1f}s...")280 await asyncio.sleep(pausa)281 282 await asyncio.sleep(2.0)283 await browser.close()284 285 # ── Report per hotel ───────────────────────────────────────────────────────286 log.info("=" * 65)287 log.info("📊 REPORT PER HOTEL")288 for h_name, hs in hotel_stats.items():289 total = hs["total"]290 found = hs["found"]291 verified = hs["verified"]292 soldout = hs["soldout"]293 pct = (found / max(total, 1)) * 100294 line = (295 f" [{h_name:<25}] "296 f"found={found}/{total} ({pct:.0f}%) "297 f"verified={verified}/{total} "298 f"soldout={soldout}"299 )300 if pct < 50 and total > 0:301 log.warning(line + " ← WARNING")302 else:303 log.info(line)304 305 log.info("-" * 65)306 g = global_stats307 log.info(308 f" TOTALE: found={g['found']}/{g['total']} "309 f"({(g['found'] / max(g['total'], 1)) * 100:.1f}%) "310 f"verified={g['verified']} soldout={g['soldout']}"311 )312 log.info("=" * 65)313 314 315async def main():316 await esegui_ciclo()317 318 319if __name__ == "__main__":320 asyncio.run(main())