gitea-mirror/FastAnime
1
0
Fork 0
mirror of https://github.com/Benexl/FastAnime.git synced 2026-04-28 11:53:08 -07:00
Code Issues Packages Projects Releases Wiki Activity

feat: plugins system

Browse Source
This commit is contained in:
Benexl
2025-08-18 16:02:37 +03:00
parent eaedf3268d
commit 88b707e060
18 changed files with 1356 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

319
PLUGINS.md Normal file
View File

@@ -0,0 +1,319 @@
# Viu Plugin Development Guide
This guide explains how to create plugins for viu, the terminal-based anime streaming tool.
## Overview
Viu supports four types of plugins:
- **Providers**: Add support for new anime streaming websites
- **Players**: Add support for new media players
- **Selectors**: Add support for new interactive selection tools
- **Commands**: Add new CLI commands to viu
## Plugin Structure
Every plugin must be a Git repository with the following structure:
```
your-plugin-repo/
├── plugin.info.toml # Plugin metadata (required)
├── your_module.py # Your plugin implementation
├── config.toml # Default configuration (optional)
├── requirements.txt # Dependencies (optional)
├── utils.py # Additional modules (optional)
├── helpers/ # Subdirectories supported (optional)
│ ├── __init__.py
│ └── parser.py
└── README.md # Documentation (recommended)
```
### Multi-File Plugins
Viu supports plugins with multiple Python files. You can organize your plugin code across multiple modules and import between them normally:
```python
# In your main plugin file
from utils import helper_function
from helpers.parser import ResponseParser
class MyProvider(BaseAnimeProvider):
def __init__(self, client, **config):
self.parser = ResponseParser()
# ... rest of implementation
```
The plugin system automatically adds your plugin directory to Python's import path during loading, so relative imports work as expected.
### Plugin Manifest (`plugin.info.toml`)
Every plugin repository must contain a `plugin.info.toml` file at its root:
```toml
[plugin]
name = "My Awesome Plugin"
version = "1.0.0"
description = "Adds support for Example Anime Site"
author = "Your Name"
homepage = "https://github.com/yourname/viu-example-plugin"
requires_python = ">=3.11"
[components]
# Specify which components your plugin provides
provider = "example_provider:ExampleProvider" # format: module:class
# player = "my_player:MyPlayer" # (if providing a player)
# selector = "my_selector:MySelector" # (if providing a selector)
# command = "my_command:my_command_func" # (if providing a command)
```
## Provider Plugins
Provider plugins add support for new anime streaming websites.
### Requirements
Your provider class must inherit from `BaseAnimeProvider` and implement:
- `search(query: str) -> SearchResults`
- `get(anime_id: str) -> Anime`
- `episode_streams(anime_id: str, episode: str) -> List[Server]`
### Example Provider Plugin
**plugin.info.toml:**
```toml
[plugin]
name = "Example Anime Provider"
version = "1.0.0"
description = "Adds support for example.anime.site"
[components]
provider = "example_provider:ExampleProvider"
```
**example_provider.py:**
```python
from typing import List
from httpx import Client
# These imports work because viu adds the plugin path to sys.path
from viu_media.libs.provider.anime.base import BaseAnimeProvider
from viu_media.libs.provider.anime.types import SearchResults, Anime, Server
class ExampleProvider(BaseAnimeProvider):
HEADERS = {
"Referer": "https://example.anime.site/",
}
def __init__(self, client: Client, **config):
self.client = client
# Access plugin configuration
self.timeout = config.get("timeout", 30)
self.preferred_quality = config.get("preferred_quality", "720p")
def search(self, query: str) -> SearchResults:
# Implement search logic
response = self.client.get(f"https://example.anime.site/search?q={query}")
# Parse response and return SearchResults
return SearchResults(...)
def get(self, anime_id: str) -> Anime:
# Implement anime details fetching
response = self.client.get(f"https://example.anime.site/anime/{anime_id}")
# Parse response and return Anime
return Anime(...)
def episode_streams(self, anime_id: str, episode: str) -> List[Server]:
# Implement stream URL extraction
response = self.client.get(f"https://example.anime.site/watch/{anime_id}/{episode}")
# Parse response and return list of Server objects
return [Server(...)]
```
## Player Plugins
Player plugins add support for new media players.
### Requirements
Your player class must inherit from `BasePlayer` and implement:
- `play(media_url: str, **kwargs) -> None`
### Example Player Plugin
**plugin.info.toml:**
```toml
[plugin]
name = "Custom Player"
version = "1.0.0"
description = "Adds support for my custom media player"
[components]
player = "custom_player:CustomPlayer"
```
**custom_player.py:**
```python
import subprocess
from viu_media.libs.player.base import BasePlayer
class CustomPlayer(BasePlayer):
def __init__(self, **config):
self.executable = config.get("executable", "my-player")
self.extra_args = config.get("extra_args", [])
def play(self, media_url: str, **kwargs) -> None:
cmd = [self.executable] + self.extra_args + [media_url]
subprocess.run(cmd)
```
## Selector Plugins
Selector plugins add support for new interactive selection tools.
### Requirements
Your selector class must inherit from `BaseSelector` and implement:
- `choose(choices: List[str], **kwargs) -> str`
- `confirm(message: str, **kwargs) -> bool`
- `ask(message: str, **kwargs) -> str`
## Command Plugins
Command plugins add new CLI commands to viu.
### Example Command Plugin
**plugin.info.toml:**
```toml
[plugin]
name = "My Command"
version = "1.0.0"
description = "Adds a custom command to viu"
[components]
command = "my_command:my_command"
```
**my_command.py:**
```python
import click
@click.command()
@click.argument("arg1")
def my_command(arg1: str):
"""My custom command description."""
click.echo(f"Hello from plugin command with arg: {arg1}")
```
## Plugin Configuration
Plugins can include a default configuration file (`config.toml`) in their repository root. When a plugin is installed, this default configuration is automatically copied to the user's `~/.config/viu/plugins.config.toml` file.
**Example `config.toml` in plugin repository:**
```toml
# Default configuration for My Plugin
[my-plugin-name]
timeout = 30
preferred_quality = "720p"
custom_option = "default_value"
```
**After installation, users can customize by editing `~/.config/viu/plugins.config.toml`:**
```toml
[my-plugin-name]
timeout = 60 # Customized value
preferred_quality = "1080p" # Customized value
custom_option = "my_value" # Customized value
```
Access this configuration in your plugin constructor via the `**config` parameter.
## Installation and Usage
### For Plugin Developers
1. Create your plugin repository following the structure above
2. Test your plugin locally
3. Publish your repository on GitHub/GitLab
4. Share the installation command with users
### For Users
Install a plugin:
```bash
viu plugin add --type provider myplugin github:user/viu-myplugin
```
Configure the plugin by editing `~/.config/viu/plugins.config.toml`:
```toml
[myplugin]
option1 = "value1"
option2 = "value2"
```
Use the plugin:
```bash
viu --provider myplugin search "anime name"
```
## Dependencies
If your plugin requires additional Python packages, include a `requirements.txt` file in your repository root. Users will need to install these manually:
```bash
pip install -r requirements.txt
```
## Best Practices
1. **Error Handling**: Implement proper error handling and logging
2. **Configuration**: Make your plugin configurable through the config system
3. **Documentation**: Include a README.md with usage instructions
4. **Testing**: Test your plugin thoroughly before publishing
5. **Versioning**: Use semantic versioning for your plugin releases
6. **Compatibility**: Specify minimum Python version requirements
## Plugin Management Commands
```bash
# Install a plugin
viu plugin add --type provider myplugin github:user/viu-myplugin
# List installed plugins
viu plugin list
viu plugin list --type provider
# Update a plugin
viu plugin update --type provider myplugin
# Remove a plugin
viu plugin remove --type provider myplugin
```
## Example Plugins
Check out these example plugin repositories:
- [Example Provider Plugin](https://github.com/example/viu-example-provider)
- [Example Player Plugin](https://github.com/example/viu-example-player)
## Support
For plugin development support:
- Open an issue in the main viu repository
- Join the Discord server: https://discord.gg/C4rhMA4mmK

1
pyproject.toml
View File

@@ -11,6 +11,7 @@ dependencies = [
"inquirerpy>=0.3.4",
"pydantic>=2.11.7",
"rich>=13.9.2",
"tomli-w>=1.0.0",
]
[project.scripts]

11
uv.lock generated
View File

@@ -3553,6 +3553,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/82/4f/1695e70ceb3604f19eda9908e289c687ea81c4fecef4d90a9d1d0f2f7ae9/thefuzz-0.22.1-py3-none-any.whl", hash = "sha256:59729b33556850b90e1093c4cf9e618af6f2e4c985df193fdf3c5b5cf02ca481", size = 8245, upload-time = "2024-01-19T19:18:20.362Z" },
]
[[package]]
name = "tomli-w"
version = "1.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/19/75/241269d1da26b624c0d5e110e8149093c759b7a286138f4efd61a60e75fe/tomli_w-1.2.0.tar.gz", hash = "sha256:2dd14fac5a47c27be9cd4c976af5a12d87fb1f0b4512f81d69cce3b35ae25021", size = 7184, upload-time = "2025-01-15T12:07:24.262Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c7/18/c86eb8e0202e32dd3df50d43d7ff9854f8e0603945ff398974c1d91ac1ef/tomli_w-1.2.0-py3-none-any.whl", hash = "sha256:188306098d013b691fcadc011abd66727d3c414c571bb01b1a174ba8c983cf90", size = 6675, upload-time = "2025-01-15T12:07:22.074Z" },
]
[[package]]
name = "typing-extensions"
version = "4.14.1"
@@ -3598,6 +3607,7 @@ dependencies = [
{ name = "inquirerpy" },
{ name = "pydantic" },
{ name = "rich" },
{ name = "tomli-w" },
]
[package.optional-dependencies]
@@ -3667,6 +3677,7 @@ requires-dist = [
{ name = "pypresence", marker = "extra == 'discord'", specifier = ">=4.3.0" },
{ name = "rich", specifier = ">=13.9.2" },
{ name = "thefuzz", marker = "extra == 'standard'", specifier = ">=0.22.1" },
{ name = "tomli-w", specifier = ">=1.0.0" },
{ name = "yt-dlp", marker = "extra == 'download'", specifier = ">=2025.7.21" },
{ name = "yt-dlp", marker = "extra == 'standard'", specifier = ">=2025.7.21" },
]

