tessl/pypi-streamlink

Command-line utility and Python library for extracting video streams from various streaming services

Overview

Eval results

Files

Utilities

Name: tessl/pypi-streamlink
Author: tessl

Streamlink provides a comprehensive set of utility functions for URL manipulation, parsing, caching, and system integration. These utilities are commonly used in plugin development and stream processing.

Capabilities

URL Utilities

Functions for manipulating and working with URLs in plugin development.

def absolute_url(base, url):
    """
    Convert relative URL to absolute using base URL.
    
    Parameters:
    - base: Base URL for resolution
    - url: Relative or absolute URL
    
    Returns:
    Absolute URL string
    """

def prepend_www(url):
    """
    Add www prefix to URL if not present.
    
    Parameters:
    - url: URL to modify
    
    Returns:
    URL with www prefix added if needed
    """

def update_qsd(url, **params):
    """
    Update URL query string parameters.
    
    Parameters:
    - url: URL to modify
    - **params: Query parameters to add/update
    
    Returns:
    URL with updated query string
    """

def update_scheme(scheme, url, force=False):
    """
    Update URL scheme (protocol).
    
    Parameters:
    - scheme: New scheme (e.g., 'https://')
    - url: URL to modify
    - force: Force scheme change even if URL already has scheme
    
    Returns:
    URL with updated scheme
    """

def url_concat(base, *urls):
    """
    Concatenate URL segments properly handling slashes.
    
    Parameters:
    - base: Base URL
    - *urls: URL segments to append
    
    Returns:
    Concatenated URL
    """

def url_equal(a, b):
    """
    Compare URLs for equality ignoring insignificant differences.
    
    Parameters:
    - a: First URL
    - b: Second URL
    
    Returns:
    True if URLs are equivalent
    """

Parsing Utilities

Functions for parsing different data formats commonly encountered in streaming services.

def parse_html(data, **kwargs):
    """
    Parse HTML content using lxml.
    
    Parameters:
    - data: HTML string or bytes
    - **kwargs: Additional lxml.html parsing options
    
    Returns:
    Parsed HTML element tree
    """

def parse_json(data, **kwargs):
    """
    Parse JSON data with error handling.
    
    Parameters:
    - data: JSON string
    - **kwargs: Additional json.loads() options
    
    Returns:
    Parsed JSON object (dict, list, etc.)
    """

def parse_qsd(data, **kwargs):
    """
    Parse query string data into dictionary.
    
    Parameters:
    - data: Query string (can include leading '?')
    - **kwargs: Additional parsing options
    
    Returns:
    Dictionary of query parameters
    """

def parse_xml(data, **kwargs):
    """
    Parse XML content using lxml.
    
    Parameters:
    - data: XML string or bytes
    - **kwargs: Additional lxml.etree parsing options
    
    Returns:
    Parsed XML element tree
    """

Data Utilities

Functions for searching and manipulating data structures.

def search_dict(obj, key):
    """
    Recursively search nested dictionaries and lists for key.
    
    Parameters:
    - obj: Dictionary, list, or other object to search
    - key: Key to search for
    
    Returns:
    First value found for key, or None if not found
    """

Caching Utilities

LRU (Least Recently Used) cache implementation for efficient data caching.

class LRUCache:
    """
    Least Recently Used cache with maximum size limit.
    
    Automatically evicts least recently used items when
    maximum capacity is reached.
    """
    
    def __init__(self, maxsize=128):
        """
        Initialize LRU cache.
        
        Parameters:
        - maxsize: Maximum number of items to cache
        """
    
    def get(self, key, default=None):
        """
        Get cached value, updating access time.
        
        Parameters:
        - key: Cache key
        - default: Default value if key not found
        
        Returns:
        Cached value or default
        """
    
    def set(self, key, value):
        """
        Set cached value.
        
        Parameters:
        - key: Cache key
        - value: Value to cache
        """
    
    def clear(self):
        """Clear all cached items"""
    
    def size(self):
        """Get current cache size"""

System Utilities

System integration utilities for cross-platform functionality.

class NamedPipe:
    """
    Cross-platform named pipe implementation for streaming data.
    
    Provides a file-like interface for creating named pipes
    that can be used to stream data to external applications.
    """
    
    def __init__(self):
        """Initialize named pipe"""
    
    def open(self, mode='wb'):
        """
        Open named pipe for reading or writing.
        
        Parameters:
        - mode: File mode ('rb', 'wb', etc.)
        
        Returns:
        File-like object for pipe access
        """
    
    @property
    def path(self):
        """Path to the named pipe"""

def load_module(path):
    """
    Dynamically load Python module from file path.
    
    Parameters:
    - path: Path to Python module file
    
    Returns:
    Loaded module object
    """

Usage Examples

URL Manipulation

from streamlink.utils import absolute_url, update_qsd, url_concat

# Convert relative URLs to absolute
base = "https://example.com/videos/"
relative = "../api/stream.json"
absolute = absolute_url(base, relative)
# Result: "https://example.com/api/stream.json"

# Update query parameters
original_url = "https://api.example.com/stream?quality=720p"
updated_url = update_qsd(original_url, quality="1080p", format="hls")
# Result: "https://api.example.com/stream?quality=1080p&format=hls"

