""" DAZN EPG Downloader Scarica la guida programmi (EPG) di DAZN Italia per la data corrente. Si aggiorna automaticamente ogni giorno a mezzanotte. """ import sys import io import requests import json import os import time import logging import random import schedule from datetime import datetime, timedelta from pathlib import Path if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') if hasattr(sys.stderr, 'reconfigure'): sys.stderr.reconfigure(encoding='utf-8') # --- Configurazione --- BASE_URL = "https://epg.discovery.indazn.com/eu/v5/epgWithDatesRange" IMAGE_URL_TEMPLATE = ( "https://image.discovery.indazn.com/eu/v3/eu/none/" "{image_id}/fill/none/top/none/100/1280/720/webp/image?brand=dazn" ) PARAMS_TEMPLATE = { "country": "it", "languageCode": "it", "openBrowse": "true", "timeZoneOffset": "120", "brand": "dazn", } OUTPUT_DIR = Path(__file__).parent / "epg_data" LOG_FILE = Path(__file__).parent / "dazn_epg.log" PROXY_FILE = Path(__file__).parent / "proxydaznepg.txt" # --- Logging --- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_FILE, encoding="utf-8"), logging.StreamHandler(sys.stderr), ], ) logger = logging.getLogger(__name__) # --- Campi immagine da convertire --- # Campi a livello di Tile che contengono un singolo oggetto immagine SINGLE_IMAGE_FIELDS = [ "Image", "BackgroundImage", "PromoImage", "LogoImage", "PortraitImage", "HeroImage", ] def build_image_url(image_id: str) -> str: """Costruisce l'URL completo dell'immagine DAZN a partire dall'ID.""" return IMAGE_URL_TEMPLATE.format(image_id=image_id) def convert_image_object(img_obj: dict) -> dict: """ Converte un oggetto immagine: sostituisce il campo 'Id' con il campo 'Url' contenente il link completo. """ if img_obj and isinstance(img_obj, dict) and img_obj.get("Id"): img_obj["Url"] = build_image_url(img_obj["Id"]) return img_obj def replace_image_ids(data: dict) -> tuple[dict, int]: """ Scorre tutti i Tiles dell'EPG e sostituisce gli ID immagine con gli URL completi. Gestisce: - Campi immagine singoli (Image, BackgroundImage, ecc.) - Liste di immagini dentro Competition, Sport, ecc. Restituisce (data_modificato, numero_sostituzioni). """ count = 0 tiles = data.get("Tiles", []) for tile in tiles: # 1) Campi immagine singoli a livello di tile for field in SINGLE_IMAGE_FIELDS: img = tile.get(field) if img and isinstance(img, dict) and img.get("Id"): convert_image_object(img) count += 1 # 2) Oggetti annidati con lista "Images" (Competition, Sport, ecc.) for nested_key in ["Competition", "Sport", "TournamentCalendar"]: nested = tile.get(nested_key) if nested and isinstance(nested, dict): images_list = nested.get("Images", []) if isinstance(images_list, list): for img in images_list: if img and isinstance(img, dict) and img.get("Id"): convert_image_object(img) count += 1 # 3) LinearSchedule - potrebbe avere immagini linear = tile.get("LinearSchedule") if linear and isinstance(linear, list): for item in linear: if isinstance(item, dict): for field in SINGLE_IMAGE_FIELDS: img = item.get(field) if img and isinstance(img, dict) and img.get("Id"): convert_image_object(img) count += 1 return data, count def load_proxies() -> list[str]: """ Carica la lista dei proxy dal file proxydaznepg.txt. Ogni riga deve contenere un proxy nel formato: ip:porta http://ip:porta https://ip:porta socks5://ip:porta Le righe vuote e quelle che iniziano con # vengono ignorate. """ if not PROXY_FILE.exists(): logger.warning(f"[PROXY] File proxy non trovato: {PROXY_FILE}") return [] proxies = [] with open(PROXY_FILE, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): proxies.append(line) logger.info(f"[PROXY] Caricati {len(proxies)} proxy da {PROXY_FILE.name}") return proxies def _build_proxy_dict(proxy_str: str) -> dict: """ Costruisce il dizionario proxy per requests. Supporta i formati: - host:porta - host:porta:username:password - http://host:porta - http://user:pass@host:porta """ if "://" not in proxy_str: # Controlla se รจ formato host:porta:user:pass parts = proxy_str.split(":") if len(parts) == 4: host, port, user, password = parts proxy_str = f"http://{user}:{password}@{host}:{port}" elif len(parts) == 2: proxy_str = f"http://{proxy_str}" else: # Fallback: tratta come URL diretto proxy_str = f"http://{proxy_str}" return { "http": proxy_str, "https": proxy_str, } def download_epg(target_date: datetime | None = None) -> Path | None: """ Scarica l'EPG di DAZN per la data specificata (default: oggi). Usa i proxy dal file proxydaznepg.txt, ruotando tra di essi. Se il download fallisce, riprova con un altro proxy finche' non riesce. Salva il risultato in un file JSON nella cartella epg_data/. Restituisce il percorso del file salvato o None in caso di errore. """ if target_date is None: target_date = datetime.now() date_str = target_date.strftime("%Y-%m-%d") end_date = target_date + timedelta(days=1) end_date_str = end_date.strftime("%Y-%m-%d") params = { **PARAMS_TEMPLATE, "startDate": date_str, "endDate": end_date_str, } logger.info(f"[>] Scaricamento EPG per {date_str} ...") # Carica i proxy proxy_list = load_proxies() if not proxy_list: logger.error("[ERRORE] Nessun proxy disponibile in proxydaznepg.txt. Impossibile scaricare.") return None # Mischia l'ordine dei proxy per distribuire il carico random.shuffle(proxy_list) attempt = 0 while True: proxy_str = proxy_list[attempt % len(proxy_list)] proxy_dict = _build_proxy_dict(proxy_str) attempt += 1 # Ogni giro completo della lista, rimescola if attempt > 1 and attempt % len(proxy_list) == 1: random.shuffle(proxy_list) logger.info("[PROXY] Completato un giro di tutti i proxy, rimescolo e riprovo...") logger.info(f"[PROXY] Tentativo #{attempt} con proxy: {proxy_str}") try: response = requests.get( BASE_URL, params=params, proxies=proxy_dict, timeout=30 ) response.raise_for_status() data = response.json() # Sostituisci gli ID immagine con URL completi data, img_count = replace_image_ids(data) logger.info(f"[IMG] Sostituite {img_count} immagini con URL completi") # Crea la cartella di output se non esiste OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # Salva il JSON filename = f"dazn_epg_{date_str}.json" filepath = OUTPUT_DIR / filename with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) file_size_kb = filepath.stat().st_size / 1024 logger.info(f"[OK] EPG salvato: {filepath} ({file_size_kb:.1f} KB) [proxy: {proxy_str}]") # Stampa un riepilogo dei canali/eventi trovati print_summary(data, date_str) return filepath except requests.exceptions.HTTPError as e: logger.warning(f"[RETRY] HTTP error con proxy {proxy_str}: {e}") except requests.exceptions.ConnectionError as e: logger.warning(f"[RETRY] Connessione fallita con proxy {proxy_str}: {e}") except requests.exceptions.Timeout: logger.warning(f"[RETRY] Timeout con proxy {proxy_str}") except requests.exceptions.ProxyError as e: logger.warning(f"[RETRY] Errore proxy {proxy_str}: {e}") except requests.exceptions.RequestException as e: logger.warning(f"[RETRY] Errore richiesta con proxy {proxy_str}: {e}") except json.JSONDecodeError: logger.warning(f"[RETRY] Risposta non JSON valida con proxy {proxy_str}") # Pausa tra i tentativi per non sovraccaricare wait_time = min(5 + (attempt // len(proxy_list)) * 2, 30) logger.info(f"[RETRY] Attesa {wait_time}s prima del prossimo tentativo...") time.sleep(wait_time) def print_summary(data: dict | list, date_str: str): """Stampa un riepilogo dei contenuti EPG scaricati.""" print(f"\n{'='*60}") print(f" DAZN EPG - {date_str}") print(f"{'='*60}") if isinstance(data, dict): # Prova a estrarre info in base alla struttura dell'API entries = data.get("entries", data.get("data", data.get("events", []))) if isinstance(entries, list): print(f" Totale eventi/voci: {len(entries)}") for i, entry in enumerate(entries[:10]): title = ( entry.get("title") or entry.get("name") or entry.get("eventName") or entry.get("label") or "-" ) start = entry.get("startDate") or entry.get("start") or "" print(f" {i+1:>3}. {title} {start}") if len(entries) > 10: print(f" ... e altri {len(entries) - 10} eventi") else: print(f" Chiavi nel JSON: {list(data.keys())}") elif isinstance(data, list): print(f" Totale elementi: {len(data)}") print(f"{'='*60}\n") def cleanup_old_files(keep_days: int = 7): """Rimuove i file EPG piu' vecchi di keep_days giorni.""" if not OUTPUT_DIR.exists(): return cutoff = datetime.now() - timedelta(days=keep_days) removed = 0 for f in OUTPUT_DIR.glob("dazn_epg_*.json"): try: # Estrai la data dal nome del file date_part = f.stem.replace("dazn_epg_", "") file_date = datetime.strptime(date_part, "%Y-%m-%d") if file_date < cutoff: f.unlink() removed += 1 logger.info(f"[PULIZIA] Rimosso file vecchio: {f.name}") except (ValueError, OSError): continue if removed: logger.info(f"[PULIZIA] Completata: {removed} file rimossi") def daily_job(): """Job giornaliero: scarica EPG e pulisce i file vecchi.""" logger.info("[TIMER] Avvio job giornaliero...") download_epg() cleanup_old_files(keep_days=7) logger.info("[TIMER] Job giornaliero completato.") def main(): print("") print(" ====================================") print(" DAZN EPG Downloader - Italia") print(" Scarica automaticamente la guida") print(" programmi ogni giorno") print(" ====================================") print("") # Primo download immediato logger.info("[AVVIO] Primo download...") download_epg() cleanup_old_files(keep_days=7) # Schedula il download giornaliero a mezzanotte schedule.every().day.at("00:05").do(daily_job) logger.info("[SCHEDULE] Aggiornamento giornaliero alle 00:05") logger.info("[INFO] Premi Ctrl+C per uscire.\n") try: while True: schedule.run_pending() time.sleep(60) # Controlla ogni minuto except KeyboardInterrupt: logger.info("\n[STOP] Arresto richiesto dall'utente. Ciao!") if __name__ == "__main__": main()