""" DAZN EPG Merger Combina i dati dell'EPG ufficiale DAZN con i canali estratti via OCR dal flusso live. Usa fuzzy matching per associare i titoli OCR (spesso incompleti/sbagliati) con quelli originali DAZN. Output: EPG finale con Nome, Immagine, Canale, Ora inizio/fine. """ import sys import io import json import re import unicodedata import logging from datetime import datetime, timezone, timedelta from zoneinfo import ZoneInfo from difflib import SequenceMatcher from pathlib import Path # Fuso orario italiano IT_TZ = ZoneInfo("Europe/Rome") # Fix encoding per Windows if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') if hasattr(sys.stderr, 'reconfigure'): sys.stderr.reconfigure(encoding='utf-8') # --- Path --- BASE_DIR = Path(__file__).parent EPG_DIR = BASE_DIR / "epg_data" LOG_FILE = BASE_DIR / "dazn_merger.log" # --- 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__) # Soglia minima di similarita' per accettare un match (0.0 - 1.0) MATCH_THRESHOLD = 0.45 def utc_to_italian(utc_str: str) -> str: """ Converte un orario UTC (es. '2026-05-16T15:00:00Z') in ora italiana. Gestisce automaticamente CET (UTC+1) e CEST (UTC+2) grazie a zoneinfo. Restituisce formato ISO con offset, es. '2026-05-16T17:00:00+02:00'. """ if not utc_str: return "" try: # Parse UTC dt_utc = datetime.fromisoformat(utc_str.replace("Z", "+00:00")) # Converti a ora italiana dt_it = dt_utc.astimezone(IT_TZ) return dt_it.isoformat() except (ValueError, TypeError): return utc_str # Restituisci l'originale se non parsabile def normalize_text(text: str) -> str: """ Normalizza il testo per il confronto fuzzy: - Rimuove accenti (Nürburgring -> Nurburgring) - Lowercase - Rimuove punteggiatura e caratteri speciali - Rimuove spazi multipli """ # Rimuovi accenti: Nürburgring -> Nurburgring, 1ª -> 1a text = unicodedata.normalize("NFKD", text) text = "".join(c for c in text if not unicodedata.combining(c)) # Lowercase text = text.lower() # Rimuovi punteggiatura comune ma mantieni numeri e lettere text = re.sub(r"[|,.\-:;!?'\"()\[\]{}]", " ", text) # Rimuovi "..." e ".." text = re.sub(r"\.{2,}", " ", text) # Normalizza spazi text = re.sub(r"\s+", " ", text).strip() return text def compute_similarity(ocr_title: str, epg_title: str) -> float: """ Calcola la similarita' tra un titolo OCR e uno dell'EPG. Usa una combinazione di: 1. SequenceMatcher ratio (match globale) 2. Partial matching (il titolo OCR e' spesso troncato) 3. Bonus per parole chiave in comune """ norm_ocr = normalize_text(ocr_title) norm_epg = normalize_text(epg_title) if not norm_ocr or not norm_epg: return 0.0 # 1. Ratio globale global_ratio = SequenceMatcher(None, norm_ocr, norm_epg).ratio() # 2. Partial match: il titolo OCR potrebbe essere troncato, # quindi verifica se norm_ocr e' un sottoinsieme di norm_epg # Usa la lunghezza dell'OCR come finestra scorrevole sull'EPG if len(norm_ocr) <= len(norm_epg): best_partial = 0.0 ocr_len = len(norm_ocr) for i in range(len(norm_epg) - ocr_len + 1): window = norm_epg[i : i + ocr_len] ratio = SequenceMatcher(None, norm_ocr, window).ratio() if ratio > best_partial: best_partial = ratio partial_ratio = best_partial else: partial_ratio = global_ratio # 3. Word overlap bonus ocr_words = set(norm_ocr.split()) epg_words = set(norm_epg.split()) # Rimuovi parole troppo corte (1-2 char) che danno falsi positivi ocr_words = {w for w in ocr_words if len(w) > 2} epg_words = {w for w in epg_words if len(w) > 2} if ocr_words: word_overlap = len(ocr_words & epg_words) / len(ocr_words) else: word_overlap = 0.0 # Score finale: media pesata # Partial match ha piu' peso perche' i titoli OCR sono troncati score = (global_ratio * 0.25) + (partial_ratio * 0.45) + (word_overlap * 0.30) return score def find_best_match( ocr_event: dict, epg_tiles: list[dict], ) -> tuple[dict | None, float]: """ Trova il miglior tile EPG che corrisponde all'evento OCR. Restituisce (tile, score) o (None, 0.0) se nessun match valido. """ ocr_title = ocr_event.get("title", "") if not ocr_title or len(ocr_title) < 3: return None, 0.0 best_tile = None best_score = 0.0 for tile in epg_tiles: epg_title = tile.get("Title", "") if not epg_title: continue score = compute_similarity(ocr_title, epg_title) if score > best_score: best_score = score best_tile = tile if best_score >= MATCH_THRESHOLD: return best_tile, best_score else: return None, best_score def load_dazn_epg() -> dict | None: """Carica l'EPG DAZN piu' recente.""" today = datetime.now().strftime("%Y-%m-%d") epg_path = EPG_DIR / f"dazn_epg_{today}.json" if not epg_path.exists(): # Prova il file piu' recente files = sorted(EPG_DIR.glob("dazn_epg_????-??-??.json")) if files: epg_path = files[-1] else: logger.error("[MERGER] Nessun file EPG DAZN trovato!") return None logger.info(f"[EPG] Caricato: {epg_path.name}") with open(epg_path, "r", encoding="utf-8") as f: return json.load(f) def load_live_events() -> dict | None: """Carica gli eventi live estratti via OCR.""" live_path = EPG_DIR / "dazn_live_events.json" if not live_path.exists(): logger.error("[MERGER] File dazn_live_events.json non trovato!") return None with open(live_path, "r", encoding="utf-8") as f: return json.load(f) def get_image_url(tile: dict) -> str: """Estrae l'URL immagine migliore dal tile.""" # Priorita': Image > HeroImage > PromoImage > BackgroundImage for field in ["Image", "HeroImage", "PromoImage", "BackgroundImage"]: img = tile.get(field) if img and isinstance(img, dict) and img.get("Url"): return img["Url"] # Prova nelle immagini Competition comp = tile.get("Competition", {}) if comp and isinstance(comp, dict): images = comp.get("Images", []) for img in images: if img and img.get("Url"): return img["Url"] return "" def load_previous_merged() -> dict | None: """Carica l'EPG unito precedente come fallback.""" merged_path = EPG_DIR / "dazn_merged_epg.json" if merged_path.exists(): try: with open(merged_path, "r", encoding="utf-8") as f: return json.load(f) except Exception: return None return None def merge_epg() -> list[dict] | None: """ Pipeline principale: carica EPG + OCR, fa il matching, produce l'EPG finale. """ # 1. Carica i dati epg_data = load_dazn_epg() if not epg_data: return None live_data = load_live_events() if not live_data: return None tiles = epg_data.get("Tiles", []) ocr_events = live_data.get("events", []) prev_merged_data = load_previous_merged() prev_events = prev_merged_data.get("events", []) if prev_merged_data else [] logger.info(f"[MERGER] EPG DAZN: {len(tiles)} tiles, OCR: {len(ocr_events)} eventi") # 2. Filtra i tiles solo a quelli attualmente in onda # (Start <= now <= End) per ridurre falsi positivi now = datetime.now(timezone.utc) live_tiles = [] for tile in tiles: try: start_str = tile.get("Start", "") end_str = tile.get("End", "") if not start_str: continue start = datetime.fromisoformat(start_str.replace("Z", "+00:00")) if not end_str: # Se l'orario di fine manca, assumiamo una durata di 4 ore end = start + timedelta(hours=4) else: end = datetime.fromisoformat(end_str.replace("Z", "+00:00")) # Margine di 2 ore prima, 4 ore dopo per coprire post-partita if start - timedelta(hours=2) <= now <= end + timedelta(hours=4): live_tiles.append(tile) except (ValueError, TypeError): continue logger.info(f"[MERGER] Tiles attualmente in onda (con margine): {len(live_tiles)}") # Se pochi tile live, usa tutti i tile come fallback search_tiles = live_tiles if len(live_tiles) >= len(ocr_events) else tiles # Filtra i canali lineari permanenti 24/7 (End = anno 3000) # Es: Radio TV Serie A, Eurosport 1, Red Bull TV — non sono eventi reali search_tiles = [t for t in search_tiles if "3000-" not in (t.get("End") or "")] logger.info(f"[MERGER] Tiles dopo filtro canali permanenti: {len(search_tiles)}") # 3. Matching merged_events = [] used_tiles = set() # Evita duplicati for ocr_ev in ocr_events: # Check speciale per "Zona Serie A" ocr_title_lower = ocr_ev.get("title", "").lower() if "zona serie a" in ocr_title_lower: ocr_ev["channel"] = "ZonaSerieA" ocr_ev["channel_number"] = 0 best_tile, score = find_best_match(ocr_ev, search_tiles) # Se troviamo un match, assicuriamoci di forzare ZonaSerieA se il tile ufficiale è Zona Serie A if best_tile and "zona serie a" in best_tile["Title"].lower(): ocr_ev["channel"] = "ZonaSerieA" ocr_ev["channel_number"] = 0 if best_tile: tile_id = best_tile.get("Id", "") # Evita di assegnare lo stesso tile a piu' eventi if tile_id in used_tiles: # Riprova senza questo tile remaining = [t for t in search_tiles if t.get("Id") != tile_id] best_tile, score = find_best_match(ocr_ev, remaining) if best_tile: tile_id = best_tile.get("Id", "") if best_tile and tile_id not in used_tiles: used_tiles.add(tile_id) contestants = best_tile.get("Contestants", []) is_events_channel = "events" in ocr_ev.get("channel", "").lower() if len(contestants) == 2 and not is_events_channel: # Partita con canali dedicati per squadra (es. Serie A) # Un unico evento con channel (squadra 1) e channel2 (squadra 2) event_entry = { "title": best_tile["Title"], "image": get_image_url(best_tile), "channel": contestants[0].get("Title", ""), "channel2": contestants[1].get("Title", ""), "channel_number": ocr_ev.get("channel_number", 0), "start": utc_to_italian(best_tile.get("Start", "")), "end": utc_to_italian(best_tile.get("End", "")), "match_score": round(score, 3), "ocr_title": ocr_ev.get("title", ""), } merged_events.append(event_entry) logger.info( f"[MATCH] {score:.2f} | OCR: \"{ocr_ev['title']}\" " f"-> EPG: \"{best_tile['Title']}\" | " f"CH: {contestants[0].get('Title','')}, {contestants[1].get('Title','')}" ) else: # Evento singolo (Events X, ZonaSerieA, ecc.) event_entry = { "title": best_tile["Title"], "image": get_image_url(best_tile), "channel": ocr_ev.get("channel", ""), "channel_number": ocr_ev.get("channel_number", 0), "start": utc_to_italian(best_tile.get("Start", "")), "end": utc_to_italian(best_tile.get("End", "")), "match_score": round(score, 3), "ocr_title": ocr_ev.get("title", ""), } merged_events.append(event_entry) logger.info( f"[MATCH] {score:.2f} | OCR: \"{ocr_ev['title']}\" " f"-> EPG: \"{best_tile['Title']}\" | {ocr_ev.get('channel','')}" ) continue # Se arriviamo qui, NON c'è un match valido (o per soglia troppo bassa o mancanza di tile). # Fallback: usiamo l'evento precedente se coincide con questo canale ed è in onda. fallback_event = None chan = ocr_ev.get("channel") if chan: for prev_ev in prev_events: if prev_ev.get("channel") == chan: start_str = prev_ev.get("start") end_str = prev_ev.get("end") if start_str and end_str: try: # Controlla se è ancora in onda (con 15 min di margine di tolleranza) st = datetime.fromisoformat(start_str) en = datetime.fromisoformat(end_str) now_local = datetime.now(st.tzinfo) if st - timedelta(minutes=15) <= now_local <= en + timedelta(minutes=15): fallback_event = prev_ev break except ValueError: pass if fallback_event: logger.info(f"[CACHE] Recuperato da scansione precedente per {chan}: {fallback_event['title']}") merged_events.append({ "title": fallback_event.get("title"), "image": fallback_event.get("image"), "channel": ocr_ev.get("channel", ""), "channel_number": ocr_ev.get("channel_number", 0), "start": fallback_event.get("start"), "end": fallback_event.get("end"), "match_score": fallback_event.get("match_score", 1.0), "ocr_title": ocr_ev.get("title", ""), }) else: logger.warning( f"[NO MATCH] score={score:.2f} | OCR: \"{ocr_ev.get('title','')}\" - nessun fallback disponibile" ) merged_events.append({ "title": ocr_ev.get("title", "?"), "image": "", "channel": ocr_ev.get("channel", ""), "channel_number": ocr_ev.get("channel_number", 0), "start": "", "end": "", "match_score": 0.0, "ocr_title": ocr_ev.get("title", ""), }) # 4. Ordina per numero canale merged_events.sort(key=lambda e: e.get("channel_number", 0)) # 5. Salva il risultato finale output = { "timestamp": datetime.now().isoformat(), "source": "dazn_merged_epg", "total_events": len(merged_events), "events": merged_events, } output_path = EPG_DIR / "dazn_merged_epg.json" with open(output_path, "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=2) logger.info(f"[MERGER] EPG finale salvato: {output_path}") # Stampa riepilogo print(f"\n{'='*70}") print(f" DAZN EPG MERGED - {datetime.now().strftime('%d/%m/%Y %H:%M')}") print(f" {len(merged_events)} eventi associati") print(f"{'='*70}") print(f" {'CH':<12} {'INIZIO':>5}-{'FINE':<5} {'SCORE':>5} EVENTO") print(f" {'-'*64}") for ev in merged_events: ch = ev.get("channel", "?") start = ev.get("start", "") end = ev.get("end", "") # Formatta orari HH:MM (gia' in ora italiana) start_hm = "" end_hm = "" if start: try: dt = datetime.fromisoformat(start) start_hm = dt.strftime("%H:%M") except ValueError: start_hm = start[:5] if end: try: dt = datetime.fromisoformat(end) end_hm = dt.strftime("%H:%M") except ValueError: end_hm = end[:5] score = ev.get("match_score", 0) title = ev.get("title", "?") has_img = " [IMG]" if ev.get("image") else "" print(f" {ch:<12} {start_hm:>5}-{end_hm:<5} {score:>5.2f} {title}{has_img}") print(f"{'='*70}\n") return merged_events def main(): print("") print(" ====================================") print(" DAZN EPG Merger") print(" Associa eventi OCR all'EPG DAZN") print(" ====================================") print("") result = merge_epg() if result: logger.info(f"[DONE] Merge completato: {len(result)} eventi") else: logger.error("[DONE] Merge fallito") if __name__ == "__main__": main()