import os import sys import json import http import time import pytz import titlecase import requests import unidecode import uuid import xmltodict import base64 import argparse from rich import print as rprint from rich.tree import Tree from rich.table import Table from time import sleep from requests.adapters import HTTPAdapter, Retry from datetime import datetime, timedelta from pywidevine.decrypt.wvdecrypt import WvDecrypt from helpers.proxy_environ import proxy_env parser = argparse.ArgumentParser() parser.add_argument("-i", dest="input", help="Specify the input url") parser.add_argument('-clear',dest="clear", help='',action='store_true') parser.add_argument("--debug", help="Enable debugging", action='store_true') args = parser.parse_args() url = args.input default_path = os.getcwd() config_path = os.path.join(default_path, "config") DaznConfigFile = os.path.join(config_path, "ConfigDazn.json") JsonFile = os.path.join(config_path,"JsonFile.json") DaznTokenFile = os.path.join(config_path, "DaznTokenFile.json") EventiFile = os.path.join(config_path, "Eventi.json") def Clean(text): try: replacements = { ':': '', '&': 'and', '+': '', ';': '', 'ó': '', '[': '', "'": '', ']': '', '/': '', '//': '', '’': "'", '*': '', '<': '', '>': '', '|': '', '~': '', '#': '', '%': '', '{': '', '}': '', ',': '', '?': '' } for old_char, new_string in replacements.items(): text = text.replace(old_char, new_string) except Exception: text = text.decode('utf-8') for old_char, new_string in replacements.items(): text = text.replace(old_char, new_string) return titlecase.titlecase(text) def ascii_clear(): os.system('cls||clear') tree = Tree("""[magenta] _____ _ ____ _ _ _ _ _ |_ _| | |__ ___ / ___| _ _ | |__ ___ _ __ | \ | | (_) _ __ (_) __ _ ___ | | | '_ \ / _ \ | | | | | | | '_ \ / _ \ | '__| | \| | | | | '_ \ | | / _` | / __| | | | | | | | __/ | |___ | |_| | | |_) | | __/ | | | |\ | | | | | | | | | | (_| | \__ \ |_| |_| |_| \___| \____| \__, | |_.__/ \___| |_| |_| \_| |_| |_| |_| _/ | \__,_| |___/ |___/ |__/ Autore: Alex 2023 """) rprint(tree) def Get_DaznConfigFile(): if os.path.isfile(DaznConfigFile): with open(DaznConfigFile, "r") as source: DaznConfig = json.load(source) else: rprint("Inserisci i dati dell'account Dazn:") Email = input("Inserisci Email: ") Password = input("Enter Password: ") Platform = "web" DaznConfig = { "Email": Email, "Password": Password, "Platform": Platform } with open(DaznConfigFile, "w") as file: json.dump(DaznConfig, file) rprint("[red]Config Dazn caricato correttamente") print() return DaznConfig def pretty_print_table(data): table = Table(show_header=True, header_style="magenta") table.add_column("Dazn Info", style="cyan") table.add_column("Valori", style="green") for key, value in data.items(): table.add_row(key, str(value)) rprint(table) print() def remove_duplicates(data, key): seen = set() result = [] for item in data: if item[key] not in seen: seen.add(item[key]) result.append(item) return result def GenDevID(): return str(uuid.uuid4()) def GetAndroidLogin(debug=False): loginUrl = "https://authentication-prod.ar.indazn.com/v5/SignIn" max_retries = 3 retry_delay=5 for attempt in range(max_retries + 1): if debug: tree = Tree(f"[dark_red]Tentativo {attempt + 1}/{max_retries + 1}") rprint(tree) try: headers = { 'Content-Type': 'application/json', 'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/111.0 signin/4.24.0.25 hyper/0.8.4 (web; production; it)' } res = requests.post(url=loginUrl, json=InfoDazn, headers=headers).json() if debug: print(f"Login response: {res}") if res["Result"] == "SignedIn": tree = Tree("\n[cyan]Login avvenuto con successo...") rprint(tree) with open(DaznTokenFile, "w") as destination: destination.write(json.dumps(res)) with open(DaznTokenFile, "r") as source: TokenData = json.loads(source.read()) return TokenData else: print(res) print('\nErrore durante il Login...') if attempt < max_retries: tree = Tree(f"\n[dark_goldenrod]Login Fallito! Ritento...") rprint(tree) sleep(retry_delay) else: tree = Tree(f"\n[dark_red]Login Fallito dopo {max_retries} tentativi.") rprint(tree) sys.exit(0) except Exception as e: if debug: tree = Tree(f"[dark_red]Tentatvi di Login Fallito: {e}") rprint(tree) if attempt < max_retries: tree = Tree(f"\n[dark_goldenrod]Login Fallito! Ritento...") rprint(tree) sleep(retry_delay) else: tree = Tree(f"\n[dark_red]Impossibile effettuare l'accesso dopo {max_retries} tentativi.") rprint(tree) sys.exit(0) def LoadTokenDazn(): if os.path.isfile(DaznTokenFile): os.remove(DaznTokenFile) TokenData = GetAndroidLogin() tree = Tree("\n[bright_cyan]Token Caricato...") with open(DaznTokenFile, "r") as file: Tokenfile = json.loads(file.read()) Token = Tokenfile["AuthToken"]["Token"] rprint(tree) return Token def Get_DaznCategorys(): meta_url = 'https://rail-router.discovery.indazn.com/eu/v4/Rail?platform=web&id=Sport&country=it&languageCode=en¶ms=PageType:Home;ContentType:None' headers = { 'X-Device-Id': '933cca090c27f3e1', 'X-Session-Id': '62a8f389-8e55-44af-aa9e-6fca252e7451', 'Connection': 'Keep-Alive', 'Accept-Encoding': 'gzip', 'authorization': 'Bearer ' + str(Token), 'User-Agent': 'Dalvik/2.1.0 (Linux; U; Android 9; A5010 Build/PI) : DAZN/2.10.22 (OnePlus A5010; com.dazn; 135856; Android 28 Mobile', } res = requests.get(url=meta_url, headers=headers).json() CatList = [] for x in res['Tiles']: CatList.append({'CatID':x['Id'],'AssetId':x['AssetId'],'Title':x['Title']}) #if not x['Title'] == 'Series and Documentaries': CatList = sorted(CatList, key=lambda k: str(k["Title"])) tree = Tree(f'\n[red]Elenco delle Categorie Dazn Disponibili...\n') indice = 0 for x in CatList: indice = indice + 1 tree.add(f" [dark_goldenrod]{str(indice)}:[bold green]{str(x['Title'])}") rprint(tree) return CatList def GetSession(): header = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.131 Safari/537.36'} session = requests.Session() session.mount("https://", HTTPAdapter( max_retries=Retry( total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], ) )) session.hooks = { "response": lambda r, *_, **__: r.raise_for_status(), } session.headers.update(header) return session def Get_Channels(data): ChannelList = [] Static_Channels = ['Eurosport 1','Zona DAZN','NFL Network','Milan TV','Inter TV','Eurosport 2','Red Bull TV','Unbeaten'] if data['Title'] == 'Linear channels': api_url = 'https://rail-router.discovery.indazn.com/eu/v4/Rail?platform=web&id=c0e3fbb9-435f-465f-8c39-42f340765346&country=it&languageCode=en¶ms=PageType:Sport%3BContentType:Sport%3BContentId:'+str(data['AssetId']) elif data['Title'] == 'NFL Game Pass': api_url = 'https://rail-router.discovery.indazn.com/eu/v4/Rail?platform=web&id=Live&country=it&languageCode=it¶ms=PageType:Competition%3BContentType:Competition%3BContentId:'+str(data['AssetId']) else: api_url = 'https://rail-router.discovery.indazn.com/eu/v4/Rail?platform=web&id=Live&country=it&languageCode=en¶ms=PageType:Sport%3BContentType:Sport%3BContentId:'+str(data['AssetId']) headers = { 'X-Device-Id': '933cca090c27f3e1', 'X-Session-Id': '62a8f389-8e55-44af-aa9e-6fca252e7451', 'Connection': 'Keep-Alive', 'Accept-Encoding': 'gzip', 'authorization': 'Bearer ' + str(Token), 'User-Agent': 'Dalvik/2.1.0 (Linux; U; Android 9; A5010 Build/PI) : DAZN/2.10.22 (OnePlus A5010; com.dazn; 135856; Android 28 Mobile', } res = Session.get(url=api_url, headers=headers).json() current_time = datetime.now(pytz.timezone('Europe/Rome')) for y in res['Tiles']: ImgUrl = "https://image.discovery.indazn.com/eu/v3/eu/none/"+str(y["Image"]["Id"])+"/fill/none/top/none/85/300/168/jpg/image" ChannelList.append({ "Image":ImgUrl, "ChannelID":y["Id"], "AssetId":y["AssetId"], "AssetTypeId":y["AssetTypeId"], "EventId":y["EventId"], "Type":y["Type"], "Title":unidecode.unidecode(Clean(y["Title"])), "Ends_in": "", "Description":y["Description"], "VerifyAge":y["VerifyAge"], }) ChannelList = [x for x in ChannelList if 'CatchUp' not in x['Type']] ChannelList = [x for x in ChannelList if 'UpComing' not in x['Type']] ChannelList = [x for x in ChannelList if 'OnDemand' not in x['Ends_in']] ChannelList = sorted(ChannelList, key=lambda k: str(k["Title"])) return ChannelList def GetManifest(y, data, id): Image = y["Image"] Title = y["Title"] Description = y["Description"] Type = y["Type"] EndsIn = y["Ends_in"] VerifyAge = y["VerifyAge"] api_url = "https://api.playback.indazn.com/v5/Playback?AssetId=" + str(y["AssetId"]) + "&PlayerId=test&DrmType=WIDEVINE&Platform=web&Format=MPEG-DASH&LanguageCode=en&Model=N%2FA&Secure=true&Latitude=null&Longitude=null&Manufacturer=unknown&PlayReadyInitiator=false&MtaLanguageCode=it&AppVersion=9.27.0" headers = { "authorization": "Bearer " + str(data), "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36", "x-dazn-device": str(id), } if VerifyAge == True: headers.update({"x-age-verification-pin": "0000"}) res = requests.get(url=api_url, headers=headers).json() for x in res["PlaybackDetails"]: if "tm-live" not in x["CdnName"]: mpd_url = x["ManifestUrl"] license_url = x["LaUrl"] old = x["ManifestUrl"].rsplit("/", 3)[0] new = 'https://dca-ac-live.cdn.indazn.com' mpd_url = x["ManifestUrl"].replace(old, new) license_url = x["LaUrl"] mpd_headers = { "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en-US,en;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Origin": "https://www.dazn.com", "Pragma": "no-cache", "Referer": "https://www.dazn.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36", } resp = requests.get(url=mpd_url, headers=mpd_headers) if not resp.status_code == 200: mpd_url = mpd_url.replace("dca", "dce") return Image, Title, Description, Type, EndsIn, mpd_url, license_url def pssh_kid_grabber(data): for adaptations in data["MPD"]["Period"]["AdaptationSet"]: if adaptations["@contentType"] == "video": for protections in adaptations["ContentProtection"]: try: kid = protections["@cenc:default_KID"] except: pass if protections["@schemeIdUri"] == "urn:uuid:EDEF8BA9-79D6-4ACE-A3C8-27DCD51D21ED" or protections["@schemeIdUri"] == "urn:uuid:edef8ba9-79d6-4ace-a3c8-27dcd51d21ed": try: pssh = protections["cenc:pssh"] except KeyError: pssh = protections["pssh"]["#text"] return pssh, kid def keys_request(b64data, url): wvdecrypt = WvDecrypt(b64data) challenge = wvdecrypt.get_challenge() license_headers = {"user-agent": "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.125 Safari/537.36"} resp = requests.post(url=url, data=challenge, headers=license_headers) license_decoded = resp.content license_b64 = base64.b64encode(license_decoded) wvdecrypt.update_license(license_b64) keys = wvdecrypt.start_process() return keys def proper(keys): commandline = [] for key in keys: if key.type == 'CONTENT': #commandline.append('--key') commandline.append('{}:{}'.format(key.kid.hex(), key.key.hex())) commandline = commandline[0] KID = commandline.split(':')[0] KEY = commandline.split(':')[-1] decrypt_key = [KID + ':' + KEY] return decrypt_key def read_(): with open(JsonFile, "r") as fr: return json.load(fr) def write_(data): with open(JsonFile, "w") as fr: fr.write(json.dumps(data, indent=4)) def get_stored(): stored = [] if os.path.isfile(JsonFile): return read_() return stored def dump_keys(Image, Title, Description, Type, EndsIn, mpd_url, license_url, decrypt_key, HlsUrl): new_keys = list(formatting(Image, Title, Description, Type, EndsIn, mpd_url, license_url, decrypt_key, HlsUrl)) stored = get_stored() for key in new_keys: if not any(item["Title"] == key["Title"] for item in stored): stored.append(key) write_(stored) return new_keys def formatting(Image, Title, Description, Type, EndsIn, mpd_url, license_url, decrypt_key, HlsUrl): return [{ "Image": Image, "Title": Title, "Description": Description, "Type": Type, "EndsIn": EndsIn, "MPD": mpd_url, "license_url": license_url, "decrypt_key": key, "HlsUrl": HlsUrl, } for idx, key in enumerate(decrypt_key, start=1)] if not args.input: ascii_clear() ip = proxy_env(args).Load() InfoDazn = Get_DaznConfigFile() pretty_print_table(InfoDazn) Token = LoadTokenDazn() Eventi_json = [] Categorie = Get_DaznCategorys() Session = GetSession() for categoria in Categorie: Eventlist = Get_Channels(categoria) Eventi_json.extend(Eventlist) rprint("\n[red]Prendo i canali per ogni Categoria...") Eventi_json_unici = remove_duplicates(Eventi_json, key='Title') Eventi_json_unici = sorted(Eventi_json_unici, key=lambda k: str(k["Title"])) with open(EventiFile, 'w', encoding='utf-8') as json_file: json.dump(Eventi_json_unici, json_file, ensure_ascii=False, indent=4) DevID = GenDevID() for x in Eventi_json_unici: Image, Title, Description, Type, EndsIn, mpd_url, license_url = GetManifest(x, Token, DevID) mpd_headers = { "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en-US,en;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Origin": "https://www.dazn.com", "Pragma": "no-cache", "Referer": "https://www.dazn.com/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36", } resp = requests.get(url=mpd_url, headers=mpd_headers) mpd_data = json.loads(json.dumps(xmltodict.parse(resp.text))) pssh, kid = pssh_kid_grabber(mpd_data) keys = keys_request(pssh, license_url) decrypt_key = proper(keys) HlsUrl = "manca url Hls" dump_keys(Image, Title, Description, Type, EndsIn, mpd_url, license_url, decrypt_key, HlsUrl) with open(EventFile, 'r') as eventlist_file, open(JsonFile, 'r') as jsonfile: eventlist = json.load(eventlist_file) json_data = json.load(jsonfile) eventlist_titles = [event['Title'] for event in eventlist] filtered_json_data = [item for item in json_data if item['Title'] in eventlist_titles] with open(JsonFile, 'w') as jsonfile: json.dump(filtered_json_data, jsonfile, indent=3)