feat: enhance plugin functionality

This commit is contained in:
时雨
2026-01-06 16:54:49 +08:00
committed by GitHub
parent 31d97b2968
commit 24255744df
48 changed files with 3089 additions and 3811 deletions

View File

@@ -1 +1,17 @@
"""
Foxel 插件系统
提供 .foxpkg 插件包的安装、管理和运行时加载功能。
"""
from domain.plugins.loader import PluginLoader, PluginLoadError
from domain.plugins.service import PluginService
from domain.plugins.startup import init_plugins, load_installed_plugins
__all__ = [
"PluginLoader",
"PluginLoadError",
"PluginService",
"init_plugins",
"load_installed_plugins",
]

View File

@@ -1,76 +1,109 @@
"""
插件管理 API 路由
"""
from typing import List
from fastapi import APIRouter, Body, Request
from fastapi import APIRouter, File, Request, UploadFile
from fastapi.responses import FileResponse
from domain.audit import AuditAction, audit
from domain.plugins.service import PluginService
from domain.plugins.routes import video_player as video_player_routes
from domain.plugins.types import PluginCreate, PluginManifestUpdate, PluginOut
from domain.plugins.types import (
PluginInstallResult,
PluginOut,
)
router = APIRouter(prefix="/api/plugins", tags=["plugins"])
router.include_router(video_player_routes.router)
@router.post("", response_model=PluginOut)
@audit(
action=AuditAction.CREATE,
description="创建插件",
body_fields=["url", "enabled"],
)
async def create_plugin(request: Request, payload: PluginCreate):
return await PluginService.create(payload)
# ========== 安装 ==========
@router.post("/install", response_model=PluginInstallResult)
@audit(action=AuditAction.CREATE, description="安装插件包")
async def install_plugin(request: Request, file: UploadFile = File(...)):
"""
安装 .foxpkg 插件包
上传 .foxpkg 文件进行安装。
"""
content = await file.read()
return await PluginService.install_package(content, file.filename or "plugin.foxpkg")
# ========== 插件列表和详情 ==========
@router.get("", response_model=List[PluginOut])
@audit(action=AuditAction.READ, description="获取插件列表")
async def list_plugins(request: Request):
"""获取已安装的插件列表"""
return await PluginService.list_plugins()
@router.delete("/{plugin_id}")
@audit(action=AuditAction.DELETE, description="删除插件")
async def delete_plugin(request: Request, plugin_id: int):
await PluginService.delete(plugin_id)
@router.get("/{key_or_id}", response_model=PluginOut)
@audit(action=AuditAction.READ, description="获取插件详情")
async def get_plugin(request: Request, key_or_id: str):
"""获取单个插件详情"""
return await PluginService.get_plugin(key_or_id)
# ========== 插件管理 ==========
@router.delete("/{key_or_id}")
@audit(action=AuditAction.DELETE, description="卸载插件")
async def delete_plugin(request: Request, key_or_id: str):
"""卸载插件"""
await PluginService.delete(key_or_id)
return {"code": 0, "msg": "ok"}
@router.put("/{plugin_id}", response_model=PluginOut)
@audit(
action=AuditAction.UPDATE,
description="更新插件",
body_fields=["url", "enabled"],
)
async def update_plugin(request: Request, plugin_id: int, payload: PluginCreate):
return await PluginService.update(plugin_id, payload)
# ========== 插件资源 ==========
@router.post("/{plugin_id}/metadata", response_model=PluginOut)
@audit(
action=AuditAction.UPDATE,
description="更新插件 manifest",
body_fields=[
"key",
"name",
"version",
"open_app",
"supported_exts",
"default_bounds",
"default_maximized",
"icon",
"description",
"author",
"website",
"github",
],
)
async def update_manifest(
request: Request, plugin_id: int, manifest: PluginManifestUpdate = Body(...)
):
return await PluginService.update_manifest(plugin_id, manifest)
@router.get("/{key_or_id}/bundle.js")
async def get_bundle(request: Request, key_or_id: str):
"""获取插件前端 bundle"""
path = await PluginService.get_bundle_path(key_or_id)
return FileResponse(
path,
media_type="application/javascript",
headers={"Cache-Control": "no-store"},
)
@router.get("/{plugin_id}/bundle.js")
async def get_bundle(request: Request, plugin_id: int):
path = await PluginService.get_bundle_path(plugin_id)
return FileResponse(path, media_type="application/javascript", headers={"Cache-Control": "no-store"})
@router.get("/{key}/assets/{asset_path:path}")
async def get_asset(request: Request, key: str, asset_path: str):
"""获取插件静态资源"""
path = await PluginService.get_asset_path(key, asset_path)
# 根据扩展名确定 MIME 类型
ext = path.suffix.lower()
media_types = {
".js": "application/javascript",
".css": "text/css",
".json": "application/json",
".svg": "image/svg+xml",
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
".ico": "image/x-icon",
".woff": "font/woff",
".woff2": "font/woff2",
".ttf": "font/ttf",
".eot": "application/vnd.ms-fontobject",
".html": "text/html",
".txt": "text/plain",
".md": "text/markdown",
}
media_type = media_types.get(ext, "application/octet-stream")
return FileResponse(
path,
media_type=media_type,
headers={"Cache-Control": "public, max-age=3600"},
)

449
domain/plugins/loader.py Normal file
View File

