89 lines
2.9 KiB
Python
Executable File
89 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import youtube_dl
|
|
import streamlink
|
|
import asyncio
|
|
|
|
class DummyLogger():
|
|
def debug(self, msg):
|
|
pass
|
|
def warning(self, msg):
|
|
pass
|
|
def error(self, msg):
|
|
pass
|
|
|
|
class StreamProvider():
|
|
def __init__(self, upstream, proxy):
|
|
self.upstream = upstream
|
|
self.proxy = None
|
|
proxy = str(proxy)
|
|
if len(proxy) > 5:
|
|
self.proxy = "socks5://" + proxy
|
|
|
|
class StreamlinkRunner(StreamProvider):
|
|
def stream(self):
|
|
session = streamlink.Streamlink()
|
|
session.set_option("http-timeout", 2.0)
|
|
if self.proxy is not None:
|
|
session.set_option("https-proxy", self.proxy)
|
|
session.set_option("http-proxy", self.proxy)
|
|
streams = session.streams(self.upstream)
|
|
if streams is not None:
|
|
for key in reversed(streams):
|
|
stream = streams.get(key)
|
|
if hasattr(stream, "url"):
|
|
return stream.url
|
|
return None
|
|
async def run(self):
|
|
return await asyncio.to_thread(self.stream)
|
|
|
|
class YoutubeRunner(StreamProvider):
|
|
def stream(self):
|
|
opts = {}
|
|
opts["logger"] = DummyLogger()
|
|
if isinstance(self.proxy, str):
|
|
opts["proxy"] = self.proxy
|
|
best_url = None
|
|
with youtube_dl.YoutubeDL(opts) as ydl:
|
|
info = ydl.extract_info(self.upstream, download=False)
|
|
vformats = info.get("formats")
|
|
best_format = {}
|
|
best_format["quality"] = 0
|
|
if isinstance(vformats, list):
|
|
for vformat in vformats:
|
|
acodec = vformat.get("acodec")
|
|
vcodec = vformat.get("vcodec")
|
|
new_quality = vformat.get("quality")
|
|
old_quality = best_format.get("quality")
|
|
new_url = vformat.get("url")
|
|
if (isinstance(acodec, str) and
|
|
isinstance(vcodec, str) and
|
|
isinstance(new_quality, int) and
|
|
isinstance(old_quality, int) and
|
|
isinstance(new_url, str) and
|
|
acodec != "none" and
|
|
vcodec != "none" and
|
|
new_quality > old_quality):
|
|
best_format = vformat
|
|
best_url = new_url
|
|
return best_url
|
|
async def run(self):
|
|
return await asyncio.to_thread(self.stream)
|
|
|
|
async def ytdl_get(upstream, proxy, logger):
|
|
url = None
|
|
try:
|
|
runner = YoutubeRunner(upstream, proxy)
|
|
url = await runner.run()
|
|
except Exception as e:
|
|
logger.info(e)
|
|
return url
|
|
|
|
async def streamlink_get(upstream, proxy, logger):
|
|
url = None
|
|
try:
|
|
runner = StreamlinkRunner(upstream, proxy)
|
|
url = await runner.run()
|
|
except Exception as e:
|
|
logger.info(e)
|
|
return url
|