1
viu_media/cli/cli.py
View File

@@ -39,6 +39,7 @@ commands = {
"worker": "worker.worker",
"queue": "queue.queue",
"completions": "completions.completions",
"plugin": "plugin.plugin",
}

5
viu_media/cli/commands/plugin/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Plugin management commands for viu."""
from .cmd import plugin
__all__ = ["plugin"]

24
viu_media/cli/commands/plugin/cmd.py Normal file
View File

@@ -0,0 +1,24 @@
"""Main plugin command group."""
import click
from ...utils.lazyloader import LazyGroup
lazy_subcommands = {
"add": "add.add",
"remove": "remove.remove",
"list": "list_plugins.list_plugins",
"update": "update.update",
}
@click.group(
name="plugin",
cls=LazyGroup,
root="viu_media.cli.commands.plugin.commands",
lazy_subcommands=lazy_subcommands,
help="Manage viu plugins (providers, players, selectors, commands)"
)
def plugin() -> None:
"""Manage viu plugins."""
pass

1
viu_media/cli/commands/plugin/commands/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Plugin command implementations."""

54
viu_media/cli/commands/plugin/commands/add.py Normal file
View File

@@ -0,0 +1,54 @@
"""Add plugin command."""
import click
from rich.console import Console
from typing import cast
from viu_media.core.plugins.manager import PluginError, plugin_manager, ComponentType
console = Console()
@click.command()
@click.option(
"--type",
"plugin_type",
type=click.Choice(["provider", "player", "selector", "command"]),
required=True,
help="Type of plugin to install"
)
@click.option(
"--force",
is_flag=True,
help="Force installation, overwriting existing plugin"
)
@click.argument("name")
@click.argument("source")
def add(plugin_type: str, name: str, source: str, force: bool) -> None:
"""Install a plugin from a Git repository.
NAME: Local name for the plugin
SOURCE: Git source (e.g., 'github:user/repo' or full URL)
Examples:
viu plugin add --type provider gogoanime github:user/viu-gogoanime
viu plugin add --type player custom-mpv https://github.com/user/viu-mpv-plugin
"""
try:
console.print(f"Installing {plugin_type} plugin '{name}' from {source}...")
plugin_manager.add_plugin(cast(ComponentType, plugin_type), name, source, force=force)
console.print(f"✅ Successfully installed plugin '{name}'", style="green")
# Show configuration hint
from viu_media.core.constants import PLUGINS_CONFIG
console.print(
f"\n💡 Configure the plugin by editing: {PLUGINS_CONFIG}",
style="blue"
)
except PluginError as e:
console.print(f"❌ Failed to install plugin: {e}", style="red")
raise click.ClickException(str(e))
except Exception as e:
console.print(f"❌ Unexpected error: {e}", style="red")
raise click.ClickException(f"Unexpected error: {e}")

