Get direct downoad link from "https://streamtape.com"

streamtape.py

__all__ = ["get_st_async", "get_st"]

"""
# Async:
download_link = await get_st_async('https://streamtape.com/v/b89re9b8er94b8/Demo_Video_1.mp4')
# Sync:
download_link = get_st('https://streamtape.com/v/b89re9b8er94b8/Demo_Video_1.mp4')
"""

import asyncio
import logging
import re
from typing import Optional

import requests
from aiohttp import ClientSession

LOG = logging.getLogger(__name__)

s_t_regex = re.compile(
    r"(?<=<script>)document\.getElementById\([\w\"'-]+\)\.innerHTML\s=\s(?P<code>[\w\s&=\"'().+/?-]+);(?=</script>)"
)
s_t_js_regex = re.compile(
    r"^\"([\w&=?./-]+)\"\s\+\s\(\'([\w&=?./-]+)\'\)(?:\.substring\((\d+)\))?$"
)


def run_code(code: str) -> Optional[str]:
    try:
        import js2py
    except ImportError:
        LOG.warning('Missing required module: "js2py"')
        return
    try:
        out = js2py.eval_js(code)
    except Exception as e:
        LOG.error(f"{e.__class__.__name__}: {e}")
    else:
        return out.strip()


def search_link(text: str) -> Optional[str]:
    if match := s_t_regex.search(text):
        code = match.group("code")
        link = None

        if m := s_t_js_regex.match(code):
            # try matching js code by regex
            str_1, str_2 = m.group(1), m.group(2)
            if n := m.group(3):
                link = str_1 + str_2[int(n) :]
            else:
                link = str_1 + str_2
        elif link := run_code(code):
            # Run random js code
            pass
        else:
            LOG.debug(f"No Match for: [ {code} ]")
            return

        if link and link.startswith("//"):
            link = "http:" + link
        return link


async def get_st_async(
    url: str, session: Optional[ClientSession] = None
) -> Optional[str]:
    if session:
        close = False
    else:
        close = True
        session = ClientSession()
    async with session.get(url) as resp:
        assert resp.status == 200
        text = await resp.text()
    if close:
        await session.close()
    return await asyncio.get_event_loop().run_in_executor(None, search_link, text)


def get_st(url: str) -> Optional[str]:
    resp = requests.get(url)
    assert resp.status_code == 200
    return search_link(resp.text)