@@ -0,0 +1,449 @@
"""
插件加载器模块
负责:
1. .foxpkg 解包和验证
2. 插件文件部署
3. 后端路由动态加载
4. 处理器动态注册
"""
import io
import json
import shutil
import sys
import zipfile
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from types import ModuleType
from typing import Any, Dict, List, Optional, Tuple
from fastapi import APIRouter
from domain.plugins.types import (
ManifestProcessorConfig,
ManifestRouteConfig,
PluginManifest,
)
class PluginLoadError(Exception):
"""插件加载错误"""
pass
class PluginLoader:
"""插件加载器"""
PLUGINS_ROOT = Path("data/plugins")
# 已加载的插件模块缓存
_loaded_modules: Dict[str, ModuleType] = {}
# 已挂载的路由追踪
_mounted_routers: Dict[str, List[APIRouter]] = {}
@classmethod
def get_plugin_dir(cls, plugin_key: str) -> Path:
"""获取插件目录"""
return cls.PLUGINS_ROOT / plugin_key
@classmethod
def get_manifest_path(cls, plugin_key: str) -> Path:
"""获取插件 manifest.json 路径"""
return cls.get_plugin_dir(plugin_key) / "manifest.json"
@classmethod
def get_frontend_bundle_path(cls, plugin_key: str, entry: Optional[str] = None) -> Path:
"""获取前端 bundle 路径"""
plugin_dir = cls.get_plugin_dir(plugin_key)
if entry:
return plugin_dir / entry
# 默认位置
return plugin_dir / "frontend" / "index.js"
@classmethod
def get_asset_path(cls, plugin_key: str, asset_path: str) -> Path:
"""获取静态资源路径"""
return cls.get_plugin_dir(plugin_key) / asset_path
# ========== 解包和验证 ==========
@classmethod
def validate_manifest(cls, manifest_data: Dict[str, Any]) -> Tuple[bool, List[str]]:
"""验证 manifest 数据"""
errors: List[str] = []
# 必需字段检查
if not manifest_data.get("key"):
errors.append("manifest 缺少必需字段: key")
if not manifest_data.get("name"):
errors.append("manifest 缺少必需字段: name")
# key 格式检查Java 命名空间格式)
key = manifest_data.get("key", "")
if key:
import re
# 格式: com.example.plugin (至少两级,每级以小写字母开头,可包含小写字母和数字)
if not re.match(r"^[a-z][a-z0-9]*(\.[a-z][a-z0-9]*)+$", key):
errors.append(
"key 格式无效:必须使用命名空间格式(如 com.example.plugin"
"每个部分以小写字母开头,只能包含小写字母和数字,至少两级"
)
# 版本格式检查(简单检查)
version = manifest_data.get("version", "")
if version and not isinstance(version, str):
errors.append("version 必须是字符串")
# 验证 frontend 配置
frontend = manifest_data.get("frontend")
if frontend and isinstance(frontend, dict):
if frontend.get("entry") and not isinstance(frontend["entry"], str):
errors.append("frontend.entry 必须是字符串")
if frontend.get("styles") is not None:
if not isinstance(frontend["styles"], list) or not all(
isinstance(x, str) for x in frontend["styles"]
):
errors.append("frontend.styles 必须是字符串数组")
supported_exts = frontend.get("supportedExts") or frontend.get("supported_exts")
if supported_exts and not isinstance(supported_exts, list):
errors.append("frontend.supportedExts 必须是数组")
use_system_window = frontend.get("useSystemWindow") or frontend.get("use_system_window")
if use_system_window is not None and not isinstance(use_system_window, bool):
errors.append("frontend.useSystemWindow 必须是布尔值")
# 验证 backend 配置
backend = manifest_data.get("backend")
if backend and isinstance(backend, dict):
routes = backend.get("routes", [])
if routes:
for i, route in enumerate(routes):
if not route.get("module"):
errors.append(f"backend.routes[{i}] 缺少 module")
if not route.get("prefix"):
errors.append(f"backend.routes[{i}] 缺少 prefix")
processors = backend.get("processors", [])
if processors:
for i, proc in enumerate(processors):
if not proc.get("module"):
errors.append(f"backend.processors[{i}] 缺少 module")
if not proc.get("type"):
errors.append(f"backend.processors[{i}] 缺少 type")
return len(errors) == 0, errors
@classmethod
def unpack_foxpkg(
cls, file_content: bytes, target_key: Optional[str] = None
) -> Tuple[PluginManifest, Path]:
"""
解包 .foxpkg 文件
Args:
file_content: .foxpkg 文件内容
target_key: 可选,指定安装的插件 key覆盖 manifest 中的 key
Returns:
(manifest, plugin_dir) 元组
Raises:
PluginLoadError: 解包或验证失败
"""
try:
with zipfile.ZipFile(io.BytesIO(file_content)) as zf:
# 读取 manifest.json
try:
manifest_bytes = zf.read("manifest.json")
except KeyError:
raise PluginLoadError("插件包缺少 manifest.json")
try:
manifest_data = json.loads(manifest_bytes.decode("utf-8"))
except json.JSONDecodeError as e:
raise PluginLoadError(f"manifest.json 解析失败: {e}")
# 验证 manifest
valid, errors = cls.validate_manifest(manifest_data)
if not valid:
raise PluginLoadError(f"manifest 验证失败: {'; '.join(errors)}")
# 解析 manifest
try:
manifest = PluginManifest.model_validate(manifest_data)
except Exception as e:
raise PluginLoadError(f"manifest 解析失败: {e}")
# 确定插件 key
plugin_key = target_key or manifest.key
# 验证包内文件
cls._validate_package_files(zf, manifest)
# 部署文件
target_dir = cls.PLUGINS_ROOT / plugin_key
if target_dir.exists():
# 备份旧版本
backup_dir = cls.PLUGINS_ROOT / f"{plugin_key}.backup"
if backup_dir.exists():
shutil.rmtree(backup_dir)
shutil.move(str(target_dir), str(backup_dir))
target_dir.mkdir(parents=True, exist_ok=True)
try:
zf.extractall(target_dir)
except Exception as e:
# 恢复备份
if (cls.PLUGINS_ROOT / f"{plugin_key}.backup").exists():
shutil.rmtree(target_dir, ignore_errors=True)
shutil.move(str(cls.PLUGINS_ROOT / f"{plugin_key}.backup"), str(target_dir))
raise PluginLoadError(f"文件解压失败: {e}")
# 清理备份
backup_dir = cls.PLUGINS_ROOT / f"{plugin_key}.backup"
if backup_dir.exists():
shutil.rmtree(backup_dir, ignore_errors=True)
return manifest, target_dir
except zipfile.BadZipFile:
raise PluginLoadError("无效的插件包格式(非 ZIP 文件)")
@classmethod
def _validate_package_files(cls, zf: zipfile.ZipFile, manifest: PluginManifest) -> None:
"""验证包内文件是否完整"""
file_list = zf.namelist()
# 检查前端入口
if manifest.frontend and manifest.frontend.entry:
if manifest.frontend.entry not in file_list:
raise PluginLoadError(f"前端入口文件不存在: {manifest.frontend.entry}")
# 检查后端模块
if manifest.backend:
if manifest.backend.routes:
for route in manifest.backend.routes:
if route.module not in file_list:
raise PluginLoadError(f"路由模块不存在: {route.module}")
if manifest.backend.processors:
for proc in manifest.backend.processors:
if proc.module not in file_list:
raise PluginLoadError(f"处理器模块不存在: {proc.module}")
# ========== 路由动态加载 ==========
@classmethod
def load_route_module(cls, plugin_key: str, route_config: ManifestRouteConfig) -> APIRouter:
"""
动态加载插件路由模块
Args:
plugin_key: 插件标识
route_config: 路由配置
Returns:
加载的 APIRouter
"""
module_path = cls.get_plugin_dir(plugin_key) / route_config.module
if not module_path.exists():
raise PluginLoadError(f"路由模块不存在: {module_path}")
module_name = f"foxel_plugin_{plugin_key}_route_{module_path.stem}"
try:
spec = spec_from_file_location(module_name, module_path)
if spec is None or spec.loader is None:
raise PluginLoadError(f"无法加载路由模块: {module_path}")
module = module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
# 缓存模块
cls._loaded_modules[f"{plugin_key}:route:{route_config.module}"] = module
# 获取 router
router = getattr(module, "router", None)
if router is None:
raise PluginLoadError(f"路由模块缺少 'router' 对象: {module_path}")
if not isinstance(router, APIRouter):
raise PluginLoadError(f"'router' 不是有效的 APIRouter 实例: {module_path}")
# 创建包装路由器添加前缀
wrapper = APIRouter(prefix=route_config.prefix, tags=route_config.tags or [])
wrapper.include_router(router)
return wrapper
except PluginLoadError:
raise
except Exception as e:
raise PluginLoadError(f"加载路由模块失败 [{module_path}]: {e}")
@classmethod
def load_all_routes(cls, plugin_key: str, manifest: PluginManifest) -> List[APIRouter]:
"""加载插件的所有路由"""
routers: List[APIRouter] = []
if not manifest.backend or not manifest.backend.routes:
return routers
for route_config in manifest.backend.routes:
router = cls.load_route_module(plugin_key, route_config)
routers.append(router)
cls._mounted_routers[plugin_key] = routers
return routers
# ========== 处理器动态注册 ==========
@classmethod
def load_processor_module(
cls, plugin_key: str, processor_config: ManifestProcessorConfig
) -> None:
"""
动态加载并注册处理器模块
Args:
plugin_key: 插件标识
processor_config: 处理器配置
"""
module_path = cls.get_plugin_dir(plugin_key) / processor_config.module
if not module_path.exists():
raise PluginLoadError(f"处理器模块不存在: {module_path}")
module_name = f"foxel_plugin_{plugin_key}_processor_{module_path.stem}"
try:
spec = spec_from_file_location(module_name, module_path)
if spec is None or spec.loader is None:
raise PluginLoadError(f"无法加载处理器模块: {module_path}")
module = module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
# 缓存模块
cls._loaded_modules[f"{plugin_key}:processor:{processor_config.module}"] = module
# 获取处理器工厂
factory = getattr(module, "PROCESSOR_FACTORY", None)
if factory is None:
raise PluginLoadError(f"处理器模块缺少 'PROCESSOR_FACTORY': {module_path}")
# 获取配置 schema
config_schema = getattr(module, "CONFIG_SCHEMA", [])
processor_name = getattr(module, "PROCESSOR_NAME", processor_config.name or processor_config.type)
supported_exts = getattr(module, "SUPPORTED_EXTS", [])
# 注册到处理器注册表
from domain.processors.registry import CONFIG_SCHEMAS, TYPE_MAP
processor_type = processor_config.type
TYPE_MAP[processor_type] = factory
# 获取实例以读取属性
try:
sample = factory()
produces_file = getattr(sample, "produces_file", False)
supports_directory = getattr(sample, "supports_directory", False)
except Exception:
produces_file = False
supports_directory = False
CONFIG_SCHEMAS[processor_type] = {
"type": processor_type,
"name": processor_name,
"supported_exts": supported_exts,
"config_schema": config_schema,
"produces_file": produces_file,
"supports_directory": supports_directory,
"plugin": plugin_key, # 标记来源插件
"module_path": str(module_path),
}
except PluginLoadError:
raise
except Exception as e:
raise PluginLoadError(f"加载处理器模块失败 [{module_path}]: {e}")
@classmethod
def load_all_processors(cls, plugin_key: str, manifest: PluginManifest) -> List[str]:
"""加载插件的所有处理器,返回处理器类型列表"""
processor_types: List[str] = []
if not manifest.backend or not manifest.backend.processors:
return processor_types
for proc_config in manifest.backend.processors:
cls.load_processor_module(plugin_key, proc_config)
processor_types.append(proc_config.type)
return processor_types
# ========== 卸载 ==========
@classmethod
def unload_plugin(cls, plugin_key: str, manifest: Optional[PluginManifest] = None) -> None:
"""
卸载插件的后端组件
Args:
plugin_key: 插件标识
manifest: 可选的 manifest用于确定要卸载的组件
"""
# 卸载处理器
if manifest and manifest.backend and manifest.backend.processors:
from domain.processors.registry import CONFIG_SCHEMAS, TYPE_MAP
for proc_config in manifest.backend.processors:
proc_type = proc_config.type
if proc_type in TYPE_MAP:
del TYPE_MAP[proc_type]
if proc_type in CONFIG_SCHEMAS:
del CONFIG_SCHEMAS[proc_type]
# 清理缓存的模块
keys_to_remove = [k for k in cls._loaded_modules if k.startswith(f"{plugin_key}:")]
for key in keys_to_remove:
module = cls._loaded_modules.pop(key, None)
if module and module.__name__ in sys.modules:
del sys.modules[module.__name__]
# 清理路由追踪注意FastAPI 不支持动态移除路由,需要重启应用)
cls._mounted_routers.pop(plugin_key, None)
@classmethod
def delete_plugin_files(cls, plugin_key: str) -> None:
"""删除插件文件"""
plugin_dir = cls.get_plugin_dir(plugin_key)
if plugin_dir.exists():
shutil.rmtree(plugin_dir)
# 同时删除备份
backup_dir = cls.PLUGINS_ROOT / f"{plugin_key}.backup"
if backup_dir.exists():
shutil.rmtree(backup_dir)
# ========== 读取 manifest ==========
@classmethod
def read_manifest(cls, plugin_key: str) -> Optional[PluginManifest]:
"""从文件系统读取插件 manifest"""
manifest_path = cls.get_manifest_path(plugin_key)
if not manifest_path.exists():
return None
try:
with open(manifest_path, "r", encoding="utf-8") as f:
data = json.load(f)
return PluginManifest.model_validate(data)
except Exception:
return None

