mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 19:37:54 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
860 lines
30 KiB
Python
860 lines
30 KiB
Python
"""
|
||
装配宿主模块、托管资源及其启动和关闭顺序。
|
||
"""
|
||
|
||
import asyncio
|
||
import inspect
|
||
import sys
|
||
from typing import Any, Callable, cast
|
||
|
||
# SitesHelper涉及资源包拉取，提前引入并容错提示
|
||
try:
|
||
from app.application.site.sites import SitesHelper as _SitesHelper # noqa # pylint: disable=import-error,no-name-in-module
|
||
except ImportError as e:
|
||
error_message = f"错误: {str(e)}\n站点认证及索引相关资源导入失败，请尝试重建容器或手动拉取资源"
|
||
print(error_message, file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
from app.adapters.external.server import MoviePilotServerHelper
|
||
from app.adapters.system.host import SystemUtils
|
||
from app.application.configuration import (
|
||
TransferRetryConfig,
|
||
configure_transfer_retry_config,
|
||
get_configured_system_config,
|
||
reset_transfer_retry_config,
|
||
)
|
||
from app.application.messaging.agent import (
|
||
shutdown_web_agent_background_tasks,
|
||
wait_web_agent_background_tasks,
|
||
)
|
||
from app.application.messaging.chat import (
|
||
get_configured_agent_chat_persistence,
|
||
)
|
||
from app.application.messaging.message import (
|
||
MessageHelper,
|
||
stop_message,
|
||
)
|
||
from app.application.messaging.webagent.events import dispatch_web_agent_message_event
|
||
from app.application.module import (
|
||
ModuleRuntime,
|
||
configure_module_runtime,
|
||
reset_module_runtime,
|
||
)
|
||
from app.application.outbox import configure_outbox_dispatcher
|
||
from app.application.plugin.runtime import get_existing_plugin_manager
|
||
from app.application.security.url import close_image_proxy_block_log_coalescer
|
||
from app.application.service import configure_service_directory, reset_service_directory
|
||
from app.application.site.auth import normalize_site_auth_config
|
||
from app.command import CommandChain
|
||
from app.db.session import (
|
||
close_database,
|
||
)
|
||
from app.runtime.config import settings as legacy_settings
|
||
from app.runtime.event.binding import EventHandlerBinding
|
||
from app.runtime.events import EventManager
|
||
from app.runtime.execution import run_in_threadpool_to_completion
|
||
from app.runtime.extensions.module.manager import ModuleManager
|
||
from app.runtime.extensions.service import (
|
||
ServiceConfigHelper,
|
||
configure_service_config_reader,
|
||
reset_service_config_reader,
|
||
)
|
||
from app.runtime.log import logger
|
||
from app.runtime.settings import (
|
||
get_runtime_setting,
|
||
)
|
||
from app.runtime.tasks import get_task_registry
|
||
from app.runtime.thread import ThreadHelper
|
||
from app.schemas.category import ClassificationFactValue, ClassificationFieldDefinition
|
||
from app.schemas.message import Message
|
||
from app.schemas.types import EventType, MessageType, SystemConfigKey
|
||
from app.startup.composition.agent import (
|
||
compose_agent,
|
||
publish_agent_services,
|
||
reset_agent_services,
|
||
)
|
||
from app.startup.composition.cache import get_cache_composition
|
||
from app.startup.composition.chain import (
|
||
configure_chain_runtime_context,
|
||
configure_directory_classification_service,
|
||
configure_wallpaper_services,
|
||
reset_chain_services,
|
||
)
|
||
from app.startup.composition.classification import compose_classification
|
||
from app.startup.composition.configuration import (
|
||
compose_configuration,
|
||
publish_configuration,
|
||
reset_configuration,
|
||
)
|
||
from app.startup.composition.context import HostRuntime
|
||
from app.startup.composition.database import (
|
||
compose_database_services,
|
||
configure_database,
|
||
configure_workflow_execution_composition,
|
||
database_runtime_active,
|
||
publish_database_services,
|
||
reset_database_services,
|
||
reset_workflow_execution_composition,
|
||
start_database_runtime,
|
||
stop_database_runtime,
|
||
)
|
||
from app.startup.composition.enrichment import (
|
||
compose_classification_enrichment,
|
||
)
|
||
from app.startup.composition.network import (
|
||
configure_application_network_ports,
|
||
configure_doh_composition,
|
||
get_existing_doh_composition,
|
||
reset_application_network_ports,
|
||
stop_doh_composition,
|
||
)
|
||
from app.startup.composition.outbox import build_outbox_dispatcher, reset_outbox_services
|
||
from app.startup.composition.resource import (
|
||
configure_site_resource_versions,
|
||
reset_site_resource_composition,
|
||
)
|
||
from app.startup.composition.runtime import (
|
||
RuntimeInputs,
|
||
compose_runtime,
|
||
compose_runtime_dependencies,
|
||
publish_runtime,
|
||
reset_runtime,
|
||
)
|
||
from app.startup.composition.security import (
|
||
configure_security_access,
|
||
configure_security_services,
|
||
reset_security_access,
|
||
reset_security_services,
|
||
)
|
||
from app.startup.composition.server import configure_server_services, reset_server_services
|
||
from app.startup.initializers.agent import (
|
||
configure_agent_data_context,
|
||
init_agent,
|
||
reset_agent_data_context,
|
||
)
|
||
from app.startup.initializers.resources import (
|
||
init_managed_resources,
|
||
reset_managed_resources,
|
||
stop_managed_resources,
|
||
)
|
||
from app.startup.initializers.scheduler import (
|
||
configure_scheduler_agent_tasks,
|
||
reset_scheduler_bindings,
|
||
)
|
||
|
||
SitesHelper: Any = _SitesHelper
|
||
|
||
|
||
def _plugin_classification_fields() -> tuple[ClassificationFieldDefinition, ...]:
|
||
"""
|
||
读取已创建插件运行时的字段快照，模块初始化阶段不提前物化管理器。
|
||
"""
|
||
manager = get_existing_plugin_manager()
|
||
if manager is None:
|
||
return ()
|
||
try:
|
||
return cast(
|
||
tuple[ClassificationFieldDefinition, ...],
|
||
manager.get_classification_fields(),
|
||
)
|
||
# 插件字段目录故障不得阻断策略管理。
|
||
except Exception as error: # noqa: BLE001
|
||
logger.warning(f"读取插件分类字段目录失败，已忽略动态字段：{error}")
|
||
return ()
|
||
|
||
|
||
def _plugin_classification_facts(
|
||
media: object,
|
||
) -> dict[str, dict[str, ClassificationFactValue]]:
|
||
"""
|
||
通过已创建插件运行时校验媒体对象携带的扩展事实。
|
||
"""
|
||
manager = get_existing_plugin_manager()
|
||
if manager is None:
|
||
return {}
|
||
try:
|
||
return cast(
|
||
dict[str, dict[str, ClassificationFactValue]],
|
||
manager.get_classification_facts(media),
|
||
)
|
||
# 插件事实不得成为媒体识别硬依赖。
|
||
except Exception as error: # noqa: BLE001
|
||
logger.warning(f"读取插件扩展分类事实失败，已按字段缺失继续：{error}")
|
||
return {}
|
||
|
||
|
||
def _existing_module_manager() -> object | None:
|
||
"""
|
||
返回已创建模块管理器；测试替身或未初始化状态按无 provider 处理。
|
||
"""
|
||
getter = getattr(ModuleManager, "get_existing_instance", None)
|
||
return getter() if callable(getter) else None
|
||
|
||
|
||
def configure_runtime_data_providers() -> None:
|
||
"""
|
||
在启动组合层装配模块和配置目录的运行时读取能力。
|
||
"""
|
||
configure_service_config_reader(lambda key: get_configured_system_config().get(key))
|
||
configure_module_runtime(lambda: cast(ModuleRuntime, ModuleManager()))
|
||
configure_service_directory(
|
||
configs=ServiceConfigHelper.get_configs,
|
||
modules=lambda module_type: list(
|
||
ModuleManager().get_running_type_modules(module_type)
|
||
),
|
||
)
|
||
|
||
|
||
def reset_runtime_data_providers() -> None:
|
||
"""
|
||
按发布逆序撤销模块、服务目录和配置读取 Provider。
|
||
"""
|
||
reset_service_directory()
|
||
reset_module_runtime()
|
||
reset_service_config_reader()
|
||
|
||
|
||
def reset_event_services() -> None:
|
||
"""
|
||
撤销启动组合根登记的事件通知、监听器和宿主 resolver。
|
||
"""
|
||
event_manager = EventManager.get_existing_instance()
|
||
if event_manager is None:
|
||
return
|
||
event_manager.remove_event_listener(
|
||
EventType.NoticeMessage,
|
||
dispatch_web_agent_message_event,
|
||
)
|
||
event_manager.unregister_handler_instance_resolver("host")
|
||
event_manager.unregister_handler_instance_resolver("config_reload")
|
||
event_manager.reset_error_notifier()
|
||
|
||
|
||
def reset_tool_services() -> None:
|
||
"""
|
||
撤销工具管理器持有的 Agent 数据上下文与目录快照。
|
||
"""
|
||
from app.agent.tools.manager import moviepilot_tool_manager
|
||
|
||
moviepilot_tool_manager.reset_data_context()
|
||
|
||
|
||
def _module_provider_reset_steps() -> tuple[tuple[str, Callable[[], object]], ...]:
|
||
"""
|
||
返回当前模块 lifespan 的完整 Provider 逆序撤销清单。
|
||
"""
|
||
return (
|
||
("事件服务", reset_event_services),
|
||
("资源版本 Provider", reset_site_resource_composition),
|
||
("认证访问服务", reset_security_access),
|
||
("Chain 服务", reset_chain_services),
|
||
("托管资源引用", reset_managed_resources),
|
||
("调度器绑定", reset_scheduler_bindings),
|
||
("Agent 工具服务", reset_tool_services),
|
||
(
|
||
"Agent 服务",
|
||
lambda: reset_agent_services(
|
||
data_context_resetter=reset_agent_data_context,
|
||
),
|
||
),
|
||
("传输重试配置", reset_transfer_retry_config),
|
||
("Outbox 服务", reset_outbox_services),
|
||
("工作流执行服务", reset_workflow_execution_composition),
|
||
("中心服务", reset_server_services),
|
||
("运行时数据 Provider", reset_runtime_data_providers),
|
||
("HostRuntime 投影", reset_runtime),
|
||
("数据库服务", reset_database_services),
|
||
("认证服务", reset_security_services),
|
||
("应用网络端口", reset_application_network_ports),
|
||
("配置服务", reset_configuration),
|
||
)
|
||
|
||
|
||
def reset_module_providers() -> bool:
|
||
"""
|
||
逐项撤销模块 Provider；单项失败不跳过后续 owner，并返回整体结果。
|
||
"""
|
||
converged = True
|
||
for name, callback in _module_provider_reset_steps():
|
||
try:
|
||
callback()
|
||
except Exception as error:
|
||
logger.error("撤销%s失败：%s", name, error)
|
||
converged = False
|
||
return converged
|
||
|
||
|
||
def _existing_singleton(instance_type: type) -> object | None:
|
||
"""
|
||
读取已经物化的 Singleton，启动失败回滚时禁止反向创建新 owner。
|
||
"""
|
||
getter = getattr(instance_type, "get_existing_instance", None)
|
||
return getter() if callable(getter) else None
|
||
|
||
|
||
def _call_existing_singleton(instance_type: type | None, method_name: str) -> object:
|
||
"""
|
||
调用已存在 Singleton 的关闭方法；owner 未接入或未创建时视为已经收敛。
|
||
"""
|
||
if instance_type is None:
|
||
return True
|
||
instance = _existing_singleton(instance_type)
|
||
if instance is None:
|
||
return True
|
||
return getattr(instance, method_name)()
|
||
|
||
|
||
def notify_event_error(title: str, message: str) -> None:
|
||
"""
|
||
将事件总线错误转发到系统消息通道。
|
||
"""
|
||
MessageHelper().put(
|
||
title=title,
|
||
message=message,
|
||
role="system",
|
||
level="error",
|
||
)
|
||
|
||
|
||
def get_host_event_handler_factories() -> dict[type, Callable[[], object]]:
|
||
"""
|
||
返回所有使用事件装饰器的宿主类及其明确实例工厂。
|
||
"""
|
||
from app.chain.download import DownloadChain
|
||
from app.chain.scraping import ScrapingChain
|
||
from app.chain.search.facade import SearchChain
|
||
from app.chain.site import SiteChain
|
||
from app.chain.subscribe.facade import SubscribeChain
|
||
from app.chain.workflow import WorkflowChain
|
||
from app.command import Command
|
||
from app.scheduler.facade import Scheduler
|
||
|
||
return {
|
||
Command: Command,
|
||
DownloadChain: DownloadChain,
|
||
Scheduler: Scheduler,
|
||
ScrapingChain: ScrapingChain,
|
||
SearchChain: SearchChain,
|
||
SiteChain: SiteChain,
|
||
SubscribeChain: SubscribeChain,
|
||
WorkflowChain: WorkflowChain,
|
||
}
|
||
|
||
|
||
def get_config_reload_handler_providers(
|
||
) -> dict[type, Callable[[], object | None]]:
|
||
"""
|
||
返回未由专属运行时接管的配置重载 owner Provider。
|
||
"""
|
||
from app.chain.transfer.facade import TransferChain
|
||
from app.monitor.monitor import Monitor
|
||
from app.runtime.state import SystemHelper
|
||
|
||
# SystemHelper 不持有外部资源；闭包为当前 lifespan 保留稳定身份。
|
||
system_helper = SystemHelper()
|
||
providers: dict[type, Callable[[], object | None]] = {
|
||
SystemHelper: lambda: system_helper,
|
||
# Redis 连接 owner 只在启动时选用 Redis 缓存后存在。
|
||
**{
|
||
owner: owner.get_existing_instance
|
||
for owner in get_cache_composition().redis_owners
|
||
},
|
||
TransferChain: TransferChain.get_existing_instance,
|
||
Monitor: Monitor.get_existing_instance,
|
||
}
|
||
doh_helper = get_existing_doh_composition()
|
||
if doh_helper is not None:
|
||
providers[type(doh_helper)] = get_existing_doh_composition
|
||
plugin_manager = get_existing_plugin_manager()
|
||
if plugin_manager is not None:
|
||
providers[type(plugin_manager)] = get_existing_plugin_manager
|
||
return providers
|
||
|
||
|
||
def configure_config_reload_event_handler_resolver() -> None:
|
||
"""
|
||
绑定当前生命周期已拥有的配置重载实例，惰性 owner 未创建时跳过。
|
||
"""
|
||
providers = get_config_reload_handler_providers()
|
||
runtime_providers = (
|
||
get_existing_doh_composition,
|
||
get_existing_plugin_manager,
|
||
)
|
||
|
||
def resolve(owner_class: type) -> EventHandlerBinding | None:
|
||
"""
|
||
按类型身份读取当前 owner，不在事件分发路径构造资源。
|
||
"""
|
||
provider = providers.get(owner_class)
|
||
if provider is not None:
|
||
instance = provider()
|
||
if instance is not None and type(instance) is not owner_class:
|
||
return None
|
||
return EventHandlerBinding(
|
||
instance=instance,
|
||
owner_name=owner_class.__name__,
|
||
)
|
||
instance = next(
|
||
(
|
||
current
|
||
for runtime_provider in runtime_providers
|
||
if (current := runtime_provider()) is not None
|
||
and type(current) is owner_class
|
||
),
|
||
None,
|
||
)
|
||
if instance is None:
|
||
return None
|
||
return EventHandlerBinding(
|
||
instance=instance,
|
||
owner_name=owner_class.__name__,
|
||
)
|
||
|
||
EventManager().register_handler_instance_resolver(
|
||
"config_reload",
|
||
resolve,
|
||
)
|
||
|
||
|
||
def configure_host_event_handler_resolver() -> None:
|
||
"""
|
||
显式登记宿主内置类处理器，禁止事件总线按类名临时构造未知对象。
|
||
"""
|
||
factories = get_host_event_handler_factories()
|
||
|
||
def resolve(owner_class: type) -> EventHandlerBinding | None:
|
||
"""
|
||
按明确白名单复用单例或构造与旧路径等价的 Chain 实例。
|
||
"""
|
||
factory = factories.get(owner_class)
|
||
if factory is None:
|
||
return None
|
||
get_existing = getattr(owner_class, "get_existing_instance", None)
|
||
instance = get_existing() if callable(get_existing) else None
|
||
if instance is None:
|
||
instance = factory()
|
||
return EventHandlerBinding(
|
||
instance=instance,
|
||
owner_name=owner_class.__name__,
|
||
)
|
||
|
||
EventManager().register_handler_instance_resolver("host", resolve)
|
||
|
||
|
||
def start_frontend() -> None:
|
||
"""
|
||
启动前端服务
|
||
"""
|
||
# 仅Windows可执行文件支持内嵌nginx
|
||
if not SystemUtils.is_frozen() or not SystemUtils.is_windows():
|
||
return
|
||
# 临时Nginx目录
|
||
nginx_path = get_runtime_setting("ROOT_PATH") / "nginx"
|
||
if not nginx_path.exists():
|
||
return
|
||
# 配置目录下的Nginx目录
|
||
run_nginx_dir = get_runtime_setting("CONFIG_PATH").with_name("nginx")
|
||
if not run_nginx_dir.exists():
|
||
# 移动到配置目录
|
||
SystemUtils.move(nginx_path, run_nginx_dir)
|
||
# 启动Nginx
|
||
import subprocess
|
||
|
||
subprocess.Popen("start nginx.exe", cwd=run_nginx_dir, shell=True)
|
||
|
||
|
||
def stop_frontend() -> None:
|
||
"""
|
||
停止前端服务
|
||
"""
|
||
if not SystemUtils.is_frozen() or not SystemUtils.is_windows():
|
||
return
|
||
import subprocess
|
||
|
||
subprocess.Popen("taskkill /f /im nginx.exe", shell=True)
|
||
|
||
|
||
def clear_temp() -> None:
|
||
"""
|
||
清理临时文件和图片缓存
|
||
"""
|
||
# 清理临时目录中3天前的文件
|
||
SystemUtils.clear(
|
||
get_runtime_setting("TEMP_PATH"),
|
||
days=get_runtime_setting("TEMP_FILE_DAYS"),
|
||
)
|
||
# 清理图片缓存目录中7天前的文件
|
||
SystemUtils.clear(
|
||
get_runtime_setting("CACHE_PATH") / "images",
|
||
days=get_runtime_setting("GLOBAL_IMAGE_CACHE_DAYS"),
|
||
)
|
||
# 清理 pip/uv 包下载缓存，不接管整个 .cache 目录。
|
||
clear_package_tool_cache()
|
||
|
||
|
||
def clear_package_tool_cache() -> None:
|
||
"""
|
||
清理 pip/uv 包下载缓存，只处理 MoviePilot 管理的工具子目录。
|
||
"""
|
||
days = get_runtime_setting("PACKAGE_CACHE_DAYS")
|
||
if days <= 0:
|
||
return
|
||
tool_cache_root = get_runtime_setting("PACKAGE_CACHE_PATH")
|
||
for child in ("pip", "uv"):
|
||
cache_path = tool_cache_root / child
|
||
try:
|
||
SystemUtils.clear(cache_path, days=days)
|
||
except Exception as err:
|
||
logger.warning("清理包下载缓存失败：%s - %s", cache_path, err)
|
||
|
||
|
||
def user_auth() -> None:
|
||
"""
|
||
用户认证检查
|
||
"""
|
||
sites_helper = SitesHelper()
|
||
if sites_helper.auth_level >= 2:
|
||
return
|
||
auth_conf = get_configured_system_config().get(SystemConfigKey.UserSiteAuthParams)
|
||
if get_runtime_setting("AUTH_SITE"):
|
||
status, msg = sites_helper.check_user()
|
||
else:
|
||
normalized_auth_conf = normalize_site_auth_config(
|
||
auth_conf,
|
||
sites_helper.get_authsites(),
|
||
)
|
||
status, msg = (
|
||
sites_helper.check_user(**normalized_auth_conf)
|
||
if normalized_auth_conf
|
||
else sites_helper.check_user()
|
||
)
|
||
if status:
|
||
logger.info(f"{msg} 用户认证成功")
|
||
else:
|
||
logger.info(f"用户认证失败，{msg}")
|
||
|
||
|
||
def check_auth() -> None:
|
||
"""
|
||
检查认证状态
|
||
"""
|
||
if SitesHelper().auth_level < 2:
|
||
err_msg = "用户认证失败，站点相关功能将无法使用！"
|
||
MessageHelper().put(f"注意：{err_msg}", title="用户认证", role="system", level="warning")
|
||
CommandChain().post_message(
|
||
Message(
|
||
mtype=MessageType.Manual,
|
||
title="MoviePilot用户认证",
|
||
text=err_msg,
|
||
link=get_runtime_setting("MP_DOMAIN")("#/site"),
|
||
)
|
||
)
|
||
|
||
|
||
def update_resources() -> None:
|
||
"""
|
||
保留旧入口但不在启动后检查、下载或重启，资源由启动器预先应用。
|
||
"""
|
||
return
|
||
|
||
|
||
def close_browser_sessions() -> None:
|
||
"""
|
||
在托管资源关闭前释放所有浏览器上下文及其工作线程。
|
||
"""
|
||
from app.adapters.network.browser import BrowserSessionHelper
|
||
|
||
BrowserSessionHelper.close_all_sessions()
|
||
|
||
|
||
async def drain_events() -> bool:
|
||
"""
|
||
在插件卸载前等待已接收事件及其同步、异步处理器完成。
|
||
"""
|
||
event_manager = EventManager.get_existing_instance()
|
||
if event_manager is None:
|
||
return True
|
||
return bool(await event_manager.drain_async(seal=True))
|
||
|
||
|
||
async def settle_events() -> bool:
|
||
"""
|
||
在插件 handler 停用后结算在途事件，但保留停机 hook 的尾事件入口。
|
||
"""
|
||
event_manager = EventManager.get_existing_instance()
|
||
if event_manager is None:
|
||
return True
|
||
return bool(await event_manager.drain_async(seal=False))
|
||
|
||
|
||
async def stop_modules() -> bool:
|
||
"""
|
||
关闭模块服务，并返回全部资源 owner 是否收敛。
|
||
"""
|
||
all_converged = True
|
||
|
||
async def run_step(
|
||
name: str,
|
||
callback: Callable[[], object],
|
||
*,
|
||
offload: bool = False,
|
||
record_failure: bool = True,
|
||
) -> bool:
|
||
"""
|
||
执行单个关闭步骤，失败时继续收口并保留诚实结果。
|
||
"""
|
||
nonlocal all_converged
|
||
try:
|
||
if offload:
|
||
result = await run_in_threadpool_to_completion(callback)
|
||
else:
|
||
result = callback()
|
||
if inspect.isawaitable(result):
|
||
result = await result
|
||
converged = result is not False
|
||
if not converged:
|
||
logger.error("关闭%s未收敛，继续执行后续资源收口", name)
|
||
except asyncio.CancelledError:
|
||
logger.warning("关闭%s时收到取消请求，继续执行资源收口", name)
|
||
converged = False
|
||
except Exception as err:
|
||
logger.error(f"关闭{name}失败：{err}")
|
||
converged = False
|
||
if not converged and record_failure:
|
||
all_converged = False
|
||
return converged
|
||
|
||
await run_step("图片代理安全日志合并器", close_image_proxy_block_log_coalescer)
|
||
await run_step(
|
||
"模块",
|
||
lambda: _call_existing_singleton(ModuleManager, "shutdown"),
|
||
offload=True,
|
||
)
|
||
await run_step(
|
||
"事件消费",
|
||
lambda: _call_existing_singleton(EventManager, "stop_async"),
|
||
)
|
||
await run_step("浏览器会话", close_browser_sessions, offload=True)
|
||
await run_step("托管资源", stop_managed_resources)
|
||
await run_step(
|
||
"DoH服务",
|
||
stop_doh_composition,
|
||
offload=True,
|
||
)
|
||
await run_step(
|
||
"线程池",
|
||
lambda: _call_existing_singleton(ThreadHelper, "shutdown"),
|
||
offload=True,
|
||
)
|
||
# Web Agent 的取消 finally 可能还要写入最终展示快照，必须先完成任务收尾，再关闭写入准入。
|
||
web_agent_drained = await run_step(
|
||
"Web Agent后台任务",
|
||
shutdown_web_agent_background_tasks,
|
||
record_failure=False,
|
||
)
|
||
if not web_agent_drained:
|
||
web_agent_drained = await run_step("Web Agent后台任务收尾", wait_web_agent_background_tasks)
|
||
if web_agent_drained:
|
||
try:
|
||
persistence = get_configured_agent_chat_persistence()
|
||
except RuntimeError:
|
||
persistence = None
|
||
if persistence is None:
|
||
persistence_drained = True
|
||
else:
|
||
await run_step(
|
||
"Agent会话持久化准入",
|
||
persistence.begin_shutdown,
|
||
)
|
||
persistence_drained = await run_step(
|
||
"Agent会话持久化",
|
||
persistence.shutdown,
|
||
)
|
||
else:
|
||
persistence_drained = False
|
||
all_converged = False
|
||
logger.error("Web Agent任务未完成收尾，跳过持久化和数据库关闭以保护活动事务")
|
||
if persistence_drained:
|
||
await run_step("数据库任务", stop_database_runtime)
|
||
await run_step("前端服务", stop_frontend, offload=True)
|
||
await run_step("临时文件", clear_temp, offload=True)
|
||
if persistence_drained:
|
||
if not database_runtime_active():
|
||
await run_step("模块 Provider", reset_module_providers)
|
||
await run_step("数据库连接", close_database)
|
||
else:
|
||
all_converged = False
|
||
logger.error(
|
||
"数据库任务未收敛，保留全部 Provider 和数据库连接以供诊断与重试"
|
||
)
|
||
# Redis 连接池由全部缓存共享，前面的 Agent 收尾、数据库任务和 Provider 撤销都可能读写缓存；
|
||
# 放在最后关闭，避免关闭后又被自动重连且再无步骤收口。
|
||
await run_step(
|
||
"Redis缓存连接",
|
||
lambda: _call_existing_singleton(get_cache_composition().sync_redis_owner, "close"),
|
||
offload=True,
|
||
)
|
||
await run_step(
|
||
"异步Redis缓存连接",
|
||
lambda: _call_existing_singleton(get_cache_composition().async_redis_owner, "close"),
|
||
)
|
||
return all_converged
|
||
|
||
|
||
async def _initialize_modules() -> HostRuntime:
|
||
"""
|
||
构造模块服务并返回本次 lifespan 唯一的类型化 HostRuntime。
|
||
"""
|
||
if not ThreadHelper().reopen():
|
||
raise RuntimeError("上一应用生命周期的共享线程池尚未收敛")
|
||
configure_application_network_ports()
|
||
database_runtime = await start_database_runtime()
|
||
try:
|
||
configuration = await compose_configuration(
|
||
executor=database_runtime.worker,
|
||
settings=legacy_settings,
|
||
)
|
||
except BaseException:
|
||
try:
|
||
await stop_database_runtime()
|
||
# 清理失败不能掩盖原始启动异常。
|
||
except Exception as cleanup_error: # noqa: BLE001
|
||
logger.error(f"启动失败后的数据库任务清理失败：{cleanup_error}")
|
||
raise
|
||
publish_configuration(configuration, legacy_settings)
|
||
classification = await compose_classification(
|
||
executor=database_runtime.worker,
|
||
settings=legacy_settings,
|
||
system_config=configuration.system_config,
|
||
extra_fields_provider=_plugin_classification_fields,
|
||
extension_facts_provider=_plugin_classification_facts,
|
||
enrichment=compose_classification_enrichment(
|
||
module_manager=_existing_module_manager,
|
||
plugin_manager=get_existing_plugin_manager,
|
||
),
|
||
)
|
||
database_services = compose_database_services(
|
||
runtime=database_runtime,
|
||
system_config=configuration.system_config,
|
||
)
|
||
system_config = configuration.system_config
|
||
workflow_query = database_services.workflow_query
|
||
runtime_dependencies = compose_runtime_dependencies()
|
||
agent_composition = compose_agent(
|
||
runtime=database_runtime,
|
||
system_config=system_config,
|
||
dependencies=runtime_dependencies,
|
||
tasks=get_task_registry(),
|
||
)
|
||
if agent_composition.data.invocations is not None:
|
||
await database_runtime.worker.run(agent_composition.data.invocations.recover_running)
|
||
security_composition = configure_security_services()
|
||
runtime_composition = compose_runtime(
|
||
RuntimeInputs(
|
||
configuration=configuration,
|
||
database=database_services,
|
||
agent=agent_composition,
|
||
authentication=security_composition,
|
||
classification=classification.runtime,
|
||
classification_execution=classification.execution,
|
||
dependencies=runtime_dependencies,
|
||
tasks=get_task_registry(),
|
||
)
|
||
)
|
||
host_runtime = runtime_composition.runtime
|
||
publish_database_services(database_services)
|
||
publish_runtime(runtime_composition)
|
||
configure_runtime_data_providers()
|
||
configure_server_services(workflow_query, runtime_dependencies.subscription)
|
||
configure_workflow_execution_composition()
|
||
configure_outbox_dispatcher(build_outbox_dispatcher, submit=ThreadHelper().submit)
|
||
configure_transfer_retry_config(
|
||
lambda: TransferRetryConfig(
|
||
max_failed_retries=get_runtime_setting("TRANSFER_MAX_FAILED_RETRIES"),
|
||
)
|
||
)
|
||
configure_database()
|
||
publish_agent_services(
|
||
agent_composition,
|
||
data_context_registrar=configure_agent_data_context,
|
||
)
|
||
from app.agent.tools.manager import moviepilot_tool_manager
|
||
|
||
moviepilot_tool_manager.set_data_context(agent_composition.data)
|
||
configure_scheduler_agent_tasks(agent_composition.tasks)
|
||
# 托管资源只在这里装配声明与 adapter，具体资源仍由首个消费者显式激活。
|
||
init_managed_resources()
|
||
# 应用服务不反向依赖 Chain，由启动组合层注入壁纸来源。
|
||
configure_wallpaper_services()
|
||
# 目录选择和多级路径生成复用同一活动分类策略快照。
|
||
configure_directory_classification_service(classification.runtime)
|
||
# Chain 无参兼容入口由组合根明确提供依赖上下文；测试和新代码可直接注入替代上下文。
|
||
configure_chain_runtime_context(
|
||
dependencies=runtime_dependencies,
|
||
system_config=system_config,
|
||
configuration=configuration.runtime.chain,
|
||
classification_service=classification.execution,
|
||
system_service=host_runtime.system,
|
||
)
|
||
# 认证访问层不反向依赖数据库实现，由启动组合层注入载荷提供器。
|
||
configure_security_access()
|
||
# 初始化加密 DNS 解析能力。
|
||
configure_doh_composition()
|
||
# 站点管理
|
||
sites_helper = SitesHelper()
|
||
# 启动器已在进程拉起前应用待安装资源；这里仅注册当前版本读取器。
|
||
configure_site_resource_versions(
|
||
lambda: (sites_helper.auth_version, sites_helper.indexer_version)
|
||
)
|
||
# 用户认证
|
||
user_auth()
|
||
# 事件错误通知由启动组合层接入消息服务。
|
||
EventManager().set_error_notifier(notify_event_error)
|
||
# WebAgent 事件监听由组合根统一装配，HTTP 请求只管理自己的队列。
|
||
EventManager().add_event_listener(
|
||
EventType.NoticeMessage,
|
||
dispatch_web_agent_message_event,
|
||
)
|
||
# 宿主类处理器在启动层显式登记，事件总线不再兜底 owner_class()。
|
||
configure_host_event_handler_resolver()
|
||
# 配置 owner 按各自生命周期读取当前实例，惰性资源不会因事件而物化。
|
||
configure_config_reload_event_handler_resolver()
|
||
# 加载模块
|
||
ModuleManager()
|
||
# 启动事件消费
|
||
EventManager().start()
|
||
# 初始化共享服务端状态
|
||
await MoviePilotServerHelper.async_init_plugin_report()
|
||
await MoviePilotServerHelper.async_init_subscribe_report()
|
||
MoviePilotServerHelper.get_user_uuid()
|
||
MoviePilotServerHelper.get_github_user()
|
||
# 初始化AI智能体
|
||
await init_agent()
|
||
# 启动前端服务
|
||
start_frontend()
|
||
# 检查认证状态
|
||
check_auth()
|
||
return host_runtime
|
||
|
||
|
||
async def init_modules() -> HostRuntime:
|
||
"""
|
||
启动模块服务，并在构造中途失败时回收尚未发布的资源 owner。
|
||
"""
|
||
try:
|
||
return await _initialize_modules()
|
||
except BaseException:
|
||
try:
|
||
if stop_message() is False:
|
||
logger.error("模块启动失败后的消息资源未完全收敛")
|
||
# 清理失败不能掩盖原始启动异常。
|
||
except Exception as cleanup_error: # noqa: BLE001
|
||
logger.error(f"模块启动失败后的消息资源清理失败：{cleanup_error}")
|
||
try:
|
||
if await stop_modules() is False:
|
||
logger.error("模块启动失败后的资源 owner 未完全收敛")
|
||
# 清理失败不能掩盖原始启动异常。
|
||
except Exception as cleanup_error: # noqa: BLE001
|
||
logger.error(f"模块启动失败后的统一资源清理失败：{cleanup_error}")
|
||
raise
|