# Concatenate URL segments
api_base = "https://api.example.com"
endpoint = url_concat(api_base, "v1", "streams", "12345")
# Result: "https://api.example.com/v1/streams/12345"

HTML Parsing in Plugins

from streamlink.utils import parse_html

class ExamplePlugin(Plugin):
    def _extract_stream_url(self):
        # Fetch webpage
        res = self.session.http.get(self.url)
        
        # Parse HTML
        doc = parse_html(res.text)
        
        # Extract stream URL using XPath
        video_elements = doc.xpath(".//video[@src]")
        if video_elements:
            return video_elements[0].get("src")
        
        # Try alternative extraction
        script_elements = doc.xpath(".//script[contains(text(), 'streamUrl')]")
        for script in script_elements:
            text = script.text
            if "streamUrl" in text:
                # Extract URL from JavaScript
                start = text.find('"streamUrl":"') + 13
                end = text.find('"', start)
                return text[start:end]
        
        return None

JSON Data Processing

from streamlink.utils import parse_json, search_dict

class APIPlugin(Plugin):
    def _get_stream_data(self):
        # Fetch JSON data from API
        api_url = f"https://api.example.com/stream/{self.video_id}"
        res = self.session.http.get(api_url)
        
        # Parse JSON response
        data = parse_json(res.text)
        
        # Search nested data for stream URLs
        hls_url = search_dict(data, "hls_url")
        dash_url = search_dict(data, "dash_url")
        
        return {
            "hls": hls_url,
            "dash": dash_url,
            "title": search_dict(data, "title")
        }

Caching in Plugins

from streamlink.utils import LRUCache

class CachedPlugin(Plugin):
    # Class-level cache shared across instances
    _cache = LRUCache(maxsize=50)
    
    def _get_stream_metadata(self, video_id):
        # Check cache first
        cached_data = self._cache.get(video_id)
        if cached_data:
            return cached_data
        
        # Fetch fresh data
        api_url = f"https://api.example.com/video/{video_id}"
        res = self.session.http.get(api_url)
        metadata = parse_json(res.text)
        
        # Cache the result
        self._cache.set(video_id, metadata)
        
        return metadata

Named Pipe Usage

from streamlink.utils import NamedPipe
import subprocess

# Create named pipe for streaming to external player
pipe = NamedPipe()

# Start external player reading from pipe
player_cmd = ["vlc", pipe.path]
player_process = subprocess.Popen(player_cmd)

# Stream data to pipe
stream = streams['best']
with pipe.open('wb') as pipe_fd, stream.open() as stream_fd:
    while True:
        data = stream_fd.read(8192)
        if not data:
            break
        pipe_fd.write(data)

player_process.wait()

Query String Parsing

from streamlink.utils import parse_qsd

class RedirectPlugin(Plugin):
    def _extract_real_url(self):
        # Parse current URL query parameters
        from urllib.parse import urlparse
        parsed = urlparse(self.url)
        
        # Extract query parameters
        params = parse_qsd(parsed.query)
        
        # Get real URL from redirect parameter
        real_url = params.get('url') or params.get('redirect')
        
        if real_url:
            # Decode if URL-encoded
            from urllib.parse import unquote
            return unquote(real_url)
        
        return None

Dynamic Module Loading

from streamlink.utils import load_module
import os

def load_custom_plugins(plugin_dir):
    """Load custom plugins from directory"""
    plugins = []
    
    for filename in os.listdir(plugin_dir):
        if filename.endswith('.py') and not filename.startswith('_'):
            plugin_path = os.path.join(plugin_dir, filename)
            try:
                module = load_module(plugin_path)
                plugins.append(module)
            except Exception as e:
                print(f"Failed to load plugin {filename}: {e}")
    
    return plugins

Comprehensive URL Processing

from streamlink.utils import *

class URLProcessor:
    def process_stream_url(self, base_url, stream_path, **params):
        # Convert to absolute URL
        full_url = absolute_url(base_url, stream_path)
        
        # Update with parameters
        if params:
            full_url = update_qsd(full_url, **params)
        
        # Ensure HTTPS
        full_url = update_scheme("https://", full_url)
        
        # Add www if needed for certain domains
        if "example.com" in full_url:
            full_url = prepend_www(full_url)
        
        return full_url

# Usage in plugin
processor = URLProcessor()
stream_url = processor.process_stream_url(
    "https://example.com/videos/",
    "stream.m3u8",
    quality="720p",
    token="abc123"
)

Advanced Data Searching

from streamlink.utils import search_dict

def extract_nested_streams(api_response):
    """Extract stream URLs from complex nested API response"""
    
    # Search for various possible stream URL keys
    stream_keys = ['stream_url', 'hls_url', 'manifest_url', 'source']
    
    streams = {}
    for key in stream_keys:
        url = search_dict(api_response, key)
        if url:
            # Determine stream type from URL
            if '.m3u8' in url:
                streams['hls'] = url
            elif '.mpd' in url:
                streams['dash'] = url
            else:
                streams['http'] = url
    
    # Also search for quality-specific streams
    for quality in ['720p', '1080p', '480p']:
        quality_url = search_dict(api_response, f"{quality}_url")
        if quality_url:
            streams[quality] = quality_url
    
    return streams

Install with Tessl CLI