mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
759 lines
28 KiB
Python
759 lines
28 KiB
Python
"""
|
||
插件市场目录应用服务。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import concurrent.futures
|
||
from collections.abc import Awaitable, Callable, Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from typing import Any, Literal, Optional
|
||
|
||
from app.application.plugin.identity import (
|
||
OFFICIAL_PLUGIN_SOURCE_KEY,
|
||
PluginBindingBasis,
|
||
PluginIdentity,
|
||
TrustedPluginSourceType,
|
||
normalize_physical_plugin_id,
|
||
)
|
||
from app.application.plugin.inventory import normalize_github_plugin_source
|
||
from app.domain.plugin import check_plugin_runtime_compatibility
|
||
from app.schemas.plugin import (
|
||
Plugin,
|
||
PluginDocumentation,
|
||
PluginSourceBindingStatus,
|
||
PluginUpdateCandidate,
|
||
)
|
||
|
||
MarketLoader = Callable[[str, Optional[str], bool], Optional[dict[str, dict[str, Any]]]]
|
||
AsyncMarketLoader = Callable[
|
||
[str, Optional[str], bool],
|
||
Awaitable[Optional[dict[str, dict[str, Any]]]],
|
||
]
|
||
PluginMapper = Callable[[str, dict[str, Any], str, list[str], int, Optional[str]], Any]
|
||
ProgressCallback = Callable[..., Any]
|
||
DEFAULT_MARKET_FETCH_CONCURRENCY = 5
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class _CatalogLoadRequest:
|
||
"""
|
||
冻结同步与异步插件市场加载共用的仓库和代际参数。
|
||
|
||
:param market: 本次读取的配置市场地址
|
||
:param package_version: 请求读取的插件包代际；None 使用入口默认代际
|
||
:param force: 是否忽略本次读取或安装的缓存与版本短路
|
||
"""
|
||
|
||
market: str
|
||
package_version: Optional[str]
|
||
force: bool
|
||
|
||
|
||
def apply_declared_metadata_fallback(
|
||
plugins: Sequence[Plugin],
|
||
identities: Mapping[str, PluginIdentity],
|
||
*,
|
||
free_threaded: bool = False,
|
||
) -> list[Plugin]:
|
||
"""
|
||
用已提交快照补齐加载失败插件，不覆盖真实运行态字段。
|
||
|
||
:param free_threaded: 当前是否运行在 free-threaded 解释器；已安装卡片的运行时
|
||
不兼容提示只能来自安装时提交的声明快照，市场候选那条标注路径要联网才有。
|
||
"""
|
||
result: list[Plugin] = []
|
||
for plugin in plugins:
|
||
identity = identities.get((plugin.id or "").lower())
|
||
updates: dict[str, object] = {}
|
||
if identity is not None and identity.declared_metadata is not None:
|
||
compatible, message = check_plugin_runtime_compatibility(
|
||
identity.declared_metadata.to_json().get("runtime"),
|
||
free_threaded=free_threaded,
|
||
)
|
||
if not compatible:
|
||
updates["runtime_compatible"] = False
|
||
updates["runtime_message"] = message
|
||
if plugin.installed and not plugin.is_instance:
|
||
if identity is None:
|
||
updates["source_binding_status"] = PluginSourceBindingStatus.BINDING_REQUIRED
|
||
elif identity.trusted_source_type is TrustedPluginSourceType.UNKNOWN:
|
||
updates["source_binding_status"] = (
|
||
PluginSourceBindingStatus.LOCAL_ONLY
|
||
if identity.binding_basis is PluginBindingBasis.LOCAL_ONLY
|
||
else PluginSourceBindingStatus.BINDING_REQUIRED
|
||
)
|
||
else:
|
||
updates["source_binding_status"] = PluginSourceBindingStatus.BOUND
|
||
if (
|
||
identity is None
|
||
or identity.declared_metadata is None
|
||
or identity.declared_version is None
|
||
):
|
||
result.append(plugin.model_copy(update=updates) if updates else plugin)
|
||
continue
|
||
fallback = identity.declared_metadata.display_fallback(
|
||
installed_version=identity.declared_version
|
||
)
|
||
if not plugin.plugin_version:
|
||
updates["plugin_version"] = fallback["plugin_version"]
|
||
if (
|
||
(not plugin.plugin_name or plugin.plugin_name == plugin.id)
|
||
and "plugin_name" in fallback
|
||
):
|
||
updates["plugin_name"] = fallback["plugin_name"]
|
||
if not plugin.plugin_desc and "plugin_desc" in fallback:
|
||
updates["plugin_desc"] = fallback["plugin_desc"]
|
||
if not plugin.plugin_icon and "plugin_icon" in fallback:
|
||
updates["plugin_icon"] = fallback["plugin_icon"]
|
||
if not plugin.plugin_author and "plugin_author" in fallback:
|
||
updates["plugin_author"] = fallback["plugin_author"]
|
||
if not plugin.plugin_label and "plugin_label" in fallback:
|
||
updates["plugin_label"] = fallback["plugin_label"]
|
||
result.append(plugin.model_copy(update=updates) if updates else plugin)
|
||
return result
|
||
|
||
|
||
class PluginCatalogService:
|
||
"""
|
||
负责插件市场索引映射、并发收集、代际合并和来源去重。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
market_loader: MarketLoader,
|
||
async_market_loader: AsyncMarketLoader,
|
||
installed_plugins_provider: Callable[[], list[str]],
|
||
plugin_mapper: PluginMapper,
|
||
is_local_repo: Callable[[Optional[str]], bool],
|
||
version_compare: Callable[[str, str, str], bool],
|
||
warning: Callable[[str], Any],
|
||
error: Callable[[str], Any],
|
||
max_concurrency: int = DEFAULT_MARKET_FETCH_CONCURRENCY,
|
||
) -> None:
|
||
"""
|
||
保存市场读取、插件映射和版本比较端口。
|
||
"""
|
||
if max_concurrency < 1:
|
||
raise ValueError("插件市场读取并发必须大于 0")
|
||
self._market_loader = market_loader
|
||
self._async_market_loader = async_market_loader
|
||
self._installed_plugins_provider = installed_plugins_provider
|
||
self._plugin_mapper = plugin_mapper
|
||
self._is_local_repo = is_local_repo
|
||
self._version_compare = version_compare
|
||
self._warning = warning
|
||
self._error = error
|
||
self._max_concurrency = max_concurrency
|
||
|
||
@staticmethod
|
||
def _load_request(
|
||
market: str,
|
||
package_version: Optional[str],
|
||
force: bool,
|
||
) -> Optional[_CatalogLoadRequest]:
|
||
"""
|
||
统一拒绝空市场，并冻结一次目录加载的完整输入。
|
||
"""
|
||
if not market:
|
||
return None
|
||
return _CatalogLoadRequest(
|
||
market=market,
|
||
package_version=package_version,
|
||
force=force,
|
||
)
|
||
|
||
def _loaded_plugins(
|
||
self,
|
||
request: _CatalogLoadRequest,
|
||
online_plugins: Optional[dict[str, dict[str, Any]]],
|
||
) -> list[Any]:
|
||
"""
|
||
统一加载失败告警与在线目录到插件 DTO 的投影。
|
||
"""
|
||
if online_plugins is None:
|
||
self._warning(
|
||
f"获取{request.package_version if request.package_version else ''}插件库失败："
|
||
f"{request.market}，请检查 GitHub 网络连接"
|
||
)
|
||
return []
|
||
return self._map_plugins(
|
||
online_plugins, request.market, request.package_version
|
||
)
|
||
|
||
def load(
|
||
self,
|
||
market: str,
|
||
package_version: Optional[str] = None,
|
||
force: bool = False,
|
||
) -> list[Any]:
|
||
"""
|
||
同步读取并映射指定市场和插件代际。
|
||
"""
|
||
request = PluginCatalogService._load_request(
|
||
market, package_version, force
|
||
)
|
||
if request is None:
|
||
return []
|
||
return self._loaded_plugins(
|
||
request,
|
||
self._market_loader(
|
||
request.market, request.package_version, request.force
|
||
),
|
||
)
|
||
|
||
async def async_load(
|
||
self,
|
||
market: str,
|
||
package_version: Optional[str] = None,
|
||
force: bool = False,
|
||
) -> list[Any]:
|
||
"""
|
||
异步读取并映射指定市场和插件代际。
|
||
"""
|
||
request = PluginCatalogService._load_request(
|
||
market, package_version, force
|
||
)
|
||
if request is None:
|
||
return []
|
||
online_plugins = await self._async_market_loader(
|
||
request.market,
|
||
request.package_version,
|
||
request.force,
|
||
)
|
||
return self._loaded_plugins(request, online_plugins)
|
||
|
||
def collect(
|
||
self,
|
||
*,
|
||
markets: list[str],
|
||
compatible_flags: list[str],
|
||
force: bool,
|
||
loader: Callable[[str, Optional[str], bool], list[Any]],
|
||
) -> list[Any]:
|
||
"""
|
||
并发读取多个市场和代际，并按稳定优先级合并。
|
||
"""
|
||
with concurrent.futures.ThreadPoolExecutor(
|
||
max_workers=self._max_concurrency
|
||
) as executor:
|
||
futures_meta: dict[
|
||
concurrent.futures.Future[Any],
|
||
tuple[int, bool, int],
|
||
] = {}
|
||
for market_index, market in enumerate(markets):
|
||
base_future = executor.submit(loader, market, None, force)
|
||
futures_meta[base_future] = (market_index, False, 0)
|
||
for flag_priority, flag in enumerate(compatible_flags):
|
||
higher_future = executor.submit(loader, market, flag, force)
|
||
futures_meta[higher_future] = (
|
||
market_index,
|
||
True,
|
||
flag_priority,
|
||
)
|
||
|
||
collected = []
|
||
for future in concurrent.futures.as_completed(futures_meta):
|
||
try:
|
||
plugins = future.result()
|
||
# 单个市场失败不能中断全量刷新
|
||
except Exception as error: # noqa: BLE001
|
||
self._error(f"获取插件市场数据失败：{str(error)}")
|
||
plugins = []
|
||
market_index, is_higher, flag_priority = futures_meta[future]
|
||
collected.append((
|
||
market_index,
|
||
is_higher,
|
||
flag_priority,
|
||
plugins or [],
|
||
))
|
||
|
||
collected.sort(key=lambda item: (item[0], 0 if item[1] else 1, item[2]))
|
||
higher_plugins: list[Any] = []
|
||
base_plugins: list[Any] = []
|
||
for _market_index, is_higher, _flag_priority, plugins in collected:
|
||
(higher_plugins if is_higher else base_plugins).extend(plugins)
|
||
return self.merge(higher_plugins, base_plugins, markets)
|
||
|
||
async def async_collect(
|
||
self,
|
||
*,
|
||
markets: list[str],
|
||
compatible_flags: list[str],
|
||
force: bool,
|
||
loader: Callable[
|
||
[str, Optional[str], bool],
|
||
Awaitable[list[Any]],
|
||
],
|
||
preserve_sources: bool = False,
|
||
progress_callback: Optional[ProgressCallback] = None,
|
||
) -> list[Any]:
|
||
"""
|
||
异步读取多个市场和代际，并按调用方需要合并或保留仓库候选。
|
||
"""
|
||
semaphore = asyncio.Semaphore(self._max_concurrency)
|
||
|
||
async def fetch(
|
||
market: str,
|
||
package_version: Optional[str],
|
||
result_version: str,
|
||
task_index: int,
|
||
) -> tuple[int, str, list[Any]]:
|
||
"""
|
||
读取一个市场代际并保留创建时的稳定任务序号。
|
||
"""
|
||
async with semaphore:
|
||
plugins = await loader(market, package_version, force)
|
||
return task_index, result_version, plugins or []
|
||
|
||
tasks: list[asyncio.Task[tuple[int, str, list[Any]]]] = []
|
||
for market in markets:
|
||
tasks.append(asyncio.create_task(
|
||
fetch(market, None, "base_version", len(tasks)),
|
||
name="plugin.catalog.fetch",
|
||
))
|
||
for flag in compatible_flags:
|
||
tasks.append(asyncio.create_task(
|
||
fetch(market, flag, "higher_version", len(tasks)),
|
||
name="plugin.catalog.fetch",
|
||
))
|
||
|
||
try:
|
||
higher_plugins: list[Any] = []
|
||
base_plugins: list[Any] = []
|
||
if tasks:
|
||
total_tasks = len(tasks)
|
||
finished_tasks = 0
|
||
task_results: dict[int, tuple[str, list[Any]]] = {}
|
||
if progress_callback:
|
||
progress_callback(
|
||
value=0,
|
||
text=f"开始刷新插件市场，共 {total_tasks} 个请求 ...",
|
||
data={"total": total_tasks, "finished": 0},
|
||
)
|
||
for completed_task in asyncio.as_completed(tasks):
|
||
try:
|
||
task_index, version, plugins = await completed_task
|
||
task_results[task_index] = (version, plugins)
|
||
except Exception as err:
|
||
self._error(f"获取插件市场数据失败：{str(err)}")
|
||
finished_tasks += 1
|
||
if progress_callback:
|
||
progress_callback(
|
||
value=finished_tasks / total_tasks * 100,
|
||
text=(
|
||
f"插件市场请求（{finished_tasks}/{total_tasks}）"
|
||
"处理完成"
|
||
),
|
||
data={"total": total_tasks, "finished": finished_tasks},
|
||
)
|
||
for task_index in sorted(task_results):
|
||
version, plugins = task_results[task_index]
|
||
target = higher_plugins if version == "higher_version" else base_plugins
|
||
target.extend(plugins)
|
||
|
||
result = (
|
||
self.merge_by_source(higher_plugins, base_plugins, markets)
|
||
if preserve_sources
|
||
else self.merge(higher_plugins, base_plugins, markets)
|
||
)
|
||
if progress_callback:
|
||
progress_callback(value=100, text="插件市场缓存刷新完成")
|
||
return result
|
||
finally:
|
||
for task in tasks:
|
||
if not task.done():
|
||
task.cancel()
|
||
if tasks:
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
def merge(
|
||
self,
|
||
higher_plugins: list[Any],
|
||
base_plugins: list[Any],
|
||
markets: list[str],
|
||
) -> list[Any]:
|
||
"""
|
||
按代际、来源顺序和版本合并插件目录。
|
||
"""
|
||
all_plugins = list(higher_plugins)
|
||
higher_keys = {
|
||
(normalize_physical_plugin_id(plugin.id), plugin.plugin_version)
|
||
for plugin in higher_plugins
|
||
}
|
||
all_plugins.extend(
|
||
plugin
|
||
for plugin in base_plugins
|
||
if (
|
||
normalize_physical_plugin_id(plugin.id),
|
||
plugin.plugin_version,
|
||
) not in higher_keys
|
||
)
|
||
|
||
def repo_order(plugin: Any) -> int:
|
||
"""
|
||
本地来源排在远程市场之后，远程来源保持配置顺序。
|
||
"""
|
||
if self._is_local_repo(plugin.repo_url):
|
||
return len(markets) + 1
|
||
if plugin.repo_url in markets:
|
||
return markets.index(plugin.repo_url)
|
||
return len(markets)
|
||
|
||
deduplicated: dict[tuple[Optional[str], Optional[str]], Any] = {}
|
||
for plugin in sorted(all_plugins, key=repo_order):
|
||
key = (
|
||
normalize_physical_plugin_id(plugin.id),
|
||
plugin.plugin_version,
|
||
)
|
||
exists = deduplicated.get(key)
|
||
if not exists or (
|
||
self._is_local_repo(exists.repo_url)
|
||
and not self._is_local_repo(plugin.repo_url)
|
||
):
|
||
deduplicated[key] = plugin
|
||
|
||
result_by_id: dict[str, Any] = {}
|
||
for plugin in sorted(deduplicated.values(), key=repo_order):
|
||
normalized_id = normalize_physical_plugin_id(plugin.id)
|
||
exists = result_by_id.get(normalized_id)
|
||
if not exists \
|
||
or self._version_compare(
|
||
plugin.plugin_version,
|
||
">",
|
||
exists.plugin_version,
|
||
) \
|
||
or (
|
||
plugin.plugin_version == exists.plugin_version
|
||
and not self._is_local_repo(exists.repo_url)
|
||
and self._is_local_repo(plugin.repo_url)
|
||
):
|
||
result_by_id[normalized_id] = plugin
|
||
return list(result_by_id.values())
|
||
|
||
def merge_by_source(
|
||
self,
|
||
higher_plugins: list[Any],
|
||
base_plugins: list[Any],
|
||
markets: list[str],
|
||
) -> list[Any]:
|
||
"""
|
||
每个仓库和代际保留同一插件的最高版本，供来源准入继续决策。
|
||
"""
|
||
higher_keys = {
|
||
(
|
||
plugin.repo_url,
|
||
normalize_physical_plugin_id(plugin.id),
|
||
plugin.plugin_version,
|
||
)
|
||
for plugin in higher_plugins
|
||
}
|
||
all_plugins = list(higher_plugins)
|
||
all_plugins.extend(
|
||
plugin
|
||
for plugin in base_plugins
|
||
if (
|
||
plugin.repo_url,
|
||
normalize_physical_plugin_id(plugin.id),
|
||
plugin.plugin_version,
|
||
) not in higher_keys
|
||
)
|
||
|
||
def repo_order(plugin: Any) -> int:
|
||
"""
|
||
保留配置的仓库顺序，未配置来源排在末尾。
|
||
"""
|
||
if plugin.repo_url in markets:
|
||
return markets.index(plugin.repo_url)
|
||
return len(markets)
|
||
|
||
result_by_source: dict[tuple[Optional[str], str, Optional[str]], Any] = {}
|
||
for plugin in sorted(all_plugins, key=repo_order):
|
||
key = (
|
||
plugin.repo_url,
|
||
normalize_physical_plugin_id(plugin.id),
|
||
plugin.package_version,
|
||
)
|
||
exists = result_by_source.get(key)
|
||
if not exists or self._version_compare(
|
||
plugin.plugin_version,
|
||
">",
|
||
exists.plugin_version,
|
||
):
|
||
result_by_source[key] = plugin
|
||
return list(result_by_source.values())
|
||
|
||
def _map_plugins(
|
||
self,
|
||
online_plugins: dict[str, dict[str, Any]],
|
||
market: str,
|
||
package_version: Optional[str],
|
||
) -> list[Any]:
|
||
"""
|
||
把一个市场索引映射为宿主插件 DTO。
|
||
"""
|
||
installed_plugins = self._installed_plugins_provider()
|
||
result: list[Any] = []
|
||
add_time = len(online_plugins)
|
||
for plugin_id, plugin_info in online_plugins.items():
|
||
plugin = self._plugin_mapper(
|
||
plugin_id,
|
||
plugin_info,
|
||
market,
|
||
installed_plugins,
|
||
add_time,
|
||
package_version,
|
||
)
|
||
if plugin:
|
||
result.append(plugin)
|
||
add_time -= 1
|
||
return result
|
||
|
||
|
||
class PluginCatalogQuery:
|
||
"""
|
||
组合运行态、来源身份和市场候选，生成插件页目录快照。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
installed_plugins: Callable[[], list[Plugin]],
|
||
local_plugins: Callable[[], list[Plugin]],
|
||
local_repo_plugins: Callable[[], list[Plugin]],
|
||
online_candidates: Callable[[bool], Awaitable[list[Plugin]]],
|
||
process_plugins: Callable[[list[Plugin], list[Plugin]], list[Plugin]],
|
||
identities: Callable[[list[str]], Awaitable[list[PluginIdentity]]],
|
||
free_threaded: Callable[[], bool],
|
||
readme_url: Optional[Callable[[str, str, Optional[str]], Awaitable[Optional[str]]]] = None,
|
||
) -> None:
|
||
"""
|
||
保存运行态目录、来源身份读取和解释器变体判定窄端口。
|
||
"""
|
||
self._installed_plugins = installed_plugins
|
||
self._local_plugins = local_plugins
|
||
self._local_repo_plugins = local_repo_plugins
|
||
self._online_candidates = online_candidates
|
||
self._process_plugins = process_plugins
|
||
self._identities = identities
|
||
self._free_threaded = free_threaded
|
||
self._readme_url = readme_url
|
||
|
||
async def documentation(self, plugin_id: str, repo_url: str) -> PluginDocumentation:
|
||
"""
|
||
为所选市场来源返回显式说明链接，缺省时按候选的真实代际探测 README。
|
||
|
||
:param plugin_id: 源插件 ID，不要求已安装
|
||
:param repo_url: 用户当前查看或选择安装的仓库；本地来源只使用显式链接
|
||
:return: 文档入口；不在目录中的来源、缺失文档均返回空入口
|
||
"""
|
||
candidates = (
|
||
self._local_repo_plugins()
|
||
if repo_url.startswith("local://")
|
||
else await self._online_candidates(False)
|
||
)
|
||
matches = [
|
||
plugin for plugin in candidates
|
||
if (plugin.id or "").lower() == plugin_id.lower()
|
||
and (plugin.repo_url or "").rstrip("/").lower() == repo_url.rstrip("/").lower()
|
||
]
|
||
# 同一仓库可能同时提供多代索引，沿用市场卡片的版本与代际选择规则。
|
||
candidate = next(iter(self._process_plugins(matches, [])), None)
|
||
if candidate is None:
|
||
return PluginDocumentation()
|
||
if candidate.plugin_readme:
|
||
return PluginDocumentation(url=candidate.plugin_readme)
|
||
if self._readme_url is None or repo_url.startswith("local://"):
|
||
return PluginDocumentation()
|
||
return PluginDocumentation(
|
||
url=await self._readme_url(candidate.id or plugin_id, candidate.repo_url or repo_url, candidate.package_version),
|
||
)
|
||
|
||
async def query(
|
||
self, *, state: str = "all", force: bool = False
|
||
) -> list[Plugin]:
|
||
"""
|
||
按 installed、market 或 all 返回插件页所需稳定目录。
|
||
"""
|
||
installed_plugins = self._installed_plugins()
|
||
if state == "installed":
|
||
return await self._with_declared_metadata(installed_plugins)
|
||
|
||
local_plugins = self._local_plugins()
|
||
not_installed = [plugin for plugin in local_plugins if not plugin.installed]
|
||
local_repo_plugins = self._local_repo_plugins()
|
||
online_plugins = await self._online_candidates(force)
|
||
installed_ids = [plugin.id for plugin in installed_plugins if plugin.id]
|
||
candidates = (
|
||
self._process_plugins(online_plugins + local_repo_plugins, [])
|
||
if online_plugins or local_repo_plugins
|
||
else []
|
||
)
|
||
candidates = await self.project_update_candidates(
|
||
online_plugins,
|
||
candidates,
|
||
installed_ids,
|
||
)
|
||
if not candidates:
|
||
if state == "market":
|
||
return not_installed
|
||
return await self._with_declared_metadata(installed_plugins) + not_installed
|
||
|
||
installed_keys = set(installed_ids)
|
||
market_plugins = [
|
||
plugin
|
||
for plugin in candidates
|
||
if plugin.id not in installed_keys or plugin.has_update
|
||
]
|
||
market_ids = {plugin.id for plugin in market_plugins}
|
||
market_plugins.extend(
|
||
plugin for plugin in not_installed if plugin.id not in market_ids
|
||
)
|
||
if state == "market":
|
||
return market_plugins
|
||
return await self._with_declared_metadata(installed_plugins) + market_plugins
|
||
|
||
async def project_update_candidates(
|
||
self,
|
||
online_plugins: list[Plugin],
|
||
plugins: list[Plugin],
|
||
installed_ids: list[str],
|
||
) -> list[Plugin]:
|
||
"""
|
||
已绑定插件仅提示绑定仓库更新，并投影候选来源信息。
|
||
"""
|
||
identities = await self._identities(installed_ids)
|
||
identity_by_id = {item.normalized_plugin_id: item for item in identities}
|
||
installed_keys = {plugin_id.lower() for plugin_id in installed_ids}
|
||
result: list[Plugin] = []
|
||
for plugin in plugins:
|
||
if (
|
||
not plugin.id
|
||
or plugin.id.lower() not in installed_keys
|
||
or not plugin.has_update
|
||
):
|
||
result.append(plugin)
|
||
continue
|
||
identity = identity_by_id.get(plugin.id.lower())
|
||
plugin = self._prefer_bound_update(plugin, online_plugins, identity)
|
||
result.append(_project_update_candidate(plugin, identity))
|
||
return result
|
||
|
||
async def _with_declared_metadata(
|
||
self, plugins: list[Plugin]
|
||
) -> list[Plugin]:
|
||
"""
|
||
为未加载插件补齐已提交的展示声明。
|
||
"""
|
||
identities = await self._identities(
|
||
[plugin.id for plugin in plugins if plugin.id]
|
||
)
|
||
return apply_declared_metadata_fallback(
|
||
plugins,
|
||
{identity.normalized_plugin_id: identity for identity in identities},
|
||
free_threaded=self._free_threaded(),
|
||
)
|
||
|
||
def _prefer_bound_update(
|
||
self,
|
||
plugin: Plugin,
|
||
online_plugins: list[Plugin],
|
||
identity: PluginIdentity | None,
|
||
) -> Plugin:
|
||
"""
|
||
按绑定来源收敛更新标记，不修改共享的市场候选快照。
|
||
"""
|
||
if identity is None or not identity.trusted_source_key or not plugin.id:
|
||
return plugin
|
||
trusted_source_key = identity.trusted_source_key
|
||
plugin_id = plugin.id
|
||
bound_updates = [
|
||
candidate
|
||
for candidate in online_plugins
|
||
if _matches_trusted_source(candidate, plugin_id, trusted_source_key)
|
||
]
|
||
preferred = self._process_plugins(bound_updates, []) if bound_updates else []
|
||
return preferred[0] if preferred else plugin.model_copy(update={"has_update": False})
|
||
|
||
|
||
_catalog_query: PluginCatalogQuery | None = None
|
||
|
||
|
||
def configure_plugin_catalog_query(query: PluginCatalogQuery) -> None:
|
||
"""
|
||
由启动组合根发布当前 lifespan 的插件目录查询。
|
||
"""
|
||
global _catalog_query
|
||
_catalog_query = query
|
||
|
||
|
||
def get_plugin_catalog_query() -> PluginCatalogQuery:
|
||
"""
|
||
返回已经由启动组合根装配的插件目录查询。
|
||
"""
|
||
if _catalog_query is None:
|
||
raise RuntimeError("插件目录查询尚未完成初始化")
|
||
return _catalog_query
|
||
|
||
|
||
def reset_plugin_catalog_query() -> None:
|
||
"""
|
||
清除当前 lifespan 的插件目录查询，供停机和隔离测试使用。
|
||
"""
|
||
global _catalog_query
|
||
_catalog_query = None
|
||
|
||
|
||
def _matches_trusted_source(
|
||
plugin: Plugin,
|
||
plugin_id: str,
|
||
trusted_source_key: str,
|
||
) -> bool:
|
||
"""
|
||
判断市场候选是否为指定插件的可信绑定仓库更新。
|
||
"""
|
||
if (
|
||
not plugin.id
|
||
or plugin.id.lower() != plugin_id.lower()
|
||
or not plugin.has_update
|
||
or not plugin.repo_url
|
||
):
|
||
return False
|
||
try:
|
||
source_key, _ = normalize_github_plugin_source(plugin.repo_url)
|
||
except ValueError:
|
||
return False
|
||
return str(source_key) == trusted_source_key
|
||
|
||
|
||
def _project_update_candidate(
|
||
plugin: Plugin, identity: PluginIdentity | None
|
||
) -> Plugin:
|
||
"""
|
||
在隔离副本上附加来源类型和绑定关系。
|
||
"""
|
||
if not plugin.repo_url or not plugin.plugin_version:
|
||
return plugin
|
||
try:
|
||
source_key, repo_url = normalize_github_plugin_source(plugin.repo_url)
|
||
except ValueError:
|
||
return plugin
|
||
source_type: Literal["official", "third_party"] = (
|
||
"official" if source_key == OFFICIAL_PLUGIN_SOURCE_KEY else "third_party"
|
||
)
|
||
return plugin.model_copy(
|
||
update={
|
||
"update_candidate": PluginUpdateCandidate(
|
||
source_type=source_type,
|
||
source_key=source_key,
|
||
repo_url=repo_url,
|
||
version=plugin.plugin_version,
|
||
is_bound=bool(
|
||
identity and identity.trusted_source_key == source_key
|
||
),
|
||
)
|
||
},
|
||
)
|