mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
242 lines
7.6 KiB
Python
242 lines
7.6 KiB
Python
"""
|
||
已提交插件载荷的版本化 package 声明快照。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
_SCHEMA_VERSION = 1
|
||
_RUNTIME_FIELD_PATTERN = re.compile(r"^v[1-9][0-9]*t?$")
|
||
_MANIFEST_TEXT_FIELDS = (
|
||
"name",
|
||
"description",
|
||
"icon",
|
||
"author",
|
||
"system_version",
|
||
)
|
||
|
||
|
||
def _optional_text(
|
||
value: object,
|
||
*,
|
||
max_length: int | None = None,
|
||
) -> str | None:
|
||
"""
|
||
把外部可选文本规范为非空字符串。
|
||
"""
|
||
if not isinstance(value, str):
|
||
return None
|
||
normalized = value.strip()
|
||
if max_length is not None and len(normalized) > max_length:
|
||
raise ValueError(f"插件声明文本长度不能超过 {max_length}")
|
||
return normalized or None
|
||
|
||
|
||
def _labels(value: object) -> list[str]:
|
||
"""
|
||
把历史逗号字符串和字符串数组统一为有序标签列表。
|
||
"""
|
||
if isinstance(value, str):
|
||
items = value.split(",")
|
||
elif isinstance(value, list):
|
||
items = value
|
||
else:
|
||
return []
|
||
return [
|
||
normalized
|
||
for item in items
|
||
if isinstance(item, str) and (normalized := item.strip())
|
||
]
|
||
|
||
|
||
def _normalize_manifest(value: object) -> dict[str, object]:
|
||
"""
|
||
仅保留当前载荷展示和声明消费者需要的 package 字段。
|
||
"""
|
||
if not isinstance(value, Mapping):
|
||
return {}
|
||
manifest: dict[str, object] = {}
|
||
for field_name in _MANIFEST_TEXT_FIELDS:
|
||
if normalized := _optional_text(
|
||
value.get(field_name),
|
||
max_length=128 if field_name == "system_version" else None,
|
||
):
|
||
manifest[field_name] = normalized
|
||
labels = _labels(value.get("labels"))
|
||
if labels:
|
||
manifest["labels"] = labels
|
||
level = value.get("level")
|
||
if isinstance(level, int) and not isinstance(level, bool):
|
||
manifest["level"] = level
|
||
release = value.get("release")
|
||
if isinstance(release, bool):
|
||
manifest["release"] = release
|
||
return manifest
|
||
|
||
|
||
def _normalize_runtime(value: object) -> dict[str, bool]:
|
||
"""
|
||
保留可向后扩展的宿主代际和运行时变体布尔声明。
|
||
"""
|
||
if not isinstance(value, Mapping):
|
||
return {}
|
||
return {
|
||
str(field_name): field_value
|
||
for field_name, field_value in value.items()
|
||
if (
|
||
isinstance(field_name, str)
|
||
and _RUNTIME_FIELD_PATTERN.fullmatch(field_name)
|
||
and isinstance(field_value, bool)
|
||
)
|
||
}
|
||
|
||
|
||
def _encode(value: Mapping[str, object]) -> str:
|
||
"""
|
||
生成不可从外部原地修改的稳定内部表示。
|
||
"""
|
||
return json.dumps(
|
||
value,
|
||
ensure_ascii=False,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class PluginDeclaredMetadata:
|
||
"""
|
||
封装一份受限、版本化且不参与来源授信的 package 声明。
|
||
|
||
:param _encoded: 经过白名单规范化的声明元数据 JSON，构造时校验并规范编码
|
||
"""
|
||
|
||
_encoded: str = field(repr=False)
|
||
|
||
def __post_init__(self) -> None:
|
||
"""
|
||
拒绝绕过工厂构造的非规范内部状态。
|
||
"""
|
||
normalized = self._normalize_storage(json.loads(self._encoded))
|
||
object.__setattr__(self, "_encoded", _encode(normalized))
|
||
|
||
@classmethod
|
||
def from_package(
|
||
cls,
|
||
package: Mapping[str, Any],
|
||
*,
|
||
declaration_version: str | None,
|
||
manifest_matches_payload: bool,
|
||
) -> "PluginDeclaredMetadata":
|
||
"""
|
||
从安装候选生成与载荷一起提交的受限声明快照。
|
||
"""
|
||
runtime = _normalize_runtime(package)
|
||
value = {
|
||
"schema_version": _SCHEMA_VERSION,
|
||
"declaration_version": _optional_text(declaration_version),
|
||
"manifest_matches_payload": manifest_matches_payload,
|
||
"manifest": _normalize_manifest(package),
|
||
"runtime": runtime,
|
||
}
|
||
return cls(_encode(value))
|
||
|
||
@classmethod
|
||
def from_storage(cls, value: object) -> "PluginDeclaredMetadata":
|
||
"""
|
||
校验数据库 JSON，并忽略当前合同未消费的额外字段。
|
||
"""
|
||
return cls(_encode(cls._normalize_storage(value)))
|
||
|
||
@staticmethod
|
||
def _normalize_storage(value: object) -> dict[str, object]:
|
||
"""
|
||
验证快照结构，同时宽容丢弃非法可选展示字段。
|
||
"""
|
||
if not isinstance(value, Mapping):
|
||
raise ValueError("插件声明快照必须是 JSON 对象")
|
||
if value.get("schema_version") != _SCHEMA_VERSION:
|
||
raise ValueError("插件声明快照 schema_version 不受支持")
|
||
manifest_matches_payload = value.get("manifest_matches_payload")
|
||
if not isinstance(manifest_matches_payload, bool):
|
||
raise ValueError("插件声明快照必须说明 manifest 是否对应当前载荷")
|
||
return {
|
||
"schema_version": _SCHEMA_VERSION,
|
||
"declaration_version": _optional_text(
|
||
value.get("declaration_version")
|
||
),
|
||
"manifest_matches_payload": manifest_matches_payload,
|
||
"manifest": _normalize_manifest(value.get("manifest")),
|
||
"runtime": _normalize_runtime(value.get("runtime")),
|
||
}
|
||
|
||
@property
|
||
def declaration_version(self) -> str | None:
|
||
"""
|
||
返回生成该快照的 package 条目版本。
|
||
"""
|
||
return _optional_text(self.to_json().get("declaration_version"))
|
||
|
||
@property
|
||
def manifest_matches_payload(self) -> bool:
|
||
"""
|
||
说明该 package 声明是否与当前已提交载荷对应。
|
||
"""
|
||
return bool(self.to_json()["manifest_matches_payload"])
|
||
|
||
def runtime_support(self, runtime_name: str) -> bool | None:
|
||
"""
|
||
读取一个规范运行时声明；缺省保持未声明语义。
|
||
"""
|
||
if not _RUNTIME_FIELD_PATTERN.fullmatch(runtime_name):
|
||
raise ValueError("运行时声明必须使用 v<数字> 或 v<数字>t")
|
||
runtime = self.to_json().get("runtime")
|
||
if not isinstance(runtime, dict):
|
||
return None
|
||
value = runtime.get(runtime_name)
|
||
return value if isinstance(value, bool) else None
|
||
|
||
def display_fallback(self, *, installed_version: str) -> dict[str, str]:
|
||
"""
|
||
生成插件加载失败时可安全补齐的展示字段。
|
||
"""
|
||
manifest = self.to_json().get("manifest")
|
||
fallback = {
|
||
"plugin_version": installed_version,
|
||
}
|
||
if not isinstance(manifest, dict):
|
||
return fallback
|
||
display_fields = {
|
||
"name": "plugin_name",
|
||
"description": "plugin_desc",
|
||
"icon": "plugin_icon",
|
||
"author": "plugin_author",
|
||
}
|
||
for source_name, target_name in display_fields.items():
|
||
value = manifest.get(source_name)
|
||
if isinstance(value, str):
|
||
fallback[target_name] = value
|
||
labels = manifest.get("labels")
|
||
if isinstance(labels, list):
|
||
label = ",".join(item for item in labels if isinstance(item, str))
|
||
if label:
|
||
fallback["plugin_label"] = label
|
||
return fallback
|
||
|
||
def to_json(self) -> dict[str, object]:
|
||
"""
|
||
返回可持久化且与内部状态相互隔离的 JSON 副本。
|
||
"""
|
||
value: object = json.loads(self._encoded)
|
||
if not isinstance(value, dict):
|
||
raise ValueError("插件声明快照内部状态必须是 JSON 对象")
|
||
return {
|
||
str(field_name): field_value
|
||
for field_name, field_value in value.items()
|
||
}
|