#!/usr/bin/env python3 """ ==================================================================== DAZN - Manager & DRM Extractor (Widevine WVD) Auto-Extractor: Live & Linear Streams (DAZN 1, Plus, ecc.) Output: dazn_channels.json & dazn_playlist.m3u ==================================================================== """ import os import sys import re import json import time import base64 import uuid import hashlib import argparse import xml.etree.ElementTree as ET from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional, Tuple, Any from urllib.parse import urlparse, parse_qs, urlencode, urlunparse, unquote import socket import subprocess import shutil import atexit # Prova ad importare curl_cffi per impersonazione TLS Chrome (bypassa CloudFront/WAF) try: from curl_cffi import requests as cffi_requests HAS_CURL_CFFI = True except ImportError: import requests as cffi_requests HAS_CURL_CFFI = False import requests from pywidevine import PSSH, Cdm, Device # Garantisci supporto UTF-8 su terminali Windows if sys.platform == 'win32': try: sys.stdout.reconfigure(encoding='utf-8', errors='replace') sys.stderr.reconfigure(encoding='utf-8', errors='replace') except Exception: pass # ── Colori ANSI per il terminale ────────────────────────────────── GREEN = '\033[32m' YELLOW = '\033[33m' RED = '\033[31m' CYAN = '\033[36m' BLUE = '\033[34m' MAGENTA = '\033[35m' BOLD = '\033[1m' DIM = '\033[2m' RESET = '\033[0m' # ── Costanti & File di Sistema (Percorsi assoluti) ───────────────── SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_FILE = os.path.join(SCRIPT_DIR, "dazn_config.json") TOKEN_FILE = os.path.join(SCRIPT_DIR, "dazn_tokens.json") CACHE_FILE = os.path.join(SCRIPT_DIR, "pssh_cache.json") WVD_FILE = os.path.join(SCRIPT_DIR, "WVD.wvd") PROXY_FILE = os.path.join(SCRIPT_DIR, "proxy.txt") OUTPUT_JSON_FILE = os.path.join(SCRIPT_DIR, "dazn_channels.json") PROXY_DATA_FILE = os.path.join(SCRIPT_DIR, "dazn_proxy_data.json") PLAYLIST_FILE = os.path.join(SCRIPT_DIR, "dazn_playlist.m3u") PLAYLIST_PROXY_FILE = os.path.join(SCRIPT_DIR, "dazn_playlist_proxy.m3u") PROXY_SCRIPT_FILE= os.path.join(SCRIPT_DIR, "dazn_proxy.py") SERIEA_MPD_FILE = os.path.join(SCRIPT_DIR, "radio_tv_seriea.mpd") # ── Set globale degli asset bloccati per Eligibility (non ritentare) ── _eligibility_blocked_ids: set = set() # Radio TV Serie A (Lega / Deltatre): stesso lineare, niente DRM DAZN SERIEA_RADIO_ASSET_ID = "1daica1elq2701s7zajjchy47s" SERIEA_RADIO_VIDEO_ID = "661803d9-7d20-4569-9870-b0add6708f1e" SERIEA_RADIO_VIDEO_SOURCE = ( "https://legaseriea-live-prd.cdn.netrw.it/out/v1/s-prd/ch-01-prd-s-v2/" "dash-ch-01-prd-s-hd/index.mpd" ) SERIEA_ENTITLEMENT_URL = "https://product-services-cdn.deltatre.digital/seriea/entitlement/api/v2/main/videos" SERIEA_HEADERS = { "Content-Type": "application/json", "Accept": "application/json, text/plain, */*", "Origin": "https://www.legaseriea.it", "Referer": "https://www.legaseriea.it/", "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36" ), } # Discovery principale: Rail Router (eventi live + palinsesto lineari). EPG solo fallback. RAIL_LIVE_URL = ( "https://rail-router.discovery.indazn.com/eu/v11/Rail" "?platform=web&id=LiveAndNextNew&country=it&brand=dazn&languageCode=it" "¶ms=PageType:Home%3BContentType:None" ) RAIL_LINEAR_URL = ( "https://rail-router.discovery.indazn.com/eu/v11/Rail" "?platform=web&id=Livetvschedule&country=it&brand=dazn&languageCode=it" ) RAIL_HEADERS = { "accept": "application/json, text/plain, */*", "origin": "https://www.dazn.com", "referer": "https://www.dazn.com/", "user-agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36" ), } DAZN_IMAGE_URL = "https://image.discovery.indazn.com/eu/v2/eu/image/?id=" # Lineari 24/7 da tenere oltre Radio TV Serie A (locandina = image-header) EXTRA_LINEAR_CHANNELS = [ { "id": "4hqcjx5zv91f1bdx71io4bnrl", "name": "DAZN 1", "logo": DAZN_IMAGE_URL + "4hqcjx5zv91f1bdx71io4bnrl_image-header_pIt_1788259786000", }, { "id": "1jdysh48i12eb1cdny7b2dgma5", "name": "Inter TV", "logo": DAZN_IMAGE_URL + "1jdysh48i12eb1cdny7b2dgma5_image-header_pIt_1750150731000", }, { "id": "xkl4tjojz1s7zrvhrly6sy2n", "name": "Milan TV", "logo": DAZN_IMAGE_URL + "xkl4tjojz1s7zrvhrly6sy2n_image-header_pIt_1661948656000", }, { "id": "vowt32ah7gwx1pz824i5ntrwz", "name": "Red Bull TV", "logo": DAZN_IMAGE_URL + "vowt32ah7gwx1pz824i5ntrwz_image-header_pIt_1712069226000", }, { "id": "19oeo2wgzq1vj1rpi9oh86yhbh", "name": "Unbeaten", "logo": DAZN_IMAGE_URL + "19oeo2wgzq1vj1rpi9oh86yhbh_image-header_pIt_1711552796000", }, ] EXTRA_LINEAR_TITLES = {c["name"].strip().lower() for c in EXTRA_LINEAR_CHANNELS} EXTRA_LINEAR_TITLES.add("redbull tv") EXTRA_LINEAR_IDS = {c["id"] for c in EXTRA_LINEAR_CHANNELS} # ── Gestione Processo Proxy Integrato ───────────────────────────── _PROXY_PROCESS = None def is_port_open(port: int, host: str = "127.0.0.1") -> bool: try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(0.5) return s.connect_ex((host, port)) == 0 except Exception: return False def ensure_proxy_running(port: int = 8443) -> None: global _PROXY_PROCESS if is_port_open(port): log(f"📡 DAZN Proxy già attivo e in ascolto sulla porta {port}.", GREEN) return if not os.path.exists(PROXY_SCRIPT_FILE): log(f"⚠️ Script proxy non trovato in {PROXY_SCRIPT_FILE}", YELLOW) return log(f"🚀 Avvio automatico DAZN HTTPS Proxy sulla porta {port}...", CYAN) try: _PROXY_PROCESS = subprocess.Popen( [sys.executable, PROXY_SCRIPT_FILE], cwd=SCRIPT_DIR ) log(f"✅ DAZN Proxy avviato in background (PID: {_PROXY_PROCESS.pid})", GREEN) except Exception as e: log(f"❌ Impossibile avviare dazn_proxy.py: {e}", RED) def stop_proxy() -> None: global _PROXY_PROCESS if _PROXY_PROCESS and _PROXY_PROCESS.poll() is None: log("🛑 Arresto DAZN Proxy...", YELLOW) try: _PROXY_PROCESS.terminate() _PROXY_PROCESS.wait(timeout=2) except Exception: try: _PROXY_PROCESS.kill() except Exception: pass atexit.register(stop_proxy) # Intervallo di aggiornamento predefinito: 3 ore (in secondi) DEFAULT_REFRESH_SECONDS = 3 * 3600 # System ID ufficiale Widevine WIDEVINE_SYSTEM_ID = "edef8ba9-79d6-4ace-a3c8-27dcd51d21ed" WIDEVINE_SYSTEM_ID_BYTES = bytes.fromhex("edef8ba979d64acea3c827dcd51d21ed") pssh_cache: Dict[str, Any] = {} # ── Helper Logging & Banner ─────────────────────────────────────── def log(msg: str, color: str = RESET) -> None: print(f"{color}[{datetime.now().strftime('%H:%M:%S')}] {msg}{RESET}", flush=True) def print_banner() -> None: banner = f"""{YELLOW} ____ _ ______ _ _ | _ \\ / \\ |__ / \\| \\ | | | | | | / _ \\ / /| \\| \\| | | |_| |/ ___ \\ / /_| |\\ \\ | |____//_/ \\_\\/____|_| \\_| {BLUE}{'═' * 70}{RESET} {CYAN} ◆ DAZN Auto-Manager · Widevine Extractor & Decryptor ◆{RESET} {BLUE}{'═' * 70}{RESET}""" print(banner) # ── Gestione Proxy (proxy.txt o dazn_config.json) ───────────────── class ProxyManager: """Carica, analizza, valida e ruota i proxy da proxy.txt o da dazn_config.json""" def __init__(self, filepath: str = PROXY_FILE): self.filepath = filepath self.proxies: List[str] = [] self.config_proxy: Optional[Dict[str, str]] = None self.current_idx = 0 self.load() def parse_proxy_line(self, line: str) -> Optional[str]: line = line.strip() if not line or line.startswith('#'): return None if line.startswith(('http://', 'https://', 'socks5://', 'socks5h://')): return line parts = line.split(':') if len(parts) == 4: host, port, user, pwd = parts return f"http://{user}:{pwd}@{host}:{port}" elif len(parts) == 2: return f"http://{parts[0]}:{parts[1]}" elif '@' in line: return f"http://{line}" return None def set_config_proxy(self, proxy_conf: Any) -> None: if isinstance(proxy_conf, dict): http_p = proxy_conf.get('http') or proxy_conf.get('https') https_p = proxy_conf.get('https') or proxy_conf.get('http') if http_p: self.config_proxy = {"http": http_p, "https": https_p or http_p} elif isinstance(proxy_conf, str) and proxy_conf.strip(): parsed = self.parse_proxy_line(proxy_conf.strip()) or proxy_conf.strip() self.config_proxy = {"http": parsed, "https": parsed} def load(self, config_proxy: Any = None) -> None: self.proxies = [] if config_proxy is not None: self.set_config_proxy(config_proxy) if os.path.exists(self.filepath): try: with open(self.filepath, 'r', encoding='utf-8', errors='ignore') as f: for line in f: parsed = self.parse_proxy_line(line) if parsed and parsed not in self.proxies: self.proxies.append(parsed) except Exception: pass self.current_idx = 0 def get_dict(self) -> Optional[Dict[str, str]]: if self.proxies: proxy_url = self.proxies[self.current_idx % len(self.proxies)] return {"http": proxy_url, "https": proxy_url} if self.config_proxy: return self.config_proxy return None def rotate(self) -> Optional[Dict[str, str]]: if self.proxies: self.current_idx = (self.current_idx + 1) % len(self.proxies) return self.get_dict() def count(self) -> int: return len(self.proxies) if self.proxies else (1 if self.config_proxy else 0) def test_all(self) -> List[Tuple[str, bool, str, float]]: results = [] list_to_test = list(self.proxies) if not list_to_test and self.config_proxy: list_to_test.append(self.config_proxy['http']) for p in list_to_test: proxies_dict = {"http": p, "https": p} t0 = time.time() try: r = requests.get('https://api.ipify.org?format=json', proxies=proxies_dict, timeout=10) dur = round((time.time() - t0) * 1000, 1) if r.ok: ip = r.json().get('ip', 'OK') results.append((p, True, ip, dur)) else: results.append((p, False, f"HTTP {r.status_code}", dur)) except Exception as e: dur = round((time.time() - t0) * 1000, 1) results.append((p, False, str(e)[:30], dur)) return results proxy_manager = ProxyManager(PROXY_FILE) # ── Gestione Cache PSSH & Configurazione ─────────────────────────── def load_config() -> Dict[str, Any]: default_config = { "dazn": { "token": "", "email": "", "password": "", "device_id": "57b6a530-70a4-4d65-b412-46a1a5c22b38", "daznid": "66125bf7-0b92-47fc-b618-93cfea67f55c", "app_version": "0.150.1-hotfix.f8b078689", "player_id": "@dazn/peng-html5-core/web/web", "language": "it", "proxy": { "http": "", "https": "" }, "headers": { "referer": "https://www.dazn.com/", "sec-ch-ua": '"Chromium";v="154", "Microsoft Edge";v="154", "Not A(Brand";v="99"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/154.0.0.0 Safari/537.36 Edg/154.0.0.0", "x-brand": "DAZN" }, "license_headers": { "content-type": "application/octet-stream", "referer": "https://www.dazn.com/", "sec-ch-ua": '"Chromium";v="154", "Microsoft Edge";v="154", "Not A(Brand";v="99"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/154.0.0.0 Safari/537.36 Edg/154.0.0.0", "x-brand": "DAZN" }, "channels": [dict(c) for c in EXTRA_LINEAR_CHANNELS] }, "loop": { "refresh_interval_hours": 3 } } if not os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(default_config, f, indent=2, ensure_ascii=False) cfg = default_config else: try: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: cfg = json.load(f) if 'dazn' not in cfg: cfg['dazn'] = default_config['dazn'] except Exception: cfg = default_config if 'dazn' in cfg and 'proxy' in cfg['dazn']: proxy_manager.set_config_proxy(cfg['dazn']['proxy']) return cfg def save_config(cfg: Dict[str, Any]) -> None: try: current = {} if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'r', encoding='utf-8') as f: current = json.load(f) current.update(cfg) with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(current, f, indent=2, ensure_ascii=False) except Exception as e: log(f"Errore salvataggio config: {e}", RED) def load_cache() -> None: global pssh_cache if os.path.exists(CACHE_FILE): try: with open(CACHE_FILE, 'r', encoding='utf-8') as f: pssh_cache = json.load(f) except Exception: pssh_cache = {} def save_cache() -> None: with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(pssh_cache, f, indent=2, ensure_ascii=False) # ── Gestione Token JWT DAZN ─────────────────────────────────────── def decode_jwt_payload(token: str) -> Dict[str, Any]: """Decodifica il payload di un JWT DAZN senza richiedere librerie esterne""" try: parts = token.strip().split('.') if len(parts) != 3: return {} payload_b64 = parts[1] rem = len(payload_b64) % 4 if rem > 0: payload_b64 += '=' * (4 - rem) decoded = base64.urlsafe_b64decode(payload_b64) return json.loads(decoded.decode('utf-8')) except Exception: return {} def check_token_status(token: str) -> Tuple[bool, int, Dict[str, Any]]: """Verifica validità e scadenza del JWT DAZN""" payload = decode_jwt_payload(token) if not payload: return False, 0, {} exp = payload.get('exp', 0) now = int(time.time()) remaining = exp - now is_valid = remaining > 60 # Almeno 1 minuto di validità residua return is_valid, remaining, payload import random import string def dazn_login(email: str, password: str, proxy_dict: Optional[Dict[str, str]] = None) -> Tuple[str, Optional[str]]: """Effettua il login su DAZN e restituisce lo stato (SUCCESS, RETRY, FAIL) e il token.""" url = "https://authentication-prod.ar.indazn.com/v5/SignIn" import hashlib h = hashlib.md5(email.encode()).hexdigest() s_val = h[:4] k_val = h[4:7] payload = { "Email": email, "Password": password, "Platform": "web", "DeviceId": f"0{s_val}e5ed", "ProfilingSessionId": f"2bdz6s9k{k_val}ydg5ujkhink9" } headers = { "accept": "*/*", "accept-language": "en-GB,en;q=0.9,it;q=0.8", "cache-control": "no-cache", "content-type": "application/json", "origin": "https://www.dazn.com", "referer": "https://www.dazn.com/", "sec-ch-ua": '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"', "sec-ch-ua-mobile": "?0", "sec-ch-ua-platform": '"Windows"', "sec-fetch-dest": "empty", "sec-fetch-mode": "cors", "sec-fetch-site": "cross-site", "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36", "x-dazn-ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 signin/undefined hyper/0.14.0 (web; production; it)" } log(f"🔑 Tentativo di Login in corso per {email}...", CYAN) try: # Usa curl_cffi per bypassare CloudFront WAF s = create_session(proxy_dict) r = s.post(url, json=payload, headers=headers, timeout=15) text = r.text log(f"🔍 DEBUG Risposta Login per {email}: HTTP {r.status_code} - {text[:300]}", DIM) # KEYCHECK Logic (OpenBullet) if "Request blocked" in text or "CloudFront" in text or r.status_code == 403: log("🛑 CloudFront WAF ha bloccato la richiesta! Riprovo...", YELLOW) return "RETRY", None if "InvalidPassword" in text or "AccountBlocked" in text: return "FAIL", None if "Request limiting policy has been breached" in text or "Security policy has been breached" in text: return "FAIL", None if "SignedIn" in text or "Token" in text: try: token = r.json().get("Token") or r.json().get("AuthToken", {}).get("Token") if token: return "SUCCESS", token except Exception: pass # Fallback check if r.status_code == 202: return "RETRY", None if r.ok: try: token = r.json().get("Token") or r.json().get("AuthToken", {}).get("Token") if token: return "SUCCESS", token except Exception: pass return "FAIL", None except Exception as e: log(f"❌ Eccezione durante il Login: {e}", RED) return "RETRY", None def dazn_refresh_token(token: str, user_id: Optional[str] = None, device_id: Optional[str] = None, proxy_dict: Optional[Dict[str, str]] = None) -> Optional[str]: """Registra il dispositivo ed ottiene un token autorizzato al playback chiamando RefreshAccessToken""" payload = decode_jwt_payload(token) uid = user_id or (payload.get("user") if payload else "") dev_str = device_id or (payload.get("deviceId") if payload else "") short_dev = dev_str.split("-")[-1].replace("|dazn", "") if "-" in dev_str else "000711a5aa" url = "https://ott-authz-bff-prod.ar.indazn.com/v5/RefreshAccessToken" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Origin": "https://www.dazn.com", "Referer": "https://www.dazn.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/154.0.0.0 Safari/537.36 Edg/154.0.0.0", "x-brand": "dazn", "x-daznid": uid, "x-session-id": str(uuid.uuid4()) } try: s = create_session(proxy_dict) r = s.post(url, json={"DeviceId": short_dev}, headers=headers, timeout=15) if r.ok: new_tok = r.json().get("AuthToken", {}).get("Token") if new_tok: return new_tok except Exception as e: log(f"⚠️ Errore RefreshAccessToken: {e}", YELLOW) return None def get_token(config: Dict[str, Any], cli_token: Optional[str] = None) -> str: """Recupera il token attivo dal file tokens, dalla config, oppure esegue il login se configurato""" token = cli_token if not token and os.path.exists(TOKEN_FILE): try: t_data = json.load(open(TOKEN_FILE, 'r', encoding='utf-8')) t_cand = t_data.get('token', '') if t_cand: val, rem, _ = check_token_status(t_cand) if val: token = t_cand else: log(f"⚠️ Token salvato in {TOKEN_FILE} è scaduto (scaduto da {-rem}s)", YELLOW) except Exception: pass if not token: token = config.get('dazn', {}).get('token', '') if token: val, rem, payload = check_token_status(token) if not val: token = "" # Forza il reset se è scaduto anche quello della config # Se non c'è un token valido e abbiamo le credenziali, tentiamo il login if not token: accounts_file = os.path.join(os.path.dirname(__file__), 'accounts.txt') if os.path.exists(accounts_file): log(f"⏳ Nessun token valido trovato, avvio Login automatico da {accounts_file}...", YELLOW) try: with open(accounts_file, 'r', encoding='utf-8') as f: lines = [l.strip() for l in f if ':' in l and not l.startswith('#')] except Exception: lines = [] for line in lines: email, password = line.split(':', 1) retries = 3 while retries > 0: proxy_dict = proxy_manager.get_dict() status, new_token = dazn_login(email, password, proxy_dict) if status == "SUCCESS" and new_token: reg_token = dazn_refresh_token(new_token, proxy_dict=proxy_dict) or new_token token = reg_token log("✅ Login completato e dispositivo registrato! Nuovo token acquisito.", GREEN) try: with open(TOKEN_FILE, 'w', encoding='utf-8') as f: json.dump({"token": token}, f, indent=2) except Exception: pass break elif status == "RETRY": log(f"⚠️ Rate limit o errore per {email}, cambio proxy e riprovo... ({retries-1} tentativi rimasti)", YELLOW) proxy_manager.rotate() retries -= 1 time.sleep(2) else: log(f"❌ Account {email} fallito (Invalid o Blocked), passo al successivo...", RED) break if token: break if not token: log("❌ Nessun account in accounts.txt è risultato valido o funzionante.", RED) else: # Fallback legacy: prova config.json email = config.get('dazn', {}).get('email', '') password = config.get('dazn', {}).get('password', '') if email and password: proxy_dict = proxy_manager.get_dict() st, tk = dazn_login(email, password, proxy_dict) if tk: token = tk try: with open(TOKEN_FILE, 'w', encoding='utf-8') as f: json.dump({"token": token}, f, indent=2) except Exception: pass if token: val, rem, payload = check_token_status(token) if val: mins = rem // 60 hrs = mins // 60 log(f"🔑 Token DAZN valido · Scadenza tra {hrs}h {mins % 60}m (ViewerID: {payload.get('viewerId', 'N/A')})", GREEN) # Sincronizza token_file t_info = { "token": token, "viewer_id": payload.get("viewerId", ""), "daznid": payload.get("user", ""), "device_id": payload.get("deviceId", ""), "country": payload.get("country", "it"), "user_status": payload.get("userstatus", ""), "issued": payload.get("issued", 0), "exp": payload.get("exp", 0), "updated_at": datetime.now().isoformat() } with open(TOKEN_FILE, 'w', encoding='utf-8') as f: json.dump(t_info, f, indent=2, ensure_ascii=False) else: log(f"⚠️ ATTENZIONE: Il token fornito è SCADUTO o NON VALIDO!", RED) log(f"Aggiorna il token in {CONFIG_FILE} o usa --token ", YELLOW) return token # ── Client HTTP con TLS Impersonation ────────────────────────────── # La licenza DAZN accetta il CDM Amazon L3 solo se il TLS sembra Android. # Chrome desktop + WVD Fire Stick = 10802. chrome99_android = 200 (test 2026-10-10). _ANDROID_IMPERSONATE = ("chrome99_android", "chrome131_android") def create_session(proxy_dict: Optional[Dict[str, str]] = None) -> Any: """Sessione Chrome desktop: login, EPG, manifest.""" if HAS_CURL_CFFI: s = cffi_requests.Session(impersonate="chrome120") else: s = requests.Session() if proxy_dict: s.proxies = proxy_dict return s def create_android_session(proxy_dict: Optional[Dict[str, str]] = None) -> Any: """Sessione TLS Android: playback + license. Non serve la Fire Stick.""" if HAS_CURL_CFFI: last_err = None for imp in _ANDROID_IMPERSONATE: try: s = cffi_requests.Session(impersonate=imp) if proxy_dict: s.proxies = proxy_dict return s except Exception as e: last_err = e continue if last_err: log(f"⚠️ Impersonate Android non disponibile ({last_err}), fallback Chrome", YELLOW) return create_session(proxy_dict) # ── HTTP dalla Fire Stick (BoringSSL Android). Il PC prende 10802 ── _ADB_BIN: Optional[str] = None _STICK_READY: Optional[bool] = None _STICK_LOGGED = False def _find_adb() -> Optional[str]: candidates = [ os.path.expandvars(r"%LOCALAPPDATA%\Android\Sdk\platform-tools\adb.exe"), os.path.join(SCRIPT_DIR, "adb.exe"), ] for p in candidates: if p and os.path.isfile(p): return p return shutil.which("adb") def stick_available() -> bool: """True se c'è una Fire Stick/Android collegata via adb.""" global _ADB_BIN, _STICK_READY, _STICK_LOGGED if os.environ.get("DAZN_NO_STICK", "").strip().lower() in ("1", "true", "yes"): _STICK_READY = False return False if _STICK_READY is not None: return _STICK_READY _ADB_BIN = _find_adb() if not _ADB_BIN: _STICK_READY = False return False try: r = subprocess.run([_ADB_BIN, "devices"], capture_output=True, timeout=8) lines = (r.stdout or b"").decode("utf-8", "replace").splitlines() _STICK_READY = any( ("\tdevice" in ln) or ln.strip().endswith(" device") for ln in lines[1:] ) except Exception: _STICK_READY = False if _STICK_READY and not _STICK_LOGGED: log("📡 Fire Stick collegata: playback e licenza passano dal device (TLS Android)", GREEN) _STICK_LOGGED = True return bool(_STICK_READY) def stick_http(method: str, url: str, headers: Dict[str, str], body: Optional[bytes] = None) -> Tuple[int, bytes]: """GET/POST eseguiti da curl sulla stick (niente proxy mitm, cert insecure).""" if not _ADB_BIN: return 0, b"" local_cfg = os.path.join(SCRIPT_DIR, "_curl.cfg") local_body = os.path.join(SCRIPT_DIR, "_curl.body") local_out = os.path.join(SCRIPT_DIR, "_curl.out") remote_cfg = "/sdcard/dazn_curl.cfg" remote_body = "/sdcard/dazn_body.bin" remote_out = "/sdcard/dazn_out.bin" lines = [ f'url = "{url}"', f'request = "{method}"', "insecure", "silent", "show-error", 'noproxy = "*"', f'output = "{remote_out}"', 'write-out = "%{http_code}"', ] for k, v in headers.items(): vv = str(v).replace("\\", "\\\\").replace('"', '\\"') lines.append(f'header = "{k}: {vv}"') if body is not None: with open(local_body, "wb") as fh: fh.write(body) lines.append(f'data-binary = "@{remote_body}"') with open(local_cfg, "w", encoding="utf-8", newline="\n") as fh: fh.write("\n".join(lines) + "\n") try: subprocess.run([_ADB_BIN, "push", local_cfg, remote_cfg], capture_output=True, timeout=20) if body is not None: subprocess.run([_ADB_BIN, "push", local_body, remote_body], capture_output=True, timeout=20) r = subprocess.run( [ _ADB_BIN, "shell", "env", "http_proxy=", "https_proxy=", "HTTP_PROXY=", "HTTPS_PROXY=", "curl", "-K", remote_cfg, ], capture_output=True, timeout=45, ) code_s = (r.stdout or b"").decode("utf-8", "replace").strip() subprocess.run([_ADB_BIN, "pull", remote_out, local_out], capture_output=True, timeout=20) raw = open(local_out, "rb").read() if os.path.exists(local_out) else b"" try: code = int(code_s) except ValueError: code = 0 return code, raw except Exception: return 0, b"" # ── Risoluzione Stream & DRM Widevine ───────────────────────────── def is_widevine_pssh(pssh_b64: str) -> bool: """Verifica se una stringa PSSH Base64 appartiene a Widevine""" try: raw = base64.b64decode(pssh_b64) if len(raw) >= 28 and raw[12:28] == WIDEVINE_SYSTEM_ID_BYTES: return True p = PSSH(pssh_b64) return str(p.system_id).lower() == WIDEVINE_SYSTEM_ID except Exception: return False def extract_widevine_pssh(mpd_content: str) -> Tuple[Optional[str], Optional[str]]: """Estrae il blocco PSSH Widevine e il relativo KID dal manifesto DASH MPD""" try: root = ET.fromstring(mpd_content) except Exception as e: log(f"Errore parsing XML del manifesto DASH: {e}", RED) return None, None candidates = [] adaptation_sets = [elem for elem in root.iter() if elem.tag.endswith('AdaptationSet')] if not adaptation_sets: adaptation_sets = [root] for aset in adaptation_sets: mime = (aset.attrib.get('mimeType') or aset.attrib.get('contentType') or '').lower() is_video = 'video' in mime or 'mp4' in mime cps = [elem for elem in aset.iter() if elem.tag.endswith('ContentProtection')] for cp in cps: scheme_id = cp.attrib.get('schemeIdUri', '').lower() is_wv_scheme = WIDEVINE_SYSTEM_ID in scheme_id pssh_nodes = [elem for elem in cp.iter() if elem.tag.endswith('pssh')] for node in pssh_nodes: text = (node.text or '').strip() if not text: continue if is_wv_scheme or is_widevine_pssh(text): candidates.append({ 'pssh': text, 'is_video': is_video, 'is_wv_scheme': is_wv_scheme }) if not candidates: for elem in root.iter(): if elem.tag.endswith('pssh') and elem.text and elem.text.strip(): text = elem.text.strip() if is_widevine_pssh(text): candidates.append({ 'pssh': text, 'is_video': True, 'is_wv_scheme': True }) if not candidates: return None, None candidates.sort(key=lambda c: (c['is_video'], c['is_wv_scheme']), reverse=True) best_pssh = candidates[0]['pssh'] expected_kid = None try: p = PSSH(best_pssh) if p.key_ids: first_kid = p.key_ids[0] if hasattr(first_kid, 'hex') and callable(getattr(first_kid, 'hex')): expected_kid = first_kid.hex() elif hasattr(first_kid, 'hex') and isinstance(first_kid.hex, str): expected_kid = first_kid.hex elif isinstance(first_kid, bytes): expected_kid = first_kid.hex() else: expected_kid = str(first_kid).replace('-', '') except Exception: pass return best_pssh, expected_kid def build_playback_url(channel: Dict[str, Any], config: Dict[str, Any], viewer_id: str) -> str: """Costruisce l'endpoint Playback con SessionId e timestamp freschi""" dazn_cfg = config.get('dazn', {}) asset_id = channel.get('id', '4hqcjx5zv91f1bdx71io4bnrl') app_version = dazn_cfg.get('app_version', '0.150.1-hotfix.f8b078689') player_id = dazn_cfg.get('player_id', '@dazn/peng-html5-core/tv-legacy/exoplayer-androidtv') lang = dazn_cfg.get('language', 'it') # Fingerprint del client dichiarata al playback: configurabile da # dazn_config.json per poter sperimentare profili diversi (es. android # per abbinarsi a un CDM Android). Default = identici al player web reale. # Profili VERIFICATI da catture reali (2026-10-10): # - web/Chrome154 desktop (licenza 200): player_id= # @dazn/peng-html5-core/web/web, platform=web, model=unknown, # manufacturer=microsoft, secure=true, capabilities=dd,ddp,hcst,mta # - androidtv/Fire TV Stick (licenza 200 SOLO con identità L1 hardware): # player_id=@dazn/peng-html5-core/tv-legacy/exoplayer-androidtv, # platform=androidtv, model=Fire TV Stick, manufacturer=amzn, # secure=true, capabilities=hevc,mta, # UA=Mozilla/5.0 (Linux; Android 11; Fire TV Stick Build/RQ3A.211001.001; # wv) ... DAZN_HTML_ANDROIDTV, origin=https://tv.dazn.com # NB: il profilo androidtv NON basta con un CDM L3 software: il server # accetta solo l'identità hardware (L1/TEE) del device. platform = dazn_cfg.get('platform', 'androidtv') model = dazn_cfg.get('model', 'Fire TV Stick') secure = dazn_cfg.get('secure', 'true') manufacturer = dazn_cfg.get('manufacturer', 'amzn') playready_init = dazn_cfg.get('playready_initiator', 'false') capabilities = dazn_cfg.get('capabilities', 'hevc,mta') v_id = viewer_id or 'unknown' ts = int(time.time() * 1000) rnd_hex = uuid.uuid4().hex[:6].upper() session_id = f"{ts}-{v_id}-{asset_id}-{rnd_hex}" params = { "AppVersion": app_version, "DrmType": "WIDEVINE", "Format": "MPEG-DASH", "PlayerId": player_id, "Platform": platform, "Model": model, "Secure": secure, "Manufacturer": manufacturer, "PlayReadyInitiator": playready_init, "Capabilities": capabilities, "AssetId": asset_id, "MtaLanguageCode": "", "LanguageCode": lang, "SessionId": session_id } return f"https://api.playback.indazn.com/v5/Playback?{urlencode(params)}" def request_playback_info(playback_url: str, token: str, config: Dict[str, Any], proxy_dict: Optional[Dict[str, str]]) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: """Effettua la chiamata all'endpoint Playback DAZN con headers personalizzati e bypass WAF""" dazn_cfg = config.get('dazn', {}) payload = decode_jwt_payload(token) user_id = (payload.get("user") if payload else None) or dazn_cfg.get("daznid", "790f7b77-25f4-4cd8-bc68-15c600c9eb89") device_id = (payload.get("deviceId") if payload else None) or dazn_cfg.get("device_id") or str(uuid.uuid4()) hdr_cfg = dazn_cfg.get("headers", {}) headers = { "Authorization": f"Bearer {token}", "Origin": hdr_cfg.get("origin", "https://tv.dazn.com"), "Referer": hdr_cfg.get("referer", "https://tv.dazn.com/"), "user-agent": hdr_cfg.get( "user-agent", "Mozilla/5.0 (Linux; Android 11; Fire TV Stick Build/RQ3A.211001.001; wv) " "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/105.0.5195.79 " "Mobile Safari/537.36 DAZN_HTML_ANDROIDTV", ), "x-brand": "DAZN", "x-correlation-id": str(uuid.uuid4()), "x-dazn-device": device_id, "x-daznid": user_id, } if hdr_cfg.get("x-requested-with"): headers["x-requested-with"] = hdr_cfg["x-requested-with"] s = create_android_session(proxy_dict) try: r = s.get(playback_url, headers=headers, timeout=25) if not r.ok: ctype = r.headers.get('Content-Type', '') try: err_json = r.json() oerr = err_json.get('odata.error', {}) if isinstance(err_json, dict) else {} msg = (oerr.get('message', {}) or {}).get('value') or r.text[:200] code = oerr.get('code', '') if code: msg = f"[{code}] {msg}" except Exception: msg = r.text[:200] # Blocchi edge/WAF (pagina HTML CloudFront): messaggio pulito e riconoscibile if 'html' in ctype.lower() or 'Request blocked' in r.text or 'CloudFront' in r.text: msg = "Request blocked dall'edge CloudFront (probabile throttle IP: fai pausa 15-30 min)" return None, f"HTTP {r.status_code}: {msg}" return r.json(), None except Exception as e: return None, str(e) def get_keys_wvd(pssh_b64: str, asset_id: str, la_url: str, config: Dict[str, Any], tokens_payload: Dict[str, Any], expected_kid: Optional[str] = None, proxy_dict: Optional[Dict[str, str]] = None, token: Optional[str] = None) -> Tuple[Optional[str], Optional[str], Optional[str]]: """Decifra la chiave di contenuto Widevine usando WVD.wvd e l'endpoint di Licenza DAZN""" global pssh_cache if expected_kid: cache_key = f"{asset_id}:{expected_kid.lower()}" else: cache_key = f"{asset_id}:{hashlib.sha256(pssh_b64.encode()).hexdigest()[:16]}" if cache_key in pssh_cache: cached = pssh_cache[cache_key] return cached.get('kid'), cached.get('key'), None wvd_folder = os.path.join(SCRIPT_DIR, "wvd") # Lista deterministica dei CDM disponibili (ordine alfabetico): se un # CDM viene rifiutato dal server licenze si prova con il successivo. # MAX 5 tentativi: oltre è solo rumore che triggera il rate-limit 429. _MAX_CDM_ATTEMPTS = 5 candidates: List[str] = [] if os.path.exists(wvd_folder) and os.path.isdir(wvd_folder): for f in sorted(os.listdir(wvd_folder)): if f.endswith(".wvd"): candidates.append(os.path.join(wvd_folder, f)) if os.path.exists(WVD_FILE): candidates.append(WVD_FILE) # Deduplica per contenuto: lo stesso dump copiato in due path (es. wvd/ # e root) verrebbe provato due volte, raddoppiando i 403 verso il server # licenze senza alcuna possibilità in più di successo. seen_hashes: set = set() deduped: List[str] = [] for c in candidates: try: with open(c, 'rb') as fh: h = hashlib.md5(fh.read()).hexdigest() except Exception: h = c if h in seen_hashes: log(f"⏭️ CDM duplicato saltato: {os.path.relpath(c, SCRIPT_DIR)} (identico a uno già in lista)", DIM) continue seen_hashes.add(h) deduped.append(c) candidates = deduped[:_MAX_CDM_ATTEMPTS] if not candidates: log(f"ERRORE: Nessun file CDM trovato in {wvd_folder} o {WVD_FILE}!", RED) return None, None, "CDM missing" dazn_cfg = config.get('dazn', {}) user_id = tokens_payload.get("user") or dazn_cfg.get("daznid", "790f7b77-25f4-4cd8-bc68-15c600c9eb89") # Header della Fire Stick (cattura mitm 2026-10-10, licenza 200): # Authorization Bearer + Content-Type + UA DAZN_HTML_ANDROIDTV. # Niente Origin/Referer/x-brand/privacy-mode: la stick manda la challenge # in chiaro (~2111 B). Chrome-privacy (2437 B) su questo CDM prende 10802. stick_ua = ( "Mozilla/5.0 (Linux; Android 11; Fire TV Stick Build/RQ3A.211001.001; wv) " "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/105.0.5195.79 " "Mobile Safari/537.36 DAZN_HTML_ANDROIDTV" ) lic_headers = { "Content-Type": "application/octet-stream", "User-Agent": dazn_cfg.get("license_headers", {}).get("user-agent", stick_ua), "Accept-Encoding": "gzip", } if token: lic_headers["Authorization"] = f"Bearer {token}" last_err: Optional[str] = None for target_wvd in candidates: cdm = None session = None try: device = Device.load(target_wvd) # NOTA: nessun "spoof" del client_id. Il vecchio codice forzava # type=CHROME e riscriveva i client_info (company/model/device): # la challenge resta comunque firmata con la chiave privata del # WVD originale, quindi il server Widevine vede un certificato # manomesso e risponde 10802 "License denied". Si usa il device # così com'è (il tuo Chrome reale è L3 e funziona, quindi un # WVD L3 valido deve funzionare senza travestimenti). try: dev_type = getattr(device.client_id, 'type', '?') dev_level = getattr(device, 'security_level', '?') except Exception: dev_type, dev_level = '?', '?' log(f"🔑 CDM in uso: {os.path.relpath(target_wvd, SCRIPT_DIR)} (type={dev_type}, level={dev_level})", DIM) cdm = Cdm.from_device(device) session = cdm.open() # TLS Android (curl_cffi chrome99_android): stesso CDM Amazon, niente Fire Stick. # Chrome desktop / requests nativo su questo WVD = 10802. s = create_android_session(proxy_dict) log(f"🔍 DEBUG License URL: {la_url}", DIM) challenge = cdm.get_license_challenge(session, PSSH(pssh_b64), privacy_mode=False) log(f"🔍 DEBUG Challenge size: {len(challenge)} bytes (plaintext={b'amzn' in challenge})", DIM) r = s.post(la_url, headers=lic_headers, data=challenge, timeout=30) if not r.ok: error_text = f"HTTP {r.status_code}: {r.text[:200]}" log(f"Chiamata License fallita ({error_text}) [{os.path.relpath(target_wvd, SCRIPT_DIR)}]", RED) last_err = error_text if stick_available() and any(k in error_text.lower() for k in ("10802", "not allowed to receive")): code, lic_bytes = stick_http("POST", la_url, lic_headers, body=challenge) if code != 200: continue else: continue else: lic_bytes = r.content cdm.parse_license(session, lic_bytes) content_keys: List[Tuple[str, str]] = [] for k in cdm.get_keys(session): if k.type != 'SIGNING': k_kid = k.kid.hex if hasattr(k.kid, 'hex') and isinstance(k.kid.hex, str) else (k.kid.hex() if hasattr(k.kid, 'hex') else str(k.kid).replace('-', '')) k_key = k.key.hex() if callable(getattr(k.key, 'hex', None)) else getattr(k.key, 'hex', str(k.key)) content_keys.append((k_kid.lower(), k_key.lower())) if not content_keys: log("Nessuna chiave di contenuto trovata nella licenza DRM", RED) last_err = "No keys" continue kid, key = None, None if expected_kid: for c_kid, c_key in content_keys: if c_kid == expected_kid.lower(): kid, key = c_kid, c_key break if not kid and content_keys: kid, key = content_keys[0] if kid and key: pssh_cache[cache_key] = {'kid': kid, 'key': key} save_cache() log(f"Chiave DRM estratta con successo: {kid}:{key}", GREEN) return kid, key, None except Exception as e: log(f"Errore licenza DRM Widevine [{os.path.relpath(target_wvd, SCRIPT_DIR)}]: {e}", RED) last_err = str(e) continue finally: try: if cdm is not None and session is not None: cdm.close(session) except Exception: pass return None, None, last_err or "License denied da tutti i CDM" def is_radio_tv_serie_a(channel: Dict[str, Any]) -> bool: """True solo per Radio TV Serie A Lega (mai un evento DAZN).""" if not channel: return False if channel.get("source") == "seriea_radio": return True if channel.get("id") == SERIEA_RADIO_ASSET_ID: return True name = re.sub(r"\s+", " ", str(channel.get("name") or "").strip().lower()) return name == "radio tv serie a" def _seriea_public_mpd_url() -> str: """Link HTTPS stabile all'MPD riscritto (niente file://).""" try: cfg = load_config() or {} override = ( (cfg.get("dazn") or {}).get("seriea_mpd_url") or cfg.get("seriea_mpd_url") or "" ).strip() if override: return override except Exception: pass script = SCRIPT_DIR.replace("\\", "/").rstrip("/") m = re.match(r"^/var/www/([^/]+)/(.+)$", script) if m: return f"https://{m.group(1)}/{m.group(2)}/radio_tv_seriea.mpd" return "https://addictedbytheproject.nl/esdrongo/dazn/radio_tv_seriea.mpd" def _seriea_cdn_headers() -> Dict[str, str]: return { "Accept": "*/*", "Origin": "https://www.legaseriea.it", "Referer": "https://www.legaseriea.it/", "User-Agent": SERIEA_HEADERS["User-Agent"], } def _rewrite_seriea_mpd_xml(xml: str, content_url: str) -> str: """BaseURL = path Chrome (senza @syna-token / :443); hdnts sui segmenti.""" parsed = urlparse(content_url) netloc = parsed.netloc.replace(":443", "") dir_url = f"{parsed.scheme}://{netloc}{parsed.path.rsplit('/', 1)[0]}/" qs = parsed.query.replace("&", "&") xml = re.sub(r"[^<]*", "", xml, flags=re.I) xml = re.sub( r"(]*>)", r"\1\n " + dir_url + r"", xml, count=1, flags=re.I, ) if not qs: return xml def _add_qs(match: re.Match) -> str: key, val = match.group(1), match.group(2) if "hdnts=" in val: return match.group(0) sep = "&" if "?" in val else "?" return f'{key}="{val}{sep}{qs}"' return re.sub(r'\b(initialization|media)="([^"]+)"', _add_qs, xml) def write_seriea_vlc_mpd(content_url: str) -> Optional[str]: """MPD locale per VLC: segmenti su path hdnts Chrome, senza 307 :443/@syna-token.""" if not content_url: return None hdrs = _seriea_cdn_headers() try: r = requests.get(content_url, headers=hdrs, timeout=15, allow_redirects=False) if r.status_code in (301, 302, 303, 307, 308): loc = (r.headers.get("Location") or "").replace(":443/", "/") if loc.startswith("/"): p = urlparse(content_url) loc = f"{p.scheme}://{p.netloc.replace(':443', '')}{loc}" r = requests.get(loc, headers=hdrs, timeout=15, allow_redirects=True) elif not r.ok: r = requests.get(content_url, headers=hdrs, timeout=15, allow_redirects=True) if not r.ok or " Optional[Dict[str, Any]]: """Token Lega: @syna-token nel path (40s) oppure hdnts in query.""" url = (url or "").replace(":443/", "/") if not url: return None parsed = urlparse(url) syna = "" m = re.search(r"/@syna-token=([^/]+)/", parsed.path) if m: syna = unquote(m.group(1)) token = {"Name": "syna-token", "Value": syna, "ExpirySeconds": 40} else: hdnts = (parse_qs(parsed.query).get("hdnts") or [""])[0] if not hdnts: return None token = {"Name": "hdnts", "Value": hdnts, "ExpirySeconds": 40} base_path = parsed.path.rsplit("/", 1)[0] if parsed.path else "" return { "content_url": url, "base_manifest_url": url, "base_cdn_url": f"{parsed.scheme}://{parsed.netloc}{base_path}/" if parsed.netloc else "", "cdn_token": token, "base_query": parsed.query, } def request_seriea_entitlement() -> Tuple[Optional[str], Optional[Dict[str, Any]], Optional[str]]: """POST Deltatre Open Auth. SessionId/PlaybackSessionId random a ogni chiamata.""" sid = str(uuid.uuid4()) psid = str(uuid.uuid4()).upper() payload = { "Type": 1, "VideoId": SERIEA_RADIO_VIDEO_ID, "VideoSource": SERIEA_RADIO_VIDEO_SOURCE, "VideoKind": "live", "AssetState": "2", "PlayerType": "HTML5", "VideoSourceFormat": "DASH", "VideoSourceName": "Desktop-DASH", "DRMType": "", "AuthType": "Open", "ContentKeyData": "", "SessionId": sid, "PlaybackSessionId": psid, "Other": f"{sid}|web_browser", "User": "", } try: r = requests.post(SERIEA_ENTITLEMENT_URL, json=payload, headers=SERIEA_HEADERS, timeout=15) if not r.ok: return None, None, f"HTTP {r.status_code}: {r.text[:180]}" data = r.json() content_url = (data.get("ContentUrl") or "").replace(":443/", "/") if data.get("Response") != "OK" or not content_url: return None, None, data.get("Message") or f"Entitlement {data.get('Response')}" # Stesso URL del Network Chrome: index.mpd?hdnts=st=...~exp=...~hmac=...&hdcore= # Non seguire il 307 verso /@syna-token= (VLC lo gestisce male / token già bruciato). pack = _pack_seriea_url(content_url) if not pack: return None, None, f"Token Lega mancante (got {content_url[:160]})" return pack["content_url"], pack, None except Exception as e: return None, None, str(e) def extract_seriea_radio(channel: Dict[str, Any]) -> Dict[str, Any]: """Flusso Radio TV Serie A da Deltatre (Clear, niente Widevine/DAZN).""" channel_name = channel.get("name") or "Radio TV Serie A" log(f"📻 [{channel_name}] Flusso Lega Serie A (Deltatre, no DRM)...", CYAN) content_url, pack, err = request_seriea_entitlement() if err or not pack: log(f"❌ [{channel_name}] Entitlement Lega fallito: {err}", RED) return { "id": channel.get("id"), "name": channel_name, "status": "ERROR", "error": err or "Entitlement Lega fallito", "logo": channel.get("logo", ""), "event_id": channel.get("event_id", ""), "epg": channel.get("epg"), } chrome_url = pack["content_url"] write_seriea_vlc_mpd(chrome_url) play_url = _seriea_public_mpd_url() base = pack["base_manifest_url"] token = pack["cdn_token"] parsed = urlparse(base) base_path = parsed.path.rsplit("/", 1)[0] if parsed.path else "" base_cdn = f"{parsed.scheme}://{parsed.netloc}{base_path}/" if parsed.netloc else "" cid = channel.get("id") or SERIEA_RADIO_ASSET_ID log(f"✅ [{channel_name}] Manifest Lega Chrome/hdnts OK", GREEN) log(f" VLC: {play_url}", DIM) return { "logo": channel.get("logo", ""), "event_id": channel.get("event_id", ""), "proxy_manifest": play_url, "name": channel_name, "clear_key": None, "epg": channel.get("epg"), "_internal": { "id": cid, "status": "OK", "source": "seriea_radio", "manifest_url": play_url, "chrome_manifest": chrome_url, "base_manifest_url": chrome_url, "base_cdn_url": base_cdn, "cdn_name": "legaseriea-netrw", "cdn_token": token, "resolution": "hd", "actual_resolution": "hd", "kid": None, "key": None, "key_string": None, "license_url": "", "extracted_at": datetime.now().isoformat(), "mpv_command": f'mpv "{play_url}" --user-agent="{SERIEA_HEADERS["User-Agent"]}" --referrer="https://www.legaseriea.it/"', "ffplay_command": f'ffplay -user_agent "{SERIEA_HEADERS["User-Agent"]}" -headers "Referer: https://www.legaseriea.it/" "{play_url}"', }, } def classify_playback_error(err: Optional[str]) -> str: """Classifica un errore Playback in 3 categorie: - 'throttle': blocco edge/IP (429, CloudFront, 10405...). Insistere peggiora il blocco: la run va fermata, non ritentata. - 'account': diniego legato a QUESTO account (Pin, geo/country, entitlement, concorrenza...). Inutile riprovare con lo stesso token, ma ha senso far provare la rotazione con un altro account. - 'transient': tutto il resto (errori rete, 5xx...). Ha senso riprovare.""" e = (err or "").lower() if any(k in e for k in ("429", "too many requests", "rate limit", "rate-limit", "security policy has been breached", "request limiting", "request blocked", "cloudfront", "10405")): return "throttle" if any(k in e for k in ("pin", "not valid", "no articles", "not found", "404", "eligibility", "entitlement", "geo", "concurrency", "not allowed", "forbidden", "denied", "inactive", "signedinactive", "401", "403", "10400", "trust 2fa", "2fa required")): return "account" return "transient" def is_trust_2fa_error(err: Optional[str]) -> bool: """True se DAZN ha rifiutato il playback perché l'account richiede Trust 2FA.""" e = (err or "").lower() return "10400" in e or "trust 2fa" in e or "2fa required" in e def extract_channel(channel: Dict[str, Any], config: Dict[str, Any], token: str, tokens_payload: Dict[str, Any], max_retries: int = 8) -> Dict[str, Any]: """ Esegue l'estrazione completa per un singolo canale DAZN con retry intelligente: se un flusso restituisce 'Solo Manifest' (nessuna chiave DRM al primo colpo), riprova automaticamente esplorando tutti i CDN disponibili e ripetendo la richiesta finché la chiave non viene estratta con successo. """ proxy_dict = proxy_manager.get_dict() viewer_id = tokens_payload.get('viewerId', '') channel_name = channel.get('name', 'DAZN Channel') last_composed_manifest_url = "" last_raw_manifest_url = "" last_cdn_name = "" last_resolution = "fhd" last_actual_resolution = "unknown" last_cdn_token = {} last_la_url = "" last_cookie_hdr = "" last_mpv_cmd = "" last_ffplay_cmd = "" last_asset = {} kid, key = None, None license_refreshed = False for attempt in range(1, max_retries + 1): playback_url = build_playback_url(channel, config, viewer_id) if attempt == 1: log(f"Richiesta Playback per '{channel_name}'...", CYAN) else: log(f"🔁 [{channel_name}] Tentativo {attempt}/{max_retries}: riprovo estrazione Playback & DRM...", YELLOW) playback_data, err = request_playback_info(playback_url, token, config, proxy_dict) if err or not playback_data: # Se l'errore indica dispositivo non registrato, effettua la registrazione e aggiorna il token if err and any(k in str(err).lower() for k in ("registered", "device is not", "register", "not allowed to play")): log(f"📱 Registrazione dispositivo richiesta per '{channel_name}', chiamo RefreshAccessToken...", CYAN) user_id = tokens_payload.get('user', '') device_id = tokens_payload.get('deviceId', '') ref_tok = dazn_refresh_token(token, user_id=user_id, device_id=device_id, proxy_dict=proxy_dict) if ref_tok: token = ref_tok _, _, tokens_payload = check_token_status(ref_tok) viewer_id = tokens_payload.get('viewerId', viewer_id) log(f"✅ Dispositivo registrato! Riprovo Playback per '{channel_name}'...", GREEN) time.sleep(1.0) continue # Eligibility not allowed: marca per esclusione totale dal JSON e non ritentare if err and "Eligibility" in str(err): log(f"⛔ [{channel_name}] Eligibility not allowed — evento escluso dal JSON.", YELLOW) _eligibility_blocked_ids.add(channel.get("id")) return { "id": channel.get("id"), "name": channel_name, "status": "ELIGIBILITY_BLOCKED", "error": err } # Classificazione errore (fail-fast): # - throttle edge/IP -> stop immediato, insistere peggiora il blocco # - diniego legato a QUESTO account (Pin, geo, entitlement...) -> # niente retry, la rotazione account ne proverà un altro err_kind = classify_playback_error(err) if err_kind == "throttle": log(f"⛔ [{channel_name}] Throttle edge/IP, niente retry: {err}", RED) return { "id": channel.get("id"), "name": channel_name, "status": "THROTTLED", "error": err } if err_kind == "account": status = "ACCOUNT_2FA" if is_trust_2fa_error(err) else "ERROR" if status == "ACCOUNT_2FA": log(f"🔒 [{channel_name}] Trust 2FA su questo account, provo un altro: {err}", YELLOW) else: log(f"🚫 [{channel_name}] Diniego per questo account, passo oltre senza retry: {err}", YELLOW) return { "id": channel.get("id"), "name": channel_name, "status": status, "error": err } # Errori temporanei — riprova fino a max_retries if err: log(f"⚠️ [{channel_name}] Errore temporaneo: {err} — riprovo...", YELLOW) if attempt < max_retries: time.sleep(1.5) continue log(f"Errore Playback: {err}", RED) return { "id": channel.get("id"), "name": channel_name, "status": "ERROR", "error": err } asset = playback_data.get("Asset", {}) if asset.get("Title"): channel_name = asset.get("Title") details_list = playback_data.get("PlaybackDetails", []) last_asset = asset if not details_list: if attempt < max_retries: time.sleep(1.5) continue log(f"Nessun PlaybackDetails disponibile nella risposta", RED) return { "id": channel.get("id"), "name": channel_name, "status": "NO_STREAMS", "error": "Nessun PlaybackDetails trovato" } # Ordina i CDN secondo le preferenze: DaznEdge (primario nel web player), Fastly e 'ac', poi Akamai pref_order = ['dca-de-live', 'dcb-de-live', 'de', 'dca-fs-live', 'dcb-fs-live', 'dcs-fs-live', 'dct-fs-live', 'fs', 'dcs-ac-live', 'dct-ac-live', 'dca-ac-live', 'ac', 'dct-ak-live', 'dcs-ak-live', 'ak'] def cdn_score(d): c_name = d.get('CdnName', '').lower() for idx, pref in enumerate(pref_order): if pref in c_name: return idx return 999 sorted_details = sorted(details_list, key=cdn_score) # DEBUG: dump tutti i PlaybackDetails per diagnosticare il token corretto if attempt == 1: log(f"🔍 DEBUG PlaybackDetails: {len(sorted_details)} CDN disponibili", DIM) for idx, det in enumerate(sorted_details): cdn_n = det.get("CdnName", "?") la = det.get("LaUrl", "") ct = det.get("CdnToken", {}) ct_val = ct.get("Value", "")[:60] if isinstance(ct, dict) else str(ct)[:60] has_jwt = "eyJ" in la keys = list(det.keys()) log(f" [{idx}] CDN={cdn_n} | LaUrl_has_jwt={'YES' if has_jwt else 'NO'} | Token={ct_val}... | Keys={keys}", DIM) s = create_session(proxy_dict) dazn_headers = config.get("dazn", {}).get("headers", {}) ua = dazn_headers.get("user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/154.0.0.0 Safari/537.36 Edg/154.0.0.0") json_pssh = (playback_data.get("License") or {}).get("Pssh") if json_pssh and not kid: d0 = sorted_details[0] la0 = d0.get("LaUrl", "") ct0 = d0.get("CdnToken", {}) tn0 = ct0.get("Name", "dazn-token") if isinstance(ct0, dict) else "dazn-token" tv0 = ct0.get("Value", "") if isinstance(ct0, dict) else (ct0 if isinstance(ct0, str) else "") if la0 and tv0: la0 = f"{la0}{'&' if '?' in la0 else '?'}{tn0}={tv0}" if la0: kid, key, drm_err = get_keys_wvd( pssh_b64=json_pssh, asset_id=channel.get("id", asset.get("Id", "")), la_url=la0, config=config, tokens_payload=tokens_payload, expected_kid=None, proxy_dict=proxy_dict, token=token, ) if not kid and drm_err and any(k in str(drm_err).lower() for k in ("too many requests", "429", "rate limit", "rate-limit")): log(f"⏳ [{channel_name}] Rate limit server licenze (429): mi fermo su questo canale per non peggiorare il blocco.", YELLOW) return {"id": channel.get("id"), "name": channel_name, "status": "ERROR", "error": drm_err} if not kid and drm_err and any(k in str(drm_err).lower() for k in ("not allowed to receive a license", "10802", "license denied")): log(f"⛔ [{channel_name}] 10802 su questo CDM: inutile altri CDN/retry.", YELLOW) return {"id": channel.get("id"), "name": channel_name, "status": "ERROR", "error": drm_err} for d in sorted_details: raw_manifest_url = d.get("ManifestUrl", "") la_url = d.get("LaUrl", "") cdn_token = d.get("CdnToken", {}) resolution = d.get("Resolution", "fhd") cdn_name = d.get("CdnName", "") token_name = cdn_token.get("Name", "dazn-token") token_val = cdn_token.get("Value", "") if token_name and token_val: sep = '&' if '?' in raw_manifest_url else '?' composed_manifest_url = f"{raw_manifest_url}{sep}{token_name}={token_val}" # Aggiungi dazn-token anche alla license URL if la_url: la_sep = '&' if '?' in la_url else '?' la_url = f"{la_url}{la_sep}{token_name}={token_val}" else: composed_manifest_url = raw_manifest_url mpd_headers = { "Origin": "https://www.dazn.com", "Referer": "https://www.dazn.com/", "User-Agent": ua, "sec-ch-ua": dazn_headers.get("sec-ch-ua", '"Chromium";v="154", "Microsoft Edge";v="154", "Not A(Brand";v="99"'), "sec-ch-ua-mobile": dazn_headers.get("sec-ch-ua-mobile", "?0"), "sec-ch-ua-platform": dazn_headers.get("sec-ch-ua-platform", '"Windows"') } if token_name and token_val: mpd_headers["Cookie"] = f"{token_name}={token_val}" try: r_mpd = s.get(composed_manifest_url, headers=mpd_headers, timeout=15) mpd_text = r_mpd.text if r_mpd.ok else "" except Exception: mpd_text = "" actual_resolution = "unknown" max_bandwidth = 0 if mpd_text: try: root = ET.fromstring(mpd_text) for aset in root.findall('.//{urn:mpeg:dash:schema:mpd:2011}AdaptationSet'): if aset.get('contentType') == 'video': max_h = 0 max_fps = 25 for rep in aset.findall('{urn:mpeg:dash:schema:mpd:2011}Representation'): h = rep.get('height') fps = rep.get('frameRate') bw = rep.get('bandwidth') if h and h.isdigit(): if int(h) >= max_h: max_h = int(h) if fps and fps.isdigit(): max_fps = int(fps) if bw and bw.isdigit() and int(bw) > max_bandwidth: max_bandwidth = int(bw) if max_h > 0: actual_resolution = f"{max_h}p{max_fps if max_fps > 25 else ''}" break except Exception: pass pssh_b64, expected_kid = extract_widevine_pssh(mpd_text) if pssh_b64 and la_url and not kid: kid, key, drm_err = get_keys_wvd( pssh_b64=pssh_b64, asset_id=channel.get("id", asset.get("Id", "")), la_url=la_url, config=config, tokens_payload=tokens_payload, expected_kid=expected_kid, proxy_dict=proxy_dict, token=token ) if not kid and drm_err and any(k in str(drm_err).lower() for k in ("too many requests", "429", "rate limit", "rate-limit")): log(f"⏳ [{channel_name}] Rate limit server licenze (429): mi fermo su questo canale per non peggiorare il blocco.", YELLOW) return { "id": channel.get("id"), "name": channel_name, "status": "ERROR", "error": drm_err } if not kid and drm_err and any(k in str(drm_err).lower() for k in ("not allowed to receive a license", "10802", "license denied")): log(f"⛔ [{channel_name}] 10802 su questo CDM: inutile altri CDN/retry.", YELLOW) return { "id": channel.get("id"), "name": channel_name, "status": "ERROR", "error": drm_err } # Salva gli ultimi dettagli validi per il fallback last_composed_manifest_url = composed_manifest_url last_raw_manifest_url = raw_manifest_url last_cdn_name = cdn_name last_resolution = resolution last_actual_resolution = actual_resolution last_cdn_token = cdn_token last_la_url = la_url cookie_hdr = f"{token_name}={token_val}" if (token_name and token_val) else "" last_cookie_hdr = cookie_hdr mpv_cmd = f'mpv "{composed_manifest_url}" --user-agent="{ua}"' if cookie_hdr: mpv_cmd += f' --http-header-fields="Cookie: {cookie_hdr}"' if kid and key: mpv_cmd += f' --demuxer-lavf-o="decryption_key={kid}:{key}"' ffplay_cmd = f'ffplay -user_agent "{ua}"' if cookie_hdr: ffplay_cmd += f' -headers "Cookie: {cookie_hdr}"' ffplay_cmd += f' -cenc_decryption_key {key} "{composed_manifest_url}"' else: ffplay_cmd = f'ffplay -user_agent "{ua}"' if cookie_hdr: ffplay_cmd += f' -headers "Cookie: {cookie_hdr}"' ffplay_cmd += f' "{composed_manifest_url}"' last_mpv_cmd = mpv_cmd last_ffplay_cmd = ffplay_cmd if kid and key: bw_str = f" @ {max_bandwidth // 1000} kbps" if max_bandwidth else "" log(f"Risoluzione stream: {actual_resolution}{bw_str} (Profilo API: {resolution.upper()})", GREEN if "1080" in actual_resolution else YELLOW) break if kid and key: break if attempt < max_retries: log(f"⚠️ [{channel_name}] Solo Manifest (chiavi DRM non ancora pronte). Nuovo tentativo {attempt + 1}/{max_retries} tra 2s...", YELLOW) time.sleep(2) status = "OK" if (kid and key) else ("MANIFEST_ONLY" if last_composed_manifest_url else "ERROR") result_entry = { "logo": channel.get("logo", ""), "event_id": last_asset.get("EventId", ""), "proxy_manifest": f"https://addictedbytheproject.nl:8443/play/{channel.get('id', last_asset.get('Id'))}/stream.mpd", "name": channel_name, "clear_key": f"{kid}:{key}" if (kid and key) else None, "epg": channel.get("epg"), } # Campi interni necessari per il funzionamento del proxy e delle playlist (non salvati nel JSON) result_entry["_internal"] = { "id": channel.get("id", last_asset.get("Id")), "status": status, "manifest_url": last_composed_manifest_url, "base_manifest_url": last_raw_manifest_url, "cdn_name": last_cdn_name, "resolution": last_resolution, "actual_resolution": last_actual_resolution, "cdn_token": last_cdn_token, "license_url": last_la_url, "kid": kid, "key": key, "key_string": f"{kid}:{key}" if (kid and key) else None, "cookie_header": f"Cookie: {last_cookie_hdr}" if last_cookie_hdr else None, "mpv_cmd": last_mpv_cmd, "mpv_command": last_mpv_cmd, "ffplay_command": last_ffplay_cmd, "extracted_at": datetime.now().isoformat() } return result_entry # ── Funzione Principale di Estrazione e Salvataggio ──────────────── def extract_and_save_all(cli_token: Optional[str] = None, channels: Optional[List[Dict[str, Any]]] = None, include_events: bool = True) -> Dict[str, Any]: """Estrae tutti i canali/eventi configurati, risolve le chiavi DRM e aggiorna JSON & M3U""" config = load_config() load_cache() proxy_manager.load(config_proxy=config.get('dazn', {}).get('proxy')) token = get_token(config, cli_token) if not token: log("ERRORE CRITICO: Nessun token DAZN disponibile! Configuralo in dazn_config.json", RED) return {} _, _, tokens_payload = check_token_status(token) # Se i canali non vengono passati, usa la configurazione base + eventi Rail (EPG fallback) if channels is None: base_channels = config.get('dazn', {}).get('channels', []).copy() if include_events: live_events = fetch_live_events() channels = _merge_live_into_base(base_channels, live_events) else: channels = base_channels channels = _ensure_static_seriea(channels) if not channels: log("Nessun canale o evento da estrarre", YELLOW) return {} # Filtra gli asset bloccati per Eligibility channels = [ch for ch in channels if ch.get("id") not in _eligibility_blocked_ids] log(f"🚀 Avvio estrazione di {len(channels)} canali/eventi DAZN...", CYAN) results = [] m3u_lines = ["#EXTM3U x-tvg-url=\"\""] found = 0 first_ffplay_cmd = None first_mpv_cmd = None # Helper: ottieni status da result (può essere al top level per errori, o in _internal per successi) def _get_status(r): return r.get("status") or r.get("_internal", {}).get("status") def _get_internal(r, key, default=""): return r.get("_internal", {}).get(key, default) accounts_file = os.path.join(os.path.dirname(__file__), 'accounts.txt') rotation_accounts = [] if os.path.exists(accounts_file): try: with open(accounts_file, 'r', encoding='utf-8') as f: rotation_accounts = [l.strip() for l in f if ':' in l and not l.startswith('#')] except Exception: pass rotation_idx = 0 alt_token_cache = {} failed_accounts = set() throttle_streak = 0 throttled_abort = False for i, ch in enumerate(channels, 1): name = ch.get('name', ch.get('id', f'Canale {i}')) print(f" [{i:2d}/{len(channels)}] {name:<25}", end=" ", flush=True) res = None if is_radio_tv_serie_a(ch): res = extract_seriea_radio(ch) st0 = res.get("status") or res.get("_internal", {}).get("status") if st0 == "OK": print(f"✅ OK (Lega/Deltatre, no DRM)") else: print(f"❌ Errore ({res.get('error', 'entitlement')})") results.append(res) continue attempts_allowed = len(rotation_accounts) if rotation_accounts else 1 tried_emails = set() current_email = "" for _acc_try in range(attempts_allowed): current_token = token current_payload = tokens_payload current_email = "" if rotation_accounts: acct_line = None for _skip in range(len(rotation_accounts)): cand = rotation_accounts[rotation_idx % len(rotation_accounts)] rotation_idx += 1 email, password = cand.split(":", 1) if email in failed_accounts or email in tried_emails: continue acct_line = cand current_email = email break if not acct_line: break tried_emails.add(current_email) if current_email in alt_token_cache: current_token, current_payload = alt_token_cache[current_email] print(f"({current_email.split('@')[0]})", end=" ", flush=True) else: print(f"(login {current_email.split('@')[0]})...", end=" ", flush=True) proxy_dict = proxy_manager.get_dict() status, login_token = dazn_login(current_email, password, proxy_dict) if status == "SUCCESS" and login_token: reg_token = dazn_refresh_token(login_token, proxy_dict=proxy_dict) or login_token current_token = reg_token _, _, current_payload = check_token_status(current_token) alt_token_cache[current_email] = (current_token, current_payload) else: print(f"[fail]", end=" ", flush=True) if status == "FAIL": failed_accounts.add(current_email) continue res = extract_channel(ch, config, current_token, current_payload) st = _get_status(res) if st == "ACCOUNT_2FA": who = current_email.split("@")[0] if current_email else "token principale" if current_email: failed_accounts.add(current_email) alt_token_cache.pop(current_email, None) log(f"\n🔒 [{name}] Trust 2FA su {who}: account scartato, provo il prossimo...", YELLOW) if rotation_accounts: continue break if st in ("OK", "ELIGIBILITY_BLOCKED"): break # Altri errori: se ci sono altri account, ruota if rotation_accounts: who = current_email.split("@")[0] if current_email else "account" log(f"\n⚠️ [{name}] Non riuscito con {who}, provo account alternativo da accounts.txt...", YELLOW) continue break # Se tutti i tentativi sono falliti, prova un fallback disperato col token principale # (mai in caso di throttle: sarebbero solo richieste in più verso un IP già bloccato) st_now = _get_status(res) if res else None skip_fallback = st_now in ("OK", "ELIGIBILITY_BLOCKED", "THROTTLED") or ( st_now == "ACCOUNT_2FA" and not rotation_accounts ) if not skip_fallback: log(f"\n⚠️ [{name}] Tutti gli account hanno fallito (o fallito login). Tento fallback col token principale...", YELLOW) res = extract_channel(ch, config, token, tokens_payload) st = _get_status(res) if res: results.append(res) st = _get_status(res) if st != "THROTTLED": throttle_streak = 0 if st == "THROTTLED": throttle_streak += 1 print(f"{RED}⛔ Throttle edge/IP ({throttle_streak}/2){RESET}") if throttle_streak >= 2: log("🛑 Throttle IP confermato: fermo la run per non peggiorare il blocco. Fai pausa 15-30 minuti e riprova.", RED) throttled_abort = True break elif st == "ELIGIBILITY_BLOCKED": print(f"{YELLOW}⛔ Eligibility not allowed (escluso){RESET}") elif st == "ACCOUNT_2FA": print(f"{YELLOW}🔒 Trust 2FA (tutti gli account){RESET}") elif st == "OK": kid = _get_internal(res, "kid", "") key = _get_internal(res, "key", "") print(f"{GREEN}✅ OK ({kid[:8]}...:{key[:8]}...){RESET}") found += 1 elif st == "MANIFEST_ONLY": print(f"{YELLOW}⚠️ Solo Manifest (in coda per sweep di retry){RESET}") else: print(f"{RED}❌ Errore ({res.get('error', 'N/A')}){RESET}") # Sweep dedicato per canali rimasti a 'Solo Manifest': riprova ad estrarre le chiavi manifest_only_indices = [idx for idx, r in enumerate(results) if _get_status(r) == "MANIFEST_ONLY"] if throttled_abort: log("⏭️ Sweep di retry saltati (throttle IP: niente altre richieste).", YELLOW) if manifest_only_indices and not throttled_abort: log(f"🔄 Rilevati {len(manifest_only_indices)} canali con 'Solo Manifest'. Avvio sweep di retry dedicato...", CYAN) time.sleep(2) for idx in manifest_only_indices: target_ch = channels[idx] ch_title = target_ch.get('name', 'Evento') log(f"🔁 Secondo tentativo di estrazione chiavi per '{ch_title}'...", YELLOW) retry_res = extract_channel(target_ch, config, token, tokens_payload, max_retries=6) if _get_status(retry_res) == "OK": results[idx] = retry_res found += 1 log(f"✅ Chiave recuperata con successo per '{ch_title}': {_get_internal(retry_res, 'key_string')}", GREEN) # Sweep dedicato per canali con 'Concurrency max locations violation': usa account alternativi concurrency_indices = [ idx for idx, r in enumerate(results) if r.get("error") and "Concurrency" in str(r.get("error", "")) ] if concurrency_indices and not throttled_abort: # Carica tutti gli account disponibili da accounts.txt accounts_file = os.path.join(os.path.dirname(__file__), 'accounts.txt') alt_accounts = [] if os.path.exists(accounts_file): try: with open(accounts_file, 'r', encoding='utf-8') as f: alt_accounts = [l.strip() for l in f if ':' in l and not l.startswith('#')] except Exception: pass if not alt_accounts: log(f"⚠️ {len(concurrency_indices)} canali bloccati per Concurrency ma nessun account alternativo in accounts.txt.", YELLOW) else: log(f"🔄 Sweep Concurrency: {len(concurrency_indices)} canali bloccati. Provo con account alternativi da accounts.txt...", CYAN) # Cicla gli account disponibili per i canali bloccati alt_token_cache = {} # email -> (token, payload) — evita login multipli per lo stesso account skip_accounts = set(failed_accounts) account_idx = 0 for idx in concurrency_indices: target_ch = channels[idx] ch_title = target_ch.get('name', 'Evento') # Prova fino a max 3 account diversi per canale resolved = False attempts_per_channel = min(3, len(alt_accounts)) for _ in range(attempts_per_channel): acct_line = alt_accounts[account_idx % len(alt_accounts)] account_idx += 1 email, password = acct_line.split(':', 1) if email in skip_accounts: continue # Login (o usa token già ottenuto per questo account) if email in alt_token_cache: alt_token, alt_payload = alt_token_cache[email] else: log(f"🔑 [{ch_title}] Login con account alternativo {email}...", CYAN) proxy_dict = proxy_manager.get_dict() status, alt_token = dazn_login(email, password, proxy_dict) if status != "SUCCESS" or not alt_token: log(f"❌ Login fallito per {email}, provo il prossimo account...", RED) alt_token_cache[email] = (None, {}) skip_accounts.add(email) continue _, _, alt_payload = check_token_status(alt_token) alt_token_cache[email] = (alt_token, alt_payload) log(f"✅ Login riuscito con {email}", GREEN) if not alt_token: continue # Prova a estrarre il canale con il token alternativo log(f"🔁 [{ch_title}] Retry con account {email}...", YELLOW) retry_res = extract_channel(target_ch, config, alt_token, alt_payload, max_retries=3) st = _get_status(retry_res) if st == "OK": results[idx] = retry_res found += 1 log(f"✅ [{ch_title}] Recuperato con successo via {email}: {_get_internal(retry_res, 'key_string')}", GREEN) resolved = True break elif st == "MANIFEST_ONLY": results[idx] = retry_res log(f"⚠️ [{ch_title}] Solo Manifest via {email} (senza chiavi DRM)", YELLOW) resolved = True break elif st == "ACCOUNT_2FA" or is_trust_2fa_error(retry_res.get("error")): log(f"🔒 [{ch_title}] Trust 2FA su {email}, account scartato, provo il prossimo...", YELLOW) skip_accounts.add(email) alt_token_cache.pop(email, None) failed_accounts.add(email) elif retry_res.get("error") and "Concurrency" in str(retry_res.get("error", "")): log(f"⏳ [{ch_title}] Ancora Concurrency con {email}, provo un altro account...", DIM) else: results[idx] = retry_res log(f"❌ [{ch_title}] Errore con {email}: {retry_res.get('error', 'N/A')}", RED) break if not resolved: log(f"⚠️ [{ch_title}] Non recuperabile con nessun account alternativo.", YELLOW) # Riepilogo finale Concurrency still_concurrency = sum(1 for r in results if r.get("error") and "Concurrency" in str(r.get("error", ""))) if still_concurrency: log(f"⚠️ {still_concurrency} canali ancora bloccati per Concurrency dopo aver provato tutti gli account.", YELLOW) # Costruzione delle playlist M3U (Diretta CDN e Proxy Locale) m3u_lines = ["#EXTM3U x-tvg-url=\"\""] proxy_m3u_lines = ["#EXTM3U x-tvg-url=\"\""] ua = config.get("dazn", {}).get("headers", {}).get("user-agent", "") for ch, res in zip(channels, results): st = _get_status(res) if st in ("OK", "MANIFEST_ONLY"): kid = _get_internal(res, "kid", "") key = _get_internal(res, "key", "") manifest = _get_internal(res, "manifest_url", "") proxy_manifest = res.get("proxy_manifest", "") logo = res.get("logo") or ch.get("logo", "") logo_attr = f' tvg-logo="{logo}"' if logo else "" is_live_event = bool(ch.get("epg")) is_radio = is_radio_tv_serie_a(ch) or _get_internal(res, "source") == "seriea_radio" if is_radio: group_name = "Radio TV Serie A" elif _is_linear_channel(ch): group_name = "DAZN Canali" elif is_live_event: group_name = "DAZN Eventi Live" else: group_name = "DAZN Canali" if not first_ffplay_cmd and st == "OK": first_ffplay_cmd = _get_internal(res, "ffplay_command") first_mpv_cmd = _get_internal(res, "mpv_command") # 1. Playlist M3U Diretta CDN (Widevine) token_val = _get_internal(res, "cdn_token", {}).get("Value", "") if isinstance(_get_internal(res, "cdn_token", {}), dict) else "" token_name = _get_internal(res, "cdn_token", {}).get("Name", "dazn-token") if isinstance(_get_internal(res, "cdn_token", {}), dict) else "dazn-token" cookie_hdr = f"&Cookie={token_name}={token_val}" if (token_name and token_val) else "" ch_id = _get_internal(res, "id", "") m3u_lines.append(f'#EXTINF:-1 tvg-id="{ch_id}" tvg-name="{res.get("name")}"{logo_attr} group-title="{group_name}",{res.get("name")}') m3u_lines.append('#KODIPROP:inputstream=inputstream.adaptive') m3u_lines.append('#KODIPROP:inputstream.adaptive.manifest_type=mpd') m3u_lines.append('#KODIPROP:inputstream.adaptive.manifest_update_parameter=full') m3u_lines.append('#KODIPROP:inputstream.adaptive.live_delay=8') m3u_lines.append('#KODIPROP:inputstream.adaptive.live_delay_type=fixed') if kid and key: m3u_lines.append('#KODIPROP:inputstream.adaptive.license_type=com.widevine.alpha') m3u_lines.append(f'#KODIPROP:inputstream.adaptive.license_key={kid}:{key}') if is_radio: m3u_lines.append(f'#KODIPROP:inputstream.adaptive.stream_headers=User-Agent={SERIEA_HEADERS["User-Agent"]}&Referer=https://www.legaseriea.it/') m3u_lines.append(f'#EXTVLCOPT:http-user-agent={SERIEA_HEADERS["User-Agent"]}') m3u_lines.append('#EXTVLCOPT:http-referrer=https://www.legaseriea.it/') elif ua: m3u_lines.append(f'#KODIPROP:inputstream.adaptive.stream_headers=User-Agent={ua}&Referer=https://www.dazn.com/{cookie_hdr}') m3u_lines.append(f'#EXTVLCOPT:http-user-agent={ua}') m3u_lines.append('#EXTVLCOPT:http-referrer=https://www.dazn.com/') m3u_lines.append(manifest) # Radio TV Serie A: MPD Lega diretto, mai proxy DAZN if is_radio: proxy_m3u_lines.append(f'#EXTINF:-1 tvg-id="{ch_id}" tvg-name="{res.get("name")}"{logo_attr} group-title="{group_name}",{res.get("name")}') proxy_m3u_lines.append('#KODIPROP:inputstream=inputstream.adaptive') proxy_m3u_lines.append('#KODIPROP:inputstream.adaptive.manifest_type=mpd') proxy_m3u_lines.append(f'#KODIPROP:inputstream.adaptive.stream_headers=User-Agent={SERIEA_HEADERS["User-Agent"]}&Referer=https://www.legaseriea.it/') proxy_m3u_lines.append(manifest) else: proxy_m3u_lines.append(f'#EXTINF:-1 tvg-id="{ch_id}" tvg-name="{res.get("name")}"{logo_attr} group-title="{group_name}",{res.get("name")}') proxy_m3u_lines.append('#KODIPROP:inputstream=inputstream.adaptive') proxy_m3u_lines.append('#KODIPROP:inputstream.adaptive.manifest_type=mpd') if kid and key: proxy_m3u_lines.append('#KODIPROP:inputstream.adaptive.license_type=clearkey') proxy_m3u_lines.append(f'#KODIPROP:inputstream.adaptive.license_key={kid}:{key}') proxy_m3u_lines.append(proxy_manifest) # Filtra i risultati: escludi ELIGIBILITY_BLOCKED completamente clean_channels = [] # Per dazn_channels.json (solo i 6 campi utente) proxy_channels = [] # Per dazn_proxy_data.json (dati completi per il proxy) for r in results: st = _get_status(r) if st in ("OK", "MANIFEST_ONLY"): internal = r.get("_internal", {}) # JSON pulito: solo i 6 campi richiesti dall'utente clean_channels.append({ "logo": r.get("logo", ""), "event_id": r.get("event_id", ""), "proxy_manifest": r.get("proxy_manifest", ""), "name": r.get("name", ""), "clear_key": r.get("clear_key"), "epg": r.get("epg"), }) # Dati completi per il proxy proxy_entry = { "logo": r.get("logo", ""), "event_id": r.get("event_id", ""), "proxy_manifest": r.get("proxy_manifest", ""), "name": r.get("name", ""), "clear_key": r.get("clear_key"), "epg": r.get("epg"), } proxy_entry.update(internal) if internal.get("source") == "seriea_radio": continue proxy_channels.append(proxy_entry) # I canali falliti escludono quelli bloccati per Eligibility failed_channels = [r for r in results if _get_status(r) not in ("OK", "MANIFEST_ONLY", "ELIGIBILITY_BLOCKED")] eligibility_count = sum(1 for r in results if _get_status(r) == "ELIGIBILITY_BLOCKED") if eligibility_count: log(f"⛔ {eligibility_count} eventi esclusi per Eligibility not allowed (non saranno ritentati).", YELLOW) now_ts = datetime.now().isoformat() # Salva dazn_channels.json (JSON pulito con solo i 6 campi) output_data = { "updated_at": now_ts, "total_channels": len(channels), "keys_found": found, "channels": clean_channels, } with open(OUTPUT_JSON_FILE, 'w', encoding='utf-8') as f: json.dump(output_data, f, indent=2, ensure_ascii=False) # Salva dazn_proxy_data.json (dati completi per il proxy) proxy_output = { "updated_at": now_ts, "total_channels": len(channels), "keys_found": found, "proxy_used": proxy_manager.get_dict(), "channels": proxy_channels, "failed_channels": failed_channels } with open(PROXY_DATA_FILE, 'w', encoding='utf-8') as f: json.dump(proxy_output, f, indent=2, ensure_ascii=False) # Salva dazn_playlist.m3u (Diretto CDN) with open(PLAYLIST_FILE, 'w', encoding='utf-8') as f: f.write("\n".join(m3u_lines) + "\n") # Salva dazn_playlist_proxy.m3u (Proxy Locale HTTPS) with open(PLAYLIST_PROXY_FILE, 'w', encoding='utf-8') as f: f.write("\n".join(proxy_m3u_lines) + "\n") # Salva file batch pronti all'uso per ffplay e mpv if first_ffplay_cmd: ffplay_bat = os.path.join(SCRIPT_DIR, "play_ffplay.bat") with open(ffplay_bat, 'w', encoding='utf-8') as f: f.write(f"@echo off\n{first_ffplay_cmd}\npause\n") if first_mpv_cmd: mpv_bat = os.path.join(SCRIPT_DIR, "play_mpv.bat") with open(mpv_bat, 'w', encoding='utf-8') as f: f.write(f"@echo off\n{first_mpv_cmd}\npause\n") log(f"💾 Salvato con successo in: {BOLD}{OUTPUT_JSON_FILE}{RESET} ({found}/{len(channels)} canali attivi)", GREEN) log(f"📋 Playlist M3U aggiornata in: {BOLD}{PLAYLIST_FILE}{RESET}", GREEN) if first_ffplay_cmd: log(f"🎬 Launcher generati: play_ffplay.bat e play_mpv.bat", CYAN) log(f"▶️ Comando FFplay:\n{first_ffplay_cmd}", YELLOW) return output_data # Cache discovery (Rail primario, EPG fallback) — valida 5 minuti _epg_cache_data = None _epg_cache_time = 0 _rail_cache_data: Dict[str, Any] = {} _rail_cache_time: Dict[str, float] = {} def _parse_dazn_dt(value: Any) -> Optional[datetime]: if not value or not isinstance(value, str): return None try: return datetime.strptime(value.replace("Z", "").split(".")[0], "%Y-%m-%dT%H:%M:%S") except Exception: return None def _nested_title(obj: Any) -> str: if isinstance(obj, dict): return obj.get("Title") or obj.get("Name") or "" return str(obj or "") def _tile_image_id(tile: Dict[str, Any]) -> str: for key in ("Image", "LogoImage", "HeroImage", "BackgroundImage"): img = tile.get(key) if isinstance(img, dict) and img.get("Id"): return str(img["Id"]) if isinstance(img, str) and img: return img return "" def _is_radio_tv_tile(tile: Dict[str, Any]) -> bool: if tile.get("AssetId") == SERIEA_RADIO_ASSET_ID: return True title = re.sub(r"\s+", " ", str(tile.get("Title") or "").strip().lower()) return title == "radio tv serie a" def _static_seriea_channel() -> Dict[str, Any]: return { "id": SERIEA_RADIO_ASSET_ID, "name": "Radio TV Serie A", "logo": DAZN_IMAGE_URL + "1daica1elq2701s7zajjchy47s_image-header_pIt_1786353830000", "source": "seriea_radio", "playback_url": "", } def _ensure_static_seriea(channels: List[Dict[str, Any]]) -> List[Dict[str, Any]]: out = [ch for ch in channels if not is_radio_tv_serie_a(ch)] out.append(_static_seriea_channel()) return out def rewrite_seriea_direct_url(manifest: str) -> None: """Sostituisce solo l'URL diretto di Radio TV nelle M3U (niente proxy).""" if not manifest: return pat = re.compile( r'(#EXTINF:-1[^\n]*tvg-name="Radio TV Serie A"[^\n]*\n(?:#.*\n)*)(?:https://[^\s]+|file://[^\s]+)', re.IGNORECASE, ) for path in (PLAYLIST_FILE, PLAYLIST_PROXY_FILE): if not os.path.isfile(path): continue try: with open(path, "r", encoding="utf-8") as f: text = f.read() new, n = pat.subn(r"\1" + manifest, text, count=1) if n: with open(path, "w", encoding="utf-8") as f: f.write(new) except Exception as e: log(f"Avviso rewrite Radio TV M3U: {e}", YELLOW) def refresh_seriea_direct_playlists() -> bool: url, pack, err = request_seriea_entitlement() if err or not url: log(f"Radio TV Lega: {err}", YELLOW) return False write_seriea_vlc_mpd(url) play_url = _seriea_public_mpd_url() rewrite_seriea_direct_url(play_url) try: if os.path.isfile(OUTPUT_JSON_FILE): with open(OUTPUT_JSON_FILE, "r", encoding="utf-8") as f: data = json.load(f) for ch in data.get("channels") or []: if is_radio_tv_serie_a(ch): ch["proxy_manifest"] = play_url with open(OUTPUT_JSON_FILE, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception: pass return True def _is_keep_linear_tile(tile: Dict[str, Any]) -> bool: if _is_radio_tv_tile(tile): return True if tile.get("AssetId") in EXTRA_LINEAR_IDS: return True title = re.sub(r"\s+", " ", str(tile.get("Title") or "").strip().lower()) return title in EXTRA_LINEAR_TITLES def _is_linear_channel(ch: Dict[str, Any]) -> bool: if ch.get("linear_channel") or ch.get("id") in EXTRA_LINEAR_IDS: return True name = re.sub(r"\s+", " ", str(ch.get("name") or "").strip().lower()) if name in EXTRA_LINEAR_TITLES: return True end = str((ch.get("epg") or {}).get("end") or "") try: return int(end[:4]) >= 2100 except Exception: return False def _merge_live_into_base( base_channels: List[Dict[str, Any]], live_events: List[Dict[str, Any]] ) -> List[Dict[str, Any]]: live_by_id = {ev.get("id"): ev for ev in live_events if ev.get("id")} out: List[Dict[str, Any]] = [] seen = set() for ch in base_channels: merged = dict(ch) ev = live_by_id.get(merged.get("id")) if ev: if not merged.get("logo") and ev.get("logo"): merged["logo"] = ev["logo"] for k in ("epg", "linear_channel", "now_program", "event_id", "competition", "label"): if ev.get(k) and not merged.get(k): merged[k] = ev[k] out.append(merged) if merged.get("id"): seen.add(merged["id"]) for ev in live_events: if ev.get("id") and ev["id"] not in seen: out.append(ev) seen.add(ev["id"]) return out def _tile_to_channel(tile: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Converte un tile Rail/EPG in canale estraibile, o None se non è un live attivo.""" if not isinstance(tile, dict): return None if _is_radio_tv_tile(tile): return None typ = tile.get("Type") or "" display = tile.get("DisplayType") or "" if typ in ("CatchUp", "Highlights", "Navigation") or display in ("CatchUp", "Highlights"): return None if typ != "Live" and display != "Live": return None asset_id = tile.get("AssetId") if not asset_id: return None start_dt = _parse_dazn_dt(tile.get("Start") or tile.get("EventStartTime") or tile.get("MediaStartTime")) end_dt = _parse_dazn_dt(tile.get("End")) if start_dt and not end_dt: end_dt = start_dt + timedelta(hours=3) if not start_dt or not end_dt: return None is_radio = _is_radio_tv_tile(tile) # Lineari 24/7 perpetui saltati, tranne Radio TV e i canali extra (DAZN 1, club TV, ecc.) if end_dt.year >= 2100 and not _is_keep_linear_tile(tile): return None now_utc = datetime.now(timezone.utc).replace(tzinfo=None) if not ((start_dt - timedelta(minutes=15)) <= now_utc <= (end_dt + timedelta(minutes=15))): return None competition = _nested_title(tile.get("Competition")) label = tile.get("Label") or "" title = tile.get("Title") or f"Event {asset_id}" img_id = _tile_image_id(tile) now_program = "" ls = tile.get("LinearSchedule") if isinstance(tile.get("LinearSchedule"), dict) else {} nowp = ls.get("Now") if isinstance(ls, dict) else None if isinstance(nowp, dict): now_program = nowp.get("Title") or "" return { "id": asset_id, "name": title, "logo": f"{DAZN_IMAGE_URL}{img_id}" if img_id else "", "playback_url": "", "competition": competition, "label": label, "source": "seriea_radio" if is_radio else "", "event_id": tile.get("EventId") or "", "linear_channel": title if tile.get("IsLinear") else "", "now_program": now_program, "epg": { "start": start_dt.isoformat(), "end": end_dt.isoformat(), }, } def _tiles_to_channels(tiles: Any) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] seen = set() if not isinstance(tiles, list): return out for tile in tiles: ch = _tile_to_channel(tile) if not ch or ch["id"] in seen: continue seen.add(ch["id"]) out.append(ch) return out def _fetch_rail_json(url: str, cache_key: str) -> Optional[Dict[str, Any]]: cached = _rail_cache_data.get(cache_key) cached_at = _rail_cache_time.get(cache_key, 0) if cached and (time.time() - cached_at) < 300: return cached proxy_dict = proxy_manager.get_dict() try: r = requests.get(url, headers=RAIL_HEADERS, proxies=proxy_dict, timeout=15) if not r.ok: log(f"Errore Rail Router ({cache_key}): HTTP {r.status_code}", RED) return None data = r.json() _rail_cache_data[cache_key] = data _rail_cache_time[cache_key] = time.time() return data except Exception as e: log(f"Errore Rail Router ({cache_key}): {e}", RED) return None def fetch_live_events_from_rails() -> Optional[List[Dict[str, Any]]]: """Discovery primaria: LiveAndNextNew (eventi) + Livetvschedule (lineari). None se entrambe falliscono.""" live_data = _fetch_rail_json(RAIL_LIVE_URL, "LiveAndNextNew") linear_data = _fetch_rail_json(RAIL_LINEAR_URL, "Livetvschedule") if live_data is None and linear_data is None: return None live_tiles = (live_data or {}).get("Tiles") or [] linear_tiles = (linear_data or {}).get("Tiles") or [] channels = _tiles_to_channels(live_tiles) seen = {ch["id"] for ch in channels} extra = 0 for ch in _tiles_to_channels(linear_tiles): if ch["id"] in seen: continue seen.add(ch["id"]) channels.append(ch) extra += 1 log( f"🔎 Rail Check: LiveAndNextNew {len(live_tiles)} tile, " f"Livetvschedule {len(linear_tiles)} tile. " f"Trovati {len(channels)} eventi/canali LIVE attivi" + (f" (+{extra} da palinsesto lineare)." if extra else "."), DIM, ) return channels def fetch_live_events_from_epg() -> List[Dict[str, Any]]: """Fallback: EPG ufficiale DAZN (epgWithDatesRange).""" global _epg_cache_data, _epg_cache_time url = "https://epg.discovery.indazn.com/eu/v5/epgWithDatesRange" now_utc = datetime.now(timezone.utc).replace(tzinfo=None) start_date_str = (now_utc - timedelta(days=1)).strftime("%Y-%m-%d") end_date_str = (now_utc + timedelta(days=1)).strftime("%Y-%m-%d") try: offset_sec = datetime.now().astimezone().utcoffset().total_seconds() tz_offset = str(int(offset_sec // 60)) except Exception: tz_offset = "120" params = { "country": "it", "languageCode": "it", "openBrowse": "true", "timeZoneOffset": tz_offset, "brand": "dazn", "startDate": start_date_str, "endDate": end_date_str, } if _epg_cache_data and (time.time() - _epg_cache_time) < 300: data = _epg_cache_data else: proxy_dict = proxy_manager.get_dict() try: r = requests.get(url, params=params, proxies=proxy_dict, timeout=15) if not r.ok: log(f"Errore download EPG DAZN: HTTP {r.status_code}", RED) return [] data = r.json() _epg_cache_data = data _epg_cache_time = time.time() log("📥 EPG scaricato con successo dal server DAZN (aggiornato ogni 5 min).", GREEN) except Exception as e: log(f"Errore eccezione EPG: {e}", RED) return [] tiles = data.get("Tiles", []) live_channels = _tiles_to_channels(tiles) log(f"🔎 EPG Check: {len(tiles)} programmi in palinsesto. Trovati {len(live_channels)} eventi speciali LIVE attivi.", DIM) return live_channels def fetch_live_events() -> List[Dict[str, Any]]: """Eventi/canali live attivi: Rail Router principale, EPG solo se le rail falliscono.""" rail_events = fetch_live_events_from_rails() if rail_events is not None: return rail_events log("⚠️ Rail Router non disponibile, fallback su EPG epgWithDatesRange.", YELLOW) return fetch_live_events_from_epg() # ── Loop Automatico ─────────────────────────────────────────────── def run_auto_loop(interval_hours: int = 3, cli_token: Optional[str] = None, include_events: bool = True) -> None: """Esegue l'estrazione all'avvio e ripete periodicamente monitorando Rail/EPG e canali""" print_banner() config = load_config() proxy_manager.load(config_proxy=config.get('dazn', {}).get('proxy')) # Leggi intervallo programmato da config (default 3 ore) refresh_hours = config.get('loop', {}).get('refresh_interval_hours', interval_hours) refresh_seconds = refresh_hours * 3600 # 20s: Radio TV Lega ha token ~40s da rinnovare sull'M3U diretto check_interval = 20 # Conserviamo la lista originale dei canali statici base_channels = config.get('dazn', {}).get('channels', []).copy() last_known_live_ids = set() last_full_extraction_time = 0.0 last_seriea_refresh = 0.0 force_extraction = True # Avvia automaticamente il proxy DAZN locale / remoto ensure_proxy_running(port=8443) while True: try: now = time.time() # 1. Recupera eventi Live dal Rail Router (EPG solo fallback) if include_events: live_events = fetch_live_events() current_live_ids = {ev["id"] for ev in live_events} else: live_events = [] current_live_ids = set() # Se ci sono nuovi eventi o ne sono finiti alcuni, aggiorna la configurazione if current_live_ids != last_known_live_ids: if len(current_live_ids) > len(last_known_live_ids): log(f"🚨 Trovati {len(current_live_ids) - len(last_known_live_ids)} NUOVI EVENTI LIVE nella discovery!", MAGENTA) elif len(current_live_ids) < len(last_known_live_ids): log(f"ℹ️ {len(last_known_live_ids) - len(current_live_ids)} eventi live sono terminati.", DIM) config['dazn']['channels'] = _ensure_static_seriea( _merge_live_into_base(base_channels, live_events) ) last_known_live_ids = current_live_ids force_extraction = True # 2. Rinnovo dinamico CDN gestito on-demand dal proxy (dazn_proxy.py) # Le chiavi ClearKey sono permanenti. Il rinnovo dei token CDN a 200s viene # gestito on-demand dal proxy solo per i canali attivamente riprodotti, evitando # di saturare tutti gli account DAZN con estrazioni globali continue ogni 3 minuti. # Controllo Intervallo Programmato (es. ogni 3 ore da dazn_config.json) if (now - last_full_extraction_time) >= refresh_seconds: if last_full_extraction_time > 0: log(f"⏰ Intervallo programmato ({refresh_hours}h) trascorso. Rinnovo generale canali...", CYAN) force_extraction = True # 3. Estrazione solo all'avvio, su nuovi/finiti eventi, o alla scadenza programmata if force_extraction: print(f"\n{BLUE}{'═' * 70}{RESET}") log("🔄 CICLO DI ESTRAZIONE CHIAVI & MANIFEST (Live + Lineari)", CYAN) print(f"{BLUE}{'═' * 70}{RESET}") target_channels = config['dazn']['channels'] output_data = extract_and_save_all(cli_token=cli_token, channels=target_channels, include_events=False) last_full_extraction_time = time.time() force_extraction = False log(f"✨ Estrazione completata! Monitoraggio Rail/EPG attivo (prossimo controllo tra {check_interval}s).", GREEN) else: log(f"💤 Nessun nuovo evento live nella discovery. In attesa (prossimo controllo tra {check_interval}s)...", DIM) # Radio TV Lega: token ~40s, rinnovo URL diretto (niente proxy, niente DAZN) if (now - last_seriea_refresh) >= 20: if refresh_seriea_direct_playlists(): last_seriea_refresh = time.time() # Controllo liveness del proxy if not is_port_open(8443): ensure_proxy_running(port=8443) # Il bot dorme sempre per check_interval (60s) time.sleep(check_interval) except KeyboardInterrupt: print(f"\n\n{YELLOW}🛑 Bot arrestato dall'utente.{RESET}\n") stop_proxy() break except Exception as e: log(f"Errore durante il ciclo: {e}", RED) time.sleep(30) # ── Main Entrypoint ─────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="DAZN - DRM Widevine Stream & Key Extractor") parser.add_argument("-c", "--once", "--channels", dest="once", action="store_true", help="Esegui una sola volta ed esci") parser.add_argument("--loop", action="store_true", help="Esegui in loop continuo con intervallo programmato") parser.add_argument("--token", type=str, default=None, help="Specifica o aggiorna il JWT Bearer Token") parser.add_argument("--asset", type=str, default=None, help="Estrai solo uno specifico AssetId") parser.add_argument("--url", type=str, default=None, help="Usa uno specifico URL Playback personalizzato") parser.add_argument("--test-proxy", action="store_true", help="Testa tutti i proxy configurati ed esci") parser.add_argument("--clear-cache", action="store_true", help="Cancella la cache delle chiavi pssh_cache.json") parser.add_argument("--no-events", action="store_true", help="Non includere eventi live dalla discovery Rail/EPG (solo canali statici)") args = parser.parse_args() if args.clear_cache: if os.path.exists(CACHE_FILE): os.remove(CACHE_FILE) log("Cache pssh_cache.json eliminata con successo!", GREEN) return if args.test_proxy: cfg = load_config() proxy_manager.load(config_proxy=cfg.get('dazn', {}).get('proxy')) log(f"Test di {proxy_manager.count()} proxy in corso...", CYAN) results = proxy_manager.test_all() for p, ok, ip_msg, dur in results: col = GREEN if ok else RED log(f" {p} -> {col}{ip_msg}{RESET} ({dur}ms)") return if args.asset or args.url: print_banner() cfg = load_config() token = get_token(cfg, args.token) _, _, tokens_payload = check_token_status(token) target_ch = { "id": args.asset or "custom_asset", "name": f"Asset {args.asset or 'Custom'}", "playback_url": args.url or "" } if args.asset: for ev in fetch_live_events(): if ev.get("id") == args.asset: target_ch.update(ev) break if is_radio_tv_serie_a(target_ch): res = extract_seriea_radio(target_ch) else: res = extract_channel(target_ch, cfg, token, tokens_payload) print("\n" + json.dumps(res, indent=2, ensure_ascii=False)) return include_events = not args.no_events if args.once: print_banner() extract_and_save_all(cli_token=args.token, include_events=include_events) return # Di default se eseguito interattivamente o con --loop, avvia il loop automatico run_auto_loop(cli_token=args.token, include_events=include_events) if __name__ == '__main__': main()