89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""Resolve Doodstream embed players into direct download URLs."""
|
|
|
|
import random
|
|
import re
|
|
import string
|
|
import time
|
|
from typing import Any
|
|
|
|
import requests
|
|
from fake_useragent import UserAgent
|
|
|
|
from .Provider import Provider
|
|
|
|
# Precompiled regex patterns to extract the ``pass_md5`` endpoint and the
|
|
# session token embedded in the obfuscated player script. Compiling once keeps
|
|
# repeated invocations fast and documents the parsing intent.
|
|
PASS_MD5_PATTERN = re.compile(r"\$\.get\('([^']*/pass_md5/[^']*)'")
|
|
TOKEN_PATTERN = re.compile(r"token=([a-zA-Z0-9]+)")
|
|
|
|
|
|
class Doodstream(Provider):
|
|
"""Doodstream video provider implementation."""
|
|
|
|
def __init__(self):
|
|
self.RANDOM_USER_AGENT = UserAgent().random
|
|
|
|
def get_link(
|
|
self, embedded_link: str, timeout: int
|
|
) -> tuple[str, dict[str, Any]]:
|
|
"""
|
|
Extract direct download link from Doodstream embedded player.
|
|
|
|
Args:
|
|
embedded_link: URL of the embedded Doodstream player
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
Tuple of (direct_link, headers)
|
|
"""
|
|
headers = {
|
|
"User-Agent": self.RANDOM_USER_AGENT,
|
|
"Referer": "https://dood.li/",
|
|
}
|
|
|
|
def extract_data(pattern: re.Pattern[str], content: str) -> str | None:
|
|
"""Extract data using a compiled regex pattern."""
|
|
match = pattern.search(content)
|
|
return match.group(1) if match else None
|
|
|
|
def generate_random_string(length: int = 10) -> str:
|
|
"""Generate random alphanumeric string."""
|
|
charset = string.ascii_letters + string.digits
|
|
return "".join(random.choices(charset, k=length))
|
|
|
|
# WARNING: SSL verification disabled for doodstream compatibility
|
|
# This is a known limitation with this streaming provider
|
|
response = requests.get(
|
|
embedded_link,
|
|
headers=headers,
|
|
timeout=timeout,
|
|
verify=True, # Changed from False for security
|
|
)
|
|
response.raise_for_status()
|
|
|
|
pass_md5_url = extract_data(PASS_MD5_PATTERN, response.text)
|
|
if not pass_md5_url:
|
|
raise ValueError(f"pass_md5 URL not found using {embedded_link}.")
|
|
|
|
full_md5_url = f"https://dood.li{pass_md5_url}"
|
|
|
|
token = extract_data(TOKEN_PATTERN, response.text)
|
|
if not token:
|
|
raise ValueError(f"Token not found using {embedded_link}.")
|
|
|
|
md5_response = requests.get(
|
|
full_md5_url, headers=headers, timeout=timeout, verify=True
|
|
)
|
|
md5_response.raise_for_status()
|
|
video_base_url = md5_response.text.strip()
|
|
|
|
random_string = generate_random_string(10)
|
|
expiry = int(time.time())
|
|
|
|
direct_link = (
|
|
f"{video_base_url}{random_string}?token={token}&expiry={expiry}"
|
|
)
|
|
|
|
return direct_link, headers
|