mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-14 09:51:28 +08:00
e47d7defe7
全局日志等级是一个开关，调试单个插件时只能整体下调。一旦下调，被调试插件那几行 日志会淹没在其他几十个插件与宿主自身的 DEBUG 输出里，找不出来；不下调又什么都 看不见。插件实例数量只会继续增长，这个矛盾不会自己消失。 因此把等级下放到实例：plugininstance 行加 log_level / log_expires_at 两列，本体与 分身共用同一张表、同一套语义。等级落在实例自己那一行而不是另起一张按实例 ID 索引 的表——它和业务参数一样是用户按实例设置的东西，同一个生命周期，实例被删时应当随行 一起消失。清除覆盖后若本体行只剩一对身份列，整行按 carries_only_identity 回收，与 清空业务参数同一规则。 覆盖带失效时间，因为它是调试设置不是长期配置：开了 DEBUG 忘记关，下次谁也想不起 某个插件为什么一直在刷日志。过期判定在读取时惰性执行并就地清理，不另起后台线程去 扫一张最多几十条的表。expires_at 在写入侧先归一到 UTC 再分发给缓存与落盘两处：客户端 提交的时间可能不带时区，两处各按进程时区折算一次，就会与读取口返回的 UTC 值差出一个 时区，表现为「提交的失效时刻和读回来的对不上」。 运行期用 ContextVar 在宿主自己控制的四类调用点绑定当前实例——构造与 init_plugin、 事件处理器、定时服务回调、插件声明的 HTTP 端点——而不是沿用 logger() 已有的栈帧内省。 内省认的是栈顶的模块与类，插件通过宿主公共方法转发调用时栈顶是宿主而不是发起调用的 插件，会把日志记到错误的来源上；分身更甚：分身共享源码，类的 __qualname__ 保持源类名 不变，内省根本分不出是哪一个分身在打日志。内省结果继续只用于文件路由，不参与等级判定。 等级过滤必须排在全局等级判定之前。官方现状是先按全局等级短路、之后才做内省，先短路 就会把实例调低的等级挡在门外，后面再怎么识别来源都看不到这条日志。 回调在注册时被捕获、稍后才由调度器或 FastAPI 调用，那时已经不在任何绑定作用域内， 故这两处改用 wrap_for_plugin_instance 把绑定包进回调自身；同一实例重复包装直接返回 原对象，避免插件缓存声明时包装链随重载次数无限增长。 进程内覆盖表按实例 ID 常驻，删分身时必须一并清掉：只删库里那一行的话，同一进程内用 相同后缀重建的分身会继承上一个分身的等级，界面显示「跟随全局」而实际仍按旧等级输出。 进程重启后反向从库预热缓存。 迁移链在 F1 之后追加：c4e1a7b9d2f6（3.0.35）-> 487f7e681955（3.0.36 加两列）。 顺带补了 agent 策略里缺失的 media.classification.policy.get 的 template 字段说明： 少了它，api_mcp_schema.json 与技能文档的两个生成脚本都跑不起来。
412 lines
15 KiB
Python
412 lines
15 KiB
Python
"""插件运行时持久化端口。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Awaitable, Callable
|
||
from datetime import datetime
|
||
from typing import Any, Optional
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from app.runtime.extensions.plugin.database import PluginDatabase
|
||
from app.schemas.plugin import PluginInstance
|
||
from app.schemas.types import SystemConfigKey
|
||
|
||
|
||
ConfigReader = Callable[[Any], Any]
|
||
ConfigWriter = Callable[[Any, Any], Any]
|
||
AsyncConfigWriter = Callable[[Any, Any], Awaitable[Any]]
|
||
ConfigDeleter = Callable[[Any], bool]
|
||
PluginDataDeleter = Callable[[str], Any]
|
||
# 日志等级覆盖端口的载荷：`(等级名, 失效时间)`，两者皆为 None 即未设置覆盖
|
||
LogLevelOverride = tuple[Optional[str], Optional[datetime]]
|
||
LogLevelReader = Callable[[str], LogLevelOverride]
|
||
LogLevelWriter = Callable[[str, Optional[str], Optional[datetime]], None]
|
||
PluginExists = Callable[[str], bool]
|
||
|
||
|
||
def _empty_read(_key: Any) -> Any:
|
||
"""组合根尚未装配时返回空配置。"""
|
||
return None
|
||
|
||
|
||
def _ignore_write(_key: Any, _value: Any) -> None:
|
||
"""组合根尚未装配时忽略同步配置写入。"""
|
||
|
||
|
||
async def _ignore_async_write(_key: Any, _value: Any) -> None:
|
||
"""组合根尚未装配时忽略异步配置写入。"""
|
||
|
||
|
||
def _ignore_delete(_key: Any) -> bool:
|
||
"""组合根尚未装配时报告配置未删除。"""
|
||
return False
|
||
|
||
|
||
def _ignore_plugin_data_delete(_plugin_id: str) -> None:
|
||
"""组合根尚未装配时忽略插件数据删除。"""
|
||
|
||
|
||
def _no_log_level(_instance_id: str) -> LogLevelOverride:
|
||
"""组合根尚未装配时报告实例没有日志等级覆盖。"""
|
||
return (None, None)
|
||
|
||
|
||
def _ignore_log_level_write(
|
||
_instance_id: str,
|
||
_level: Optional[str],
|
||
_expires_at: Optional[datetime],
|
||
) -> None:
|
||
"""组合根尚未装配时忽略日志等级落盘。"""
|
||
|
||
|
||
class PluginStorage:
|
||
"""封装插件运行时所需的最小持久化能力。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
read: ConfigReader = _empty_read,
|
||
write: ConfigWriter = _ignore_write,
|
||
async_write: AsyncConfigWriter = _ignore_async_write,
|
||
delete: ConfigDeleter = _ignore_delete,
|
||
delete_data: PluginDataDeleter = _ignore_plugin_data_delete,
|
||
read_log_level: LogLevelReader = _no_log_level,
|
||
write_log_level: LogLevelWriter = _ignore_log_level_write,
|
||
) -> None:
|
||
"""保存由启动组合根提供的读写函数。"""
|
||
self._read = read
|
||
self._write = write
|
||
self._async_write = async_write
|
||
self._delete = delete
|
||
self._delete_data = delete_data
|
||
self._read_log_level = read_log_level
|
||
self._write_log_level = write_log_level
|
||
|
||
def read(self, key: Any) -> Any:
|
||
"""读取插件运行时配置。"""
|
||
return self._read(key)
|
||
|
||
def write(self, key: Any, value: Any) -> Any:
|
||
"""同步保存插件运行时配置。"""
|
||
return self._write(key, value)
|
||
|
||
async def async_write(self, key: Any, value: Any) -> Any:
|
||
"""异步保存插件运行时配置。"""
|
||
return await self._async_write(key, value)
|
||
|
||
def delete(self, key: Any) -> bool:
|
||
"""删除插件运行时配置。"""
|
||
return self._delete(key)
|
||
|
||
def delete_data(self, plugin_id: str) -> Any:
|
||
"""删除指定插件的业务数据。"""
|
||
return self._delete_data(plugin_id)
|
||
|
||
def read_log_level(self, instance_id: str) -> LogLevelOverride:
|
||
"""读取实例配置里登记的日志等级覆盖与失效时间。"""
|
||
return self._read_log_level(instance_id)
|
||
|
||
def write_log_level(
|
||
self,
|
||
instance_id: str,
|
||
level: Optional[str],
|
||
expires_at: Optional[datetime],
|
||
) -> None:
|
||
"""把日志等级覆盖写进该实例的配置行。"""
|
||
self._write_log_level(instance_id, level, expires_at)
|
||
|
||
|
||
class PluginConfigStore:
|
||
"""封装插件配置键、存在性和强制删除规则。"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
storage: Callable[[], "PluginStorage"],
|
||
database: Callable[[], PluginDatabase],
|
||
plugin_exists: PluginExists,
|
||
key_prefix: str = "plugin.%s",
|
||
) -> None:
|
||
"""保存持久化端口、自有数据库端口和运行态插件查询端口。"""
|
||
self._storage = storage
|
||
self._database = database
|
||
self._plugin_exists = plugin_exists
|
||
self._key_prefix = key_prefix
|
||
|
||
def _key(self, plugin_id: str) -> str:
|
||
"""构造插件配置在统一配置存储中的键。"""
|
||
return self._key_prefix % plugin_id
|
||
|
||
def read(self, plugin_id: str) -> dict:
|
||
"""读取配置并过滤历史空键。"""
|
||
if not self._plugin_exists(plugin_id):
|
||
return {}
|
||
config = self._storage().read(self._key(plugin_id))
|
||
return {
|
||
key: value
|
||
for key, value in (config or {}).items()
|
||
if key
|
||
}
|
||
|
||
def write(self, plugin_id: str, config: dict, force: bool = False) -> bool:
|
||
"""保存配置，默认拒绝不存在插件的配置写入。"""
|
||
if not force and not self._plugin_exists(plugin_id):
|
||
return False
|
||
self._storage().write(self._key(plugin_id), config)
|
||
return True
|
||
|
||
async def async_write(
|
||
self,
|
||
plugin_id: str,
|
||
config: dict,
|
||
force: bool = False,
|
||
) -> bool:
|
||
"""异步保存配置并保持同步写入的存在性规则。"""
|
||
if not force and not self._plugin_exists(plugin_id):
|
||
return False
|
||
await self._storage().async_write(self._key(plugin_id), config)
|
||
return True
|
||
|
||
def delete(self, plugin_id: str, force: bool = False) -> bool:
|
||
"""删除配置并保持停止插件后的强制删除能力。"""
|
||
if not force and not self._plugin_exists(plugin_id):
|
||
return False
|
||
return self._storage().delete(self._key(plugin_id))
|
||
|
||
def read_log_level(self, instance_id: str) -> LogLevelOverride:
|
||
"""读取实例的日志等级覆盖；它与业务参数同属该实例的配置。"""
|
||
return self._storage().read_log_level(instance_id)
|
||
|
||
def write_log_level(
|
||
self,
|
||
instance_id: str,
|
||
level: Optional[str],
|
||
expires_at: Optional[datetime],
|
||
) -> None:
|
||
"""写入实例的日志等级覆盖。"""
|
||
self._storage().write_log_level(instance_id, level, expires_at)
|
||
|
||
def delete_data(self, plugin_id: str, force: bool = False) -> bool:
|
||
"""删除插件业务数据与自有数据库，并保持旧的布尔结果合同。"""
|
||
if not force and not self._plugin_exists(plugin_id):
|
||
return False
|
||
self._storage().delete_data(plugin_id)
|
||
self._database().destroy(plugin_id)
|
||
return True
|
||
|
||
|
||
InstanceReader = Callable[[str], "PluginInstance | None"]
|
||
InstanceLister = Callable[[], "list[PluginInstance]"]
|
||
InstanceSourceLister = Callable[[str], "list[PluginInstance]"]
|
||
InstanceWriter = Callable[["PluginInstance"], None]
|
||
InstanceDeleter = Callable[[str], bool]
|
||
|
||
|
||
def _empty_instance_get(_instance_id: str) -> PluginInstance | None:
|
||
"""组合根尚未装配时返回空实例描述。"""
|
||
return None
|
||
|
||
|
||
def _empty_instance_list() -> list[PluginInstance]:
|
||
"""组合根尚未装配时返回空实例列表。"""
|
||
return []
|
||
|
||
|
||
def _empty_instance_list_by_source(_source_plugin_id: str) -> list[PluginInstance]:
|
||
"""组合根尚未装配时返回空实例列表。"""
|
||
return []
|
||
|
||
|
||
def _ignore_instance_save(_instance: PluginInstance) -> None:
|
||
"""组合根尚未装配时忽略实例描述写入。"""
|
||
|
||
|
||
def _ignore_instance_delete(_instance_id: str) -> bool:
|
||
"""组合根尚未装配时报告实例描述未删除。"""
|
||
return False
|
||
|
||
|
||
class PluginInstanceDirectory:
|
||
"""封装插件实例表的持久化能力。
|
||
|
||
分身与源插件本体共用同一张表、同一套读写原语，两者靠 ``instance_id`` 是否等于
|
||
``source_plugin_id`` 区分；本类不做角色过滤，角色隔离由调用方
|
||
（``PluginInstanceStore``）负责，因为只有调用方知道当前服务的是分身清单还是
|
||
本体自身的那一行。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
get: InstanceReader = _empty_instance_get,
|
||
list_all: InstanceLister = _empty_instance_list,
|
||
list_by_source: InstanceSourceLister = _empty_instance_list_by_source,
|
||
save: InstanceWriter = _ignore_instance_save,
|
||
delete: InstanceDeleter = _ignore_instance_delete,
|
||
) -> None:
|
||
"""保存由启动组合根提供的实例表读写函数。"""
|
||
self._get = get
|
||
self._list_all = list_all
|
||
self._list_by_source = list_by_source
|
||
self._save = save
|
||
self._delete = delete
|
||
|
||
def get(self, instance_id: str) -> PluginInstance | None:
|
||
"""按实例 ID 读取单条描述，不区分分身与本体。"""
|
||
return self._get(instance_id)
|
||
|
||
def list_all(self) -> list[PluginInstance]:
|
||
"""列出表中全部描述，不区分分身与本体。"""
|
||
return self._list_all()
|
||
|
||
def list_by_source(self, source_plugin_id: str) -> list[PluginInstance]:
|
||
"""按源插件 ID 列出其全部描述，不区分分身与本体。"""
|
||
return self._list_by_source(source_plugin_id)
|
||
|
||
def save(self, instance: PluginInstance) -> None:
|
||
"""新增或更新一条描述，以 ``instance_id`` 为稳定键。"""
|
||
self._save(instance)
|
||
|
||
def delete(self, instance_id: str) -> bool:
|
||
"""按实例 ID 删除一行，连同其配置，返回删除前是否存在。"""
|
||
return self._delete(instance_id)
|
||
|
||
|
||
class PluginInstanceStore:
|
||
"""管理共享源码的分身实例描述，并把源插件本体自身的那一行隔离在视图之外。
|
||
|
||
两类记录同存一张表，靠 ``instance_id`` 是否等于 ``source_plugin_id`` 区分：
|
||
本类只服务分身，本体行（它承载插件自身的业务参数）读不到也改不到。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
storage: Callable[[], "PluginStorage"],
|
||
directory: Callable[[], PluginInstanceDirectory],
|
||
) -> None:
|
||
"""保存独立表持久化端口，以及旧 systemconfig 单键端口供兜底导入使用。"""
|
||
self._storage = storage
|
||
self._directory = directory
|
||
self._bootstrap_checked = False
|
||
|
||
def _ensure_bootstrapped(self) -> None:
|
||
"""旧 systemconfig 单键的内容向独立表兜底导入一次，且只导入一次。
|
||
|
||
alembic 迁移已经搬过一轮，这里兜的是「库结构升级后又用旧版本写过分身」这类
|
||
回滚往返。导入完成后落一个持久化标记，判据不能是「表当前为空」：旧键刻意保留
|
||
作回滚依据、从不清理，用户把分身全部删光后表就会重新变空，下次进程启动会把
|
||
已删除的分身整批导回来，删一次复活一次。进程内另外维护一个已检查标志，避免
|
||
每次访问都为此多打一次查询。
|
||
|
||
:raise Exception: 导入失败时向上抛出，不吞掉持久化层错误
|
||
"""
|
||
if self._bootstrap_checked:
|
||
return
|
||
self._bootstrap_checked = True
|
||
storage = self._storage()
|
||
if storage.read(SystemConfigKey.PluginInstancesImported):
|
||
return
|
||
directory = self._directory()
|
||
if not directory.list_all():
|
||
for instance in self._legacy_instances().values():
|
||
directory.save(instance)
|
||
storage.write(SystemConfigKey.PluginInstancesImported, True)
|
||
|
||
def _legacy_instances(self) -> dict[str, PluginInstance]:
|
||
"""解析旧 systemconfig 单键里的实例描述，兼容历史字典与列表两种载荷形态。"""
|
||
raw_instances = self._storage().read(SystemConfigKey.PluginInstances) or {}
|
||
if isinstance(raw_instances, list):
|
||
entries = {
|
||
item.get("instance_id"): item
|
||
for item in raw_instances
|
||
if isinstance(item, dict) and item.get("instance_id")
|
||
}
|
||
elif isinstance(raw_instances, dict):
|
||
entries = raw_instances
|
||
else:
|
||
return {}
|
||
|
||
instances: dict[str, PluginInstance] = {}
|
||
for instance_id, raw_instance in entries.items():
|
||
try:
|
||
payload = dict(raw_instance) if isinstance(raw_instance, dict) else {}
|
||
payload.setdefault("instance_id", instance_id)
|
||
instance = PluginInstance.model_validate(payload)
|
||
instances[instance.instance_id] = instance
|
||
except (TypeError, ValidationError):
|
||
continue
|
||
return instances
|
||
|
||
def all(self) -> dict[str, PluginInstance]:
|
||
"""读取全部登记的分身实例，不含源插件本体自身的那一行。"""
|
||
self._ensure_bootstrapped()
|
||
return {
|
||
record.instance_id: record
|
||
for record in self._directory().list_all()
|
||
if not record.is_host
|
||
}
|
||
|
||
def get(self, instance_id: str) -> PluginInstance | None:
|
||
"""读取指定分身实例描述；本体自身的那一行不会从这里返回。"""
|
||
self._ensure_bootstrapped()
|
||
record = self._directory().get(instance_id)
|
||
return record if record is not None and not record.is_host else None
|
||
|
||
def save(self, instance: PluginInstance) -> None:
|
||
"""新增或更新分身实例描述，并以实例 ID 作为稳定持久化键。
|
||
|
||
分身的实例 ID 必须区别于其源插件 ID：两者相等的那一行表示的是本体自身，
|
||
按分身写入会把本体承载的业务参数顶掉。
|
||
"""
|
||
self._ensure_bootstrapped()
|
||
if instance.is_host:
|
||
raise ValueError(
|
||
f"分身实例 {instance.instance_id} 的 ID 不能等于其源插件 ID"
|
||
)
|
||
self._directory().save(instance)
|
||
|
||
def delete(self, instance_id: str) -> bool:
|
||
"""删除指定分身实例描述连同其配置，返回删除前是否存在。"""
|
||
self._ensure_bootstrapped()
|
||
record = self.get(instance_id)
|
||
if record is None:
|
||
return False
|
||
return self._directory().delete(record.instance_id)
|
||
|
||
def for_source(self, source_plugin_id: str) -> list[PluginInstance]:
|
||
"""按持久化顺序返回引用同一源码插件的全部分身，不含本体自身那一行。"""
|
||
self._ensure_bootstrapped()
|
||
return [
|
||
record
|
||
for record in self._directory().list_by_source(source_plugin_id)
|
||
if not record.is_host
|
||
]
|
||
|
||
|
||
_plugin_storage = PluginStorage()
|
||
_plugin_instance_directory = PluginInstanceDirectory()
|
||
|
||
|
||
def configure_plugin_instance_directory(directory: PluginInstanceDirectory) -> None:
|
||
"""由启动组合根替换插件实例表持久化实现。"""
|
||
global _plugin_instance_directory
|
||
_plugin_instance_directory = directory
|
||
|
||
|
||
def get_plugin_instance_directory() -> PluginInstanceDirectory:
|
||
"""返回当前插件实例表持久化端口。"""
|
||
return _plugin_instance_directory
|
||
|
||
|
||
def configure_plugin_storage(storage: PluginStorage) -> None:
|
||
"""由启动组合根替换插件运行时持久化实现。"""
|
||
global _plugin_storage
|
||
_plugin_storage = storage
|
||
|
||
|
||
def get_plugin_storage() -> PluginStorage:
|
||
"""返回当前插件运行时持久化端口。"""
|
||
return _plugin_storage
|