View File

@@ -1,2 +0,0 @@
"""插件专属服务端路由集合。"""

View File

@@ -1,142 +0,0 @@
import json
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from api.response import success
from domain.auth.service import get_current_active_user
router = APIRouter(
prefix="/video-player",
tags=["plugins"],
dependencies=[Depends(get_current_active_user)],
)
DATA_ROOT = Path("data/.video")
def _read_json(path: Path) -> Dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def _file_mtime_iso(path: Path) -> str:
try:
ts = path.stat().st_mtime
except FileNotFoundError:
return ""
return datetime.fromtimestamp(ts, tz=UTC).isoformat()
def _extract_title(payload: Dict[str, Any]) -> str:
detail = (payload.get("tmdb") or {}).get("detail") or {}
if payload.get("type") == "tv":
return str(detail.get("name") or detail.get("original_name") or "")
return str(detail.get("title") or detail.get("original_title") or "")
def _extract_year(payload: Dict[str, Any]) -> Optional[str]:
detail = (payload.get("tmdb") or {}).get("detail") or {}
value = detail.get("first_air_date") if payload.get("type") == "tv" else detail.get("release_date")
if not value or not isinstance(value, str):
return None
return value[:4] if len(value) >= 4 else value
def _extract_genres(payload: Dict[str, Any]) -> List[str]:
detail = (payload.get("tmdb") or {}).get("detail") or {}
genres = detail.get("genres") or []
out: List[str] = []
if isinstance(genres, list):
for g in genres:
if isinstance(g, dict) and g.get("name"):
out.append(str(g["name"]))
return out
def _summarize(item_id: str, payload: Dict[str, Any], mtime_iso: str) -> Dict[str, Any]:
detail = (payload.get("tmdb") or {}).get("detail") or {}
media_type = payload.get("type") or "unknown"
episodes = payload.get("episodes") or []
seasons = {e.get("season") for e in episodes if isinstance(e, dict) and e.get("season") is not None}
return {
"id": item_id,
"type": media_type,
"title": _extract_title(payload),
"year": _extract_year(payload),
"overview": detail.get("overview"),
"poster_path": detail.get("poster_path"),
"backdrop_path": detail.get("backdrop_path"),
"genres": _extract_genres(payload),
"tmdb_id": (payload.get("tmdb") or {}).get("id"),
"source_path": payload.get("source_path"),
"scraped_at": payload.get("scraped_at"),
"updated_at": mtime_iso,
"episodes_count": len(episodes) if isinstance(episodes, list) else 0,
"seasons_count": len(seasons),
"vote_average": detail.get("vote_average"),
"vote_count": detail.get("vote_count"),
}
def _iter_library_files() -> List[tuple[str, Path]]:
files: List[tuple[str, Path]] = []
for sub in ("tv", "movie"):
folder = DATA_ROOT / sub
if not folder.exists():
continue
for p in folder.glob("*.json"):
if not p.is_file():
continue
files.append((sub, p))
return files
@router.get("/library")
async def list_library(
q: str | None = Query(None, description="搜索关键字(标题/简介)"),
media_type: str | None = Query(None, alias="type", description="tv 或 movie"),
):
items: List[Dict[str, Any]] = []
keyword = (q or "").strip().lower()
type_filter = (media_type or "").strip().lower()
if type_filter and type_filter not in {"tv", "movie"}:
raise HTTPException(status_code=400, detail="type must be tv or movie")
for _sub, path in _iter_library_files():
item_id = path.stem
try:
payload = _read_json(path)
except Exception:
continue
if type_filter and str(payload.get("type") or "").lower() != type_filter:
continue
summary = _summarize(item_id, payload, _file_mtime_iso(path))
if keyword:
haystack = f"{summary.get('title') or ''} {summary.get('overview') or ''}".lower()
if keyword not in haystack:
continue
items.append(summary)
items.sort(key=lambda x: x.get("updated_at") or "", reverse=True)
return success(items)
@router.get("/library/{item_id}")
async def get_library_item(item_id: str):
candidates = [
DATA_ROOT / "tv" / f"{item_id}.json",
DATA_ROOT / "movie" / f"{item_id}.json",
]
path = next((p for p in candidates if p.exists()), None)
if not path:
raise HTTPException(status_code=404, detail="Item not found")
payload = _read_json(path)
payload["id"] = item_id
payload["updated_at"] = _file_mtime_iso(path)
return success(payload)

