stream-api/backend/stream_providers.py

#!/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()
        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 get_ytdl(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 get_streamlink(upstream, proxy, logger):
    url = None
    try:
        runner = StreamlinkRunner(upstream, proxy)
        url = await runner.run()
    except Exception as e:
        logger.info(e)
    return url

async def get_any(upstream, proxy, logger):
    tasks = []
    tasks.append(asyncio.create_task(get_streamlink(upstream, proxy, logger)))
    tasks.append(asyncio.create_task(get_ytdl(upstream, proxy, logger)))
    result = None
    for task in asyncio.as_completed(tasks, timeout=5.0):
        result = await task
        if isinstance(result, str):
            break
    for task in tasks:
        if not task.done():
            task.cancel()
    return result