mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 19:37:54 +08:00
408 lines
18 KiB
Python
408 lines
18 KiB
Python
"""
|
||
插件分身创建运行时用例。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
from collections.abc import Callable
|
||
from typing import Any, NamedTuple, Optional, Protocol
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from app.schemas.plugin import PluginInstance, PluginRuntimeStatus
|
||
|
||
# 自动分配的后缀从 2 起算：源插件本体在用户眼里就是第 1 个实例，分身接着往下排
|
||
_FIRST_AUTO_SUFFIX = 2
|
||
# 在「已登记分身数」之上再多探测的号数，取值见 PluginCloneService._allocate_suffix
|
||
_AUTO_SUFFIX_PROBE_SLACK = 1000
|
||
|
||
|
||
def _first_validation_message(error: ValidationError) -> str:
|
||
"""
|
||
取校验错误里的首条说明，供回给用户的单行原因使用。
|
||
"""
|
||
details = error.errors()
|
||
return str(details[0].get("msg")) if details else str(error)
|
||
|
||
|
||
class InstanceIdTaken(Protocol):
|
||
"""
|
||
候选实例 ID 的占用判据。
|
||
"""
|
||
|
||
def __call__(
|
||
self,
|
||
instance_id: str,
|
||
*,
|
||
ignore_instance_row: bool = False,
|
||
) -> bool:
|
||
"""
|
||
判断该 ID 是否已被某个插件身份占住。
|
||
|
||
:param instance_id: 候选实例 ID
|
||
:param ignore_instance_row: 是否把「实例表里已有这一行」排除在占用之外。恢复
|
||
一个已停用的分身时为真：那一行正是本次要拿回来的东西，它自己不构成冲突
|
||
:return: 该 ID 是否已被占用
|
||
"""
|
||
|
||
|
||
class _Reservation(NamedTuple):
|
||
"""
|
||
占位阶段已经定下来的事实，供后续加载与回滚判断来历。
|
||
|
||
:param clone_id: 已预留的插件分身标识
|
||
:param restoring: 是否恢复已有停用实例，而非创建新记录
|
||
"""
|
||
|
||
clone_id: str
|
||
restoring: bool
|
||
|
||
|
||
class _ClonePlan(NamedTuple):
|
||
"""
|
||
落库之前定下的分身方案：写哪一行，以及这一行是新建还是恢复。
|
||
|
||
:param reservation: 本次创建或恢复操作取得的预留事实
|
||
:param instance: 待写入数据库的完整插件实例记录
|
||
"""
|
||
|
||
reservation: _Reservation
|
||
instance: PluginInstance
|
||
|
||
|
||
class PluginCloneService:
|
||
"""
|
||
创建共享源码的虚拟插件实例并协调失败回滚。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
plugin_class: Callable[[str], Optional[Any]],
|
||
instance_id_taken: InstanceIdTaken,
|
||
get_instance: Callable[[str], Optional[PluginInstance]],
|
||
instances_for_source: Callable[[str], list[PluginInstance]],
|
||
source_plugin_id: Callable[[str], str],
|
||
save_instance: Callable[[PluginInstance], Any],
|
||
delete_instance: Callable[[str], bool],
|
||
disable_instance: Callable[[str], bool],
|
||
read_config: Callable[[str], dict[str, object]],
|
||
save_config: Callable[[str, dict[str, object]], bool],
|
||
delete_config: Callable[[str], bool],
|
||
reload_plugin: Callable[[str], Any],
|
||
remove_plugin: Callable[[str], Any],
|
||
log: Any,
|
||
) -> None:
|
||
"""
|
||
保存实例描述、持久化和运行态端口。
|
||
"""
|
||
self._plugin_class = plugin_class
|
||
self._instance_id_taken = instance_id_taken
|
||
self._get_instance = get_instance
|
||
self._instances_for_source = instances_for_source
|
||
self._source_plugin_id = source_plugin_id
|
||
self._save_instance = save_instance
|
||
self._delete_instance = delete_instance
|
||
self._disable_instance = disable_instance
|
||
self._read_config = read_config
|
||
self._save_config = save_config
|
||
self._delete_config = delete_config
|
||
self._reload_plugin = reload_plugin
|
||
self._remove_plugin = remove_plugin
|
||
self._logger = log
|
||
# 「挑一个没被占用的 ID」与「把这个 ID 占下来」之间存在窗口，两个并发的创建
|
||
# 请求会在窗口里选出同一个 ID，随后一个覆盖另一个的实例行。创建是低频的管理
|
||
# 动作，用一把进程内互斥把占位串行化即可，代价远小于让两个分身共用一行
|
||
self._reservation_lock = threading.Lock()
|
||
|
||
def clone(
|
||
self,
|
||
*,
|
||
plugin_id: str,
|
||
suffix: Optional[str] = None,
|
||
name: str,
|
||
description: str,
|
||
version: Optional[str] = None,
|
||
icon: Optional[str] = None,
|
||
restore_previous: bool = True,
|
||
) -> tuple[bool, str]:
|
||
"""
|
||
创建虚拟分身，复制隔离配置并保持默认禁用语义。
|
||
|
||
:param plugin_id: 源插件ID
|
||
:param suffix: 追加到源插件ID后的分身后缀；留空时自动分配最小可用序号
|
||
:param name: 分身展示名称，恢复时留空表示沿用停用前登记的那一份
|
||
:param description: 分身展示描述，留空时的处理同 ``name``
|
||
:param version: 旧客户端仍会带上；分身始终跟随源插件版本，此处不消费
|
||
:param icon: 分身展示图标，留空时的处理同 ``name``
|
||
:param restore_previous: 该后缀名下留有一个已停用的分身时是否沿用它的业务
|
||
参数；为假时改按源插件模板重建一份配置
|
||
:return: 是否成功与分身ID或失败原因
|
||
"""
|
||
del version
|
||
rejection = self._reject_invalid_source(plugin_id)
|
||
if rejection:
|
||
return False, rejection
|
||
|
||
# 实例行落库、备配置与首次加载同属一次创建，共用这一个异常与清理边界：落库留在
|
||
# 边界之外时，写到一半失败既不会按约定回报 (False, 原因)，也不会触发回滚，一行
|
||
# 没有配置、也没有运行态的实例就此留在表里
|
||
reservation: Optional[_Reservation] = None
|
||
try:
|
||
with self._reservation_lock:
|
||
plan, message = self._plan(
|
||
plugin_id=plugin_id,
|
||
suffix=suffix,
|
||
name=name,
|
||
description=description,
|
||
icon=icon,
|
||
restore_previous=restore_previous,
|
||
)
|
||
if plan is None:
|
||
return False, message
|
||
self._claim(plan)
|
||
# 记下来历的时机就是占位成功的时机：此后这一行确定归本次所有，锁外的
|
||
# 清理不会误伤别人。落库失败的清理由 _claim 在锁内自己做完
|
||
reservation = plan.reservation
|
||
self._activate(
|
||
reservation,
|
||
plugin_id=plugin_id,
|
||
restore_previous=restore_previous,
|
||
)
|
||
action = "恢复" if reservation.restoring else "创建"
|
||
self._logger.info(f"插件分身 {reservation.clone_id} {action}成功")
|
||
return True, reservation.clone_id
|
||
# 分身创建的任一步骤失败都要释放已取得的预留并回滚已写入状态。
|
||
except Exception as error: # noqa: BLE001
|
||
# 没占到位就没有任何东西落下来，无需也无从回滚
|
||
if reservation is not None:
|
||
self._rollback(
|
||
reservation.clone_id,
|
||
purge_instance=not reservation.restoring,
|
||
)
|
||
self._logger.error(f"创建插件分身失败：{error}")
|
||
return False, f"创建插件分身失败：{error}"
|
||
|
||
def _reject_invalid_source(self, plugin_id: str) -> str:
|
||
"""
|
||
判断给定插件能否作为分身来源，可以时返回空串。
|
||
|
||
分身不能再生分身：分身共享源插件的源码，它自己并不是一份源码。放行的话新实例
|
||
会挂到源插件名下，后缀却是按分身 ID 算的，自动分配因而会在一个与占用判据对不上
|
||
的号段里挑号。
|
||
"""
|
||
if not plugin_id:
|
||
return "插件ID不能为空"
|
||
if self._plugin_class(plugin_id) is None:
|
||
return f"原插件 {plugin_id} 不存在"
|
||
source_id = self._source_plugin_id(plugin_id)
|
||
if source_id != plugin_id:
|
||
return (
|
||
f"{plugin_id} 是分身实例，不能作为分身来源，"
|
||
f"请对源插件 {source_id} 创建分身"
|
||
)
|
||
return ""
|
||
|
||
def _plan(
|
||
self,
|
||
*,
|
||
plugin_id: str,
|
||
suffix: Optional[str],
|
||
name: str,
|
||
description: str,
|
||
icon: Optional[str],
|
||
restore_previous: bool,
|
||
) -> tuple[Optional[_ClonePlan], str]:
|
||
"""
|
||
定下分身 ID 与要写出的实例行，或给出拒绝原因；本方法自身不写任何东西。
|
||
|
||
调用方须持有占位锁：判存是「先读后写」的前半段，两个并发创建各自读到「未占用」
|
||
就会落到同一行上。
|
||
"""
|
||
resolved_suffix = (suffix or "").strip().lower() or self._allocate_suffix(plugin_id)
|
||
if not resolved_suffix:
|
||
return None, f"插件 {plugin_id} 的可用分身后缀已耗尽，请手动指定一个"
|
||
clone_id = f"{plugin_id}{resolved_suffix}"
|
||
# 已停用的分身仍留着自己那一行，业务参数原样挂在上面：同后缀重建就是把它连同
|
||
# 配置一起拿回来，而不是撞车。归属对不上的行不算可恢复——它是别的源插件的分身
|
||
previous = self._get_instance(clone_id)
|
||
restoring = (
|
||
previous is not None
|
||
and not previous.is_enabled
|
||
and previous.source_plugin_id == plugin_id
|
||
)
|
||
# 恢复只豁免待恢复的那一行自身。停用从不删行，那一行可以在表里躺很久，其间同名
|
||
# 的真实插件完全可能出现（装进类注册表、进安装清单、或在磁盘上留下插件包）；此时
|
||
# 放行等于让分身顶掉一个真实插件的身份，那个插件此后连装都装不回来
|
||
if self._instance_id_taken(clone_id, ignore_instance_row=restoring):
|
||
return None, (
|
||
f"分身插件 {clone_id} 的 ID 已被其他插件占用，无法恢复"
|
||
if restoring
|
||
else f"分身插件 {clone_id} 已存在"
|
||
)
|
||
|
||
instance, message = self._build_instance(
|
||
clone_id=clone_id,
|
||
source_plugin_id=plugin_id,
|
||
name=name,
|
||
description=description,
|
||
icon=icon,
|
||
previous=previous if restoring else None,
|
||
inherit_display=restore_previous,
|
||
)
|
||
if instance is None:
|
||
return None, message
|
||
return _ClonePlan(
|
||
reservation=_Reservation(clone_id=clone_id, restoring=restoring),
|
||
instance=instance,
|
||
), ""
|
||
|
||
def _claim(self, plan: _ClonePlan) -> None:
|
||
"""
|
||
把实例行写出去占住这个 ID，写入失败时就地清理并把异常抛出。
|
||
|
||
调用方须持有占位锁，清理也必须在锁内做完：锁一放开，另一个同后缀请求立刻就能
|
||
把这个 ID 占下来并成功落库，而按 ID 清理分不出那一行是谁写的，删掉的会是它的
|
||
行、配置和运行态。占位成功之后则不必再持锁——实例行此时已经挡住同 ID 的后续
|
||
请求（判存认这一行），这个 ID 确定归本次所有，直到本次自己把它清掉。
|
||
|
||
:param plan: 本次要写出的实例行与它的来历
|
||
:raise Exception: 原样抛出持久化层的写入错误，交由 :meth:`clone` 统一回报
|
||
"""
|
||
try:
|
||
self._save_instance(plan.instance)
|
||
except Exception:
|
||
self._rollback(
|
||
plan.reservation.clone_id,
|
||
purge_instance=not plan.reservation.restoring,
|
||
)
|
||
raise
|
||
|
||
def _allocate_suffix(self, plugin_id: str) -> str:
|
||
"""
|
||
为新分身分配一个最小可用的数字后缀，无号可用时返回空串。
|
||
|
||
占用判据与显式指定后缀走的是同一个 ``instance_id_taken``，自动分配因此不可能
|
||
挑中一个手填时会被判成「已存在」的 ID。已停用的分身同样占位：重用它的 ID 是
|
||
「恢复」而不是新建，用户没点恢复就不该凭空拿到上一个分身留下的配置。
|
||
|
||
探测上限随该插件已登记的分身数一起抬高。定成一个常数的话，分身数量越过它之后
|
||
每次自动分配都报「后缀已耗尽」，而下一个号明明空着；按鸽巢原理，N 个已登记分身
|
||
最多占掉 N 个号，再多探测一段余量必然能撞上空位。上限仍然有限：占用判据若因端口
|
||
故障恒为真，无界的循环会让请求线程原地打转并一直扣着占位锁。
|
||
"""
|
||
probe_limit = len(self._instances_for_source(plugin_id)) + _AUTO_SUFFIX_PROBE_SLACK
|
||
for index in range(_FIRST_AUTO_SUFFIX, _FIRST_AUTO_SUFFIX + probe_limit):
|
||
if not self._instance_id_taken(f"{plugin_id}{index}"):
|
||
return str(index)
|
||
return ""
|
||
|
||
def _build_instance(
|
||
self,
|
||
*,
|
||
clone_id: str,
|
||
source_plugin_id: str,
|
||
name: str,
|
||
description: str,
|
||
icon: Optional[str],
|
||
previous: Optional[PluginInstance],
|
||
inherit_display: bool,
|
||
) -> tuple[Optional[PluginInstance], str]:
|
||
"""
|
||
构造实例描述，并在任何写入之前拦下不合法的实例 ID。
|
||
|
||
合法性由 ``PluginInstance`` 自己判定，这里不另抄一份规则；构造排在落库之前，
|
||
非法 ID 因而不会先写出一行再靠回滚擦掉。
|
||
|
||
:param previous: 正在被恢复的那一行，非恢复时为 None
|
||
:param inherit_display: 恢复时展示信息是否沿用停用前登记的那一份
|
||
"""
|
||
if previous is not None and inherit_display:
|
||
name = name or (previous.plugin_name or "")
|
||
description = description or (previous.plugin_desc or "")
|
||
icon = icon or (previous.plugin_icon or "")
|
||
try:
|
||
instance = PluginInstance(
|
||
instance_id=clone_id,
|
||
source_plugin_id=source_plugin_id,
|
||
plugin_name=name or None,
|
||
plugin_desc=description or None,
|
||
plugin_icon=icon or None,
|
||
# 恢复时保留该行原有的默认目标置位；新建的分身不该凭空成为默认目标
|
||
is_default_target=(
|
||
previous.is_default_target if previous is not None else False
|
||
),
|
||
# 新建的分身就是要拿去跑的；启用位是装载判据，留空会让它建出来却不加载
|
||
is_enabled=True,
|
||
)
|
||
except ValidationError as error:
|
||
return None, f"分身实例 ID {clone_id} 不合法：{_first_validation_message(error)}"
|
||
return instance, ""
|
||
|
||
def _activate(
|
||
self,
|
||
reservation: _Reservation,
|
||
*,
|
||
plugin_id: str,
|
||
restore_previous: bool,
|
||
) -> None:
|
||
"""
|
||
准备配置并完成首次加载，失败一律抛出，由 :meth:`clone` 统一回滚。
|
||
"""
|
||
# 恢复留存配置时不得用源插件模板盖掉它，那正是用户要拿回来的东西
|
||
if not (reservation.restoring and restore_previous):
|
||
self._rebuild_config(reservation.clone_id, source_plugin_id=plugin_id)
|
||
if self._reload_plugin(reservation.clone_id) is PluginRuntimeStatus.LOAD_FAILED:
|
||
raise RuntimeError("虚拟实例加载失败")
|
||
|
||
def _rebuild_config(self, clone_id: str, *, source_plugin_id: str) -> None:
|
||
"""
|
||
按源插件当前的配置模板重建分身配置，模板为空时清掉分身原有的那一份。
|
||
|
||
源插件没有配置与源插件配置为空字典都落在「模板里没有可继承的业务参数」这一档，
|
||
两者都必须清除：只是跳过写入的话，已停用分身留存的旧参数会原样活下来并在重载后
|
||
继续生效，用户要的「按模板重建一份」于是变成了「沿用旧配置」，且毫无提示。
|
||
|
||
:param clone_id: 分身实例 ID
|
||
:param source_plugin_id: 提供配置模板的源插件 ID
|
||
:raise RuntimeError: 配置写入被持久化层拒绝
|
||
"""
|
||
template = self._read_config(source_plugin_id)
|
||
if not template:
|
||
self._delete_config(clone_id)
|
||
return
|
||
clone_config = dict(template)
|
||
# 分身是照着源插件的参数建出来的，业务开关一律置假：它还没被用户配过，直接开跑
|
||
# 等于拿着别人的参数上线
|
||
clone_config["enable"] = False
|
||
clone_config["enabled"] = False
|
||
if not self._save_config(clone_id, clone_config):
|
||
raise RuntimeError("虚拟实例配置保存失败")
|
||
|
||
def _rollback(self, clone_id: str, *, purge_instance: bool) -> None:
|
||
"""
|
||
逐项清理失败实例，单个清理错误不得阻断其余回滚。
|
||
|
||
:param clone_id: 本次创建的分身ID
|
||
:param purge_instance: 实例行与配置是否由本次创建产生、可随之抹掉；恢复一个
|
||
已停用的分身时为假——那一行连同用户留在上面的业务参数是特意保留的，一次
|
||
加载失败不该把它毁掉，只需把启用位退回停用，下次仍可再试一遍恢复
|
||
"""
|
||
rollback_steps: list[tuple[str, Callable[[str], Any]]] = [
|
||
("运行态", self._remove_plugin),
|
||
]
|
||
if purge_instance:
|
||
rollback_steps.append(("实例描述", self._delete_instance))
|
||
rollback_steps.append(("配置", self._delete_config))
|
||
else:
|
||
rollback_steps.append(("启用位", self._disable_instance))
|
||
for label, rollback in rollback_steps:
|
||
try:
|
||
rollback(clone_id)
|
||
# 某一步回滚失败仍需继续其余清理步骤，并记录需要后续恢复的状态。
|
||
except Exception as rollback_error: # noqa: BLE001
|
||
self._logger.warning(
|
||
f"回滚插件分身 {clone_id} 的{label}失败：{rollback_error}"
|
||
)
|