View File

@@ -1,138 +1,273 @@
"""
插件服务模块
负责插件的安装、卸载等管理操作
"""
import contextlib
import re
import logging
import shutil
from pathlib import Path
from typing import List, Optional, Union
import aiofiles
import httpx
from fastapi import HTTPException
from domain.plugins.types import PluginCreate, PluginManifestUpdate, PluginOut
from domain.plugins.loader import PluginLoadError, PluginLoader
from domain.plugins.types import (
PluginInstallResult,
PluginManifest,
PluginOut,
)
from models.database import Plugin
logger = logging.getLogger(__name__)
class PluginService:
"""插件服务"""
_plugins_root = Path("data/plugins")
@classmethod
def _folder_name(cls, rec: Plugin) -> str:
if rec.key:
safe = re.sub(r"[^A-Za-z0-9_.-]", "_", rec.key)
return safe or str(rec.id)
return str(rec.id)
# ========== 工具方法 ==========
@classmethod
def _bundle_dir_from_rec(cls, rec: Plugin) -> Path:
return cls._plugins_root / cls._folder_name(rec) / "current"
def _get_plugin_dir(cls, plugin_key: str) -> Path:
"""获取插件目录"""
return cls._plugins_root / plugin_key
@classmethod
def _bundle_path_from_rec(cls, rec: Plugin) -> Path:
return cls._bundle_dir_from_rec(rec) / "index.js"
def _get_bundle_path(cls, rec: Plugin) -> Path:
"""获取前端 bundle 路径"""
plugin_dir = cls._get_plugin_dir(rec.key)
# 从 manifest 读取
if rec.manifest:
frontend = rec.manifest.get("frontend", {})
entry = frontend.get("entry")
if entry:
return plugin_dir / entry
# 默认位置
return plugin_dir / "frontend" / "index.js"
@classmethod
async def _download_bundle(cls, rec: Plugin, url: str) -> None:
dest_dir = cls._bundle_dir_from_rec(rec)
dest_dir.mkdir(parents=True, exist_ok=True)
dest_path = cls._bundle_path_from_rec(rec)
tmp_path = dest_path.with_suffix(".tmp")
try:
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
async with client.stream("GET", url) as resp:
resp.raise_for_status()
async with aiofiles.open(tmp_path, "wb") as f:
async for chunk in resp.aiter_bytes(chunk_size=65536):
if not chunk:
continue
await f.write(chunk)
tmp_path.replace(dest_path)
except Exception:
with contextlib.suppress(Exception):
if tmp_path.exists():
tmp_path.unlink()
raise
@classmethod
async def _ensure_bundle(cls, plugin_id: int) -> Path:
rec = await cls._get_or_404(plugin_id)
bundle_path = cls._bundle_path_from_rec(rec)
if bundle_path.exists():
return bundle_path
legacy = cls._plugins_root / str(rec.id) / "current" / "index.js"
if legacy.exists():
return legacy
raise HTTPException(status_code=404, detail="Plugin bundle not found")
@classmethod
async def get_bundle_path(cls, plugin_id: int) -> Path:
return await cls._ensure_bundle(plugin_id)
@classmethod
async def create(cls, payload: PluginCreate) -> PluginOut:
rec = await Plugin.create(**payload.model_dump())
try:
await cls._download_bundle(rec, rec.url)
except Exception as exc:
with contextlib.suppress(Exception):
await rec.delete()
raise HTTPException(status_code=400, detail=f"Failed to fetch plugin: {exc}")
return PluginOut.model_validate(rec)
@classmethod
async def list_plugins(cls) -> list[PluginOut]:
rows = await Plugin.all().order_by("-id")
return [PluginOut.model_validate(r) for r in rows]
@classmethod
async def _get_or_404(cls, plugin_id: int) -> Plugin:
rec = await Plugin.get_or_none(id=plugin_id)
async def _get_by_key_or_404(cls, key: str) -> Plugin:
"""通过 key 获取插件,不存在则返回 404"""
rec = await Plugin.get_or_none(key=key)
if not rec:
raise HTTPException(status_code=404, detail="Plugin not found")
return rec
@classmethod
async def delete(cls, plugin_id: int) -> None:
rec = await cls._get_or_404(plugin_id)
await rec.delete()
with contextlib.suppress(Exception):
dirs = {cls._bundle_dir_from_rec(rec).parent, cls._plugins_root / str(rec.id)}
for plugin_dir in dirs:
if plugin_dir.exists():
shutil.rmtree(plugin_dir)
async def _get_by_key_or_id(cls, key_or_id: Union[str, int]) -> Plugin:
"""通过 key 或 ID 获取插件"""
# 尝试作为 ID
if isinstance(key_or_id, int) or (isinstance(key_or_id, str) and key_or_id.isdigit()):
plugin_id = int(key_or_id)
rec = await Plugin.get_or_none(id=plugin_id)
if rec:
return rec
# 尝试作为 key
if isinstance(key_or_id, str):
rec = await Plugin.get_or_none(key=key_or_id)
if rec:
return rec
raise HTTPException(status_code=404, detail="Plugin not found")
# ========== 安装 ==========
@classmethod
async def update(cls, plugin_id: int, payload: PluginCreate) -> PluginOut:
rec = await cls._get_or_404(plugin_id)
url_changed = rec.url != payload.url
if url_changed:
try:
await cls._download_bundle(rec, payload.url)
except Exception as exc:
raise HTTPException(status_code=400, detail=f"Failed to fetch plugin: {exc}")
rec.url = payload.url
rec.enabled = payload.enabled
await rec.save()
return PluginOut.model_validate(rec)
async def install_package(cls, file_content: bytes, filename: str) -> PluginInstallResult:
"""
安装 .foxpkg 插件包
Args:
file_content: 插件包内容
filename: 文件名
Returns:
安装结果
"""
errors: List[str] = []
try:
# 解包
manifest, plugin_dir = PluginLoader.unpack_foxpkg(file_content)
plugin_key = manifest.key
# 检查是否已存在
existing = await Plugin.get_or_none(key=plugin_key)
if existing:
# 更新现有插件
logger.info(f"更新插件: {plugin_key}")
rec = existing
else:
# 创建新插件
logger.info(f"安装新插件: {plugin_key}")
rec = Plugin(key=plugin_key)
# 更新字段
rec.name = manifest.name
rec.version = manifest.version
rec.description = manifest.description
rec.author = manifest.author
rec.website = manifest.website
rec.github = manifest.github
rec.license = manifest.license
rec.manifest = manifest.model_dump(mode="json")
# 从 manifest.frontend 提取前端配置
if manifest.frontend:
rec.open_app = manifest.frontend.open_app or False
rec.supported_exts = manifest.frontend.supported_exts
rec.default_bounds = manifest.frontend.default_bounds
rec.default_maximized = manifest.frontend.default_maximized
rec.icon = manifest.frontend.icon
@classmethod
async def update_manifest(
cls, plugin_id: int, manifest: PluginManifestUpdate
) -> PluginOut:
rec = await cls._get_or_404(plugin_id)
old_dir = cls._bundle_dir_from_rec(rec).parent
updates = manifest.model_dump(exclude_none=True)
if updates:
for key, value in updates.items():
setattr(rec, key, value)
await rec.save()
new_dir = cls._bundle_dir_from_rec(rec).parent
if rec.key and new_dir != old_dir:
candidate_dir = old_dir if old_dir.exists() else (cls._plugins_root / str(rec.id))
if candidate_dir.exists():
new_dir.parent.mkdir(parents=True, exist_ok=True)
with contextlib.suppress(Exception):
if new_dir.exists():
shutil.rmtree(new_dir)
shutil.move(str(candidate_dir), str(new_dir))
# 加载后端组件(如果有)
loaded_routes: List[str] = []
loaded_processors: List[str] = []
if manifest.backend:
# 加载路由
if manifest.backend.routes:
try:
from main import app
routers = PluginLoader.load_all_routes(plugin_key, manifest)
for router in routers:
app.include_router(router)
loaded_routes.append(router.prefix)
except PluginLoadError as e:
errors.append(f"路由加载失败: {e}")
logger.error(f"插件 {plugin_key} 路由加载失败: {e}")
except Exception as e:
errors.append(f"路由加载失败: {e}")
logger.exception(f"插件 {plugin_key} 路由加载异常")
# 加载处理器
if manifest.backend.processors:
try:
processor_types = PluginLoader.load_all_processors(plugin_key, manifest)
loaded_processors = processor_types
except PluginLoadError as e:
errors.append(f"处理器加载失败: {e}")
logger.error(f"插件 {plugin_key} 处理器加载失败: {e}")
except Exception as e:
errors.append(f"处理器加载失败: {e}")
logger.exception(f"插件 {plugin_key} 处理器加载异常")
# 更新加载状态
rec.loaded_routes = loaded_routes if loaded_routes else None
rec.loaded_processors = loaded_processors if loaded_processors else None
await rec.save()
return PluginInstallResult(
success=True,
plugin=PluginOut.model_validate(rec),
message="安装成功" if not errors else "安装完成,但有部分组件加载失败",
errors=errors if errors else None,
)
except PluginLoadError as e:
logger.error(f"插件安装失败: {e}")
return PluginInstallResult(
success=False,
message=str(e),
errors=[str(e)],
)
except Exception as e:
logger.exception("插件安装异常")
return PluginInstallResult(
success=False,
message=f"安装失败: {e}",
errors=[str(e)],
)
# ========== 查询 ==========
@classmethod
async def list_plugins(cls) -> List[PluginOut]:
"""获取所有插件列表"""
rows = await Plugin.all().order_by("-id")
for rec in rows:
try:
manifest = PluginLoader.read_manifest(rec.key)
if manifest:
rec.manifest = manifest.model_dump(mode="json")
except Exception:
continue
return [PluginOut.model_validate(r) for r in rows]
@classmethod
async def get_plugin(cls, key_or_id: Union[str, int]) -> PluginOut:
"""获取单个插件详情"""
rec = await cls._get_by_key_or_id(key_or_id)
try:
manifest = PluginLoader.read_manifest(rec.key)
if manifest:
rec.manifest = manifest.model_dump(mode="json")
except Exception:
pass
return PluginOut.model_validate(rec)
@classmethod
async def get_bundle_path(cls, key_or_id: Union[str, int]) -> Path:
"""获取插件前端 bundle 路径"""
rec = await cls._get_by_key_or_id(key_or_id)
bundle_path = cls._get_bundle_path(rec)
if not bundle_path.exists():
raise HTTPException(status_code=404, detail="Plugin bundle not found")
return bundle_path
@classmethod
async def get_asset_path(cls, key: str, asset_path: str) -> Path:
"""获取插件静态资源路径"""
rec = await cls._get_by_key_or_404(key)
plugin_dir = cls._get_plugin_dir(rec.key)
# 安全检查:防止路径遍历
asset_path = asset_path.lstrip("/")
if ".." in asset_path:
raise HTTPException(status_code=400, detail="Invalid asset path")
full_path = plugin_dir / asset_path
if not full_path.exists():
raise HTTPException(status_code=404, detail="Asset not found")
# 确保路径在插件目录内
try:
full_path.resolve().relative_to(plugin_dir.resolve())
except ValueError:
raise HTTPException(status_code=400, detail="Invalid asset path")
return full_path
# ========== 管理操作 ==========
@classmethod
async def delete(cls, key_or_id: Union[str, int]) -> None:
"""删除/卸载插件"""
rec = await cls._get_by_key_or_id(key_or_id)
# 获取 manifest 用于卸载组件
manifest: Optional[PluginManifest] = None
if rec.manifest:
try:
manifest = PluginManifest.model_validate(rec.manifest)
except Exception:
pass
# 卸载后端组件
if manifest:
PluginLoader.unload_plugin(rec.key, manifest)
# 删除数据库记录
await rec.delete()
# 删除文件
with contextlib.suppress(Exception):
plugin_dir = cls._get_plugin_dir(rec.key)
if plugin_dir.exists():
shutil.rmtree(plugin_dir)
logger.info(f"插件 {rec.key} 已卸载")

