mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-02 21:34:32 +08:00
342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""
|
||
事件 payload、可见性、投递与可靠性契约清单。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from enum import StrEnum
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, ValidationError
|
||
|
||
import app.schemas.event as event_schemas
|
||
from app.schemas.mediaserver import WebhookEventInfo
|
||
from app.schemas.types import ChainEventType, EventType
|
||
from app.schemas.workflow import ActionContext
|
||
|
||
|
||
class EventDelivery(StrEnum):
|
||
"""
|
||
描述事件当前的交付保证。
|
||
"""
|
||
|
||
# 进程内即时交付，不保证重启后重放
|
||
EPHEMERAL = "ephemeral"
|
||
# 发布前要求由持久化发件箱保障交付
|
||
DURABLE_REQUIRED = "durable_required"
|
||
|
||
|
||
class EventVisibility(StrEnum):
|
||
"""
|
||
描述事件是否属于插件公开扩展面。
|
||
"""
|
||
|
||
# 仅供宿主内部消费者使用
|
||
HOST_ONLY = "host_only"
|
||
# 允许插件订阅的公开事件
|
||
PLUGIN_PUBLIC = "plugin_public"
|
||
# 仅投递给指定插件实例
|
||
TARGET_PLUGIN = "target_plugin"
|
||
|
||
|
||
class EventErrorBehavior(StrEnum):
|
||
"""
|
||
描述 handler 异常后的传播规则。
|
||
"""
|
||
|
||
# 隔离当前处理器异常，继续其他处理器
|
||
ISOLATE = "isolate"
|
||
# 遇到处理器异常时中断当前调用链
|
||
STOP_CHAIN = "stop_chain"
|
||
# 将处理器失败交给通知策略处理
|
||
NOTIFY = "notify"
|
||
|
||
|
||
class EventPayloadMode(StrEnum):
|
||
"""
|
||
描述事件 payload 是运行时对象还是稳定快照。
|
||
"""
|
||
|
||
# 保留进程内运行时对象语义
|
||
RUNTIME = "runtime"
|
||
# 使用可稳定校验的载荷快照
|
||
SNAPSHOT = "snapshot"
|
||
# 保留扩展方定义的载荷结构
|
||
EXTENSION = "extension"
|
||
|
||
|
||
class EventValidationMode(StrEnum):
|
||
"""
|
||
描述契约校验失败后的兼容策略。
|
||
"""
|
||
|
||
# 校验失败只记录诊断，保留兼容调用
|
||
DIAGNOSTIC = "diagnostic"
|
||
# 校验失败即拒绝继续投递
|
||
STRICT = "strict"
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class EventContract:
|
||
"""
|
||
冻结一个事件的 payload 与运行语义。
|
||
|
||
:param event_name: 事件注册名称
|
||
:param payload_model: 完整载荷的校验模型；为 None 时不执行该模型校验
|
||
:param payload_contract: 面向消费者描述的载荷协议名称
|
||
:param mode: 事件的分发模式
|
||
:param visibility: 事件面向宿主、全部插件或指定插件的可见范围
|
||
:param delivery: 事件需要的交付保证
|
||
:param error_behavior: 处理器异常后的传播或隔离策略
|
||
:param ordering: 处理器调用顺序约束
|
||
:param input_model: 链式事件输入模型；未声明时为 None
|
||
:param output_model: 链式事件输出模型；未声明时为 None
|
||
:param schema_version: 载荷结构版本，默认从 1 开始
|
||
:param payload_mode: 载荷保留运行时对象、冻结快照或扩展对象的策略
|
||
:param validation_mode: 校验失败时记录诊断或严格拒绝的策略
|
||
:param sensitive_fields: 输出诊断时需要脱敏的字段路径集合
|
||
:param legacy_reason: 保留历史契约的原因；没有历史兼容说明时为 None
|
||
"""
|
||
|
||
event_name: str
|
||
payload_model: type[BaseModel] | None
|
||
payload_contract: str
|
||
mode: str
|
||
visibility: EventVisibility
|
||
delivery: EventDelivery
|
||
error_behavior: EventErrorBehavior
|
||
ordering: str
|
||
input_model: type[BaseModel] | None = None
|
||
output_model: type[BaseModel] | None = None
|
||
schema_version: int = 1
|
||
payload_mode: EventPayloadMode = EventPayloadMode.RUNTIME
|
||
validation_mode: EventValidationMode = EventValidationMode.DIAGNOSTIC
|
||
sensitive_fields: tuple[str, ...] = ()
|
||
legacy_reason: str | None = None
|
||
|
||
|
||
_PAYLOAD_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = {
|
||
EventType.ConfigChanged: event_schemas.ConfigChangeEventData,
|
||
EventType.PluginReload: event_schemas.PluginReloadEventData,
|
||
EventType.PluginAction: event_schemas.PluginActionEventData,
|
||
EventType.PluginTriggered: event_schemas.PluginTriggeredEventData,
|
||
EventType.CommandExcute: event_schemas.CommandExecuteEventData,
|
||
EventType.SiteDeleted: event_schemas.SiteEventData,
|
||
EventType.SiteUpdated: event_schemas.SiteEventData,
|
||
EventType.SiteRefreshed: event_schemas.SiteEventData,
|
||
EventType.AgentTokensUsage: event_schemas.AgentTokensUsageEventData,
|
||
EventType.SubscribeAdded: event_schemas.SubscribeAddedEventData,
|
||
EventType.SubscribeDeleted: event_schemas.SubscribeDeletedEventData,
|
||
EventType.SubscribeModified: event_schemas.SubscribeModifiedEventData,
|
||
EventType.DownloadAdded: event_schemas.DownloadAddedContractData,
|
||
EventType.TransferComplete: event_schemas.TransferResultContractData,
|
||
EventType.TransferFailed: event_schemas.TransferResultContractData,
|
||
EventType.SubtitleTransferComplete: event_schemas.TransferResultContractData,
|
||
EventType.SubtitleTransferFailed: event_schemas.TransferResultContractData,
|
||
EventType.AudioTransferComplete: event_schemas.TransferResultContractData,
|
||
EventType.AudioTransferFailed: event_schemas.TransferResultContractData,
|
||
EventType.HistoryDeleted: event_schemas.HistoryDeletedEventData,
|
||
EventType.DownloadFileDeleted: event_schemas.DownloadFileDeletedEventData,
|
||
EventType.DownloadDeleted: event_schemas.DownloadDeletedEventData,
|
||
EventType.UserMessage: event_schemas.UserMessageEventData,
|
||
EventType.WebhookMessage: WebhookEventInfo,
|
||
EventType.NoticeMessage: event_schemas.NoticeMessageEventData,
|
||
EventType.SubscribeComplete: event_schemas.SubscribeCompleteEventData,
|
||
EventType.SystemError: event_schemas.SystemErrorEventData,
|
||
EventType.MetadataScrape: event_schemas.MetadataScrapeContractData,
|
||
EventType.ModuleReload: event_schemas.EmptyEventData,
|
||
EventType.MessageAction: event_schemas.MessageActionEventData,
|
||
EventType.WorkflowExecute: event_schemas.WorkflowExecuteEventData,
|
||
ChainEventType.PluginDataReset: event_schemas.PluginDataResetEventData,
|
||
ChainEventType.AuthVerification: event_schemas.AuthCredentials,
|
||
ChainEventType.AuthIntercept: event_schemas.AuthInterceptCredentials,
|
||
ChainEventType.CommandRegister: event_schemas.CommandRegisterEventData,
|
||
ChainEventType.TransferRename: event_schemas.TransferRenameEventData,
|
||
ChainEventType.TransferRenameBuild: event_schemas.TransferRenameBuildEventData,
|
||
ChainEventType.TransferIntercept: event_schemas.TransferInterceptEventData,
|
||
ChainEventType.TransferOverwriteCheck: event_schemas.TransferOverwriteCheckEventData,
|
||
ChainEventType.ResourceSelection: event_schemas.ResourceSelectionContractData,
|
||
ChainEventType.ResourceDownload: event_schemas.ResourceDownloadContractData,
|
||
ChainEventType.DiscoverSource: event_schemas.DiscoverSourceEventData,
|
||
ChainEventType.MediaRecognizeConvert: event_schemas.MediaRecognizeConvertEventData,
|
||
ChainEventType.RecommendSource: event_schemas.RecommendSourceEventData,
|
||
ChainEventType.StorageOperSelection: event_schemas.StorageOperSelectionEventData,
|
||
ChainEventType.AgentLLMProvider: event_schemas.AgentLLMProviderEventData,
|
||
ChainEventType.SubscribeEpisodesRefresh: event_schemas.SubscribeEpisodesRefreshEventData,
|
||
ChainEventType.SubscribeCompletionCheck: event_schemas.SubscribeCompletionCheckContractData,
|
||
ChainEventType.NameRecognize: event_schemas.NameRecognizeEventData,
|
||
ChainEventType.MusicNameRecognize: event_schemas.MusicNameRecognizeEventData,
|
||
ChainEventType.MediaRecognize: event_schemas.MediaRecognizeEventData,
|
||
ChainEventType.MusicMediaRecognize: event_schemas.MusicMediaRecognizeEventData,
|
||
ChainEventType.WorkflowExecution: ActionContext,
|
||
}
|
||
|
||
_SNAPSHOT_EVENTS = {
|
||
EventType.DownloadAdded,
|
||
EventType.TransferComplete,
|
||
EventType.TransferFailed,
|
||
EventType.SubtitleTransferComplete,
|
||
EventType.SubtitleTransferFailed,
|
||
EventType.AudioTransferComplete,
|
||
EventType.AudioTransferFailed,
|
||
EventType.MetadataScrape,
|
||
ChainEventType.ResourceSelection,
|
||
ChainEventType.ResourceDownload,
|
||
ChainEventType.SubscribeCompletionCheck,
|
||
}
|
||
|
||
_INPUT_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = {
|
||
ChainEventType.ResourceSelection: event_schemas.ResourceSelectionInputContractData,
|
||
ChainEventType.ResourceDownload: event_schemas.ResourceDownloadInputContractData,
|
||
ChainEventType.SubscribeCompletionCheck: event_schemas.SubscribeCompletionCheckInputContractData,
|
||
}
|
||
|
||
_OUTPUT_MODELS: dict[EventType | ChainEventType, type[BaseModel]] = {
|
||
ChainEventType.ResourceSelection: event_schemas.ResourceSelectionOutputContractData,
|
||
ChainEventType.ResourceDownload: event_schemas.ResourceDownloadOutputContractData,
|
||
ChainEventType.SubscribeCompletionCheck: event_schemas.SubscribeCompletionCheckOutputContractData,
|
||
}
|
||
|
||
_DURABLE_REQUIRED = {
|
||
EventType.SubscribeAdded,
|
||
EventType.SubscribeModified,
|
||
EventType.SubscribeDeleted,
|
||
EventType.SubscribeComplete,
|
||
EventType.DownloadAdded,
|
||
EventType.TransferComplete,
|
||
EventType.TransferFailed,
|
||
EventType.SubtitleTransferComplete,
|
||
EventType.SubtitleTransferFailed,
|
||
EventType.AudioTransferComplete,
|
||
EventType.AudioTransferFailed,
|
||
}
|
||
_TARGET_PLUGIN = {EventType.PluginAction, EventType.PluginTriggered}
|
||
_HOST_ONLY = {EventType.SystemError, EventType.ConfigChanged, EventType.ModuleReload}
|
||
_SENSITIVE_FIELDS: dict[EventType | ChainEventType, tuple[str, ...]] = {
|
||
ChainEventType.AuthVerification: ("password", "token", "mfa_code"),
|
||
ChainEventType.AuthIntercept: ("token",),
|
||
ChainEventType.AgentLLMProvider: ("api_key",),
|
||
}
|
||
|
||
|
||
def _build_contract(event_type: EventType | ChainEventType) -> EventContract:
|
||
"""
|
||
按 enum 类别和首批 model 映射构造完整契约。
|
||
"""
|
||
payload_model = _PAYLOAD_MODELS.get(event_type)
|
||
is_chain = isinstance(event_type, ChainEventType)
|
||
visibility = EventVisibility.PLUGIN_PUBLIC
|
||
if event_type in _TARGET_PLUGIN:
|
||
visibility = EventVisibility.TARGET_PLUGIN
|
||
elif event_type in _HOST_ONLY:
|
||
visibility = EventVisibility.HOST_ONLY
|
||
return EventContract(
|
||
event_name=f"{event_type.__class__.__name__}.{event_type.name}",
|
||
payload_model=payload_model,
|
||
payload_contract=payload_model.__name__ if payload_model else "legacy_dict",
|
||
mode="chain" if is_chain else "broadcast",
|
||
visibility=visibility,
|
||
delivery=(
|
||
EventDelivery.DURABLE_REQUIRED
|
||
if event_type in _DURABLE_REQUIRED
|
||
else EventDelivery.EPHEMERAL
|
||
),
|
||
error_behavior=(
|
||
EventErrorBehavior.STOP_CHAIN if is_chain else EventErrorBehavior.NOTIFY
|
||
),
|
||
ordering="priority_serial" if is_chain else "priority_queue",
|
||
input_model=_INPUT_MODELS.get(event_type, payload_model),
|
||
output_model=_OUTPUT_MODELS.get(event_type),
|
||
payload_mode=(
|
||
EventPayloadMode.SNAPSHOT
|
||
if event_type in _SNAPSHOT_EVENTS
|
||
else EventPayloadMode.RUNTIME
|
||
),
|
||
validation_mode=EventValidationMode.DIAGNOSTIC,
|
||
sensitive_fields=_SENSITIVE_FIELDS.get(event_type, ()),
|
||
legacy_reason=(
|
||
None
|
||
if payload_model
|
||
else "现有插件 payload 尚未收敛为稳定 model，保留原始 dict ABI"
|
||
),
|
||
)
|
||
|
||
|
||
_ALL_EVENT_TYPES: tuple[EventType | ChainEventType, ...] = (
|
||
*tuple(EventType),
|
||
*tuple(ChainEventType),
|
||
)
|
||
|
||
EVENT_CONTRACTS: dict[EventType | ChainEventType, EventContract] = {
|
||
event_type: _build_contract(event_type)
|
||
for event_type in _ALL_EVENT_TYPES
|
||
}
|
||
|
||
|
||
def normalize_event_type(
|
||
event_type: EventType | ChainEventType | str,
|
||
) -> EventType | ChainEventType | str:
|
||
"""
|
||
把旧 SDK 传入的已知字符串恢复为 enum，未知扩展值保持原样。
|
||
"""
|
||
if not isinstance(event_type, str):
|
||
return event_type
|
||
for enum_type in (EventType, ChainEventType):
|
||
try:
|
||
return enum_type(event_type)
|
||
except ValueError:
|
||
continue
|
||
return event_type
|
||
|
||
|
||
def get_event_contract(
|
||
event_type: EventType | ChainEventType | str,
|
||
) -> EventContract:
|
||
"""
|
||
返回已登记事件的完整契约，兼容传入 enum value 字符串。
|
||
"""
|
||
normalized = normalize_event_type(event_type)
|
||
if isinstance(normalized, str):
|
||
raise KeyError(normalized)
|
||
return EVENT_CONTRACTS[normalized]
|
||
|
||
|
||
def describe_event_validation_error(model: type[BaseModel], error: Exception) -> str:
|
||
"""
|
||
只输出模型名和错误数量，避免校验输入、异常文本或动态字段名泄露凭据。
|
||
"""
|
||
if isinstance(error, ValidationError):
|
||
return f"{model.__name__}: {error.error_count()} validation error(s)"
|
||
return f"{model.__name__}: {type(error).__name__}"
|
||
|
||
|
||
def validate_event_payload(
|
||
event_type: EventType | ChainEventType | str,
|
||
payload: Any,
|
||
) -> tuple[str, ...]:
|
||
"""
|
||
在发送边界诊断首批 typed payload，保持原对象和插件 dict 形状不变。
|
||
"""
|
||
try:
|
||
contract = get_event_contract(event_type)
|
||
models = tuple(
|
||
model
|
||
for model in (contract.input_model, contract.output_model)
|
||
if model is not None
|
||
)
|
||
except KeyError:
|
||
# 动态插件在旧 ABI 下可能使用宿主枚举之外的字符串事件。
|
||
return ()
|
||
if not models or payload is None:
|
||
return ()
|
||
problems: list[str] = []
|
||
for model in models:
|
||
if isinstance(payload, model):
|
||
continue
|
||
try:
|
||
model.model_validate(payload)
|
||
except (ValidationError, TypeError, ValueError) as error:
|
||
problems.append(describe_event_validation_error(model, error))
|
||
return tuple(problems)
|