#!/usr/bin/python3 """ Yallo TV - Script per o11 Token anonimo rinnovato automaticamente ad ogni chiamata Azioni: channels | manifest | cdm """ import sys, os, base64, json, uuid import requests from logging.handlers import RotatingFileHandler import logging # --------------------------------------------------------------------------- # Parametri # --------------------------------------------------------------------------- def parse_params(params, name, default=""): for p in params: parts = p.split("=", 1) if parts[0] == name and len(parts) == 2: return parts[1] return default action = parse_params(sys.argv, 'action') channel_path = parse_params(sys.argv, 'id') encoded_challenge = parse_params(sys.argv, 'challenge') proxy = parse_params(sys.argv, 'proxy') # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- log_filename = f"yallo_{channel_path.replace('/', '_')}.log" if channel_path else "yallo.log" logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ RotatingFileHandler(log_filename, maxBytes=5*1024*1024, backupCount=3), logging.StreamHandler(), ] ) # --------------------------------------------------------------------------- # Costanti # --------------------------------------------------------------------------- USER_AGENT = "okhttp/5.1.0" API_BASE = "https://api.production.k8s.y3o.tv" DATA_BASE = "https://data.production.web.y3o.tv" proxies = {"http": proxy, "https": proxy} if proxy else {} HEADERS_BASE = { "connection": "Keep-Alive", "host": "api.production.k8s.y3o.tv", "user-agent": USER_AGENT, "x-tenant": "yallo", } # --------------------------------------------------------------------------- # Token anonimo (rinnovato ogni chiamata — scade ogni 24h) # --------------------------------------------------------------------------- def get_anonymous_token() -> str: """Ottiene un token JWT anonimo fresco dalle API Yallo.""" session = requests.Session() # Step 1: crea utente anonimo headers = {**HEADERS_BASE, "x-device-ads-id": str(uuid.uuid4())} resp = session.post( f"{API_BASE}/kamino/users/anonymous/", headers=headers, proxies=proxies, timeout=10 ) if not resp.ok: logging.error(f"Errore creazione utente anonimo: {resp.text}") sys.exit(1) user_id = resp.json().get("id") if not user_id: logging.error("user_id non trovato nella risposta") sys.exit(1) # Step 2: login con user_id → ottieni JWT resp2 = session.get( f"{API_BASE}/kamino/auth/login", headers=headers, params={"user_id": user_id, "device_type_name": "androidtv"}, proxies=proxies, timeout=10 ) if not resp2.ok: logging.error(f"Errore login anonimo: {resp2.text}") sys.exit(1) token = resp2.json().get("client_jwt_token") logging.debug(f"Token JWT ottenuto: {token[:30]}...") return token # --------------------------------------------------------------------------- # Canali # --------------------------------------------------------------------------- def get_channels(): resp = requests.get( f"{DATA_BASE}/channel/full.json", proxies=proxies, timeout=10 ) if not resp.ok: logging.error(f"Errore fetch canali: {resp.text}") sys.exit(1) channels = [] # Canali italiani identificati tramite short_name o path CANALI_IT = { "la7", "canale-5", "italia-1", "rai-news", "rai-1", "rai-2", "rai-3", "rai-4", "rai-5", "rai-gulp", "rai-sport", "rai-scuola", "rai-movie", "rai-yoyo", "rai-storia", "rai-premium", "rete4", "cielo", "giallo", "dmax-italia", "20-mediaset", "canale-italia", "focus", "marcopolo", "nove", "super", "uninettuno-university-tv", "la5", "iris", "italia-2", "La7 Cinema", "k2", "real-time", "sky-tg24", "boing", "cartoonito", "frisbee", "tv8", "class-tv-moda", "euronews-it", "rsi-la-1", "rsi-la-2", "teleticino", "ric", "tgcom24", "topcrime", "cine34", "5star", "sportitalia", "hgtv-italy", "mediaset-extra", "dmax", "la7", "tv-25", "rtl-1025-tv", "sport1", "tele-5", "food-network-italia", "sportdigital1", "wedo-movies", "wedo-big-stories", "filmgold", "wedo-sports", "top-serien", "top-true-crime", "top-filme", "top-sci-fi", "comedy-shows", "xplore", "27-twentyseven", "tgcom24", "topcrime", "cine34", "5star", "sportitalia-hd", "hgtv-italy", "mediaset-extra", } for ch in resp.json().get("result", []): try: short = ch.get("short_name", "") name_it = ch["long_name"].get("it") path = ch["naboo_urls"]["streams"] # Filtra: short_name nella whitelist OPPURE path contiene indicatori italiani path_it = any(x in path for x in [ "rai-", "rai_", "rainews", "raigulp", "canale-5", "italia-1", "italia_2", "rete-4", "la-7", "la_7", "cielo", "giallo_it", "dmaxitalia", "20_mediaset_it", "canale_italia", "focus_it", "marcopolo_it", "nove_it", "super_it", "uninettuno_it", "la_5", "iris", "k2_it", "real_time_it", "sky_tg24_it", "boing", "cartoonito_it", "frisbee_it", "mtvitalia", "class_tv_moda", "euronews_italian", "rsi-la", "teleticino", "rtl102_5", "sportitalia_it", "food_network_it", "hgtv_it", "mediaset_extra", "tgcom24_it", "topcrime_it", "cine34_it", "boing_plus_it", ]) if not (short in CANALI_IT or path_it): continue if not name_it: continue channels.append({"name": name_it, "path": path}) except (KeyError, TypeError): continue return channels # --------------------------------------------------------------------------- # Risorse streaming (manifest + licenza) # --------------------------------------------------------------------------- def get_resources(path: str, token: str, force_license: bool = False) -> dict: url = f"{API_BASE}/naboo{path}?timeshift=true" headers = { **HEADERS_BASE, "accept-encoding": "br,gzip", "accept-language": "it", "authorization": f"Bearer {token}", } resp = requests.get(url, headers=headers, proxies=proxies, timeout=15) if not resp.ok: logging.error(f"Errore get_resources ({resp.status_code}): {resp.text}") sys.exit(1) data = resp.json() manifest = data.get("url", "") license_url = data.get("license_url", "") # Se c'è DRM ma non lo vogliamo → prova versione chiara if "dashenc" in manifest and not force_license: clear_manifest = manifest.replace("dashenc", "dash").replace("m_drm_widevine.mpd", "m.mpd") test = requests.get(clear_manifest, timeout=10) if test.ok: logging.debug("Stream chiaro disponibile — nessun DRM necessario") return {"manifest": clear_manifest, "license_url": None} return {"manifest": manifest, "license_url": license_url} # --------------------------------------------------------------------------- # Licenza CDM # --------------------------------------------------------------------------- def post_license(encoded_challenge: str, license_url: str) -> str: challenge = base64.b64decode(encoded_challenge) headers = { "User-Agent": USER_AGENT, "Content-Type": "application/octet-stream", } resp = requests.post(license_url, headers=headers, data=challenge, proxies=proxies, timeout=15) if not resp.ok: logging.error(f"Errore licenza CDM ({resp.status_code}): {resp.text}") sys.exit(1) return base64.b64encode(resp.content).decode() # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- logging.error(f"Avvio con action='{action}' id='{channel_path}'") if action == "channels": items = get_channels() channels = [] for item in items: channels.append({ "Name": item["name"], "Mode": "live", "SessionManifest": True, "ManifestScript": f"id={item['path']}", "Autostart": False, "CdmType": "widevine", "CdmMode": "internal", "UseCdm": True, "Cdm": f"id={item['path']}", "Video": "best", }) print(json.dumps({"Channels": channels}, indent=2, ensure_ascii=False)) elif action == "manifest": token = get_anonymous_token() # token fresco ogni volta resources = get_resources(channel_path, token) print(json.dumps({"ManifestUrl": resources["manifest"]}, indent=2)) elif action == "cdm": token = get_anonymous_token() # token fresco ogni volta resources = get_resources(channel_path, token, force_license=True) licence = post_license(encoded_challenge, resources["license_url"]) print(licence) else: logging.error( "Azione non valida. Usa:\n" " python3 yallo.py action=channels\n" " python3 yallo.py action=manifest id=/live/...\n" " python3 yallo.py action=cdm id=/live/... challenge=\n" )