116
domain/plugins/startup.py Normal file
View File

@@ -0,0 +1,116 @@
"""
插件启动加载模块
负责在应用启动时加载所有已安装的插件
"""
import logging
from typing import TYPE_CHECKING, List, Tuple
from domain.plugins.loader import PluginLoadError, PluginLoader
from domain.plugins.types import PluginManifest
if TYPE_CHECKING:
from fastapi import FastAPI
logger = logging.getLogger(__name__)
async def load_installed_plugins(app: "FastAPI") -> Tuple[int, List[str]]:
"""
加载所有已安装的插件
Args:
app: FastAPI 应用实例
Returns:
(成功加载数量, 错误列表)
"""
from models.database import Plugin
errors: List[str] = []
loaded_count = 0
try:
plugins = await Plugin.all()
except Exception as e:
logger.error(f"查询插件列表失败: {e}")
return 0, [f"查询插件列表失败: {e}"]
for plugin in plugins:
if not plugin.key:
continue
try:
# 获取 manifest
manifest = None
if plugin.manifest:
try:
manifest = PluginManifest.model_validate(plugin.manifest)
except Exception:
# 尝试从文件系统读取
manifest = PluginLoader.read_manifest(plugin.key)
else:
manifest = PluginLoader.read_manifest(plugin.key)
if not manifest:
logger.warning(f"插件 {plugin.key} 缺少 manifest跳过加载")
continue
# 加载后端路由
loaded_routes: List[str] = []
if manifest.backend and manifest.backend.routes:
try:
routers = PluginLoader.load_all_routes(plugin.key, manifest)
for router in routers:
app.include_router(router)
loaded_routes.append(router.prefix)
logger.info(f"插件 {plugin.key} 加载了 {len(routers)} 个路由")
except PluginLoadError as e:
errors.append(f"插件 {plugin.key} 路由加载失败: {e}")
logger.error(f"插件 {plugin.key} 路由加载失败: {e}")
# 加载处理器
loaded_processors: List[str] = []
if manifest.backend and manifest.backend.processors:
try:
processor_types = PluginLoader.load_all_processors(plugin.key, manifest)
loaded_processors = processor_types
logger.info(f"插件 {plugin.key} 注册了 {len(processor_types)} 个处理器")
except PluginLoadError as e:
errors.append(f"插件 {plugin.key} 处理器加载失败: {e}")
logger.error(f"插件 {plugin.key} 处理器加载失败: {e}")
# 更新数据库记录
plugin.loaded_routes = loaded_routes if loaded_routes else None
plugin.loaded_processors = loaded_processors if loaded_processors else None
await plugin.save()
loaded_count += 1
logger.info(f"插件 {plugin.key} 加载完成")
except Exception as e:
error_msg = f"插件 {plugin.key} 加载异常: {e}"
errors.append(error_msg)
logger.exception(error_msg)
return loaded_count, errors
async def init_plugins(app: "FastAPI") -> None:
"""
初始化插件系统
在应用启动时调用
"""
logger.info("开始加载已安装插件...")
loaded_count, errors = await load_installed_plugins(app)
if errors:
logger.warning(f"插件加载完成,共 {loaded_count} 个成功,{len(errors)} 个错误")
for error in errors:
logger.warning(f" - {error}")
else:
logger.info(f"插件加载完成,共 {loaded_count} 个插件")

