tessl/pypi-streamlink

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

Overview

Eval results

Files

Plugin System

Name: tessl/pypi-streamlink
Author: tessl

Streamlink's plugin system enables support for hundreds of streaming services through a unified, extensible architecture. Plugins use decorators to define URL patterns and command-line arguments, making it easy to add support for new streaming services.

Capabilities

Plugin Base Class

The foundation class that all plugins inherit from, providing common functionality for URL matching, option handling, and stream extraction.

class Plugin:
    def __init__(self, session, url, options=None):
        """
        Initialize plugin instance.
        
        Parameters:
        - session: Streamlink session instance
        - url: URL that matched this plugin
        - options: Optional Options instance for plugin-specific settings
        """

    def set_option(self, key: str, value) -> None:
        """
        Set plugin-specific option.
        
        Parameters:
        - key: Option name
        - value: Option value
        """

    def get_option(self, key: str):
        """
        Get plugin-specific option value.
        
        Parameters:
        - key: Option name
        
        Returns:
        Option value or None if not set
        """

    def streams(self, stream_types=None, sorting_excludes=None) -> dict[str, Stream]:
        """
        Extract streams from the plugin's URL.
        
        Parameters:
        - stream_types: List of stream types to include (default: all)
        - sorting_excludes: List of stream qualities to exclude from sorting
        
        Returns:
        Dictionary mapping quality names to Stream instances
        
        Raises:
        NoStreamsError: If no streams are found
        PluginError: If extraction fails
        """

    def stream_weight(self, stream) -> tuple:
        """
        Calculate stream quality weight for sorting.
        
        Parameters:
        - stream: Stream instance to weigh
        
        Returns:
        Tuple representing stream quality weight for sorting
        """

Plugin instances have several important attributes available:

# Plugin attributes available in all plugin instances
session: Streamlink        # Streamlink session instance
options: Options           # Plugin options
cache: Cache              # Plugin cache object  
matchers: list            # URL pattern matchers
arguments: Arguments      # Plugin argument definitions
matches: list             # Current URL match results
matcher: Pattern          # Current matching pattern
match: Match              # Current match result

# Metadata attributes (can be set by plugins)
id: str                   # Unique plugin identifier
title: str                # Stream title
author: str               # Stream author/channel
category: str             # Stream category

Plugin Decorators

Decorators for defining plugin URL patterns and command-line arguments.

def pluginmatcher(pattern, priority=NORMAL_PRIORITY, name=None):
    """
    Decorator to define URL pattern matchers for plugins.
    
    Parameters:
    - pattern: Regular expression pattern to match URLs  
    - priority: Matcher priority (higher = checked first)
    - name: Optional name for the matcher
    
    Usage:
    @pluginmatcher(r'https?://example\.com/(\w+)')
    class ExamplePlugin(Plugin):
        ...
    """

def pluginargument(name, **kwargs):
    """
    Decorator to define command-line arguments for plugins.
    
    Parameters:
    - name: Argument name (converted to CLI format)
    - **kwargs: argparse.ArgumentParser.add_argument() parameters
    
    Usage:
    @pluginargument('username', help='Account username')
    @pluginargument('password', help='Account password', sensitive=True)
    class ExamplePlugin(Plugin):
        ...
    """

Priority Constants

Priority levels for URL pattern matching, controlling the order plugins are checked.

HIGH_PRIORITY = 30      # Highest priority - checked first
NORMAL_PRIORITY = 20    # Default priority for most plugins  
LOW_PRIORITY = 10       # Lower priority
NO_PRIORITY = 0         # Lowest priority - checked last

Plugin Options Classes

Classes for managing plugin arguments and options.

class PluginOptions(Options):
    """Alias for Options class used in plugin contexts"""

class PluginArguments:
    """Collection of plugin command-line arguments"""

class PluginArgument:
    """Individual plugin argument definition with CLI integration"""

Usage Examples

Basic Plugin Structure

import re
from streamlink.plugin import Plugin, pluginmatcher, pluginargument
from streamlink.stream import HTTPStream

@pluginmatcher(r'https?://example\.com/watch/(\w+)')
@pluginargument('quality', choices=['720p', '1080p'], default='best',
                help='Stream quality to extract')
