#!/usr/bin/env python3 """ ==================================================================== TIM VISION - Manager & DRM Extractor (Widevine WVD) Auto-Extractor: 18 Canali Discovery ed Eurosport Output: canalidiscoveryeeurosporttim.json (Aggiornamento ogni 3 giorni) ==================================================================== """ import os import sys import re import json import time import base64 import uuid import hashlib import xml.etree.ElementTree as ET from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple, Any from urllib.parse import urlparse, parse_qs, urlencode, urlunparse 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 basati sulla cartella dello script) ── SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_FILE = os.path.join(SCRIPT_DIR, "timvision_config.json") TOKEN_FILE = os.path.join(SCRIPT_DIR, "timvision_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, "canalidiscoveryeeurosporttim.json") PLAYLIST_FILE = os.path.join(SCRIPT_DIR, "timvision_playlist.m3u") # Intervallo di aggiornamento: 3 giorni (in secondi) REFRESH_INTERVAL_SECONDS = 3 * 24 * 3600 # 259.200 secondi (3 giorni) UA = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ) # ── Lista Canali Richiesti (18 Canali) ───────────────────────────── CHANNELS = [ {"epgid": "90300133", "name": "LA7 HD", "wsx": "la7", "logo": "https://img.timvision.it/channels/la7.png"}, {"epgid": "90320074", "name": "NOVE", "wsx": "novehd", "logo": "https://img.timvision.it/channels/nove.png"}, {"epgid": "90300553", "name": "LA7 Cinema HD", "wsx": "la7dhd", "logo": "https://img.timvision.it/channels/la7d.png"}, {"epgid": "90020075", "name": "Real Time", "wsx": "realtimehd", "logo": "https://img.timvision.it/channels/realtime.png"}, {"epgid": "90020062", "name": "Food Network", "wsx": "foodnetworkhd", "logo": "https://img.timvision.it/channels/foodnetwork.png"}, {"epgid": "90020037", "name": "Discovery", "wsx": "warnertv", "logo": "https://img.timvision.it/channels/warnertv.png"}, {"epgid": "90120080", "name": "Giallo", "wsx": "giallo", "logo": "https://img.timvision.it/channels/giallo.png"}, {"epgid": "90120077", "name": "K2", "wsx": "k2", "logo": "https://img.timvision.it/channels/k2.png"}, {"epgid": "90120078", "name": "Frisbee", "wsx": "frisbee", "logo": "https://img.timvision.it/channels/frisbee.png"}, {"epgid": "90020076", "name": "DMAX", "wsx": "dmaxhd", "logo": "https://img.timvision.it/channels/dmax.png"}, {"epgid": "90020061", "name": "HGTV", "wsx": "hgtvhd", "logo": "https://img.timvision.it/channels/hgtv.png"}, {"epgid": "90020060", "name": "Discovery Turbo", "wsx": "motortrendhd", "logo": "https://img.timvision.it/channels/motortrend.png"}, {"epgid": "90020000", "name": "Eurosport 1", "wsx": "eurosport1", "logo": "https://img.timvision.it/channels/eurosport1.png"}, {"epgid": "90020001", "name": "Eurosport 2", "wsx": "eurosport2", "logo": "https://img.timvision.it/channels/eurosport2.png"}, {"epgid": "90020002", "name": "Eurosport 3", "wsx": "eurosport3", "logo": "https://img.timvision.it/channels/eurosport3.png"}, {"epgid": "90020003", "name": "Eurosport 4", "wsx": "eurosport4", "logo": "https://img.timvision.it/channels/eurosport4.png"}, {"epgid": "90020004", "name": "Eurosport 5", "wsx": "eurosport5", "logo": "https://img.timvision.it/channels/eurosport5.png"}, {"epgid": "90020005", "name": "Eurosport 6", "wsx": "eurosport6", "logo": "https://img.timvision.it/channels/eurosport6.png"}, ] 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} _____ ___ __ ____ _____ ____ ___ ___ _ _ |_ _|_ _| \\/ \\ \\ / /_ _/ ___|_ _/ _ \\| \\ | | | | | || |\\/| |\\ \\ / / | |\\___ \\| | | | | \\| | | | | || | | | \\ V / | | ___) | | |_| | |\\ | |_| |___|_| |_| \\_/ |___|____/___\\___/|_| \\_| {BLUE}{'═' * 70}{RESET} {CYAN} ◆ TIM VISION Auto-Manager · Discovery & Eurosport · Loop 3 Giorni ◆{RESET} {BLUE}{'═' * 70}{RESET}""" print(banner) # ── Gestione Proxy (proxy.txt) ──────────────────────────────────── # ── Gestione Proxy (proxy.txt o timvision_config.json) ──────────── class ProxyManager: """Carica, analizza, valida e ruota i proxy da proxy.txt o da timvision_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 # Se ha già lo schema (http://, https://, socks5://) if line.startswith(('http://', 'https://', 'socks5://', 'socks5h://')): return line parts = line.split(':') # Formato: host:port:username:password (es. DataImpulse / Webshare) if len(parts) == 4: host, port, user, pwd = parts return f"http://{user}:{pwd}@{host}:{port}" # Formato: username:password:host:port elif len(parts) == 4 and '@' not in line: return f"http://{parts[0]}:{parts[1]}@{parts[2]}:{parts[3]}" # Formato: host:port elif len(parts) == 2: return f"http://{parts[0]}:{parts[1]}" # Formato con @ elif '@' in line: return f"http://{line}" return None def set_config_proxy(self, proxy_conf: Any) -> None: """Imposta il proxy dal blocco di configurazione timvision_config.json""" 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]]: """Testa tutti i proxy su ipify per verificare latenza e IP pubblico""" 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 = { "timvision": { "proxy": { "http": "", "https": "" }, "portail_id": "vbdTj7eb6aM.", "pass_id": "", "live_token": "", "user_key_id": "", "device_key_id": "" } } if not os.path.exists(CONFIG_FILE): json.dump(default_config, open(CONFIG_FILE, 'w'), indent=2) cfg = default_config else: try: cfg = json.load(open(CONFIG_FILE)) if 'timvision' not in cfg: cfg['timvision'] = default_config['timvision'] except Exception: cfg = default_config # Imposta automaticamente il proxy da timvision_config.json if 'timvision' in cfg and 'proxy' in cfg['timvision']: proxy_manager.set_config_proxy(cfg['timvision']['proxy']) return cfg def save_config(cfg: Dict[str, Any]) -> None: # Preserva eventuali altre sezioni esistenti nel file (xaccel, monitor, ecc.) try: current = {} if os.path.exists(CONFIG_FILE): current = json.load(open(CONFIG_FILE)) current.update(cfg) with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(current, f, indent=2, ensure_ascii=False) except Exception: with open(CONFIG_FILE, 'w', encoding='utf-8') as f: json.dump(cfg, f, indent=2, ensure_ascii=False) def load_cache() -> None: global pssh_cache if os.path.exists(CACHE_FILE): try: pssh_cache = json.load(open(CACHE_FILE)) except Exception: pssh_cache = {} def save_cache() -> None: json.dump(pssh_cache, open(CACHE_FILE, 'w'), indent=2) # ── Autenticazione TIMVISION & Token Management ─────────────────── def create_token(config: Dict[str, Any], pass_id: Optional[str] = None, email: Optional[str] = None, password: Optional[str] = None) -> Tuple[str, str, Dict[str, Any]]: """Chiama l'endpoint createToken di CPTech tramite proxy""" portail_id = config['timvision'].get('portail_id', 'vbdTj7eb6aM.') if not pass_id: pass_id = config['timvision'].get('pass_id', '') if not email: email = config['timvision'].get('email', '') if not password: password = config['timvision'].get('password', '') data = { 'portailId': portail_id, 'media': 'web', 'vect': 'INTERNET', 'passIdType': 'pass', 'noCache': 'false' } if pass_id: data['passId'] = pass_id if email and password: data['login'] = email data['password'] = password headers = { 'User-Agent': UA, 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', 'Content-Type': 'application/x-www-form-urlencoded' } proxy = proxy_manager.get_dict() r = requests.post( 'https://auth-api-provider-cdn.prod.pass.tim.cptech.pro/provider/services/tiita-it/public/createToken', headers=headers, data=data, proxies=proxy, timeout=25 ) r.raise_for_status() resp = r.json()['response'] return resp['passToken'], resp.get('passId', ''), resp.get('userData', {}) def init_live_tv(config: Dict[str, Any], pass_token: str) -> Tuple[str, str, str, Dict[str, Any]]: """Inizializza la sessione LiveTV e ottiene il Bearer LiveToken""" device_key_id = f"{int(time.time()*1000)}-{uuid.uuid4().hex[:12]}" user_key_id = str(uuid.uuid4()) payload = { "ServiceRequest": { "InData": { "UseRmuTokenPlaceHolder": True, "PassData": {"Id": 0, "Token": pass_token}, "UserKeyId": user_key_id, "DeviceKeyId": device_key_id, "PDSData": {"GroupTypes": "1;4"}, "DeviceSpecificities": {"BpkCompatibility": True} } } } headers = { 'User-Agent': UA, 'Content-Type': 'application/json', 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', 'Accept': 'application/json, text/plain, */*' } proxy = proxy_manager.get_dict() r = requests.post( 'https://dtd-pfsbo-slc-ltv-api-eu-west-1.prod.bo.tim.cptech.pro/api/V4/zones/tiita/locations/it/devices/3/apps/1/jobs/InitLiveTV', json=payload, headers=headers, proxies=proxy, timeout=30 ) r.raise_for_status() out = r.json()['ServiceResponse']['OutData'] return out['LiveToken'], user_key_id, device_key_id, out def do_auth(config: Dict[str, Any], email: Optional[str] = None, password: Optional[str] = None, pass_id: Optional[str] = None) -> Dict[str, Any]: """Esegue il flusso completo di autenticazione e sincronizza sia timvision_tokens.json che timvision_config.json""" log('Autenticazione TIM VISION in corso...', CYAN) # Assicura che il proxy configurato sia attivo if 'timvision' in config and 'proxy' in config['timvision']: proxy_manager.set_config_proxy(config['timvision']['proxy']) if email is not None: config['timvision']['email'] = email if password is not None: config['timvision']['password'] = password if pass_id is not None: config['timvision']['pass_id'] = pass_id pass_token, new_pass_id, user_data = create_token( config, pass_id=config['timvision'].get('pass_id'), email=config['timvision'].get('email'), password=config['timvision'].get('password') ) log('passToken generato con successo!', GREEN) live_token, user_key_id, device_key_id, live_out = init_live_tv(config, pass_token) log('LiveToken generato con successo!', GREEN) final_pass_id = new_pass_id or config['timvision'].get('pass_id', '') tokens = { 'live_token': live_token, 'pass_token': pass_token, 'pass_id': final_pass_id, 'user_key_id': user_key_id, 'device_key_id': device_key_id, 'user_data': user_data, 'timestamp': time.time(), 'updated_at': datetime.now().isoformat() } # Salva su timvision_tokens.json with open(TOKEN_FILE, 'w', encoding='utf-8') as f: json.dump(tokens, f, indent=2, ensure_ascii=False) # Sincronizza anche timvision_config.json config['timvision']['live_token'] = live_token config['timvision']['pass_id'] = final_pass_id config['timvision']['user_key_id'] = user_key_id config['timvision']['device_key_id'] = device_key_id save_config(config) log(f'Token salvati e sincronizzati in {TOKEN_FILE} e {CONFIG_FILE}', GREEN) return tokens def get_tokens(config: Dict[str, Any], force_refresh: bool = False) -> Dict[str, Any]: """Recupera i token esistenti dalla cache/config se validi (< 8 ore) o li rigenera""" if 'timvision' in config and 'proxy' in config['timvision']: proxy_manager.set_config_proxy(config['timvision']['proxy']) if not force_refresh and os.path.exists(TOKEN_FILE): try: t = json.load(open(TOKEN_FILE)) if time.time() - t.get('timestamp', 0) < 3600 * 8 and t.get('live_token'): # Sincronizza user_key_id e device_key_id se assenti tv_cfg = config.get('timvision', {}) if 'user_key_id' not in t and 'user_key_id' in tv_cfg: t['user_key_id'] = tv_cfg['user_key_id'] if 'device_key_id' not in t and 'device_key_id' in tv_cfg: t['device_key_id'] = tv_cfg['device_key_id'] return t except Exception: pass # Fallback su timvision_config.json se contiene token e chiavi recenti tv_cfg = config.get('timvision', {}) if not force_refresh and tv_cfg.get('live_token') and tv_cfg.get('user_key_id') and tv_cfg.get('device_key_id'): if os.path.exists(TOKEN_FILE): try: t = json.load(open(TOKEN_FILE)) if time.time() - t.get('timestamp', 0) < 3600 * 8: return { 'live_token': tv_cfg['live_token'], 'pass_token': tv_cfg.get('pass_token', t.get('pass_token', '')), 'pass_id': tv_cfg.get('pass_id', t.get('pass_id', '')), 'user_key_id': tv_cfg['user_key_id'], 'device_key_id': tv_cfg['device_key_id'], 'timestamp': t.get('timestamp', time.time()) } except Exception: pass return do_auth(config) # ── Risoluzione Stream & DRM Widevine ───────────────────────────── def get_mpd_url(wsx_url: str, live_token: str) -> Optional[str]: """Interroga il server RMU di TIMVISION per ottenere l'URL dell'MPD""" headers = { 'User-Agent': UA, 'Authorization': f"Bearer {live_token}", 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/', 'xx-operator': 'pc', 'xx-service': 'mycanal' } proxy = proxy_manager.get_dict() try: r = requests.get(wsx_url, headers=headers, proxies=proxy, timeout=20, allow_redirects=True) if not r.ok: return None if '.mpd' in r.url: return r.url data = r.json() def find_url(obj: Any) -> Optional[str]: if isinstance(obj, dict): for k, v in obj.items(): if k in ('src', 'url', 'href') and isinstance(v, str) and v.startswith('http'): return v res = find_url(v) if res: return res elif isinstance(obj, list): for i in obj: res = find_url(i) if res: return res return None return find_url(data) except Exception as e: log(f"Errore recupero MPD: {e}", RED) return None def get_manifest(mpd_url: str, live_token: str) -> Optional[str]: """Scarica il manifesto MPD XML""" headers = { 'User-Agent': UA, 'Authorization': f"Bearer {live_token}" } proxy = proxy_manager.get_dict() try: r = requests.get(mpd_url, headers=headers, proxies=proxy, timeout=25) return r.text if r.ok else None except Exception: return None # System ID ufficiale Widevine WIDEVINE_SYSTEM_ID = "edef8ba9-79d6-4ace-a3c8-27dcd51d21ed" WIDEVINE_SYSTEM_ID_BYTES = bytes.fromhex("edef8ba979d64acea3c827dcd51d21ed") def is_widevine_pssh(pssh_b64: str) -> bool: """Verifica se una stringa PSSH Base64 appartiene a Widevine controllando il System ID""" 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 specificamente il blocco PSSH Widevine e il relativo KID dal manifesto DASH MPD. Ritorna una tupla: (pssh_b64, expected_kid). Prioritizza i blocchi PSSH appartenenti all'AdaptationSet video. """ 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: # Fallback globale su qualsiasi tag pssh che sia Widevine 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 # Ordina: prima le tracce video e con schemeIdUri esplicito candidates.sort(key=lambda c: (c['is_video'], c['is_wv_scheme']), reverse=True) best_pssh = candidates[0]['pssh'] # Estrai il KID atteso direttamente dal box PSSH Widevine expected_kid = None try: p = PSSH(best_pssh) if p.key_ids: first_kid = p.key_ids[0] if hasattr(first_kid, 'hex'): expected_kid = first_kid.hex if isinstance(first_kid.hex, str) else 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 extract_pssh(mpd_content: str) -> List[str]: """Compatibilità all'indietro: estrae solo i blocchi PSSH validi Widevine""" pssh, _ = extract_widevine_pssh(mpd_content) return [pssh] if pssh else [] def get_keys_wvd(pssh_b64: str, epgid: str, tokens: Dict[str, Any], expected_kid: Optional[str] = None) -> Tuple[Optional[str], Optional[str]]: """Decifra la chiave di contenuto Widevine usando WVD.wvd e l'endpoint GetLicence""" global pssh_cache # Cache key dinamica basata sul KID atteso o sull'hash della PSSH (risolve il bug del prefisso fisso) if expected_kid: cache_key = f"{epgid}:{expected_kid.lower()}" else: cache_key = f"{epgid}:{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') if not os.path.exists(WVD_FILE): log(f"ERRORE: File {WVD_FILE} non trovato nella cartella di lavoro!", RED) return None, None try: device = Device.load(WVD_FILE) cdm = Cdm.from_device(device) session = cdm.open() challenge = cdm.get_license_challenge(session, PSSH(pssh_b64)) challenge_b64 = base64.b64encode(challenge).decode() payload = { "ServiceRequest": { "InData": { "EpgId": int(epgid), "LiveToken": tokens['live_token'], "UserKeyId": tokens['user_key_id'], "DeviceKeyId": tokens['device_key_id'], "ChallengeInfo": challenge_b64, "Mode": "MKPL", "PersistentLicense": "false" } } } headers = { 'User-Agent': UA, 'Content-Type': 'application/json', 'Origin': 'https://timvisiontv.tim.it', 'Referer': 'https://timvisiontv.tim.it/' } proxy = proxy_manager.get_dict() r = requests.post( 'https://dtd-pfsbo-slc-ltv-api-eu-west-1.prod.bo.tim.cptech.pro/api/V4/zones/tiita/devices/31/apps/1/jobs/GetLicence', json=payload, headers=headers, proxies=proxy, timeout=30 ) if not r.ok: log(f'GetLicence fallita (HTTP {r.status_code})', RED) cdm.close(session) return None, None resp = r.json() service_resp = resp.get('ServiceResponse', {}) if service_resp.get('Status') != '0' and 'OutData' not in service_resp: err_data = service_resp.get('ErrorData', {}) log(f"Errore licenza DRM: {err_data.get('TechnicalStatus')} ({err_data.get('Details')})", RED) cdm.close(session) return None, None license_b64 = service_resp['OutData']['LicenseInfo'] license_bytes = base64.b64decode(license_b64) cdm.parse_license(session, license_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())) cdm.close(session) if not content_keys: log(f"Nessuna chiave di contenuto trovata nella licenza DRM", RED) return None, None # Cerca la chiave corrispondente all'expected_kid estratto dal PSSH 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 # Fallback sulla prima chiave di contenuto disponibile 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: {kid}:{key}', GREEN) return kid, key except Exception as e: log(f'Errore decifratura WVD: {e}', RED) return None, None def get_channel_info(channel: Dict[str, Any], config: Dict[str, Any], tokens: Dict[str, Any]) -> Tuple[Optional[str], Optional[str], Optional[str]]: """Estrae l'MPD URL e le chiavi KID:KEY per un canale""" wsx_url = f"https://dtd-pfsbo-ottd-rmu-front-eu-west-1.prod.bo.tim.cptech.pro/plfiles/v2/ital/dash-ssl/{channel['wsx']}.json?device=PC&capabilities=0x4" mpd_url = get_mpd_url(wsx_url, tokens['live_token']) if not mpd_url: log(f"{channel['name']}: MPD non disponibile", RED) return None, None, None mpd = get_manifest(mpd_url, tokens['live_token']) if not mpd: log(f"{channel['name']}: Manifest non scaricabile", RED) return mpd_url, None, None pssh_b64, expected_kid = extract_widevine_pssh(mpd) if not pssh_b64: log(f"{channel['name']}: PSSH Widevine non trovato nel manifesto", RED) return mpd_url, None, None kid, key = get_keys_wvd(pssh_b64, channel['epgid'], tokens, expected_kid=expected_kid) return mpd_url, kid, key # ── Funzione Principale di Estrazione e Salvataggio ──────────────── def extract_and_save_all(force_token_refresh: bool = False) -> Dict[str, Any]: """ Estrae tutti i 18 canali, risolve le chiavi DRM e salva il file canalidiscoveryeeurosporttim.json e la playlist timvision_playlist.m3u """ config = load_config() load_cache() proxy_manager.load(config_proxy=config.get('timvision', {}).get('proxy')) tokens = get_tokens(config, force_refresh=force_token_refresh) log(f"🚀 Avvio estrazione di tutti i {len(CHANNELS)} canali Discovery ed Eurosport...", CYAN) results = [] m3u_lines = ["#EXTM3U x-tvg-url=\"\""] found = 0 for i, ch in enumerate(CHANNELS, 1): print(f" [{i:2d}/{len(CHANNELS)}] {ch['name']:<22}", end=" ", flush=True) mpd_url, kid, key = get_channel_info(ch, config, tokens) if kid and key: p = urlparse(mpd_url) q = parse_qs(p.query) q['decryption_key'] = [f"{kid}:{key}"] q['v'] = ['best'] fhd_url = urlunparse((p.scheme, p.netloc, p.path, p.params, urlencode(q, doseq=True), p.fragment)).replace('%3A', ':') q['v'] = ['720p'] hd_url = urlunparse((p.scheme, p.netloc, p.path, p.params, urlencode(q, doseq=True), p.fragment)).replace('%3A', ':') q['v'] = ['480p'] sd_url = urlunparse((p.scheme, p.netloc, p.path, p.params, urlencode(q, doseq=True), p.fragment)).replace('%3A', ':') results.append({ "epgid": ch["epgid"], "name": ch["name"], "wsx": ch["wsx"], "logo": ch.get("logo", ""), "status": "OK", "manifest": mpd_url, "kid": kid, "key": key, "key_string": f"{kid}:{key}", "stream_url_fhd": fhd_url, "stream_url_hd": hd_url, "stream_url_sd": sd_url, "mpv_command": f'mpv "{mpd_url}" --demuxer-lavf-o="decryption_key={kid}:{key}"', "ffplay_command": f'ffplay -decryption_key {kid}:{key} "{mpd_url}"' }) # Playlist M3U Entry m3u_lines.append(f'#EXTINF:-1 tvg-id="{ch["epgid"]}" tvg-name="{ch["name"]}" tvg-logo="{ch.get("logo", "")}" group-title="TIMVISION",{ch["name"]}') m3u_lines.append('#KODIPROP:inputstream=inputstream.adaptive') m3u_lines.append('#KODIPROP:inputstream.adaptive.manifest_type=mpd') m3u_lines.append('#KODIPROP:inputstream.adaptive.license_type=com.widevine.alpha') m3u_lines.append(f'#KODIPROP:inputstream.adaptive.license_key={kid}:{key}') m3u_lines.append(fhd_url) print(f"{GREEN}✅ OK ({kid[:8]}...){RESET}") found += 1 elif mpd_url: results.append({ "epgid": ch["epgid"], "name": ch["name"], "wsx": ch["wsx"], "logo": ch.get("logo", ""), "status": "NO_KEYS", "manifest": mpd_url, "kid": None, "key": None, "key_string": None, "stream_url_fhd": mpd_url }) m3u_lines.append(f'#EXTINF:-1 tvg-id="{ch["epgid"]}" tvg-name="{ch["name"]}" tvg-logo="{ch.get("logo", "")}" group-title="TIMVISION",{ch["name"]}') m3u_lines.append(mpd_url) print(f"{YELLOW}⚠️ Solo Manifest{RESET}") else: results.append({ "epgid": ch["epgid"], "name": ch["name"], "wsx": ch["wsx"], "logo": ch.get("logo", ""), "status": "ERROR", "manifest": None, "kid": None, "key": None }) print(f"{RED}❌ Errore{RESET}") output_data = { "updated_at": datetime.now().isoformat(), "next_update": (datetime.now() + timedelta(seconds=REFRESH_INTERVAL_SECONDS)).isoformat(), "total_channels": len(CHANNELS), "keys_found": found, "proxy_used": proxy_manager.get_dict(), "channels": results } # Salva canalidiscoveryeeurosporttim.json with open(OUTPUT_JSON_FILE, 'w', encoding='utf-8') as f: json.dump(output_data, f, indent=2, ensure_ascii=False) # Salva anche timvision_playlist.m3u with open(PLAYLIST_FILE, 'w', encoding='utf-8') as f: f.write("\n".join(m3u_lines) + "\n") log(f"💾 Salvato con successo in: {BOLD}{OUTPUT_JSON_FILE}{RESET} ({found}/{len(CHANNELS)} canali)", GREEN) log(f"📋 Playlist M3U aggiornata in: {BOLD}{PLAYLIST_FILE}{RESET}", GREEN) return output_data # ── Loop Automatico con Standby di 3 Giorni ─────────────────────── def run_auto_loop() -> None: """ Esegue l'estrazione all'avvio, entra in standby e aggiorna le chiavi ogni 3 giorni. """ print_banner() config = load_config() proxy_manager.load(config_proxy=config.get('timvision', {}).get('proxy')) proxy_count = proxy_manager.count() proxy_desc = "proxy da timvision_config.json" if proxy_manager.config_proxy and not proxy_manager.proxies else f"{proxy_count} proxy" log(f"Proxy attivo: {BOLD}{proxy_desc}{RESET}", CYAN) while True: try: print(f"\n{BLUE}{'═' * 70}{RESET}") log("🔄 CICLO DI AGGIORNAMENTO CHIAVI & CANALI", CYAN) print(f"{BLUE}{'═' * 70}{RESET}") # Esegui estrazione completa extract_and_save_all(force_token_refresh=False) next_refresh = datetime.now() + timedelta(seconds=REFRESH_INTERVAL_SECONDS) print(f"\n{GREEN}{'═' * 70}{RESET}") log(f"✨ Estrazione completata! Il bot entra in STANDBY.", GREEN) log(f"⏰ Prossimo aggiornamento programmato il: {BOLD}{next_refresh.strftime('%d/%m/%Y alle ore %H:%M:%S')}{RESET} (tra 3 giorni)", YELLOW) print(f"{DIM}Premi CTRL+C per arrestare il bot in qualsiasi momento.{RESET}") print(f"{GREEN}{'═' * 70}{RESET}\n") # Standby loop con log orario start_time = time.time() while (time.time() - start_time) < REFRESH_INTERVAL_SECONDS: remaining_seconds = int(REFRESH_INTERVAL_SECONDS - (time.time() - start_time)) remaining_days = remaining_seconds // 86400 remaining_hours = (remaining_seconds % 86400) // 3600 remaining_minutes = (remaining_seconds % 3600) // 60 # Dormi per intervalli di 1 ora (3600s) o tempo rimanente sleep_chunk = min(3600, remaining_seconds) time.sleep(sleep_chunk) if remaining_seconds > 60: log(f"💤 Standby attivo · Prossimo aggiornamento tra: {remaining_days}g {remaining_hours}h {remaining_minutes}m", DIM) except KeyboardInterrupt: print(f"\n\n{YELLOW}🛑 Bot arrestato dall'utente.{RESET}\n") break except Exception as e: log(f"Errore durante il ciclo: {e}", RED) log("Nuovo tentativo tra 5 minuti...", YELLOW) time.sleep(300) if __name__ == '__main__': run_auto_loop()