View File

@@ -1,48 +1,119 @@
from typing import Any, Dict, List, Optional
from pydantic import AliasChoices, BaseModel, ConfigDict, Field
from pydantic import BaseModel, ConfigDict, Field
class PluginCreate(BaseModel):
url: str = Field(min_length=1)
enabled: bool = True
# ========== Manifest 相关类型 ==========
class PluginManifestUpdate(BaseModel):
class ManifestFrontend(BaseModel):
"""manifest.json 中的 frontend 配置"""
model_config = ConfigDict(populate_by_name=True, extra="ignore")
key: Optional[str] = None
name: Optional[str] = None
version: Optional[str] = None
entry: Optional[str] = Field(default=None, description="前端入口文件路径")
styles: Optional[List[str]] = Field(default=None, description="前端样式文件路径列表(相对插件根目录)")
open_app: Optional[bool] = Field(
default=None,
validation_alias=AliasChoices("open_app", "openApp"),
alias="openApp",
description="是否支持独立打开",
)
supported_exts: Optional[List[str]] = Field(
default=None,
validation_alias=AliasChoices("supported_exts", "supportedExts"),
alias="supportedExts",
description="支持的文件扩展名列表",
)
default_bounds: Optional[Dict[str, Any]] = Field(
default=None,
validation_alias=AliasChoices("default_bounds", "defaultBounds"),
alias="defaultBounds",
description="默认窗口尺寸",
)
default_maximized: Optional[bool] = Field(
default=None,
validation_alias=AliasChoices("default_maximized", "defaultMaximized"),
alias="defaultMaximized",
description="是否默认最大化",
)
icon: Optional[str] = None
description: Optional[str] = None
author: Optional[str] = None
website: Optional[str] = None
github: Optional[str] = None
icon: Optional[str] = Field(default=None, description="图标路径")
use_system_window: Optional[bool] = Field(
default=None,
alias="useSystemWindow",
description="是否使用系统窗口",
)
class ManifestRouteConfig(BaseModel):
"""manifest.json 中的路由配置"""
model_config = ConfigDict(extra="ignore")
module: str = Field(..., description="路由模块路径")
prefix: str = Field(..., description="路由前缀")
tags: Optional[List[str]] = Field(default=None, description="API 标签")
class ManifestProcessorConfig(BaseModel):
"""manifest.json 中的处理器配置"""
model_config = ConfigDict(extra="ignore")
module: str = Field(..., description="处理器模块路径")
type: str = Field(..., description="处理器类型标识")
name: Optional[str] = Field(default=None, description="处理器显示名称")
class ManifestBackend(BaseModel):
"""manifest.json 中的 backend 配置"""
model_config = ConfigDict(extra="ignore")
routes: Optional[List[ManifestRouteConfig]] = Field(default=None, description="路由列表")
processors: Optional[List[ManifestProcessorConfig]] = Field(
default=None, description="处理器列表"
)
class ManifestDependencies(BaseModel):
"""manifest.json 中的依赖配置"""
model_config = ConfigDict(extra="ignore")
python: Optional[str] = Field(default=None, description="Python 版本要求")
packages: Optional[List[str]] = Field(default=None, description="Python 包依赖列表")
class PluginManifest(BaseModel):
"""完整的 manifest.json 结构"""
model_config = ConfigDict(populate_by_name=True, extra="ignore")
foxpkg: str = Field(default="1.0", description="foxpkg 格式版本")
key: str = Field(..., min_length=1, description="插件唯一标识")
name: str = Field(..., min_length=1, description="插件名称")
version: str = Field(default="1.0.0", description="插件版本")
description: Optional[str] = Field(default=None, description="插件描述")
i18n: Optional[Dict[str, Dict[str, str]]] = Field(
default=None,
description="多语言信息name/description例如{'en': {'name': '...', 'description': '...'}}",
)
author: Optional[str] = Field(default=None, description="作者")
website: Optional[str] = Field(default=None, description="网站")
github: Optional[str] = Field(default=None, description="GitHub 地址")
license: Optional[str] = Field(default=None, description="许可证")
frontend: Optional[ManifestFrontend] = Field(default=None, description="前端配置")
backend: Optional[ManifestBackend] = Field(default=None, description="后端配置")
dependencies: Optional[ManifestDependencies] = Field(default=None, description="依赖配置")
# ========== API 请求/响应类型 ==========
class PluginOut(BaseModel):
"""插件输出模型"""
id: int
url: str
enabled: bool
key: str
open_app: bool = False
key: Optional[str] = None
name: Optional[str] = None
version: Optional[str] = None
supported_exts: Optional[List[str]] = None
@@ -53,5 +124,20 @@ class PluginOut(BaseModel):
author: Optional[str] = None
website: Optional[str] = None
github: Optional[str] = None
license: Optional[str] = None
# 新增字段
manifest: Optional[Dict[str, Any]] = None
loaded_routes: Optional[List[str]] = None
loaded_processors: Optional[List[str]] = None
model_config = ConfigDict(from_attributes=True)
class PluginInstallResult(BaseModel):
"""安装结果"""
success: bool
plugin: Optional[PluginOut] = None
message: Optional[str] = None
errors: Optional[List[str]] = None

