Last active
September 25, 2024 14:29
-
-
Save metinsanli/3bfd5df9de81cc55b23d507e0cd32583 to your computer and use it in GitHub Desktop.
This python script writes football broadcast links taken from dadylivehd.com to the m3u file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
from http.client import HTTPResponse | |
from urllib.error import HTTPError | |
from urllib.request import urlopen, Request | |
from concurrent.futures import ThreadPoolExecutor | |
from re import compile, search, findall, DOTALL, MULTILINE | |
from logging import DEBUG, StreamHandler, getLogger, Formatter | |
from json import loads, dump | |
# Logging Setup | |
lgr = getLogger(__name__) | |
lgr.setLevel(DEBUG) | |
handler = StreamHandler() | |
formatter = Formatter( | |
"%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") | |
handler.setFormatter(formatter) | |
lgr.addHandler(handler) | |
base_url = 'https://dlhd.so' | |
UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:98.0) Gecko/20100101 Firefox/98.0' | |
def create_m3u_file(channels: list[dict]) -> None: | |
m3u_file = 'tv.m3u' | |
fl = open(m3u_file, 'w') | |
lines = list[str]() | |
lines.append(f'#EXTM3U') | |
for channel in channels: | |
lines.append(f'\n\n#EXTVLCOPT:http-user-agent={channel["user_agent"]}') | |
lines.append(f'\n#EXTVLCOPT:http-referrer={channel["referer"]}') | |
lines.append( | |
f'\n#EXTINF:-1 tvg-id="{channel["ch_id"]}" tvg-name="tv-{str(channel["ch_name"]).lower().replace(" ", "-")}" tvg-language="english" group-title="SPORT",{channel["ch_name"]}') | |
lines.append(f'\n{channel["stream_url"]}') | |
fl.writelines(lines) | |
fl.close() | |
if fl.closed: | |
lgr.info( | |
f'\"{m3u_file}\" dosyasina \"{len(channels)}\" adet kanal bilgisi yazildi.') | |
def solve_from_site() -> list: | |
hdrs = {'User-Agent': UA} | |
req = Request(url=base_url+"/24-7-channels.php", headers=hdrs) | |
dlive: HTTPResponse = urlopen(url=req) | |
ptrn_channels = compile( | |
pattern=r"<div\sclass=\"grid-item\"><a\shref=\"(?P<url>.*?)\"\starget=\"_blank\"\srel=\"noopener\"><span\sstyle=\"color:\s#000000;\"><strong>(?P<name>.*?)<\/strong><\/span><\/a><\/div>", flags=DOTALL | MULTILINE) | |
channels = findall(pattern=ptrn_channels, | |
string=dlive.read().decode('utf-8')) | |
chlist = list() | |
for line in channels: | |
if len(line) == 2 and line[0] != '' and line[1] != '': | |
ch_match = search(pattern=r"m-(\d+)\.p", string=line[0]) | |
ch_id = str(ch_match.group(1)).zfill(4) if ch_match else "" | |
chlist.append({'url': line[0], 'ch_id': ch_id, 'ch_name': line[1]}) | |
return chlist | |
def merge_channel_infos(stream_infos: list, channel_infos: list) -> list: | |
resp = list() | |
for si in stream_infos: | |
ci = next((ci for ci in channel_infos if ci['url'] in si['url']), None) | |
if ci and len(ci) > 0: | |
ci.update(si) | |
resp.append(ci) | |
return resp | |
def fetch_schedule() -> set: | |
api_url = "/schedule/schedule-generated.json" | |
hdrs = {'User-Agent': UA} | |
req = Request(url=base_url+api_url, headers=hdrs, method="POST") | |
sch: HTTPResponse = urlopen(url=req) | |
raw = sch.read().decode("utf-8") | |
js = loads(raw) | |
head_node = list(js)[0] | |
soccers = js[head_node]["Soccer"] | |
for event in soccers: | |
time_of_event = event["time"] | |
pieces_of_time = time_of_event.split(":") | |
hour = int(pieces_of_time[0]) | |
minute = int(pieces_of_time[1]) | |
gmt = +3 | |
hour = hour+gmt if hour+gmt <= 24 else hour+gmt-24 | |
event["time"] = f"{hour:02d}:{minute:02d}" | |
return soccers | |
def fetch_channel(channel_url: str) -> dict | None: | |
try: | |
hdrs1 = {'Referer': base_url+'/', 'User-Agent': UA} | |
req1 = Request(url=channel_url, headers=hdrs1) | |
raw1: HTTPResponse = urlopen(url=req1) | |
resp = raw1.read().decode('utf-8') | |
url_1 = findall(r'iframe src="(.*)" width', resp) | |
hdrs2 = {'Referer': channel_url, 'User-Agent': UA} | |
req2 = Request(url=url_1[0], headers=hdrs2) | |
raw2: HTTPResponse = urlopen(url=req2) | |
resp2 = raw2.read().decode('utf-8') | |
stream_url = findall(pattern=r"source:\s\'(.*)\'", string=resp2)[-1] | |
return {'url': channel_url, 'stream_url': stream_url, 'referer': url_1[0], 'user_agent': UA} | |
except HTTPError: | |
# lgr.exception(f"URL:{channel_url} Error: ") | |
return None | |
def multi_fetch(ch_urls: list) -> list: | |
with ThreadPoolExecutor(max_workers=10) as tpe: | |
return [item for item in tpe.map(fetch_channel, ch_urls) if item is not None] | |
def run(): | |
lgr.info("Program basladi.") | |
channel_infos = solve_from_site() | |
channel_urls = sorted({base_url+ci['url'] for ci in channel_infos}) | |
stream_infos = multi_fetch(channel_urls) | |
merged_infos = merge_channel_infos(stream_infos, channel_infos) | |
create_m3u_file(merged_infos) | |
lgr.info('Program bitti.') | |
if __name__ == '__main__': | |
run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment