src/webui/services/modules.py
Rama main · réplica de lectura del repositorio original.
"""Content Module（模组）统一只读 facade（MOD-04，母方案 §50/§52/§108）。
普通用户的"模组 / 内容库（Compendium）"是产品层抽象；技术实现完全复用
既有 PluginHost 组件（PluginRuntime / ContributionRegistry /
PluginContentCatalog / AdventureSourceRegistry），**不新建数据库、不新建
lifecycle**。本模块只做三件事：
- ``list_modules``：已安装的 content-pack（含传统包与 adventure-module 模组）
的模块卡片数据（母方案 §52：名称 / 版本 / 规则目标 / 冒险数 / 内容数 /
状态 / 来源）；
- ``module_detail``：单个模组的概览 + 内容分组 + 冒险清单（§53 Tabs 的数据源）;
- ``module_content``：内容库只读详情（委托 ``PluginContentCatalog.get_content_resource``）。
边界：
- 只读：不安装、不启停、不改文件（生命周期在 LIFE 组，走既有 PluginHost API）。
- 不把插件类型混淆：provider / tool / bot-extension 等不进入模块库（§51）。
- 性能（母方案 §165）：列表只返回元数据计数；为区分 disabled 与真实零值，
会读取声明目录的 manifest / 静态贡献元数据，但不会把资源正文放入响应。
"""
from __future__ import annotations
from dataclasses import dataclass
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any
from src.knowledge.visibility import entry_visible_to_viewer
from src.plugin_host.support import (
content_delivery_mode,
content_profile,
)
from src.lorebook.module_books import module_book_display_name, module_book_id
from src.adventures import AdventureBundleLoader
from src.plugin_host.registry import ContributionRegistry
from src.version import version_below
from src.webui.services.module_validation import (
MODULE_PLUGIN_TYPE,
ModulePackageValidation,
parse_requires,
validate_module_package,
)
_MODULE_CONTENT_KINDS = (
"npc", "item", "spell", "class", "character_template", "world_template",
)
_MARKETPLACE_CONTENT_PROFILES = frozenset({"adventure-module", "content-pack"})
# 封面只从已安装包的目录里读盘，永远不出站：路径必须落在包内 assets/ 下，
# 且只接受图片后缀。未声明或文件缺失时按「没有封面」处理，前端用占位。
_MODULE_COVER_SUFFIXES = (".webp", ".png", ".jpg", ".jpeg")
_RULESET_FILTER_ALIASES = {
# Generic module code matches the registry's stable short id, without
# importing or hard-coding a concrete ruleset runtime identifier.
"dnd2024": "dnd2024",
"coc": "coc7",
"freeform": "freeform",
}
@dataclass(frozen=True)
class ModuleDependencies:
"""Explicit read-side dependencies for the module catalogue facade."""
plugin_host: Any | None
adventure_registry: Any | None
# Canonical Book sources contributed by a module are read-only catalog data;
# legacy world auto-imports are intentionally not inferred here.
lorebook_store: Any | None = None
ruleset_registry: Any | None = None
list_instances: Callable[[], list[Any]] | None = None
# FIX-01 §3.7：保护目标是"所有持久化存档"，不能只看内存 active GameInstance。
# ``list_save_metadata`` 扫描存档目录，覆盖 paused / ended / 加载失败但元数据
# 仍可读的对局；``refresh_adventure_sources`` 对应 §3.6——任何 destructive
# module action 前先刷新来源注册表，不依赖"碰巧同步过"。
list_save_metadata: Callable[[], list[dict[str, Any]]] | None = None
refresh_adventure_sources: Callable[[], None] | None = None
# 组合根声明的默认 runtime：模块声明了 ruleset_catalogs 但没写 requires 时，
# 用它判断 catalog 契约归属；不使用独立的通用剧情包默认要求。
default_runtime_requirement: Callable[[], dict[str, Any]] | None = None
# FIX-06 §8：ModulesView 的"在线模组"区块复用既有插件市场索引（不新建市场）。
# 组合根注入 ``PluginHost.marketplace_plugins``（已带 installed / 版本信息）。
list_marketplace_plugins: Callable[[], Awaitable[dict[str, Any]]] | None = None
def _installed_content_packs(plugin_host: Any) -> list[Any]:
runtimes = getattr(plugin_host, "plugins", {}) or {}
return sorted(
(
runtime
for runtime in runtimes.values()
if str(runtime.manifest.get("plugin_type") or "") == "content-pack"
),
key=lambda runtime: str(runtime.manifest.get("id") or ""),
)
def _module_cover(runtime: Any) -> str:
"""已安装内容包的封面路径：只认包内 assets/ 下的图片，缺失就当没有。"""
manifest = getattr(runtime, "manifest", {}) or {}
relative = str(manifest.get("cover") or "").strip().replace("\\", "/").lstrip("/")
if not relative.startswith("assets/") or not relative.lower().endswith(_MODULE_COVER_SUFFIXES):
return ""
root = runtime.directory.resolve()
try:
target = (root / relative).resolve()
except OSError:
return ""
if not target.is_relative_to(root) or not target.is_file() or target.is_symlink():
return ""
return relative
def module_cover_path(deps: ModuleDependencies, module_id: str) -> str:
"""已安装内容包封面的绝对路径；未声明、越界或文件缺失都按不存在处理。"""
runtime = getattr(deps.plugin_host, "plugins", {}).get(str(module_id or ""))
if runtime is None or str(runtime.manifest.get("plugin_type") or "") != MODULE_PLUGIN_TYPE:
raise KeyError("MODULE_NOT_FOUND")
relative = _module_cover(runtime)
if not relative:
raise KeyError("MODULE_COVER_MISSING")
return str((runtime.directory.resolve() / relative).resolve())
def _declared_contributions(runtime: Any) -> list[Any] | None:
try:
registry = ContributionRegistry()
return registry.register_static_plugin(runtime.manifest, runtime.directory)
except Exception: # noqa: BLE001 - unreadable inventory is explicitly unknown
return None
def _inventory_projection(runtime: Any, adventure_registry: Any) -> tuple[int | None, dict[str, int] | None, str]:
"""Count installed declarations without confusing disabled with empty."""
root = getattr(runtime, "adventure_packages_root", None)
directories = tuple(getattr(runtime, "adventure_package_directories", ()) or ())
adventure_count: int | None
try:
if root is not None:
root_path = Path(root)
if (
not root_path.is_dir()
or any(
not (root_path / directory).is_dir()
or not (root_path / directory / "manifest.json").is_file()
for directory in directories
)
):
adventure_count = None
else:
adventure_count = len(AdventureBundleLoader(root, allowed_directory_ids=directories).list(""))
else:
source = adventure_registry.source_for(
"plugin", str(runtime.manifest.get("id") or ""),
) if adventure_registry is not None else None
adventure_count = len(source.loader.list("")) if source is not None else 0
except Exception: # noqa: BLE001 - unreadable inventory is explicitly unknown
adventure_count = None
try:
contributions = _declared_contributions(runtime)
if contributions is None:
raise ValueError("declared contributions unavailable")
counts: dict[str, int] | None = {}
for item in contributions:
counts[item.kind] = counts.get(item.kind, 0) + 1
except Exception: # noqa: BLE001 - unreadable inventory is explicitly unknown
counts = None
status = "unknown" if adventure_count is None or counts is None else (
"disabled" if str(getattr(runtime, "status", "")) == "disabled" else "available"
)
return adventure_count, counts, status
def _book_entry_count(store: Any, book_id: str, *, viewer_is_gm: bool = True) -> int | None:
"""条目数只是展示信息：读不到就说不知道，不要让整页失败。"""
lister = getattr(store, "list_book_entries", None)
if not callable(lister):
return None
try:
entries = lister(str(book_id)) or []
except Exception: # noqa: BLE001 - a count is never worth failing the page
return None
if not viewer_is_gm:
# issue #708：计数与正文同一视角谓词，避免「12 条」与玩家可见数不一致。
entries = [entry for entry in entries if entry_visible_to_viewer(entry, "party")]
return len(entries)
def _module_lorebooks(
deps: ModuleDependencies, module_id: str, *, viewer_is_gm: bool = True,
) -> list[dict[str, Any]]:
"""Return canonical Lorebook sources explicitly owned by one module.
两类来源合并（issue #700），**用 ``catalog`` 明确区分**，前端据此决定
能不能编辑、能不能在资料册页打开：
- 用户书库里已物化的 module 书（legacy_autoimport 启用即落地、catalog
模组开局按需物化都会产生），带 ``source_kind=module`` 标记，``catalog``
为 false —— 它就是一本普通的、可编辑的书；
- 该模块**声明**的 lorebook 资源（只读数据，未落地）。物化与声明用同一
deterministic book id 对齐去重，所以每本书只出现一行；catalog 行带
``catalog: true`` 与结构化 ``ref``，创建页据此把它作为开局候选，
开局勾选时由服务端按声明物化。
两种行都带 ``entry_count``：模块页要能回答"包里那本 18 条的书在哪"，
只给书名不给条数的话，用户看到的是一本和空世界主册无从分辨的册子。
"""
store = getattr(deps, "lorebook_store", None)
lister = getattr(store, "list_lorebooks", None)
books: list[dict[str, Any]] = []
if callable(lister):
try:
books = lister() or []
except Exception: # noqa: BLE001 - an optional source must not hide the catalog
books = []
rows = [
{
"id": str(book.get("id") or ""),
"name": str(book.get("name") or book.get("id") or ""),
"description": str(book.get("description") or ""),
"language": str(book.get("language") or "zh-CN"),
"enabled": bool(book.get("enabled", True)),
"source_kind": "module",
"source_id": str(module_id),
"catalog": False,
"entry_count": _book_entry_count(store, str(book.get("id") or ""), viewer_is_gm=viewer_is_gm),
}
for book in books
if str(book.get("source_kind") or "") == "module"
and str(book.get("source_id") or "") == str(module_id)
]
host = getattr(deps, "plugin_host", None)
resources_lister = getattr(host, "list_content_resources", None)
if not callable(resources_lister):
return rows
try:
resources = (resources_lister("lorebook") or {}).get("lorebook", [])
except Exception: # noqa: BLE001 - declared resources must not break the page
return rows
seen_ids = {row["id"] for row in rows}
for resource in resources:
if not isinstance(resource, dict):
continue
if str(resource.get("plugin_id") or "") != str(module_id):
continue
external_id = str(resource.get("id") or "")
book_id = module_book_id(str(module_id), external_id)
if not book_id or book_id in seen_ids:
# 已物化的书走上一段的真实行（含 enabled 状态），catalog 行不再重复。
continue
entries = resource.get("entries")
seen_ids.add(book_id)
if isinstance(entries, list) and not viewer_is_gm:
# issue #708：目录行计数与就地展开正文用同一 fail-closed 谓词。
entries = [entry for entry in entries if entry_visible_to_viewer(entry, "party")]
rows.append({
"id": book_id,
# 包内文件可能没写 name：catalog 兜底成文件名，直接显示就成了
# "dagon-gm-guide"。这里与物化共用同一套可读名字兜底。
"name": module_book_display_name(resource) or external_id,
# 读取条目正文用的声明 key（文件名），与上面的 canonical book id 不同。
"key": external_id,
"description": str(resource.get("description") or ""),
"language": str(resource.get("language") or "zh-CN"),
"enabled": True,
"source_kind": "module",
"source_id": str(module_id),
"catalog": True,
"entry_count": len(entries) if isinstance(entries, list) else None,
"ref": {
"source_kind": "module",
"source_id": str(module_id),
"kind": "lorebook",
"id": book_id,
"digest": "",
},
})
return rows
def list_modules(deps: ModuleDependencies) -> dict[str, Any]:
"""模块库列表（母方案 §51：已安装；在线/本地导入由 marketplace 提供）。"""
modules: list[dict[str, Any]] = []
for runtime in _installed_content_packs(deps.plugin_host):
plugin_id = str(runtime.manifest.get("id") or "")
profile = content_profile(runtime.manifest)
adventure_count, counts, inventory_status = _inventory_projection(runtime, deps.adventure_registry)
lorebooks = _module_lorebooks(deps, plugin_id)
modules.append({
"id": plugin_id,
"name": str(runtime.manifest.get("name") or plugin_id),
"version": str(runtime.manifest.get("version") or ""),
"plugin_type": str(runtime.manifest.get("plugin_type") or ""),
"content_profile": profile,
"content_delivery_mode": content_delivery_mode(runtime.manifest),
"cover": _module_cover(runtime),
"is_module": profile == "adventure-module",
"status": str(runtime.status or ""),
"adventure_count": adventure_count,
"content_counts": counts,
"inventory_status": inventory_status,
"lorebook_count": len(lorebooks),
})
return {"ok": True, "modules": modules}
def _refresh_adventure_sources(deps: ModuleDependencies) -> None:
"""Best-effort refresh of the plugin adventure-source registry (FIX-01 §3.6).
读模型与 guard 都不能依赖"碰巧同步过 registry"（server restart / 刚启用 /
刚安装都会让来源集合过期）。刷新失败不影响只读结果。
"""
refresh = getattr(deps, "refresh_adventure_sources", None)
if not callable(refresh):
return
try:
refresh()
except Exception: # noqa: BLE001 - 刷新失败不拖垮只读目录
pass
async def module_marketplace(
deps: ModuleDependencies, *, keyword: str = "", ruleset: str = "",
) -> dict[str, Any]:
"""在线模组库（FIX-06 §8：ModulesView 的 Online / Marketplace 区块）。
复用既有插件市场索引（``PluginHost.marketplace_plugins``，不新建市场、不新建
索引），只保留 data-only ``content-pack``：模块库不是"另一个插件列表"，把
provider / tool 混进来会让用户以为它们能当模组装。
``installed`` / ``installed_version`` 由宿主给出（服务端事实），
``update_available`` 用既有 ``version_below`` 做展示级比较；市场不可达时返回
``ok=False``，UI 显示离线提示而不是空列表（空列表会被读成"没有模组"）。
"""
lister = getattr(deps, "list_marketplace_plugins", None)
if not callable(lister):
return {
"ok": False, "error_code": "MODULE_MARKETPLACE_UNAVAILABLE",
"error": "marketplace is not wired", "modules": [],
}
try:
listing = await lister()
except Exception as exc: # noqa: BLE001 - 市场故障不能变成 500
return {
"ok": False, "error_code": "MODULE_MARKETPLACE_UNAVAILABLE",
"error": str(exc), "modules": [],
}
if not listing.get("ok"):
return {
"ok": False, "error_code": "MODULE_MARKETPLACE_UNAVAILABLE",
"error": str(listing.get("error") or ""), "modules": [],
}
needle = str(keyword or "").strip().lower()
ruleset_filter = str(ruleset or "").strip().lower()
if ruleset_filter in {"", "all"}:
ruleset_filter = ""
else:
ruleset_filter = _RULESET_FILTER_ALIASES.get(ruleset_filter, ruleset_filter)
modules: list[dict[str, Any]] = []
for item in listing.get("plugins") or []:
if str(item.get("plugin_type") or "") != MODULE_PLUGIN_TYPE:
continue
profile = str(item.get("content_profile") or "content-pack")
if profile not in _MARKETPLACE_CONTENT_PROFILES:
continue
targets = [str(target) for target in (item.get("ruleset_targets") or [])]
target_ids = {target.rsplit(":", 1)[-1].lower() for target in targets}
# 只有「明确声明了别的规则」才不匹配。未声明（全网现存内容包的常态）与
# 声明为通用（system_agnostic）的包都要保留，交给 UI 打「规则未标注」徽章，
# 否则用户一选规则就看不到任何东西。
declared = bool(targets)
agnostic = item.get("system_agnostic") is True
if (
ruleset_filter
and declared
and not agnostic
and ruleset_filter not in targets
and ruleset_filter not in target_ids
):
continue
if needle and needle not in _market_search_text(item):
continue
installed_version = str(item.get("installed_version") or "")
latest = item.get("latest") if isinstance(item.get("latest"), dict) else {}
latest_version = str(latest.get("version") or item.get("version") or "")
modules.append({
"id": str(item.get("id") or ""),
"name": str(item.get("name") or ""),
"version": str(item.get("version") or ""),
"latest_version": latest_version,
"description": str(item.get("description") or ""),
"content_profile": profile,
"content_delivery_mode": str(item.get("content_delivery_mode") or ""),
"adventure_count": int(item.get("adventure_count") or 0),
"ruleset_targets": targets,
"ruleset_declared": declared,
"system_agnostic": agnostic,
"languages": list(item.get("languages") or []),
"tags": list(item.get("tags") or []),
"trust_level": str(item.get("trust_level") or ""),
"distribution": str(item.get("distribution") or ""),
"repository_url": str(item.get("repository_url") or ""),
"release_url": str(item.get("release_url") or ""),
"stars": int(item.get("stars") or 0),
# Index entries may carry author/license (author defaults to the
# repository owner); the normalizer falls back to the manifest.
"author": _author_label(item.get("author")),
"license": str(item.get("license") or "").strip(),
# Hub catalog download / rating / like counters for the shared card.
"stats": dict(item["stats"]) if isinstance(item.get("stats"), dict) else {},
"installed": bool(item.get("installed")),
"installed_version": installed_version,
"update_available": bool(
installed_version and latest_version
# version_below(minimum, current)：已装版本是否**低于**市场最新版本。
and version_below(latest_version, installed_version)
),
"installable": bool(item.get("installable", True)),
"verification_error": str(item.get("verification_error") or ""),
"needs_core_update": bool(item.get("needs_core_update")),
"min_app_version": str(item.get("min_app_version") or ""),
})
return {
"ok": True,
"modules": modules,
"total": len(modules),
"source": listing.get("source") or {},
}
def _author_label(value: Any) -> str:
"""Index entries carry a string; a manifest may carry ``{name, url}``."""
if isinstance(value, str):
return value.strip()
if isinstance(value, dict):
for key in ("name", "login", "id"):
label = value.get(key)
if isinstance(label, str) and label.strip():
return label.strip()
return ""
def _market_search_text(item: dict[str, Any]) -> str:
parts = [str(item.get("id") or ""), str(item.get("name") or "")]
parts.extend(str(tag) for tag in (item.get("tags") or []))
parts.extend(str(target) for target in (item.get("ruleset_targets") or []))
return " ".join(parts).lower()
def module_detail(deps: ModuleDependencies, module_id: str, *, viewer_is_gm: bool = True) -> dict[str, Any]:
"""单个模组详情：概览 + 内容分组计数 + 冒险清单 + 使用存档/按钮 guard。"""
runtime = getattr(deps.plugin_host, "plugins", {}).get(str(module_id or ""))
if runtime is None or str(runtime.manifest.get("plugin_type") or "") != MODULE_PLUGIN_TYPE:
return {"ok": False, "error_code": "MODULE_NOT_FOUND"}
# FIX-06 §8：详情页的"使用中的存档"与 update/disable/uninstall 按钮状态都来自
# 服务端 guard 结果（同一绑定存档判定，且顺带刷新来源注册表），前端不自行推断。
guard = module_action_guard(deps, module_id)
bound_games = list(guard.get("bound_games") or [])
actions = guard.get("actions") or {}
plugin_id = str(runtime.manifest.get("id") or "")
adventure_count, content_counts, inventory_status = _inventory_projection(runtime, deps.adventure_registry)
grouped: dict[str, list[dict[str, Any]]] = {}
contributions = _declared_contributions(runtime) or []
for item in contributions:
grouped.setdefault(item.kind, []).append({
"key": item.key,
"title": item.title,
"description": item.description,
})
adventures = _module_adventure_rows(deps, module_id)
lorebooks = _module_lorebooks(deps, module_id, viewer_is_gm=viewer_is_gm)
return {
"ok": True,
"module": {
"id": plugin_id,
"name": str(runtime.manifest.get("name") or plugin_id),
"version": str(runtime.manifest.get("version") or ""),
"content_profile": content_profile(runtime.manifest),
"content_delivery_mode": content_delivery_mode(runtime.manifest),
"status": str(runtime.status or ""),
"adventure_count": adventure_count,
"content_counts": content_counts,
"inventory_status": inventory_status,
"content": grouped,
"adventures": adventures,
"lorebooks": lorebooks,
"bound_games": bound_games,
"actions": actions,
},
}
def module_lorebooks(
deps: ModuleDependencies, module_id: str, *, viewer_is_gm: bool = True,
) -> dict[str, Any]:
"""List canonical Lorebook sources owned by an installed module."""
detail = module_detail(deps, module_id, viewer_is_gm=viewer_is_gm)
if not detail.get("ok"):
return detail
return {
"ok": True,
"module_id": str(module_id),
"lorebooks": list(detail["module"].get("lorebooks") or []),
}
def module_adventures(deps: ModuleDependencies, module_id: str) -> dict[str, Any]:
"""Dedicated adventure-list read model for one module (API-02)."""
detail = module_detail(deps, module_id)
if not detail.get("ok"):
return detail
module = detail["module"]
return {
"ok": True,
"module_id": module["id"],
"adventures": module["adventures"],
}
def module_content(
deps: ModuleDependencies, module_id: str, kind: str, key: str, *, language: str = "",
viewer_is_gm: bool = True,
) -> dict[str, Any]:
"""内容库只读详情；委托 PluginContentCatalog（不新建存储）。
issue #708：包文件是直读的，展示层必须自己补上视角过滤。非 GM 视角
（玩家座位 / share 请求）只返回公开条目——与引擎运行时同一个
fail-closed 谓词（``entry_visible_to_viewer``），空 ``visible_to`` 的
守秘人条目连标题都不出。GM 视角不做裁剪。
"""
catalog = getattr(deps.plugin_host, "content", None)
if catalog is None:
return {"ok": False, "error_code": "CONTENT_UNAVAILABLE"}
resource = catalog.get_content_resource(kind, key, plugin_id=str(module_id or ""), language=language)
if resource is None:
return {"ok": False, "error_code": "CONTENT_NOT_FOUND"}
if str(kind or "") == "lorebook" and not viewer_is_gm:
entries = resource.get("entries")
if isinstance(entries, list):
visible = [entry for entry in entries if entry_visible_to_viewer(entry, "party")]
resource = {
**resource,
"entries": visible,
"total_entries": len(entries),
"hidden_entries": len(entries) - len(visible),
}
return {"ok": True, "content": resource}
# ---- LIFE-00：安装前兼容性预览（母方案 §33/§123）--------------------------
#
# FIX-01 §3.3：预览与安装共用同一套校验（src.webui.services.module_validation），
# 不再各自维护一份规则。这里只保留 facade 形状。
def preview_module_install(
deps: ModuleDependencies,
manifest: dict[str, Any],
*,
directory: Any | None = None,
require_content_pack: bool = True,
) -> dict[str, Any]:
"""安装前兼容性预览：blockers（阻断）与 warnings（警告）分离（§57）。
``directory`` 给出已解压的包目录时同时做深度校验（Adventure 包真装载 /
ruleset catalog 真装载）；模块安装面默认要求 content-pack。
"""
validation = validate_module_package(
manifest,
deps=deps,
directory=directory,
require_content_pack=require_content_pack,
)
return validation.preview()
def validate_module_directory(
deps: ModuleDependencies,
directory: Any,
manifest: dict[str, Any],
*,
require_content_pack: bool = True,
) -> ModulePackageValidation:
"""Validate one extracted package for the module install surface (raises)."""
from src.webui.services.module_validation import validate_module_install
return validate_module_install(
directory,
manifest,
deps=deps,
require_content_pack=require_content_pack,
)
def module_compatibility(deps: ModuleDependencies, module_id: str) -> dict[str, Any]:
"""Preview the installed module's declared runtime compatibility."""
runtime = getattr(deps.plugin_host, "plugins", {}).get(str(module_id or ""))
if runtime is None or str(runtime.manifest.get("plugin_type") or "") != MODULE_PLUGIN_TYPE:
return {"ok": False, "error_code": "MODULE_NOT_FOUND"}
result = preview_module_install(deps, runtime.manifest)
return {**result, "module_id": str(runtime.manifest.get("id") or "")}
# ---- LIFE-01：绑定存档保护（母方案 §35/§36/§37/§124）------------------------
# 受保护操作 → 绑定存档存在时是否阻断。更新/禁用/卸载/覆盖安装一律默认 block；
# 母方案 §35：更新改变 bound adventure digest → BLOCK。
# FIX-01 §3.5：``overwrite``（本地覆盖安装 / 市场覆盖安装 / 后台自动更新）也是
# 改 package bytes 的破坏性操作，必须与 update 同等受保护。
PROTECTED_MODULE_ACTIONS = ("uninstall", "disable", "update", "overwrite", "stop")
# FIX-06 §8：模块详情页会呈现给用户的受保护操作（``stop`` 是 ``disable`` 在
# 插件控制路由里的内部别名，不作为独立按钮暴露）。
MODULE_ACTIONS = ("update", "disable", "uninstall", "overwrite")
def _module_adventure_rows(deps: ModuleDependencies, module_id: str) -> list[dict[str, Any]]:
"""The adventure packages a module declares, as read-model rows.
FIX-01 §3.6 / FIX-06 §8：**不依赖"碰巧同步过 registry"**（否则 server restart
之后详情页的 Adventures 段永远是空的）。来源优先级：
1. runtime 自己声明的 declared-only 包目录（安装后即存在，``discover()`` 重建，
且与 enabled/disabled 状态无关）；
2. 退回到来源注册表（plugin 来源在禁用时会被同步移除，因此只作兜底）。
"""
runtime = getattr(deps.plugin_host, "plugins", {}).get(str(module_id or ""))
if runtime is None:
return []
plugin_id = str(runtime.manifest.get("id") or "")
bundles: list[Any] | None = None
root = getattr(runtime, "adventure_packages_root", None)
directories = tuple(getattr(runtime, "adventure_package_directories", ()) or ())
if root is not None and directories:
from src.adventures import AdventureBundleLoader
try:
loader = AdventureBundleLoader(root, allowed_directory_ids=directories)
bundles = list(loader.list(""))
except Exception: # noqa: BLE001 - 坏包不拖垮详情页
bundles = []
if bundles is None:
registry = deps.adventure_registry
source = (
registry.source_for("plugin", plugin_id) if registry is not None else None
)
if source is None:
return []
try:
bundles = list(source.loader.list(""))
except Exception: # noqa: BLE001 - 坏包不拖垮详情页
return []
return [
{
"adventure_id": bundle.manifest.adventure_id,
"version": bundle.manifest.version,
"format": bundle.manifest.format,
"directory_id": bundle.root.name,
"name": str(bundle.adventure.get("tutorial", {}).get("name") or bundle.adventure.get("name") or bundle.manifest.adventure_id),
"summary": str(bundle.adventure.get("tutorial", {}).get("summary") or bundle.adventure.get("summary") or ""),
}
for bundle in bundles
]
def module_adventure_ids(deps: ModuleDependencies, module_id: str) -> set[str]:
"""The adventure ids a module package declares (protection target)."""
return {
str(row["adventure_id"])
for row in _module_adventure_rows(deps, module_id)
if row["adventure_id"]
}
def _active_bindings(deps: ModuleDependencies) -> list[dict[str, Any]]:
"""Bindings from in-memory instances (active / paused / ended)."""
rows: list[dict[str, Any]] = []
# 可选依赖：缺失/未接线的读模型按"没有内存实例"处理，不抛（与 §3.6/§3.7 一致）。
lister = getattr(deps, "list_instances", None)
for instance in (lister or (lambda: []))():
binding = getattr(instance, "adventure_binding", {}) or {}
rows.append({
"game_key": "|".join(str(part) for part in instance.game_key),
"adventure_id": str(binding.get("adventure_id") or ""),
"run_id": str(instance.run_id or ""),
"content_digest": str(binding.get("content_digest") or ""),
"source_kind": str(binding.get("source_kind") or ""),
"source_id": str(binding.get("source_id") or ""),
"state": str(getattr(getattr(instance, "state", ""), "value", "") or ""),
"metadata_readable": True,
})
return rows
def _persisted_bindings(deps: ModuleDependencies) -> list[dict[str, Any]]:
"""Bindings from persisted saves (FIX-01 §3.7).
扫描存档目录而不是内存 registry，因此重启后、ENDED 对局、以及加载/恢复失败
但元数据仍可读的存档同样受保护。
"""
scanner = getattr(deps, "list_save_metadata", None)
if not callable(scanner):
return []
rows: list[dict[str, Any]] = []
try:
entries = scanner() or []
except Exception: # noqa: BLE001 - 扫描失败不拖垮只读目录
return []
for entry in entries:
if not isinstance(entry, dict):
continue
rows.append({
"game_key": str(entry.get("game_key") or ""),
"adventure_id": str(entry.get("adventure_id") or ""),
"run_id": str(entry.get("run_id") or ""),
"content_digest": str(entry.get("content_digest") or ""),
"source_kind": str(entry.get("source_kind") or ""),
"source_id": str(entry.get("source_id") or ""),
"state": str(entry.get("state") or ""),
"metadata_readable": bool(entry.get("metadata_readable", False)),
})
return rows
def module_bound_rows(
deps: ModuleDependencies,
module_id: str,
) -> list[dict[str, Any]]:
"""Every save (memory + persisted) bound to one module, deduped by game key."""
adventure_ids = module_adventure_ids(deps, module_id)
if not adventure_ids:
return []
merged: dict[str, dict[str, Any]] = {}
for row in (*_active_bindings(deps), *_persisted_bindings(deps)):
if row["adventure_id"] not in adventure_ids:
continue
# Source-aware plugin bindings protect only their owning module. A
# legacy id-only save remains conservatively protected by every module
# that declares that id because its origin cannot be proven.
if row.get("source_kind") == "plugin" and row.get("source_id") != module_id:
continue
key = row["game_key"]
current = merged.get(key)
# 内存实例更权威（有 state / 最新 run_id）；缺失时用存档元数据补齐。
if current is None or not current.get("state"):
merged[key] = {**current, **row} if current else row
return sorted(merged.values(), key=lambda item: item["game_key"])
def module_bound_games(
deps: ModuleDependencies,
module_id: str,
) -> list[dict[str, Any]]:
"""List the games bound to one module's adventures（§35 UI 数据源）。"""
return [
{
"game_key": row["game_key"],
"adventure_id": row["adventure_id"],
"run_id": row["run_id"],
}
for row in module_bound_rows(deps, module_id)
]
def _module_bound_games_for_guard(
deps: ModuleDependencies, module_id: str,
) -> list[dict[str, Any]]:
"""Bound saves blocking a protected action — the one guard computation.
FIX-01 §3.6：guard 前先刷新 module/adventure 来源注册表，避免
"server restart → registry 为空 → destructive action fail-open"。
FIX-06 §8：UI 的按钮状态与 enforce 路径共用本函数，两者不可能给出不同结论。
"""
refresh = getattr(deps, "refresh_adventure_sources", None)
if callable(refresh):
try:
refresh()
except Exception: # noqa: BLE001 - 刷新失败不得让 guard 变成 fail-open
pass
return module_bound_games(deps, module_id)
def assert_module_action_allowed(deps: ModuleDependencies, module_id: str, action: str) -> None:
"""Guard for uninstall/disable/update/overwrite/stop（母方案 §124：默认 block）。
绑定存档存在时抛 :class:`ModuleInUse`；调用方（插件生命周期 API / 插件宿主）
把它转成结构化错误，UI 展示"哪些存档正在使用"。
"""
if action not in PROTECTED_MODULE_ACTIONS:
return
bound = _module_bound_games_for_guard(deps, module_id)
if bound:
raise ModuleInUse(module_id, action, bound)
def module_action_guard(deps: ModuleDependencies, module_id: str) -> dict[str, Any]:
"""Per-action server verdict for the module detail buttons（FIX-06 §8）。
按钮状态必须来自 server guard 结果，而不是前端猜：这里用与
:func:`assert_module_action_allowed` 完全相同的绑定存档判定，一次性给出每个
受保护 action 的 ``allowed`` / ``reason`` / ``games``。用户点击时服务端仍会
再判一次（同一函数），所以这只是"提前显示结论"，不是放行。
"""
runtime = getattr(deps.plugin_host, "plugins", {}).get(str(module_id or ""))
if runtime is None or str(runtime.manifest.get("plugin_type") or "") != MODULE_PLUGIN_TYPE:
return {"ok": False, "error_code": "MODULE_NOT_FOUND", "actions": {}}
bound = _module_bound_games_for_guard(deps, module_id)
actions: dict[str, dict[str, Any]] = {}
for action in MODULE_ACTIONS:
blocked = bool(bound)
actions[action] = {
"allowed": not blocked,
"reason": "MODULE_IN_USE" if blocked else "",
"games": list(bound) if blocked else [],
}
return {
"ok": True,
"module_id": str(runtime.manifest.get("id") or ""),
"actions": actions,
"bound_games": list(bound),
}
class ModuleInUse(ValueError):
"""The module is bound by games; the protected action must not proceed."""
def __init__(self, module_id: str, action: str, games: list[dict[str, Any]]) -> None:
self.module_id = str(module_id)
self.action = str(action)
self.games = list(games)
super().__init__(
f"module {self.module_id!r} is used by {len(self.games)} game(s); "
f"{self.action} blocked"
)
# ---- LIFE-02：Module Usage Index（母方案 §125/§169）-------------------------
def module_usages(deps: ModuleDependencies, module_id: str) -> dict[str, Any]:
"""模块使用索引：哪个模块 / 哪个冒险被哪些存档绑定（只读聚合）。
FIX-01 §3.7：与 bound-save guard 同源——内存实例 + 持久化存档元数据。
"""
runtime = getattr(deps.plugin_host, "plugins", {}).get(str(module_id or ""))
if runtime is None:
return {"ok": False, "error_code": "MODULE_NOT_FOUND"}
plugin_id = str(runtime.manifest.get("id") or "")
adventure_ids = module_adventure_ids(deps, module_id)
by_adventure: dict[str, list[dict[str, Any]]] = {
adventure_id: [] for adventure_id in sorted(adventure_ids)
}
for row in module_bound_rows(deps, module_id):
adventure_id = row["adventure_id"]
if adventure_id in by_adventure:
by_adventure[adventure_id].append({
"game_key": row["game_key"],
"run_id": row["run_id"],
"content_digest": row["content_digest"],
"state": row["state"],
"metadata_readable": row["metadata_readable"],
})
usages = [
{"adventure_id": adventure_id, "games": games}
for adventure_id, games in by_adventure.items()
]
total = sum(len(item["games"]) for item in usages)
return {"ok": True, "module_id": plugin_id, "usages": usages, "total_games": total}
__all__ = [
"MODULE_ACTIONS",
"ModuleDependencies",
"ModuleInUse",
"PROTECTED_MODULE_ACTIONS",
"assert_module_action_allowed",
"list_modules",
"module_action_guard",
"module_adventures",
"module_adventure_ids",
"module_bound_games",
"module_bound_rows",
"module_content",
"module_compatibility",
"module_detail",
"module_marketplace",
"module_usages",
"parse_requires",
"preview_module_install",
"validate_module_directory",
]