Skip to content

Instantly share code, notes, and snippets.

@metinsanli
Last active September 25, 2024 14:29
Show Gist options
  • Save metinsanli/3bfd5df9de81cc55b23d507e0cd32583 to your computer and use it in GitHub Desktop.
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.
#!/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