74
viu_media/cli/commands/plugin/commands/list_plugins.py Normal file
View File

@@ -0,0 +1,74 @@
"""List plugins command."""
import click
from rich.console import Console
from rich.table import Table
from typing import cast
from viu_media.core.plugins.manager import plugin_manager, ComponentType
console = Console()
@click.command(name="list")
@click.option(
"--type",
"plugin_type",
type=click.Choice(["provider", "player", "selector", "command"]),
help="Filter by plugin type"
)
def list_plugins(plugin_type: str) -> None:
"""List installed plugins.
Examples:
viu plugin list
viu plugin list --type provider
"""
all_plugins = plugin_manager.list_plugins()
# Filter by type if specified
if plugin_type:
plugins_to_show = {cast(ComponentType, plugin_type): all_plugins[cast(ComponentType, plugin_type)]}
else:
plugins_to_show = all_plugins
# Count total plugins
total_count = sum(len(plugins) for plugins in plugins_to_show.values())
if total_count == 0:
if plugin_type:
console.print(f"No {plugin_type} plugins installed.", style="yellow")
else:
console.print("No plugins installed.", style="yellow")
console.print("Install plugins with: viu plugin add --type <type> <name> <source>")
return
# Create table
table = Table(title="Installed Plugins")
table.add_column("Type", style="cyan")
table.add_column("Name", style="green")
table.add_column("Version", style="yellow")
table.add_column("Source", style="blue")
table.add_column("Path", style="magenta")
# Add rows
for component_type, plugins in plugins_to_show.items():
for name, plugin_info in plugins.items():
table.add_row(
component_type,
name,
plugin_info.version or "unknown",
plugin_info.source,
str(plugin_info.path)
)
console.print(table)
console.print(f"\nTotal: {total_count} plugin(s)")
# Show configuration hint if plugins exist
if total_count > 0:
from viu_media.core.constants import PLUGINS_CONFIG
console.print(
f"\n💡 Configure plugins by editing: {PLUGINS_CONFIG}",
style="blue"
)

43
viu_media/cli/commands/plugin/commands/remove.py Normal file
View File

@@ -0,0 +1,43 @@
"""Remove plugin command."""
import click
from rich.console import Console
from typing import cast
from viu_media.core.plugins.manager import PluginError, PluginNotFoundError, plugin_manager, ComponentType
console = Console()
@click.command()
@click.option(
"--type",
"plugin_type",
type=click.Choice(["provider", "player", "selector", "command"]),
required=True,
help="Type of plugin to remove"
)
@click.argument("name")
def remove(plugin_type: str, name: str) -> None:
"""Remove an installed plugin.
NAME: Name of the plugin to remove
Examples:
viu plugin remove --type provider gogoanime
viu plugin remove --type player custom-mpv
"""
try:
console.print(f"Removing {plugin_type} plugin '{name}'...")
plugin_manager.remove_plugin(cast(ComponentType, plugin_type), name)
console.print(f"✅ Successfully removed plugin '{name}'", style="green")
except PluginNotFoundError as e:
console.print(f"❌ Plugin not found: {e}", style="red")
raise click.ClickException(str(e))
except PluginError as e:
console.print(f"❌ Failed to remove plugin: {e}", style="red")
raise click.ClickException(str(e))
except Exception as e:
console.print(f"❌ Unexpected error: {e}", style="red")
raise click.ClickException(f"Unexpected error: {e}")