View File

@@ -1,396 +0,0 @@
import hashlib
import json
import os
import re
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import httpx
from domain.virtual_fs.service import VirtualFSService
from domain.virtual_fs.thumbnail import VIDEO_EXT, is_video_filename
DATA_ROOT = Path("data/.video")
TMDB_BASE_URL = "https://api.themoviedb.org/3"
def _sha1(text: str) -> str:
return hashlib.sha1(text.encode("utf-8")).hexdigest()
def _store_path(media_type: str, source_path: str) -> Path:
subdir = "tv" if media_type == "tv" else "movie"
return DATA_ROOT / subdir / f"{_sha1(source_path)}.json"
def _write_json(path: Path, payload: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
_CLEAN_TAGS_RE = re.compile(
r"\b("
r"2160p|1080p|720p|480p|4k|hdr|dv|dolby|atmos|"
r"x264|x265|h264|h265|hevc|av1|aac|dts|flac|"
r"bluray|bdrip|web[- ]?dl|webrip|dvdrip|remux|proper|repack"
r")\b",
re.IGNORECASE,
)
def _clean_query_name(raw: str) -> str:
name = raw
name = name.replace(".", " ").replace("_", " ")
name = re.sub(r"\[[^\]]*\]", " ", name)
name = re.sub(r"\([^\)]*\)", " ", name)
name = _CLEAN_TAGS_RE.sub(" ", name)
name = re.sub(r"\s+", " ", name).strip()
return name
def _guess_name_from_path(path: str, is_dir: bool) -> str:
norm = path.rstrip("/") if is_dir else path
p = Path(norm)
raw = p.name if is_dir else p.stem
return _clean_query_name(raw)
def _as_bool(value: Any, default: bool) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
if isinstance(value, int):
return value != 0
if isinstance(value, str):
v = value.strip().lower()
if v in {"1", "true", "yes", "y", "on"}:
return True
if v in {"0", "false", "no", "n", "off"}:
return False
return default
_SXXEYY_RE = re.compile(r"[Ss](\d{1,2})\s*[.\-_ ]*\s*[Ee](\d{1,3})")
_X_RE = re.compile(r"(\d{1,2})x(\d{1,3})", re.IGNORECASE)
_CN_EP_RE = re.compile(r"\s*(\d{1,3})\s*[集话]")
_CN_SEASON_RE = re.compile(r"\s*(\d{1,2})\s*季")
_SEASON_WORD_RE = re.compile(r"Season\s*(\d{1,2})", re.IGNORECASE)
_S_RE = re.compile(r"[Ss](\d{1,2})")
def _parse_season_episode(rel_path: str) -> Tuple[Optional[int], Optional[int]]:
stem = Path(rel_path).stem
m = _SXXEYY_RE.search(stem) or _SXXEYY_RE.search(rel_path)
if m:
return int(m.group(1)), int(m.group(2))
m = _X_RE.search(stem)
if m:
return int(m.group(1)), int(m.group(2))
m = _CN_EP_RE.search(stem)
if m:
episode = int(m.group(1))
season = None
for part in reversed(Path(rel_path).parts[:-1]):
sm = _CN_SEASON_RE.search(part) or _SEASON_WORD_RE.search(part) or _S_RE.search(part)
if sm:
season = int(sm.group(1))
break
return season or 1, episode
m = re.match(r"^(\d{1,3})(?!\d)", stem)
if m:
episode = int(m.group(1))
season = None
for part in reversed(Path(rel_path).parts[:-1]):
sm = _CN_SEASON_RE.search(part) or _SEASON_WORD_RE.search(part) or _S_RE.search(part)
if sm:
season = int(sm.group(1))
break
return season or 1, episode
return None, None
class TMDBClient:
def __init__(self, access_token: str | None, api_key: str | None):
self._access_token = access_token
self._api_key = api_key
@classmethod
def from_env(cls) -> "TMDBClient":
access_token = os.getenv("TMDB_ACCESS_TOKEN")
api_key = os.getenv("TMDB_API_KEY")
if not access_token and not api_key:
raise RuntimeError("缺少 TMDB_ACCESS_TOKEN 或 TMDB_API_KEY")
return cls(access_token=access_token, api_key=api_key)
def _headers(self) -> dict:
headers = {"Accept": "application/json"}
if self._access_token:
headers["Authorization"] = f"Bearer {self._access_token}"
return headers
def _merge_params(self, params: dict) -> dict:
merged = dict(params or {})
if self._api_key:
merged.setdefault("api_key", self._api_key)
return merged
async def get(self, path: str, params: dict) -> dict:
url = f"{TMDB_BASE_URL}{path}"
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.get(url, headers=self._headers(), params=self._merge_params(params))
resp.raise_for_status()
return resp.json()
class VideoLibraryProcessor:
name = "影视入库"
supported_exts = sorted(VIDEO_EXT)
config_schema = [
{
"key": "name",
"label": "手动名称(可选)",
"type": "string",
"required": False,
"placeholder": "留空则从路径提取",
},
{
"key": "language",
"label": "语言",
"type": "string",
"required": False,
"default": "zh-CN",
},
{
"key": "include_episodes",
"label": "电视剧:保存每集",
"type": "select",
"required": False,
"default": 1,
"options": [
{"label": "", "value": 1},
{"label": "", "value": 0},
],
},
]
produces_file = False
supports_directory = True
requires_input_bytes = False
async def process(self, input_bytes: bytes, path: str, config: Dict[str, Any]) -> Dict[str, Any]:
tmdb = TMDBClient.from_env()
is_dir = await VirtualFSService.path_is_directory(path)
language = str(config.get("language") or "zh-CN")
manual_name = str(config.get("name") or "").strip()
query_name = manual_name or _guess_name_from_path(path, is_dir=is_dir)
scraped_at = datetime.now(UTC).isoformat()
if is_dir:
payload, saved_to = await self._process_tv_dir(tmdb, path, query_name, language, scraped_at, config)
return {
"ok": True,
"type": "tv",
"path": path,
"tmdb_id": payload.get("tmdb", {}).get("id"),
"saved_to": str(saved_to),
}
payload, saved_to = await self._process_movie_file(tmdb, path, query_name, language, scraped_at)
return {
"ok": True,
"type": "movie",
"path": path,
"tmdb_id": payload.get("tmdb", {}).get("id"),
"saved_to": str(saved_to),
}
async def _process_movie_file(
self,
tmdb: TMDBClient,
path: str,
query_name: str,
language: str,
scraped_at: str,
) -> Tuple[dict, Path]:
search = await tmdb.get("/search/movie", {"query": query_name, "language": language})
results = search.get("results") or []
if not results:
raise RuntimeError(f"未找到电影条目:{query_name}")
chosen = results[0] or {}
movie_id = chosen.get("id")
if not movie_id:
raise RuntimeError("TMDB 搜索结果缺少 id")
detail = await tmdb.get(
f"/movie/{movie_id}",
{
"language": language,
"append_to_response": "credits,images,external_ids,videos",
},
)
payload = {
"type": "movie",
"source_path": path,
"query": {"name": query_name, "language": language},
"scraped_at": scraped_at,
"tmdb": {
"id": movie_id,
"search": {"page": search.get("page"), "total_results": search.get("total_results"), "results": results[:5]},
"detail": detail,
},
}
saved_to = _store_path("movie", path)
_write_json(saved_to, payload)
return payload, saved_to
async def _process_tv_dir(
self,
tmdb: TMDBClient,
path: str,
query_name: str,
language: str,
scraped_at: str,
config: Dict[str, Any],
) -> Tuple[dict, Path]:
search = await tmdb.get("/search/tv", {"query": query_name, "language": language})
results = search.get("results") or []
if not results:
raise RuntimeError(f"未找到电视剧条目:{query_name}")
chosen = results[0] or {}
tv_id = chosen.get("id")
if not tv_id:
raise RuntimeError("TMDB 搜索结果缺少 id")
detail = await tmdb.get(
f"/tv/{tv_id}",
{
"language": language,
"append_to_response": "credits,images,external_ids,videos",
},
)
include_episodes = _as_bool(config.get("include_episodes"), True)
episodes: List[dict] = []
seasons_detail: Dict[str, Any] = {}
if include_episodes:
episodes = await self._collect_episode_files(path)
seasons = sorted({ep["season"] for ep in episodes if ep.get("season") is not None})
for season in seasons:
seasons_detail[str(season)] = await tmdb.get(
f"/tv/{tv_id}/season/{int(season)}",
{"language": language},
)
self._attach_tmdb_episode_detail(episodes, seasons_detail)
payload = {
"type": "tv",
"source_path": path,
"query": {"name": query_name, "language": language},
"scraped_at": scraped_at,
"tmdb": {
"id": tv_id,
"search": {"page": search.get("page"), "total_results": search.get("total_results"), "results": results[:5]},
"detail": detail,
"seasons": seasons_detail,
},
"episodes": episodes,
}
saved_to = _store_path("tv", path)
_write_json(saved_to, payload)
return payload, saved_to
async def _collect_episode_files(self, dir_path: str) -> List[dict]:
adapter_instance, adapter_model, root, rel = await VirtualFSService.resolve_adapter_and_rel(dir_path)
rel = rel.rstrip("/")
list_dir = await VirtualFSService._ensure_method(adapter_instance, "list_dir")
stack: List[str] = [rel]
page_size = 200
out: List[dict] = []
while stack:
current_rel = stack.pop()
page = 1
while True:
entries, total = await list_dir(root, current_rel, page, page_size, "name", "asc")
entries = entries or []
if not entries and (total or 0) == 0:
break
for entry in entries:
name = entry.get("name")
if not name:
continue
child_rel = VirtualFSService._join_rel(current_rel, name)
if entry.get("is_dir"):
stack.append(child_rel.rstrip("/"))
continue
if not is_video_filename(name):
continue
absolute_path = VirtualFSService._build_absolute_path(adapter_model.path, child_rel)
rel_in_show = child_rel
if rel and child_rel.startswith(rel.rstrip("/") + "/"):
rel_in_show = child_rel[len(rel.rstrip("/")) + 1 :]
season, episode = _parse_season_episode(rel_in_show)
out.append(
{
"path": absolute_path,
"rel": rel_in_show,
"name": name,
"size": entry.get("size"),
"mtime": entry.get("mtime"),
"season": season,
"episode": episode,
}
)
if total is None or page * page_size >= total:
break
page += 1
return out
def _attach_tmdb_episode_detail(self, episodes: List[dict], seasons_detail: Dict[str, Any]) -> None:
episode_maps: Dict[str, Dict[int, Any]] = {}
for season_str, season_payload in (seasons_detail or {}).items():
items = (season_payload or {}).get("episodes") or []
m: Dict[int, Any] = {}
for item in items:
try:
number = int(item.get("episode_number"))
except Exception:
continue
m[number] = item
episode_maps[season_str] = m
for ep in episodes:
season = ep.get("season")
episode = ep.get("episode")
if season is None or episode is None:
continue
m = episode_maps.get(str(season))
if not m:
continue
detail = m.get(int(episode))
if detail:
ep["tmdb_episode"] = detail
PROCESSOR_TYPE = "video_library"
PROCESSOR_NAME = VideoLibraryProcessor.name
SUPPORTED_EXTS = VideoLibraryProcessor.supported_exts
CONFIG_SCHEMA = VideoLibraryProcessor.config_schema
PROCESSOR_FACTORY = lambda: VideoLibraryProcessor()