class ExamplePlugin(Plugin):
    def _get_streams(self):
        # Extract stream URL from webpage
        stream_url = self._extract_stream_url()
        
        if stream_url:
            return {
                'best': HTTPStream(self.session, stream_url)
            }
        
        return {}
    
    def _extract_stream_url(self):
        # Implementation to extract stream URL
        match = self.match
        video_id = match.group(1)
        
        # Fetch and parse webpage
        res = self.session.http.get(self.url)
        # ... parsing logic ...
        
        return extracted_url

Multiple URL Patterns

@pluginmatcher(r'https?://example\.com/watch/(\w+)', priority=HIGH_PRIORITY)
@pluginmatcher(r'https?://example\.com/live/(\w+)')
@pluginmatcher(r'https?://(?:www\.)?example\.com/v/(\w+)', name='legacy')
class ExamplePlugin(Plugin):
    def _get_streams(self):
        # Check which matcher was used
        if self.matcher and self.matcher.name == 'legacy':
            return self._handle_legacy_url()
        else:
            return self._handle_modern_url()

Plugin Arguments and Options

@pluginmatcher(r'https?://premium\.example\.com/(\w+)')
@pluginargument('username', required=True, 
                help='Premium account username')
@pluginargument('password', required=True, sensitive=True,
                help='Premium account password')  
@pluginargument('quality', choices=['480p', '720p', '1080p'], 
                default='best', help='Stream quality preference')
class PremiumExamplePlugin(Plugin):
    def _get_streams(self):
        # Access plugin arguments
        username = self.get_option('username')
        password = self.get_option('password')
        quality_pref = self.get_option('quality')
        
        # Authenticate using credentials
        if not self._authenticate(username, password):
            raise PluginError('Authentication failed')
        
        # Extract streams based on quality preference
        streams = self._extract_streams()
        
        if quality_pref != 'best':
            # Filter streams by preference
            return {k: v for k, v in streams.items() 
                   if k == quality_pref or k in ['best', 'worst']}
        
        return streams

Advanced Stream Extraction

from streamlink.stream import HLSStream, DASHStream

@pluginmatcher(r'https?://adaptive\.example\.com/(\w+)')
class AdaptiveExamplePlugin(Plugin):
    def _get_streams(self):
        streams = {}
        
        # Get stream metadata
        metadata = self._get_stream_metadata()
        
        # Add HLS streams if available
        if metadata.get('hls_url'):
            hls_streams = HLSStream.parse_variant_playlist(
                self.session, metadata['hls_url']
            )
            streams.update(hls_streams)
        
        # Add DASH streams if available  
        if metadata.get('dash_url'):
            dash_streams = DASHStream.parse_manifest(
                self.session, metadata['dash_url']
            )
            streams.update(dash_streams)
        
        return streams
    
    def _get_stream_metadata(self):
        # Extract stream URLs from API or webpage
        api_url = f"https://api.example.com/stream/{self.match.group(1)}"
        res = self.session.http.get(api_url)
        return res.json()

Plugin with Caching

@pluginmatcher(r'https?://cached\.example\.com/(\w+)')
class CachedExamplePlugin(Plugin):
    def _get_streams(self):
        video_id = self.match.group(1)
        cache_key = f"stream_data_{video_id}"
        
        # Try to get cached data
        cached_data = self.cache.get(cache_key)
        
        if cached_data is None:
            # Fetch fresh data
            cached_data = self._fetch_stream_data(video_id)
            
            # Cache for 5 minutes
            self.cache.set(cache_key, cached_data, expires=300)
        
        # Create streams from cached data
        return self._create_streams(cached_data)

Error Handling in Plugins

from streamlink.exceptions import PluginError, NoStreamsError

@pluginmatcher(r'https?://robust\.example\.com/(\w+)')
class RobustExamplePlugin(Plugin):
    def _get_streams(self):
        try:
            # Attempt to extract streams
            streams = self._extract_streams()
            
            if not streams:
                raise NoStreamsError("No streams found for this URL")
                
            return streams
            
        except requests.RequestException as err:
            raise PluginError(f"Failed to fetch stream data: {err}")
        except ValueError as err:
            raise PluginError(f"Invalid stream data format: {err}")

Install with Tessl CLI