43
viu_media/cli/commands/plugin/commands/update.py Normal file
View File

@@ -0,0 +1,43 @@
"""Update plugin command."""
import click
from rich.console import Console
from typing import cast
from viu_media.core.plugins.manager import PluginError, PluginNotFoundError, plugin_manager, ComponentType
console = Console()
@click.command()
@click.option(
"--type",
"plugin_type",
type=click.Choice(["provider", "player", "selector", "command"]),
required=True,
help="Type of plugin to update"
)
@click.argument("name")
def update(plugin_type: str, name: str) -> None:
"""Update an installed plugin by pulling from Git.
NAME: Name of the plugin to update
Examples:
viu plugin update --type provider gogoanime
viu plugin update --type player custom-mpv
"""
try:
console.print(f"Updating {plugin_type} plugin '{name}'...")
plugin_manager.update_plugin(cast(ComponentType, plugin_type), name)
console.print(f"✅ Successfully updated plugin '{name}'", style="green")
except PluginNotFoundError as e:
console.print(f"❌ Plugin not found: {e}", style="red")
raise click.ClickException(str(e))
except PluginError as e:
console.print(f"❌ Failed to update plugin: {e}", style="red")
raise click.ClickException(str(e))
except Exception as e:
console.print(f"❌ Unexpected error: {e}", style="red")
raise click.ClickException(f"Unexpected error: {e}")

6
viu_media/core/constants.py
View File

@@ -76,11 +76,17 @@ else:
USER_APPLICATIONS = Path.home() / ".local" / "share" / "applications"
LOG_FOLDER = APP_CACHE_DIR / "logs"
# Plugin system paths
PLUGINS_DIR = APP_DATA_DIR / "plugins"
PLUGINS_MANIFEST = APP_DATA_DIR / "plugins.toml"
PLUGINS_CONFIG = APP_DATA_DIR / "plugins.config.toml"
# USER_APPLICATIONS.mkdir(parents=True,exist_ok=True)
APP_DATA_DIR.mkdir(parents=True, exist_ok=True)
APP_CACHE_DIR.mkdir(parents=True, exist_ok=True)
LOG_FOLDER.mkdir(parents=True, exist_ok=True)
USER_VIDEOS_DIR.mkdir(parents=True, exist_ok=True)
PLUGINS_DIR.mkdir(parents=True, exist_ok=True)
USER_CONFIG = APP_DATA_DIR / "config.toml"

