mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 11:27:43 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
357 lines
13 KiB
Python
357 lines
13 KiB
Python
"""
|
|
Chain 无参兼容入口与跨层依赖的宿主组合根。
|
|
"""
|
|
|
|
from collections.abc import Mapping
|
|
from functools import partial
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional, cast
|
|
|
|
from app.adapters.external.server import MoviePilotServerHelper
|
|
from app.adapters.system.host import SystemUtils
|
|
from app.application.chain.context import (
|
|
ChainRuntimeContext,
|
|
configure_chain_runtime_context_provider,
|
|
)
|
|
from app.application.classification.execution import ClassificationExecutionService
|
|
from app.application.classification.reference import ClassificationCategoryResolver
|
|
from app.application.classification.runtime import ClassificationRuntime
|
|
from app.application.configuration import ChainRuntimeConfig
|
|
from app.application.directory import (
|
|
configure_directory_classification_resolver,
|
|
reset_directory_classification_resolver,
|
|
)
|
|
from app.application.image import (
|
|
configure_wallpaper_providers,
|
|
reset_wallpaper_providers,
|
|
)
|
|
from app.application.subscription.activity import SubscriptionActivityService
|
|
from app.application.system import SystemService
|
|
from app.chain._recognition import (
|
|
RecognitionSharePort,
|
|
configure_recognition_share_port,
|
|
reset_recognition_share_port,
|
|
)
|
|
from app.chain.mediaserver import MediaServerChain
|
|
from app.chain.subscribe.notify import (
|
|
SubscriptionSharePort,
|
|
configure_subscription_share_port,
|
|
reset_subscription_share_port,
|
|
)
|
|
from app.chain.tmdb import TmdbChain
|
|
from app.chain.transfer.filter import (
|
|
NetworkFilesystemPort,
|
|
configure_network_filesystem_port,
|
|
reset_network_filesystem_port,
|
|
)
|
|
from app.db.adapters.chain import TransactionalChainDurableEventWriter
|
|
from app.db.adapters.download import TransactionalDownloadFailureRepository
|
|
from app.db.adapters.mediaserver import TransactionalMediaServerRepository
|
|
from app.db.adapters.subscriptionactivity import TransactionalSubscriptionActivityRepository
|
|
from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository
|
|
from app.db.oper.message import MessageOper
|
|
from app.db.oper.systemconfig import SystemConfigOper
|
|
from app.db.session import SessionFactory, async_session_scope
|
|
from app.domain.context import MediaInfo, MusicInfo
|
|
from app.domain.meta.metabase import MetaBase
|
|
from app.runtime.cache import AsyncFileCache, FileCache
|
|
from app.runtime.events import EventManager
|
|
from app.runtime.extensions.module.dispatcher import ModuleInvocationDispatcher
|
|
from app.runtime.extensions.module.manager import ModuleManager
|
|
from app.runtime.extensions.plugin.manager import PluginManager
|
|
from app.runtime.stop import runtime_stop_state
|
|
from app.schemas.common import JsonData
|
|
from app.schemas.mediaserver import MediaServerField, MediaServerQueryOptions
|
|
from app.schemas.types import MediaType
|
|
from app.startup.composition.database import build_transactional_user_repository
|
|
from app.startup.composition.runtime import RuntimeDependencies
|
|
from app.startup.composition.subscription import (
|
|
delete_subscribe_scope,
|
|
rule_group_mutation_scope,
|
|
site_reference_mutation_scope,
|
|
subscription_completion_scope,
|
|
subscription_mutation_scope,
|
|
sync_delete_subscribe_scope,
|
|
sync_subscription_mutation_scope,
|
|
)
|
|
|
|
_event_manager_factory: Callable[[], EventManager] = EventManager
|
|
|
|
|
|
def execute_legacy_transfer_command(**kwargs: Any) -> Any:
|
|
"""
|
|
将旧 Chain 整理接口延迟转发到 TransferChain 的唯一持久化命令。
|
|
"""
|
|
from app.chain.transfer.facade import TransferChain
|
|
|
|
return TransferChain().execute_legacy_transfer_command(**kwargs)
|
|
|
|
|
|
def build_chain_runtime_context(
|
|
*,
|
|
dependencies: RuntimeDependencies,
|
|
system_config: SystemConfigOper,
|
|
configuration: Callable[[], ChainRuntimeConfig],
|
|
classification_service: ClassificationExecutionService,
|
|
system_service: SystemService,
|
|
) -> ChainRuntimeContext:
|
|
"""
|
|
创建 Chain 无参兼容入口共享的运行时对象与数据端口。
|
|
"""
|
|
return ChainRuntimeContext(
|
|
module_manager=ModuleManager(),
|
|
plugin_manager=PluginManager(),
|
|
event_manager=_event_manager_factory(),
|
|
message_oper=MessageOper(),
|
|
message_helper=dependencies.message_helper,
|
|
file_cache=FileCache(),
|
|
async_file_cache=AsyncFileCache(),
|
|
message_queue=dependencies.message_queue,
|
|
module_dispatcher_factory=ModuleInvocationDispatcher,
|
|
site_repository=dependencies.site,
|
|
subscription_activity=SubscriptionActivityService(TransactionalSubscriptionActivityRepository(SessionFactory, async_session_scope)),
|
|
subscription_repository=dependencies.subscription,
|
|
subscription_search_repository=dependencies.subscription_search,
|
|
subscription_mutation_scope=subscription_mutation_scope,
|
|
sync_subscription_mutation_scope=sync_subscription_mutation_scope,
|
|
subscription_delete_scope=delete_subscribe_scope,
|
|
sync_subscription_delete_scope=sync_delete_subscribe_scope,
|
|
subscription_completion_scope=subscription_completion_scope,
|
|
rule_group_mutation_scope=partial(
|
|
rule_group_mutation_scope,
|
|
system_config.publish_many,
|
|
),
|
|
site_reference_mutation_scope=partial(
|
|
site_reference_mutation_scope,
|
|
system_config.publish_many,
|
|
),
|
|
download_history_repository=dependencies.download_history,
|
|
transfer_history_repository=dependencies.transfer_history,
|
|
transfer_admission_repository=TransactionalTransferAdmissionRepository(SessionFactory),
|
|
transfer_execution_repository=dependencies.transfer_execution,
|
|
media_server_repository=TransactionalMediaServerRepository(SessionFactory),
|
|
download_failure_repository=TransactionalDownloadFailureRepository(SessionFactory),
|
|
user_repository=build_transactional_user_repository(),
|
|
classification_service=classification_service,
|
|
legacy_transfer_command=execute_legacy_transfer_command,
|
|
configuration=configuration(),
|
|
configuration_provider=configuration,
|
|
durable_event_writer=TransactionalChainDurableEventWriter(SessionFactory),
|
|
stop_state=runtime_stop_state,
|
|
system_service=system_service,
|
|
)
|
|
|
|
|
|
def configure_wallpaper_services() -> None:
|
|
"""
|
|
把需要 Chain 编排的壁纸来源注入图片服务。
|
|
"""
|
|
configure_wallpaper_providers(
|
|
tmdb_wallpaper=lambda: TmdbChain().get_random_wallpager(),
|
|
tmdb_wallpapers=lambda count: TmdbChain().get_trending_wallpapers(count),
|
|
mediaserver_wallpaper=lambda: MediaServerChain().get_latest_wallpaper(
|
|
query=MediaServerQueryOptions(fields=frozenset({MediaServerField.IMAGES})),
|
|
),
|
|
mediaserver_wallpapers=lambda count: MediaServerChain().get_latest_wallpapers(
|
|
count=count,
|
|
query=MediaServerQueryOptions(fields=frozenset({MediaServerField.IMAGES})),
|
|
),
|
|
)
|
|
|
|
|
|
def configure_directory_classification_service(
|
|
runtime: ClassificationRuntime,
|
|
) -> None:
|
|
"""
|
|
把活动分类策略解析能力装配给目录选择与路径生成服务。
|
|
"""
|
|
configure_directory_classification_resolver(
|
|
ClassificationCategoryResolver(runtime)
|
|
)
|
|
|
|
|
|
def configure_chain_runtime_context(
|
|
*,
|
|
dependencies: RuntimeDependencies,
|
|
system_config: SystemConfigOper,
|
|
configuration: Callable[[], ChainRuntimeConfig],
|
|
classification_service: ClassificationExecutionService,
|
|
system_service: SystemService,
|
|
) -> None:
|
|
"""
|
|
登记按需构造的 Chain 上下文，保持无参 Chain 的插件兼容合同。
|
|
"""
|
|
configure_chain_runtime_context_provider(
|
|
lambda: build_chain_runtime_context(
|
|
dependencies=dependencies,
|
|
system_config=system_config,
|
|
configuration=configuration,
|
|
classification_service=classification_service,
|
|
system_service=system_service,
|
|
)
|
|
)
|
|
|
|
|
|
def reset_chain_services() -> None:
|
|
"""
|
|
按发布逆序撤销 Chain 无参上下文和壁纸来源。
|
|
"""
|
|
configure_chain_runtime_context_provider(None)
|
|
reset_directory_classification_resolver()
|
|
reset_wallpaper_providers()
|
|
|
|
|
|
class _RecognitionShareAdapter:
|
|
"""
|
|
把 MoviePilot Server 共享识别能力适配为 Chain 窄端口。
|
|
"""
|
|
|
|
def report_recognize_share(
|
|
self,
|
|
meta: Optional[MetaBase],
|
|
mediainfo: Optional[MediaInfo | MusicInfo],
|
|
keyword_meta: Optional[MetaBase] = None,
|
|
) -> bool:
|
|
"""
|
|
同步上报共享识别结果。
|
|
"""
|
|
return bool(MoviePilotServerHelper.report_recognize_share(
|
|
meta=meta, mediainfo=mediainfo, keyword_meta=keyword_meta
|
|
))
|
|
|
|
async def async_report_recognize_share(
|
|
self,
|
|
meta: Optional[MetaBase],
|
|
mediainfo: Optional[MediaInfo | MusicInfo],
|
|
keyword_meta: Optional[MetaBase] = None,
|
|
) -> bool:
|
|
"""
|
|
异步上报共享识别结果。
|
|
"""
|
|
return bool(await MoviePilotServerHelper.async_report_recognize_share(
|
|
meta=meta, mediainfo=mediainfo, keyword_meta=keyword_meta
|
|
))
|
|
|
|
def query_recognize_share(
|
|
self,
|
|
meta: Optional[MetaBase],
|
|
mtype: Optional[MediaType] = None,
|
|
keyword_meta: Optional[MetaBase] = None,
|
|
music_type: Optional[str] = None,
|
|
) -> Optional[dict[str, Any]]:
|
|
"""
|
|
同步查询共享识别结果。
|
|
"""
|
|
result = MoviePilotServerHelper.query_recognize_share(
|
|
meta=meta,
|
|
mtype=mtype,
|
|
keyword_meta=keyword_meta,
|
|
**({"music_type": music_type} if music_type is not None else {}),
|
|
)
|
|
return result
|
|
|
|
async def async_query_recognize_share(
|
|
self,
|
|
meta: Optional[MetaBase],
|
|
mtype: Optional[MediaType] = None,
|
|
keyword_meta: Optional[MetaBase] = None,
|
|
music_type: Optional[str] = None,
|
|
) -> Optional[dict[str, Any]]:
|
|
"""
|
|
异步查询共享识别结果。
|
|
"""
|
|
result = await MoviePilotServerHelper.async_query_recognize_share(
|
|
meta=meta,
|
|
mtype=mtype,
|
|
keyword_meta=keyword_meta,
|
|
**({"music_type": music_type} if music_type is not None else {}),
|
|
)
|
|
return result
|
|
|
|
def to_recognize_params(
|
|
self,
|
|
item: Optional[dict[str, Any]],
|
|
) -> Optional[dict[str, Any]]:
|
|
"""
|
|
把服务端结果转换为本地识别参数。
|
|
"""
|
|
return MoviePilotServerHelper.to_recognize_params(item)
|
|
|
|
|
|
class _SubscriptionShareAdapter:
|
|
"""
|
|
把 MoviePilot Server 订阅共享能力适配为 Chain 窄端口。
|
|
"""
|
|
|
|
def report_added(self, payload: dict[str, Any]) -> bool:
|
|
"""
|
|
同步上报新增订阅统计。
|
|
"""
|
|
return bool(MoviePilotServerHelper.sub_reg_durable(payload))
|
|
|
|
async def async_report_added(self, payload: dict[str, Any]) -> bool:
|
|
"""
|
|
异步上报新增订阅统计。
|
|
"""
|
|
return bool(await MoviePilotServerHelper.async_sub_reg_durable(payload))
|
|
|
|
def list_shares(self) -> list[dict[str, Any]]:
|
|
"""
|
|
读取当前用户可见的订阅分享。
|
|
"""
|
|
return MoviePilotServerHelper.get_subscribe_shares()
|
|
|
|
def report_completed(self, payload: Mapping[str, JsonData]) -> bool:
|
|
"""
|
|
同步上报订阅完成统计。
|
|
"""
|
|
return bool(MoviePilotServerHelper.sub_done_durable(dict(payload)))
|
|
|
|
|
|
class _NetworkFilesystemAdapter:
|
|
"""
|
|
把宿主文件系统探测能力适配为整理 Chain 窄端口。
|
|
"""
|
|
|
|
def is_network_filesystem(
|
|
self,
|
|
path: Path,
|
|
*,
|
|
include_local_fuse: bool = False,
|
|
) -> bool:
|
|
"""
|
|
判断路径是否位于网络或指定的本地 FUSE 文件系统。
|
|
"""
|
|
return bool(SystemUtils.is_network_filesystem(
|
|
path, include_local_fuse=include_local_fuse
|
|
))
|
|
|
|
|
|
def configure_chain_port_composition() -> None:
|
|
"""
|
|
原子装配 Chain 的共享服务与文件系统端口。
|
|
"""
|
|
reset_chain_port_composition()
|
|
try:
|
|
configure_recognition_share_port(
|
|
cast(RecognitionSharePort, _RecognitionShareAdapter())
|
|
)
|
|
configure_subscription_share_port(
|
|
cast(SubscriptionSharePort, _SubscriptionShareAdapter())
|
|
)
|
|
configure_network_filesystem_port(
|
|
cast(NetworkFilesystemPort, _NetworkFilesystemAdapter())
|
|
)
|
|
except Exception:
|
|
reset_chain_port_composition()
|
|
raise
|
|
|
|
|
|
def reset_chain_port_composition() -> None:
|
|
"""
|
|
释放 Chain 技术端口，支持重复 lifespan 与启动失败回滚。
|
|
"""
|
|
reset_network_filesystem_port()
|
|
reset_subscription_share_port()
|
|
reset_recognition_share_port()
|