6
viu_media/core/plugins/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""Plugin system for viu."""
from .model import PluginComponents, PluginInfo
from .manager import PluginManager
__all__ = ["PluginInfo", "PluginComponents", "PluginManager"]

631
viu_media/core/plugins/manager.py Normal file
View File

@@ -0,0 +1,631 @@
"""Plugin manager for viu.
This module contains the PluginManager singleton that handles all plugin operations
including loading, discovery, installation, and removal.
"""
import importlib.util
import logging
import shutil
import subprocess
import sys
import tomllib
from pathlib import Path
from typing import Any, Dict, Literal, Optional, Set, Union
import tomli_w
from pydantic import ValidationError
from ..constants import PLUGINS_CONFIG, PLUGINS_DIR, PLUGINS_MANIFEST
from .model import InstalledPlugin, PluginInfo, PluginManifest
from viu_media.core.exceptions import ViuError
logger = logging.getLogger(__name__)
ComponentType = Literal["provider", "player", "selector", "command"]
class PluginError(ViuError):
"""Base exception for plugin-related errors."""
pass
class PluginNotFoundError(ViuError):
"""Raised when a requested plugin is not found."""
pass
class PluginLoadError(ViuError):
"""Raised when a plugin fails to load."""
pass
class PluginManager:
"""Manages the plugin system for viu.
This is a singleton class that handles:
- Loading and caching plugins
- Installing and removing plugins from Git repositories
- Managing plugin configurations
- Discovering available plugins
"""
_instance: Optional["PluginManager"] = None
_initialized: bool = False
def __new__(cls) -> "PluginManager":
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self) -> None:
if self._initialized:
return
self._loaded_components: Dict[str, Any] = {}
self._manifest: PluginManifest = PluginManifest()
self._plugin_configs: Dict[str, Dict[str, Any]] = {}
self._load_manifest()
self._load_plugin_configs()
self._initialized = True
def load_component(self, component_type: ComponentType, name: str) -> Any:
"""Lazy-load a plugin component by type and name.
Args:
component_type: Type of component (provider, player, selector, command)
name: Name of the component to load
Returns:
The loaded component instance or function
Raises:
PluginNotFoundError: If the plugin is not installed
PluginLoadError: If the plugin fails to load
"""
cache_key = f"{component_type}:{name}"
# Return cached component if already loaded
if cache_key in self._loaded_components:
return self._loaded_components[cache_key]
# Find the plugin in the manifest
plugins_of_type = getattr(self._manifest, f"{component_type}s")
if name not in plugins_of_type:
raise PluginNotFoundError(
f"Plugin '{name}' of type '{component_type}' is not installed"
)
plugin_entry = plugins_of_type[name]
plugin_path = plugin_entry.path
if not plugin_path.exists():
raise PluginLoadError(
f"Plugin path does not exist: {plugin_path}"
)
# Load plugin info to get component definition
try:
plugin_info = self._get_plugin_info(plugin_path)
except PluginError as e:
raise PluginLoadError(f"Failed to load plugin info: {e}") from e
# Get the component definition
component_def = getattr(plugin_info.components, component_type)
if not component_def:
raise PluginLoadError(
f"Plugin '{name}' does not provide a {component_type} component"
)
# Parse module:class format
if ":" not in component_def:
raise PluginLoadError(
f"Invalid component definition: {component_def}"
)
module_name, class_name = component_def.split(":", 1)
# Load the module
module_path = plugin_path / f"{module_name}.py"
if not module_path.exists():
raise PluginLoadError(f"Plugin module not found: {module_path}")
try:
spec = importlib.util.spec_from_file_location(
f"plugin_{name}_{module_name}", module_path
)
if spec is None or spec.loader is None:
raise PluginLoadError(f"Could not create module spec for {module_path}")
module = importlib.util.module_from_spec(spec)
# Add plugin path to sys.path temporarily for relative imports
sys.path.insert(0, str(plugin_path))
try:
spec.loader.exec_module(module)
finally:
sys.path.remove(str(plugin_path))
except Exception as e:
raise PluginLoadError(f"Failed to load module {module_path}: {e}") from e
# Get the component class/function
if not hasattr(module, class_name):
raise PluginLoadError(
f"Module {module_name} does not have {class_name}"
)
component_cls = getattr(module, class_name)
# For providers, players, and selectors, instantiate with config
if component_type in ("provider", "player", "selector"):
plugin_config = self._plugin_configs.get(name, {})
# For providers, also inject httpx client like the built-in system
if component_type == "provider":
from ...core.utils.networking import random_user_agent
from httpx import Client
headers = getattr(component_cls, "HEADERS", {})
client = Client(
headers={"User-Agent": random_user_agent(), **headers}
)
try:
component = component_cls(client, **plugin_config)
except TypeError:
# Fallback if constructor doesn't accept config
component = component_cls(client)
else:
try:
component = component_cls(**plugin_config)
except TypeError:
# Fallback if constructor doesn't accept config
component = component_cls()
else:
# For commands, just return the function
component = component_cls
# Cache and return
self._loaded_components[cache_key] = component
logger.debug(f"Loaded plugin component: {cache_key}")
return component
def add_plugin(
self,
component_type: ComponentType,
name: str,
source: str,
force: bool = False
) -> None:
"""Install a plugin from a Git repository.
Args:
component_type: Type of component the plugin provides
name: Local name for the plugin
source: Git source (e.g., "github:user/repo")
force: Whether to overwrite existing plugin
Raises:
PluginError: If installation fails
"""
plugins_of_type = getattr(self._manifest, f"{component_type}s")
# Check if plugin already exists
if name in plugins_of_type and not force:
raise PluginError(
f"Plugin '{name}' already exists. Use --force to overwrite."
)
# Determine installation path
plugin_dir = PLUGINS_DIR / f"{component_type}s" / name
# Remove existing if force is True
if plugin_dir.exists():
if force:
shutil.rmtree(plugin_dir)
else:
raise PluginError(f"Plugin directory already exists: {plugin_dir}")
# Create parent directory
plugin_dir.parent.mkdir(parents=True, exist_ok=True)
# Clone the repository
self._clone_plugin(source, plugin_dir)
# Validate plugin structure
try:
plugin_info = self._get_plugin_info(plugin_dir)
except PluginError:
# Clean up on validation failure
shutil.rmtree(plugin_dir)
raise
# Ensure plugin provides the expected component type
expected_component = getattr(plugin_info.components, component_type)
if not expected_component:
shutil.rmtree(plugin_dir)
raise PluginError(
f"Plugin does not provide a {component_type} component"
)
# Add to manifest
plugins_of_type[name] = InstalledPlugin(
source=source,
path=plugin_dir,
version=plugin_info.plugin.version
)
# Save manifest
self._save_manifest()
# Copy default config if it exists
self._install_default_config(name, plugin_dir)
logger.info(f"Successfully installed {component_type} plugin '{name}'")
def remove_plugin(self, component_type: ComponentType, name: str) -> None:
"""Remove an installed plugin.
Args:
component_type: Type of component
name: Name of the plugin to remove
Raises:
PluginNotFoundError: If plugin is not installed
PluginError: If removal fails
"""
plugins_of_type = getattr(self._manifest, f"{component_type}s")
if name not in plugins_of_type:
raise PluginNotFoundError(
f"Plugin '{name}' of type '{component_type}' is not installed"
)
plugin_entry = plugins_of_type[name]
plugin_path = plugin_entry.path
# Remove from filesystem
if plugin_path.exists():
try:
shutil.rmtree(plugin_path)
except OSError as e:
raise PluginError(f"Failed to remove plugin directory: {e}") from e
# Remove from manifest
del plugins_of_type[name]
# Remove from loaded components cache
cache_key = f"{component_type}:{name}"
self._loaded_components.pop(cache_key, None)
# Save manifest
self._save_manifest()
logger.info(f"Successfully removed {component_type} plugin '{name}'")
def update_plugin(self, component_type: ComponentType, name: str) -> None:
"""Update an installed plugin by pulling from Git.
Args:
component_type: Type of component
name: Name of the plugin to update
Raises:
PluginNotFoundError: If plugin is not installed
PluginError: If update fails
"""
plugins_of_type = getattr(self._manifest, f"{component_type}s")
if name not in plugins_of_type:
raise PluginNotFoundError(
f"Plugin '{name}' of type '{component_type}' is not installed"
)
plugin_entry = plugins_of_type[name]
plugin_path = plugin_entry.path
if not plugin_path.exists():
raise PluginError(f"Plugin path does not exist: {plugin_path}")
# Pull latest changes
try:
result = subprocess.run(
["git", "pull"],
cwd=plugin_path,
check=True,
capture_output=True,
text=True
)
logger.debug(f"Git pull output: {result.stdout}")
except subprocess.CalledProcessError as e:
raise PluginError(f"Failed to update plugin: {e.stderr}") from e
except FileNotFoundError:
raise PluginError("Git is not installed or not in PATH") from None
# Update version in manifest
try:
plugin_info = self._get_plugin_info(plugin_path)
plugin_entry.version = plugin_info.plugin.version
self._save_manifest()
except PluginError as e:
logger.warning(f"Could not update plugin version: {e}")
# Clear from cache to force reload
cache_key = f"{component_type}:{name}"
self._loaded_components.pop(cache_key, None)
logger.info(f"Successfully updated {component_type} plugin '{name}'")
def list_plugins(self) -> Dict[ComponentType, Dict[str, InstalledPlugin]]:
"""List all installed plugins grouped by type.
Returns:
Dictionary mapping component types to their installed plugins
"""
return {
"provider": dict(self._manifest.providers),
"player": dict(self._manifest.players),
"selector": dict(self._manifest.selectors),
"command": dict(self._manifest.commands),
}
def get_available_components(self, component_type: ComponentType) -> Set[str]:
"""Get names of all available components of a given type.
This includes both built-in components and installed plugins.
Args:
component_type: Type of component
Returns:
Set of component names
"""
# Get plugin names
plugins_of_type = getattr(self._manifest, f"{component_type}s")
plugin_names = set(plugins_of_type.keys())
# Add built-in component names
if component_type == "provider":
from ...libs.provider.anime.provider import PROVIDERS_AVAILABLE
builtin_names = set(PROVIDERS_AVAILABLE.keys())
elif component_type == "player":
from ...libs.player.player import PLAYERS
builtin_names = set(PLAYERS)
elif component_type == "selector":
from ...libs.selectors.selector import SELECTORS
builtin_names = set(SELECTORS)
elif component_type == "command":
# Commands would need to be handled differently as they're registered in CLI
builtin_names = set()
else:
builtin_names = set()
return plugin_names | builtin_names
def is_plugin(self, component_type: ComponentType, name: str) -> bool:
"""Check if a component is provided by a plugin.
Args:
component_type: Type of component
name: Name of the component
Returns:
True if it's a plugin, False if it's built-in
"""
plugins_of_type = getattr(self._manifest, f"{component_type}s")
return name in plugins_of_type
def _load_manifest(self) -> None:
"""Load the plugins.toml manifest file."""
if not PLUGINS_MANIFEST.exists():
logger.debug("No plugins manifest found, creating empty one")
self._save_manifest()
return
try:
with open(PLUGINS_MANIFEST, "rb") as f:
data = tomllib.load(f)
self._manifest = PluginManifest.model_validate(data)
logger.debug(f"Loaded plugins manifest with {len(self.list_plugins())} plugins")
except (OSError, ValidationError, tomllib.TOMLDecodeError) as e:
logger.error(f"Failed to load plugins manifest: {e}")
self._manifest = PluginManifest()
def _save_manifest(self) -> None:
"""Save the current manifest to plugins.toml."""
try:
# Convert Path objects to strings for TOML serialization
manifest_dict = self._manifest.model_dump()
# Convert all Path objects to strings
def convert_paths(obj: Any) -> Any:
if isinstance(obj, dict):
return {k: convert_paths(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [convert_paths(item) for item in obj]
elif isinstance(obj, Path):
return str(obj)
else:
return obj
manifest_dict = convert_paths(manifest_dict)
with open(PLUGINS_MANIFEST, "wb") as f:
tomli_w.dump(manifest_dict, f)
logger.debug("Saved plugins manifest")
except OSError as e:
logger.error(f"Failed to save plugins manifest: {e}")
raise PluginError(f"Could not save plugins manifest: {e}") from e
def _load_plugin_configs(self) -> None:
"""Load plugin configurations from plugins.config.toml."""
if not PLUGINS_CONFIG.exists():
logger.debug("No plugin configs found")
return
try:
with open(PLUGINS_CONFIG, "rb") as f:
self._plugin_configs = tomllib.load(f)
logger.debug(f"Loaded configs for {len(self._plugin_configs)} plugins")
except (OSError, tomllib.TOMLDecodeError) as e:
logger.error(f"Failed to load plugin configs: {e}")
self._plugin_configs = {}
def _get_plugin_info(self, plugin_path: Path) -> PluginInfo:
"""Load and validate plugin.info.toml from a plugin directory."""
info_file = plugin_path / "plugin.info.toml"
if not info_file.exists():
raise PluginError(f"Plugin info file not found: {info_file}")
try:
with open(info_file, "rb") as f:
data = tomllib.load(f)
return PluginInfo.model_validate(data)
except (OSError, ValidationError, tomllib.TOMLDecodeError) as e:
raise PluginError(f"Invalid plugin info file {info_file}: {e}") from e
def _parse_git_source(self, source: str) -> tuple[str, str]:
"""Parse a git source string into platform and repo.
Examples:
"github:user/repo" -> ("github.com", "user/repo")
"gitlab:user/repo" -> ("gitlab.com", "user/repo")
"https://github.com/user/repo" -> ("github.com", "user/repo")
"/path/to/local/repo" -> ("local", "/path/to/local/repo")
"file:///path/to/repo" -> ("local", "/path/to/repo")
"""
# Handle local file paths
if source.startswith("file://"):
return "local", source[7:] # Remove file:// prefix
elif source.startswith("/") or source.startswith("./") or source.startswith("../"):
return "local", source
if source.startswith("http"):
# Full URL provided
if "github.com" in source:
repo = source.split("github.com/")[-1].rstrip(".git")
return "github.com", repo
elif "gitlab.com" in source:
repo = source.split("gitlab.com/")[-1].rstrip(".git")
return "gitlab.com", repo
else:
raise PluginError(f"Unsupported git host in URL: {source}")
# Short format like "github:user/repo"
if ":" not in source:
raise PluginError(f"Invalid source format: {source}")
platform, repo = source.split(":", 1)
platform_map = {
"github": "github.com",
"gitlab": "gitlab.com",
}
if platform not in platform_map:
raise PluginError(f"Unsupported platform: {platform}")
return platform_map[platform], repo
def _clone_plugin(self, source: str, dest_path: Path) -> None:
"""Clone a plugin repository from Git."""
platform, repo = self._parse_git_source(source)
if platform == "local":
# Handle local repository - just copy the directory
import shutil
src_path = Path(repo).resolve()
if not src_path.exists():
raise PluginError(f"Local repository path does not exist: {src_path}")
if not (src_path / ".git").exists():
raise PluginError(f"Path is not a Git repository: {src_path}")
logger.info(f"Copying local Git repository from {src_path}")
try:
# Use git clone to properly copy the repository
subprocess.run(
["git", "clone", str(src_path), str(dest_path)],
check=True,
capture_output=True,
text=True
)
except subprocess.CalledProcessError as e:
raise PluginError(f"Failed to clone local repository: {e.stderr}") from e
else:
# Handle remote repository
git_url = f"https://{platform}/{repo}.git"
logger.info(f"Cloning plugin from {git_url}")
try:
subprocess.run(
["git", "clone", git_url, str(dest_path)],
check=True,
capture_output=True,
text=True
)
except subprocess.CalledProcessError as e:
raise PluginError(f"Failed to clone plugin: {e.stderr}") from e
if not dest_path.exists():
raise PluginError("Plugin cloning failed - destination directory was not created")
# Check for git command availability
try:
subprocess.run(["git", "--version"], check=True, capture_output=True)
except FileNotFoundError:
raise PluginError("Git is not installed or not in PATH") from None
def _install_default_config(self, plugin_name: str, plugin_dir: Path) -> None:
"""Install default configuration from plugin's config.toml if it exists."""
default_config_path = plugin_dir / "config.toml"
if not default_config_path.exists():
logger.debug(f"No default config found for plugin '{plugin_name}'")
return
# Load the default config
try:
with open(default_config_path, "rb") as f:
default_config = tomllib.load(f)
except (OSError, tomllib.TOMLDecodeError) as e:
logger.warning(f"Failed to load default config for plugin '{plugin_name}': {e}")
return
# Load existing plugins config or create empty dict
if PLUGINS_CONFIG.exists():
try:
with open(PLUGINS_CONFIG, "rb") as f:
existing_config = tomllib.load(f)
except (OSError, tomllib.TOMLDecodeError) as e:
logger.warning(f"Failed to load existing plugins config: {e}")
existing_config = {}
else:
existing_config = {}
# Check if plugin config already exists
if plugin_name in existing_config:
logger.debug(f"Plugin '{plugin_name}' config already exists, skipping default config installation")
return
# Merge the default config
if plugin_name in default_config:
existing_config[plugin_name] = default_config[plugin_name]
# Write the updated config
try:
with open(PLUGINS_CONFIG, "wb") as f:
tomli_w.dump(existing_config, f)
logger.info(f"Installed default configuration for plugin '{plugin_name}'")
except OSError as e:
logger.warning(f"Failed to save default config for plugin '{plugin_name}': {e}")
else:
logger.debug(f"No config section found for plugin '{plugin_name}' in default config")
# Global instance
plugin_manager = PluginManager()

91
viu_media/core/plugins/model.py Normal file
View File

@@ -0,0 +1,91 @@
"""Plugin interface definitions for viu.
This module defines the Pydantic models that represent the structure
of plugin.info.toml files and plugin configurations.
"""
from pathlib import Path
from typing import Dict, Optional
from pydantic import BaseModel, Field
class PluginComponents(BaseModel):
"""Defines the components that a plugin provides.
Each component is defined as a string in the format:
{module_name_in_repo}:{ClassName_or_factory_function}
For example:
provider = "gogo_provider:GogoProvider"
player = "my_player:MyPlayer"
selector = "my_selector:MySelector"
command = "my_command:my_command_func"
"""
provider: Optional[str] = Field(
None,
description="Provider component in format 'module:class'"
)
player: Optional[str] = Field(
None,
description="Player component in format 'module:class'"
)
selector: Optional[str] = Field(
None,
description="Selector component in format 'module:class'"
)
command: Optional[str] = Field(
None,
description="Command component in format 'module:function'"
)
class PluginMetadata(BaseModel):
"""Plugin metadata from the [plugin] section."""
name: str = Field(description="Human-readable plugin name")
version: str = Field(description="Plugin version")
description: str = Field(description="Plugin description")
author: Optional[str] = Field(None, description="Plugin author")
homepage: Optional[str] = Field(None, description="Plugin homepage URL")
requires_python: Optional[str] = Field(
None,
description="Minimum Python version required"
)
class PluginInfo(BaseModel):
"""Complete plugin information from plugin.info.toml."""
plugin: PluginMetadata = Field(description="Plugin metadata")
components: PluginComponents = Field(description="Plugin components")
class InstalledPlugin(BaseModel):
"""Represents a plugin entry in plugins.toml."""
source: str = Field(description="Git source (e.g., 'github:user/repo')")
path: Path = Field(description="Local filesystem path to the plugin")
version: Optional[str] = Field(None, description="Installed version")
class PluginManifest(BaseModel):
"""Complete plugins.toml manifest structure."""
providers: Dict[str, InstalledPlugin] = Field(
default_factory=dict,
description="Installed provider plugins"
)
players: Dict[str, InstalledPlugin] = Field(
default_factory=dict,
description="Installed player plugins"
)
selectors: Dict[str, InstalledPlugin] = Field(
default_factory=dict,
description="Installed selector plugins"
)
commands: Dict[str, InstalledPlugin] = Field(
default_factory=dict,
description="Installed command plugins"
)

10
viu_media/libs/player/player.py
View File

@@ -30,8 +30,18 @@ class PlayerFactory:
ValueError: If the player_name is not supported.
NotImplementedError: If the player is recognized but not yet implemented.
"""
from ...core.plugins.manager import plugin_manager
player_name = config.stream.player
# Check if it's a plugin first
if plugin_manager.is_plugin("player", player_name):
try:
return plugin_manager.load_component("player", player_name)
except Exception as e:
raise ValueError(f"Could not load plugin player '{player_name}': {e}") from e
# Handle built-in players
if player_name not in PLAYERS:
raise ValueError(
f"Unsupported player: '{player_name}'. Supported players are: {PLAYERS}"

32
viu_media/libs/provider/anime/provider.py
View File

@@ -1,5 +1,6 @@
import importlib
import logging
from typing import Union
from httpx import Client
@@ -21,12 +22,12 @@ class AnimeProviderFactory:
"""Factory for creating anime provider instances."""
@staticmethod
def create(provider_name: ProviderName) -> BaseAnimeProvider:
def create(provider_name: Union[ProviderName, str]) -> BaseAnimeProvider:
"""
Dynamically creates an instance of the specified anime provider.
This method imports the necessary provider module, instantiates its main class,
and injects a pre-configured HTTP client.
and injects a pre-configured HTTP client. It now also supports plugin providers.
Args:
provider_name: The name of the provider to create (e.g., 'allanime').
@@ -38,24 +39,43 @@ class AnimeProviderFactory:
ValueError: If the provider_name is not supported.
ImportError: If the provider module or class cannot be found.
"""
from ....core.plugins.manager import plugin_manager
from ....core.utils.networking import random_user_agent
# Convert to string if it's an enum
if isinstance(provider_name, ProviderName):
provider_str = provider_name.value
else:
provider_str = str(provider_name)
# Check if it's a plugin first
if plugin_manager.is_plugin("provider", provider_str):
try:
return plugin_manager.load_component("provider", provider_str)
except Exception as e:
logger.error(f"Failed to load plugin provider '{provider_str}': {e}")
raise ImportError(f"Could not load plugin provider '{provider_str}': {e}") from e
# Handle built-in providers
if provider_str.lower() not in PROVIDERS_AVAILABLE:
raise ValueError(f"Provider '{provider_str}' is not available")
# Correctly determine module and class name from the map
import_path = PROVIDERS_AVAILABLE[provider_name.value.lower()]
import_path = PROVIDERS_AVAILABLE[provider_str.lower()]
module_name, class_name = import_path.split(".", 1)
# Construct the full package path for dynamic import
package_path = f"viu_media.libs.provider.anime.{provider_name.value.lower()}"
package_path = f"viu_media.libs.provider.anime.{provider_str.lower()}"
try:
provider_module = importlib.import_module(f".{module_name}", package_path)
provider_class = getattr(provider_module, class_name)
except (ImportError, AttributeError) as e:
logger.error(
f"Failed to load provider '{provider_name.value.lower()}': {e}"
f"Failed to load provider '{provider_str}': {e}"
)
raise ImportError(
f"Could not load provider '{provider_name.value.lower()}'. "
f"Could not load provider '{provider_str}'. "
"Check the module path and class name in PROVIDERS_AVAILABLE."
) from e

10
viu_media/libs/selectors/selector.py
View File

@@ -14,8 +14,18 @@ class SelectorFactory:
"""
Factory to create a selector instance based on the configuration.
"""
from ...core.plugins.manager import plugin_manager
selector_name = config.general.selector
# Check if it's a plugin first
if plugin_manager.is_plugin("selector", selector_name):
try:
return plugin_manager.load_component("selector", selector_name)
except Exception as e:
raise ValueError(f"Could not load plugin selector '{selector_name}': {e}") from e
# Handle built-in selectors
if selector_name not in SELECTORS:
raise ValueError(
f"Unsupported selector: '{selector_name}'.Available selectors are: {SELECTORS}"