diff --git a/app/agent/memory/__init__.py b/app/agent/memory/__init__.py index 8775e34b8..f72dfc425 100644 --- a/app/agent/memory/__init__.py +++ b/app/agent/memory/__init__.py @@ -9,7 +9,7 @@ from langchain_core.messages import BaseMessage, messages_from_dict, messages_to from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() -from app.application.agentdata import AgentChatPort as AgentChatOper +from app.application.agentdata import get_agent_chat_port from app.application.messaging.chat import ( get_configured_agent_chat_persistence, get_configured_agent_chat_service, @@ -86,9 +86,9 @@ class MemoryManager: return memory.messages try: - chat = AgentChatOper().get(session_id=session_id, user_id=user_id) + chat = get_agent_chat_port().get(session_id=session_id, user_id=user_id) if not chat: - chat = AgentChatOper().get(session_id=session_id) + chat = get_agent_chat_port().get(session_id=session_id) except Exception as e: logger.debug(f"读取持久化Agent会话失败: {e}") return [] @@ -161,7 +161,7 @@ class MemoryManager: # 更新内存缓存 self.save_memory(memory) try: - AgentChatOper().save_agent_messages( + get_agent_chat_port().save_agent_messages( session_id=session_id, user_id=user_id, messages=messages_to_dict(messages), diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index e87df0b06..97e41e24e 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -79,8 +79,8 @@ from app.application.plugin.runtime import get_plugin_manager def _get_plugin_tools_revision() -> int: """读取插件工具目录修订号,避免 Agent 编排依赖具体管理器类型。""" return get_plugin_manager().get_plugin_agent_tools_revision() -from app.application.agentdata import AgentTaskPort as AgentTaskOper -from app.application.agentdata import UserPort as UserOper +from app.application.agentdata import get_agent_task_port +from app.application.agentdata import get_agent_user_port from app.application.messaging.chat import ( get_configured_agent_chat_service, get_configured_agent_chat_persistence, @@ -977,7 +977,7 @@ class MoviePilotAgent: if not self.username: return False try: - user = await UserOper().async_get_by_name(self.username) + user = await get_agent_user_port().async_get_by_name(self.username) except Exception as e: logger.error(f"检查 Agent 用户管理员身份失败: {e}") return False @@ -3514,7 +3514,7 @@ class AgentManager: """ if not settings.AI_AGENT_ENABLE: return False, "AI Agent 未启用" - oper = AgentTaskOper() + oper = get_agent_task_port() task = oper.get(task_id) if not task or not task.enabled: return False, "Agent 定时任务不存在或已停用" diff --git a/app/agent/tools/impl/_filter_rule_utils.py b/app/agent/tools/impl/_filter_rule_utils.py index 6ded2fbca..3f128b1ce 100644 --- a/app/agent/tools/impl/_filter_rule_utils.py +++ b/app/agent/tools/impl/_filter_rule_utils.py @@ -5,7 +5,7 @@ import re from typing import Any, Dict, Iterable, Optional from app.runtime.events import eventmanager -from app.application.agentdata import SubscribePort as SubscribeOper +from app.application.agentdata import get_agent_subscribe_port from app.application.configuration import get_configured_system_config from app.application.rules import RuleHelper from app.application.rules import RuleParser @@ -284,7 +284,7 @@ async def collect_rule_group_usages( continue ensure_usage(name)["used_in_global_best_version"] = True - subscribes = await SubscribeOper().async_list() + subscribes = await get_agent_subscribe_port().async_list() for subscribe in subscribes: filter_groups = subscribe.filter_groups or [] for name in filter_groups: @@ -481,7 +481,7 @@ async def rename_rule_group_references(old_name: str, new_name: str) -> dict: await save_system_config(config_key, updated) changed["global_settings"][config_key.value] = updated - subscribe_oper = SubscribeOper() + subscribe_oper = get_agent_subscribe_port() subscribes = await subscribe_oper.async_list() for subscribe in subscribes: original = subscribe.filter_groups or [] @@ -519,7 +519,7 @@ async def remove_rule_group_references(group_name: str) -> dict: await save_system_config(config_key, updated) changed["global_settings"][config_key.value] = updated - subscribe_oper = SubscribeOper() + subscribe_oper = get_agent_subscribe_port() subscribes = await subscribe_oper.async_list() for subscribe in subscribes: original = subscribe.filter_groups or [] diff --git a/app/agent/tools/impl/add_download_tasks.py b/app/agent/tools/impl/add_download_tasks.py index c877e9c22..2d1e6d74a 100644 --- a/app/agent/tools/impl/add_download_tasks.py +++ b/app/agent/tools/impl/add_download_tasks.py @@ -17,7 +17,7 @@ from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() from app.domain.context import Context from app.domain.metainfo import MetaInfo -from app.application.agentdata import SitePort as SiteOper +from app.application.agentdata import get_agent_site_port from app.application.directory import DirectoryHelper, validate_download_save_path from app.runtime.log import logger from app.schemas.file import FileURI @@ -277,7 +277,7 @@ class AddDownloadTasksTool(MoviePilotTool): failed_messages.append(f"{torrent_input} 缺少站点名称") continue - siteinfo = await SiteOper().async_get_by_name(site_name) + siteinfo = await get_agent_site_port().async_get_by_name(site_name) if not siteinfo: failed_messages.append(f"{torrent_input} 未找到站点信息 {site_name}") continue diff --git a/app/agent/tools/impl/add_subscribe.py b/app/agent/tools/impl/add_subscribe.py index 987a61fa9..bc8e943fa 100644 --- a/app/agent/tools/impl/add_subscribe.py +++ b/app/agent/tools/impl/add_subscribe.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.subscribe import SubscribeChain -from app.application.agentdata import UserPort as UserOper +from app.application.agentdata import get_agent_user_port from app.runtime.log import logger from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType, NotificationChannel from app.domain.media import normalize_music_type @@ -154,7 +154,7 @@ class AddSubscribeTool(MoviePilotTool): mapped_username = await self.run_blocking( "db", - UserOper().get_name, + get_agent_user_port().get_name, **{key: self._user_id for key in binding_keys}, ) return mapped_username or resolved_username diff --git a/app/agent/tools/impl/create_agent_task.py b/app/agent/tools/impl/create_agent_task.py index ed2dafefa..6ffedb077 100644 --- a/app/agent/tools/impl/create_agent_task.py +++ b/app/agent/tools/impl/create_agent_task.py @@ -10,8 +10,8 @@ from app.agent.tools.tags import ToolTag from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() -from app.application.agentdata import AgentChatPort as AgentChatOper -from app.application.agentdata import AgentTaskPort as AgentTaskOper +from app.application.agentdata import get_agent_chat_port +from app.application.agentdata import get_agent_task_port from app.runtime.scheduling import TimerUtils @@ -115,11 +115,11 @@ class CreateAgentTaskTool(MoviePilotTool): timezone_name=settings.TZ, require_future=True, ) - chat = AgentChatOper().get( + chat = get_agent_chat_port().get( session_id=self._session_id, user_id=self._user_id, ) - task = AgentTaskOper().add( + task = get_agent_task_port().add( name=payload.name.strip(), content=payload.content.strip(), trigger_type=payload.trigger_type, @@ -133,7 +133,7 @@ class CreateAgentTaskTool(MoviePilotTool): original_chat_id=chat.original_chat_id if chat else None, ) next_run_at = update_agent_task_job(task.id) - return AgentTaskOper.to_dict( + return get_agent_task_port().to_dict( task, next_run_at=next_run_at, timezone=settings.TZ, diff --git a/app/agent/tools/impl/delete_agent_task.py b/app/agent/tools/impl/delete_agent_task.py index aa8ef94ae..0c5396a1b 100644 --- a/app/agent/tools/impl/delete_agent_task.py +++ b/app/agent/tools/impl/delete_agent_task.py @@ -4,7 +4,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import AgentTaskPort as AgentTaskOper +from app.application.agentdata import get_agent_task_port class DeleteAgentTaskInput(BaseModel): @@ -33,7 +33,7 @@ class DeleteAgentTaskTool(MoviePilotTool): """删除当前用户的任务并移除运行时调度。""" from app.application.scheduling import remove_agent_task_job - deleted = AgentTaskOper().delete( + deleted = get_agent_task_port().delete( task_id=task_id, user_id=str(self._user_id), ) diff --git a/app/agent/tools/impl/delete_download_history.py b/app/agent/tools/impl/delete_download_history.py index adc88ea98..cf6d1159a 100644 --- a/app/agent/tools/impl/delete_download_history.py +++ b/app/agent/tools/impl/delete_download_history.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import DownloadHistoryPort as DownloadHistoryOper +from app.application.agentdata import get_agent_download_history_port from app.runtime.log import logger @@ -37,7 +37,7 @@ class DeleteDownloadHistoryTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: history_id={history_id}") try: - await DownloadHistoryOper().async_delete_history(history_id) + await get_agent_download_history_port().async_delete_history(history_id) return f"下载历史记录 ID: {history_id} 已成功删除" except Exception as e: logger.error(f"删除下载历史记录失败: {e}", exc_info=True) diff --git a/app/agent/tools/impl/delete_transfer_history.py b/app/agent/tools/impl/delete_transfer_history.py index b8506ea6d..50e58810e 100644 --- a/app/agent/tools/impl/delete_transfer_history.py +++ b/app/agent/tools/impl/delete_transfer_history.py @@ -8,7 +8,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.storage import StorageChain -from app.application.agentdata import TransferHistoryPort as TransferHistoryOper +from app.application.agentdata import get_agent_transfer_history_port from app.runtime.log import logger from app.schemas.workflow import FileItem @@ -53,7 +53,7 @@ class DeleteTransferHistoryTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: history_id={history_id}") try: - transferhis = TransferHistoryOper() + transferhis = get_agent_transfer_history_port() history = await transferhis.async_get(history_id) if not history: return f"错误:整理历史记录不存在,ID={history_id}" diff --git a/app/agent/tools/impl/query_agent_tasks.py b/app/agent/tools/impl/query_agent_tasks.py index 99a4a1269..517f93e78 100644 --- a/app/agent/tools/impl/query_agent_tasks.py +++ b/app/agent/tools/impl/query_agent_tasks.py @@ -8,7 +8,7 @@ from app.agent.tools.tags import ToolTag from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() -from app.application.agentdata import AgentTaskPort as AgentTaskOper +from app.application.agentdata import get_agent_task_port class QueryAgentTasksInput(BaseModel): @@ -52,7 +52,7 @@ class QueryAgentTasksTool(MoviePilotTool): """读取当前用户的任务及运行时下一次触发时间。""" from app.application.scheduling import get_agent_task_next_run - oper = AgentTaskOper() + oper = get_agent_task_port() if task_id: task = oper.get(task_id=task_id, user_id=str(self._user_id)) tasks = [task] if task else [] diff --git a/app/agent/tools/impl/query_download_tasks.py b/app/agent/tools/impl/query_download_tasks.py index e396ad1a4..ae45c251f 100644 --- a/app/agent/tools/impl/query_download_tasks.py +++ b/app/agent/tools/impl/query_download_tasks.py @@ -8,7 +8,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.download import DownloadChain -from app.application.agentdata import DownloadHistoryPort as DownloadHistoryOper +from app.application.agentdata import get_agent_download_history_port from app.runtime.log import logger from app.schemas.transfer import DownloaderTorrent from app.schemas.types import MUSIC_ENTITY_RECORDING, TorrentQueryStatus, media_type_to_agent @@ -143,7 +143,7 @@ class QueryDownloadTasksTool(MoviePilotTool): hashes = [torrent.hash for torrent in torrents if getattr(torrent, "hash", None)] if not hashes: return {} - return DownloadHistoryOper().get_by_hashes(hashes) + return get_agent_download_history_port().get_by_hashes(hashes) @classmethod def _query_downloads_sync( diff --git a/app/agent/tools/impl/query_plugin_data.py b/app/agent/tools/impl/query_plugin_data.py index 2eee84a62..242d52572 100644 --- a/app/agent/tools/impl/query_plugin_data.py +++ b/app/agent/tools/impl/query_plugin_data.py @@ -12,7 +12,7 @@ from app.agent.tools.impl._plugin_tool_utils import ( build_preview_payload, get_plugin_snapshot, ) -from app.application.agentdata import PluginDataPort as PluginDataOper +from app.application.agentdata import get_agent_plugin_data_port from app.runtime.log import logger @@ -73,7 +73,7 @@ class QueryPluginDataTool(MoviePilotTool): ensure_ascii=False, ) - plugin_data_oper = PluginDataOper() + plugin_data_oper = get_agent_plugin_data_port() if key: value = await plugin_data_oper.async_get_data(plugin_id, key) if value is None: diff --git a/app/agent/tools/impl/query_site_userdata.py b/app/agent/tools/impl/query_site_userdata.py index 9c295bb1b..81e1e2c57 100644 --- a/app/agent/tools/impl/query_site_userdata.py +++ b/app/agent/tools/impl/query_site_userdata.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import SitePort as SiteOper +from app.application.agentdata import get_agent_site_port from app.runtime.log import logger SITE_USERDATA_DETAIL_PREVIEW_LIMIT = 10 @@ -62,7 +62,7 @@ class QuerySiteUserdataTool(MoviePilotTool): ) try: - site_oper = SiteOper() + site_oper = get_agent_site_port() site = await site_oper.async_get(site_id) if not site: return json.dumps( diff --git a/app/agent/tools/impl/query_sites.py b/app/agent/tools/impl/query_sites.py index f45d8a24a..384b204d7 100644 --- a/app/agent/tools/impl/query_sites.py +++ b/app/agent/tools/impl/query_sites.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import SitePort as SiteOper +from app.application.agentdata import get_agent_site_port from app.runtime.log import logger @@ -58,7 +58,7 @@ class QuerySitesTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: status={status}, name={name}") try: is_admin = await self.is_admin_user() - site_oper = SiteOper() + site_oper = get_agent_site_port() # 获取所有站点(按优先级排序) sites = await site_oper.async_list() filtered_sites = [] diff --git a/app/agent/tools/impl/query_subscribe_history.py b/app/agent/tools/impl/query_subscribe_history.py index e4ab197d2..f7d8c322e 100644 --- a/app/agent/tools/impl/query_subscribe_history.py +++ b/app/agent/tools/impl/query_subscribe_history.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import SubscribeHistoryPort as SubscribeHistoryOper +from app.application.agentdata import get_agent_subscribe_history_port from app.runtime.log import logger from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaType, media_type_to_agent from app.domain.media import normalize_music_type @@ -106,7 +106,7 @@ class QuerySubscribeHistoryTool(MoviePilotTool): if MediaType.MUSIC.value not in requested_types: return "错误:music_type 仅能与 media_type='music' 或 'all' 一起使用" - subscribe_history_oper = SubscribeHistoryOper() + subscribe_history_oper = get_agent_subscribe_history_port() if name: # 有名称过滤时,获取足够多的记录在内存中过滤,不分页 fetch_count = 500 diff --git a/app/agent/tools/impl/query_subscribes.py b/app/agent/tools/impl/query_subscribes.py index 7ccfd7aa5..6a0fad952 100644 --- a/app/agent/tools/impl/query_subscribes.py +++ b/app/agent/tools/impl/query_subscribes.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import SubscribePort as SubscribeOper +from app.application.agentdata import get_agent_subscribe_port from app.runtime.log import logger from app.schemas.subscribe import Subscribe as SubscribeSchema from app.schemas.types import ( @@ -157,7 +157,7 @@ class QuerySubscribesTool(MoviePilotTool): if media_type not in ("all", "music"): return "错误:music_type 仅能与 media_type='music' 或 'all' 一起使用" - subscribe_oper = SubscribeOper() + subscribe_oper = get_agent_subscribe_port() subscribes = await subscribe_oper.async_list() filtered_subscribes = [] for sub in subscribes: diff --git a/app/agent/tools/impl/query_transfer_history.py b/app/agent/tools/impl/query_transfer_history.py index 1afa8932e..5316b6e84 100644 --- a/app/agent/tools/impl/query_transfer_history.py +++ b/app/agent/tools/impl/query_transfer_history.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import TransferHistoryPort as TransferHistoryOper +from app.application.agentdata import get_agent_transfer_history_port from app.runtime.log import logger from app.schemas.types import media_type_to_agent from app.foundation.text import cut as jieba_cut @@ -75,7 +75,7 @@ class QueryTransferHistoryTool(MoviePilotTool): # 每页固定 30 条,与工具说明保持一致,避免整理路径等字段撑大上下文。 count = 30 - transferhis = TransferHistoryOper() + transferhis = get_agent_transfer_history_port() # 处理标题搜索 if title: # 使用统一分词封装处理标题,便于替换底层实现。 diff --git a/app/agent/tools/impl/query_workflows.py b/app/agent/tools/impl/query_workflows.py index 93d0e9b5a..ee67fea83 100644 --- a/app/agent/tools/impl/query_workflows.py +++ b/app/agent/tools/impl/query_workflows.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import WorkflowPort as WorkflowOper +from app.application.agentdata import get_agent_workflow_port from app.runtime.log import logger @@ -54,7 +54,7 @@ class QueryWorkflowsTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: state={state}, name={name}, trigger_type={trigger_type}") try: - workflow_oper = WorkflowOper() + workflow_oper = get_agent_workflow_port() workflows = await workflow_oper.async_list() # 过滤工作流 diff --git a/app/agent/tools/impl/run_agent_task.py b/app/agent/tools/impl/run_agent_task.py index 60add013c..e7f4028b4 100644 --- a/app/agent/tools/impl/run_agent_task.py +++ b/app/agent/tools/impl/run_agent_task.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag -from app.application.agentdata import AgentTaskPort as AgentTaskOper +from app.application.agentdata import get_agent_task_port class RunAgentTaskInput(BaseModel): @@ -42,7 +42,7 @@ class RunAgentTaskTool(MoviePilotTool): def _get_task_state(self, task_id: int) -> tuple[str, Optional[str]]: """校验任务归属和状态,返回可执行性及任务名称。""" - task = AgentTaskOper().get( + task = get_agent_task_port().get( task_id=task_id, user_id=str(self._user_id), ) diff --git a/app/agent/tools/impl/search_subscribe.py b/app/agent/tools/impl/search_subscribe.py index 0eefa64b4..ddc810e4d 100644 --- a/app/agent/tools/impl/search_subscribe.py +++ b/app/agent/tools/impl/search_subscribe.py @@ -8,7 +8,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.subscribe import SubscribeChain -from app.application.agentdata import SubscribePort as SubscribeOper +from app.application.agentdata import get_agent_subscribe_port from app.runtime.log import logger from app.schemas.types import media_type_to_agent @@ -57,7 +57,7 @@ class SearchSubscribeTool(MoviePilotTool): try: # 先验证订阅是否存在 - subscribe_oper = SubscribeOper() + subscribe_oper = get_agent_subscribe_port() subscribe = await subscribe_oper.async_get(subscribe_id) if not subscribe: diff --git a/app/agent/tools/impl/test_site.py b/app/agent/tools/impl/test_site.py index 5a38d247b..4e19a8048 100644 --- a/app/agent/tools/impl/test_site.py +++ b/app/agent/tools/impl/test_site.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.site import SiteChain -from app.application.agentdata import SitePort as SiteOper +from app.application.agentdata import get_agent_site_port from app.runtime.log import logger @@ -33,7 +33,7 @@ class TestSiteTool(MoviePilotTool): @staticmethod def _test_site_sync(site_identifier: int) -> tuple[Optional[str], Optional[str], bool, str]: """在同步线程里执行站点联通测试,避免网络请求卡住事件循环。""" - site = SiteOper().get(site_identifier) + site = get_agent_site_port().get(site_identifier) if not site: return None, None, False, f"未找到站点:{site_identifier},请使用 query_sites 工具查询可用的站点" diff --git a/app/agent/tools/impl/update_agent_task.py b/app/agent/tools/impl/update_agent_task.py index d0f931f2d..b287f071b 100644 --- a/app/agent/tools/impl/update_agent_task.py +++ b/app/agent/tools/impl/update_agent_task.py @@ -10,7 +10,7 @@ from app.agent.tools.tags import ToolTag from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() -from app.application.agentdata import AgentTaskPort as AgentTaskOper +from app.application.agentdata import get_agent_task_port from app.runtime.scheduling import TimerUtils @@ -104,7 +104,7 @@ class UpdateAgentTaskTool(MoviePilotTool): """更新当前用户的任务并刷新运行时调度。""" from app.application.scheduling import update_agent_task_job - oper = AgentTaskOper() + oper = get_agent_task_port() task = oper.get(task_id=payload.task_id, user_id=str(self._user_id)) if not task: return None diff --git a/app/agent/tools/impl/update_site.py b/app/agent/tools/impl/update_site.py index 62e68f040..ad658ce0a 100644 --- a/app/agent/tools/impl/update_site.py +++ b/app/agent/tools/impl/update_site.py @@ -8,7 +8,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.runtime.events import eventmanager -from app.application.agentdata import SitePort as SiteOper +from app.application.agentdata import get_agent_site_port from app.runtime.log import logger from app.schemas.types import EventType from app.foundation import url as url_tools @@ -124,7 +124,7 @@ class UpdateSiteTool(MoviePilotTool): logger.info(f"执行工具: {self.name}, 参数: site_id={site_id}") try: - site_oper = SiteOper() + site_oper = get_agent_site_port() site = await site_oper.async_get(site_id) if not site: return json.dumps( diff --git a/app/agent/tools/impl/update_site_cookie.py b/app/agent/tools/impl/update_site_cookie.py index af31db387..cd7f8b31d 100644 --- a/app/agent/tools/impl/update_site_cookie.py +++ b/app/agent/tools/impl/update_site_cookie.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.site import SiteChain -from app.application.agentdata import SitePort as SiteOper +from app.application.agentdata import get_agent_site_port from app.runtime.log import logger @@ -59,7 +59,7 @@ class UpdateSiteCookieTool(MoviePilotTool): """ 在同步线程里执行站点登录和 Cookie 更新,避免网络登录阻塞协程。 """ - site = SiteOper().get(site_identifier) + site = get_agent_site_port().get(site_identifier) if not site: return None, False, f"未找到站点:{site_identifier},请使用 query_sites 工具查询可用的站点" diff --git a/app/application/agentdata.py b/app/application/agentdata.py index 0050ac3a3..6b475d2e5 100644 --- a/app/application/agentdata.py +++ b/app/application/agentdata.py @@ -125,3 +125,53 @@ def get_agent_data_ports() -> AgentDataPorts: if _ports is None: raise RuntimeError("Agent 数据端口尚未配置") return _ports + + +def get_agent_chat_port() -> Any: + """创建 Agent 会话数据端口实例。""" + return get_agent_data_ports().agent_chat() + + +def get_agent_task_port() -> Any: + """创建 Agent 定时任务数据端口实例。""" + return get_agent_data_ports().agent_task() + + +def get_agent_user_port() -> Any: + """创建 Agent 用户数据端口实例。""" + return get_agent_data_ports().user() + + +def get_agent_site_port() -> Any: + """创建 Agent 站点数据端口实例。""" + return get_agent_data_ports().site() + + +def get_agent_subscribe_port() -> Any: + """创建 Agent 订阅数据端口实例。""" + return get_agent_data_ports().subscribe() + + +def get_agent_subscribe_history_port() -> Any: + """创建 Agent 订阅历史数据端口实例。""" + return get_agent_data_ports().subscribe_history() + + +def get_agent_transfer_history_port() -> Any: + """创建 Agent 整理历史数据端口实例。""" + return get_agent_data_ports().transfer_history() + + +def get_agent_download_history_port() -> Any: + """创建 Agent 下载历史数据端口实例。""" + return get_agent_data_ports().download_history() + + +def get_agent_workflow_port() -> Any: + """创建 Agent 工作流数据端口实例。""" + return get_agent_data_ports().workflow() + + +def get_agent_plugin_data_port() -> Any: + """创建 Agent 插件数据端口实例。""" + return get_agent_data_ports().plugin_data() diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index e76bfb72c..e4dd4c0c4 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围:宿主后端;排除 `app/plugins/**` 运行时插件副本 > 规范优先级:`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档:`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度:阶段 0~6 的宿主架构能力已完成收口;API/Application 公共复杂度基线已清零,启动组合根的 SystemConfigOper 构造点已由 14 降至 1;API 进程内后台任务已完成首批统一登记,插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零,插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线;阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义;阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权;2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零,并补齐可执行 fanout 与下载器文件 DTO 边界;阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox,宿主不再调用裸线程统计入口;阶段 4 已统一七种消息渠道的宿主回环与后台执行边界;阶段 5 已补齐事件窗口聚合任务的生命周期所有权;阶段 6 已统一插件文件操作的取消完成语义;阶段 7 已统一插件协程补偿的终态等待;阶段 8 已统一宿主同步函数的异步线程池入口;阶段 9 已统一工作流运行时的宿主获取路径;阶段 10 已统一模块、插件与调度运行时的显式 getter 调用;阶段 11 已清除系统配置 getter 的 Oper 形别名;阶段 12 已完成工作流域的显式 Chain 数据端口迁移;阶段 13 已收口用户、交互与消息链的数据端口;阶段 14 已收口音乐订阅数据端口;阶段 15 已收口站点数据端口;阶段 16 已收口媒体服务器数据端口;阶段 17 已收口下载数据端口;阶段 18 已收口主订阅数据端口;阶段 19 已收口整理数据端口。 +> 实施进度:阶段 0~6 的宿主架构能力已完成收口;API/Application 公共复杂度基线已清零,启动组合根的 SystemConfigOper 构造点已由 14 降至 1;API 进程内后台任务已完成首批统一登记,插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零,插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线;阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义;阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权;2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零,并补齐可执行 fanout 与下载器文件 DTO 边界;阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox,宿主不再调用裸线程统计入口;阶段 4 已统一七种消息渠道的宿主回环与后台执行边界;阶段 5 已补齐事件窗口聚合任务的生命周期所有权;阶段 6 已统一插件文件操作的取消完成语义;阶段 7 已统一插件协程补偿的终态等待;阶段 8 已统一宿主同步函数的异步线程池入口;阶段 9 已统一工作流运行时的宿主获取路径;阶段 10 已统一模块、插件与调度运行时的显式 getter 调用;阶段 11 已清除系统配置 getter 的 Oper 形别名;阶段 12 已完成工作流域的显式 Chain 数据端口迁移;阶段 13 已收口用户、交互与消息链的数据端口;阶段 14 已收口音乐订阅数据端口;阶段 15 已收口站点数据端口;阶段 16 已收口媒体服务器数据端口;阶段 17 已收口下载数据端口;阶段 18 已收口主订阅数据端口;阶段 19 已收口整理数据端口;阶段 20 已收口 Agent 数据端口。 ## 当前复核结论(2026-08-24) @@ -212,6 +212,15 @@ - 兼容边界不变:三个 DB Oper、Proxy 类、整理公开入口、队列和生命周期、历史字段、事件 payload 以及 V2/V3 插件调用方式均未修改。 +### 长期整改阶段 20:Agent 数据端口收口(2026-08-24) + +- Agent 编排、会话记忆和 22 个工具实现原先把十种 `AgentDataPort` 代理重新别名为数据库 Oper,形成与 + Chain 命名 getter 不同的第二套数据端口用法;现在统一调用 `get_agent_*_port()`。 +- Agent 测试接缝改为替换 getter 返回的端口实例,架构门禁遍历 `app/agent/**`,禁止生产模块重新导入 + 十个兼容 Port 代理。 +- 兼容边界不变:`AgentDataPorts` 注册表、十个旧 Port 代理类、DB Oper、Agent/工具公开参数和返回值、 + 权限判断以及 V2/V3 插件调用方式均未修改。 + ### 总体判断 当前架构总体合理,已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**: diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 34654e36c..c383e886c 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -136,6 +136,10 @@ Chain consumers use the named `get_chain_*_port()` functions from `app/application/chain/data.py`; they must not alias migration-time `*PortProxy` classes back to database Oper names. Those proxy classes remain compatibility boundaries while the other established Chain domains migrate independently. +Agent orchestration, memory and tool implementations follow the same rule via +the named `get_agent_*_port()` functions from `app/application/agentdata.py`. +The legacy Agent `*Port` proxy classes remain import-compatible boundaries and +must not be reintroduced as Oper aliases in canonical Agent modules. ### Adapter boundaries diff --git a/tests/test_agent_add_subscribe_tool.py b/tests/test_agent_add_subscribe_tool.py index d90e12501..a7f21f219 100644 --- a/tests/test_agent_add_subscribe_tool.py +++ b/tests/test_agent_add_subscribe_tool.py @@ -1,5 +1,6 @@ import asyncio import unittest +from types import SimpleNamespace from unittest.mock import AsyncMock, patch from app.agent.tools.impl.add_subscribe import AddSubscribeTool @@ -28,8 +29,10 @@ class TestAgentAddSubscribeTool(unittest.TestCase): "app.agent.tools.impl.add_subscribe.SubscribeChain.async_add", new=AsyncMock(return_value=(1, "")), ) as async_add, patch( - "app.agent.tools.impl.add_subscribe.UserOper.get_name", - return_value="moviepilot-user", + "app.agent.tools.impl.add_subscribe.get_agent_user_port", + return_value=SimpleNamespace( + get_name=lambda **_kwargs: "moviepilot-user" + ), ): result = asyncio.run( tool.run( @@ -55,8 +58,8 @@ class TestAgentAddSubscribeTool(unittest.TestCase): "app.agent.tools.impl.add_subscribe.SubscribeChain.async_add", new=AsyncMock(return_value=(1, "")), ) as async_add, patch( - "app.agent.tools.impl.add_subscribe.UserOper.get_name", - return_value=None, + "app.agent.tools.impl.add_subscribe.get_agent_user_port", + return_value=SimpleNamespace(get_name=lambda **_kwargs: None), ): result = asyncio.run( tool.run( @@ -81,8 +84,8 @@ class TestAgentAddSubscribeTool(unittest.TestCase): "app.agent.tools.impl.add_subscribe.SubscribeChain.async_add", new=AsyncMock(return_value=(1, "")), ) as async_add, patch( - "app.agent.tools.impl.add_subscribe.UserOper.get_name", - return_value=None, + "app.agent.tools.impl.add_subscribe.get_agent_user_port", + return_value=SimpleNamespace(get_name=lambda **_kwargs: None), ): result = asyncio.run( tool.run( diff --git a/tests/test_agent_data_ports.py b/tests/test_agent_data_ports.py new file mode 100644 index 000000000..d698fa8bd --- /dev/null +++ b/tests/test_agent_data_ports.py @@ -0,0 +1,28 @@ +"""Agent 命名数据端口工厂测试。""" + +from app.application import agentdata + + +def test_named_agent_data_getters_use_registered_factories(monkeypatch) -> None: + """每个命名 getter 都应创建组合根登记的对应端口实例。""" + names = { + "agent_chat": agentdata.get_agent_chat_port, + "agent_task": agentdata.get_agent_task_port, + "user": agentdata.get_agent_user_port, + "site": agentdata.get_agent_site_port, + "subscribe": agentdata.get_agent_subscribe_port, + "subscribe_history": agentdata.get_agent_subscribe_history_port, + "transfer_history": agentdata.get_agent_transfer_history_port, + "download_history": agentdata.get_agent_download_history_port, + "workflow": agentdata.get_agent_workflow_port, + "plugin_data": agentdata.get_agent_plugin_data_port, + } + factories = { + name: (lambda current=name: current) + for name in names + } + monkeypatch.setattr(agentdata, "_ports", agentdata.AgentDataPorts(**factories)) + + assert {name: getter() for name, getter in names.items()} == { + name: name for name in names + } diff --git a/tests/test_agent_music_tools.py b/tests/test_agent_music_tools.py index af9247b81..87627aced 100644 --- a/tests/test_agent_music_tools.py +++ b/tests/test_agent_music_tools.py @@ -428,8 +428,10 @@ def test_add_download_preserves_album_context_and_full_coverage_marker(): "_async_resolve_cached_context", new=AsyncMock(return_value=cached_context), ), patch( - "app.agent.tools.impl.add_download_tasks.SiteOper.async_get_by_name", - new=AsyncMock(return_value=site), + "app.agent.tools.impl.add_download_tasks.get_agent_site_port", + return_value=SimpleNamespace( + async_get_by_name=AsyncMock(return_value=site) + ), ), patch.object( AddDownloadTasksTool, "_download_single_sync", @@ -479,7 +481,7 @@ def test_query_subscribe_history_uses_database_media_values_and_music_fields(mon return [record] if mtype == MediaType.MUSIC.value else [] monkeypatch.setattr( - "app.agent.tools.impl.query_subscribe_history.SubscribeHistoryOper", + "app.agent.tools.impl.query_subscribe_history.get_agent_subscribe_history_port", FakeHistoryOper, ) tool = QuerySubscribeHistoryTool(session_id="session-1", user_id="10001") @@ -513,7 +515,7 @@ def test_music_history_filter_excludes_video_records(monkeypatch): return [movie_record] if mtype == MediaType.MOVIE.value else [] monkeypatch.setattr( - "app.agent.tools.impl.query_subscribe_history.SubscribeHistoryOper", + "app.agent.tools.impl.query_subscribe_history.get_agent_subscribe_history_port", FakeHistoryOper, ) tool = QuerySubscribeHistoryTool(session_id="session-1", user_id="10001") diff --git a/tests/test_agent_plugin_tools.py b/tests/test_agent_plugin_tools.py index 2f413d6cc..2581dbdf0 100644 --- a/tests/test_agent_plugin_tools.py +++ b/tests/test_agent_plugin_tools.py @@ -422,7 +422,7 @@ def test_query_plugin_data_truncates_large_payload() -> None: return_value=_plugin_snapshot(), ), patch( - "app.agent.tools.impl.query_plugin_data.PluginDataOper", + "app.agent.tools.impl.query_plugin_data.get_agent_plugin_data_port", return_value=plugin_data_oper, ), ): diff --git a/tests/test_agent_query_subscribes.py b/tests/test_agent_query_subscribes.py index ae2104b96..6256aaacc 100644 --- a/tests/test_agent_query_subscribes.py +++ b/tests/test_agent_query_subscribes.py @@ -1,6 +1,7 @@ import asyncio import json -from unittest.mock import patch +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch from app.agent.tools.impl.query_subscribes import QuerySubscribesTool from app.db.models.subscribe import Subscribe @@ -22,8 +23,10 @@ def test_agent_query_subscribes_returns_manual_total_episode(): ) with patch( - "app.agent.tools.impl.query_subscribes.SubscribeOper.async_list", - return_value=[subscribe], + "app.agent.tools.impl.query_subscribes.get_agent_subscribe_port", + return_value=SimpleNamespace( + async_list=AsyncMock(return_value=[subscribe]) + ), ): result = asyncio.run( QuerySubscribesTool(session_id="session-1", user_id="10001").run( diff --git a/tests/test_agent_query_workflows_tool.py b/tests/test_agent_query_workflows_tool.py index 58904610e..17e372955 100644 --- a/tests/test_agent_query_workflows_tool.py +++ b/tests/test_agent_query_workflows_tool.py @@ -28,7 +28,7 @@ class TestQueryWorkflowsTool(unittest.TestCase): workflow_oper.async_list = AsyncMock(return_value=[workflow]) with patch( - "app.agent.tools.impl.query_workflows.WorkflowOper", + "app.agent.tools.impl.query_workflows.get_agent_workflow_port", return_value=workflow_oper, ): result = asyncio.run(tool.run()) diff --git a/tests/test_agent_resource_flow_permissions.py b/tests/test_agent_resource_flow_permissions.py index aace0f9ca..b217af076 100644 --- a/tests/test_agent_resource_flow_permissions.py +++ b/tests/test_agent_resource_flow_permissions.py @@ -115,7 +115,7 @@ def test_query_sites_hides_only_sensitive_fields_for_non_admin_user(): ) with patch( - "app.agent.tools.impl.query_sites.SiteOper" + "app.agent.tools.impl.query_sites.get_agent_site_port" ) as site_oper: site_oper.return_value.async_list = AsyncMock(return_value=[site]) result = asyncio.run(tool.run()) @@ -177,7 +177,7 @@ def test_query_sites_keeps_full_fields_for_admin_context(): ) with patch( - "app.agent.tools.impl.query_sites.SiteOper" + "app.agent.tools.impl.query_sites.get_agent_site_port" ) as site_oper: site_oper.return_value.async_list = AsyncMock(return_value=[site]) result = asyncio.run(tool.run()) @@ -369,7 +369,7 @@ def test_channel_agent_admin_user_id_does_not_bypass_user_lookup(): username="normal-user", ) - with patch("app.agent.orchestrator.UserOper") as user_oper: + with patch("app.agent.orchestrator.get_agent_user_port") as user_oper: user_oper.return_value.async_get_by_name.return_value = SimpleNamespace( is_superuser=False ) @@ -391,7 +391,7 @@ def test_channel_agent_rejects_local_admin_username_without_trusted_principal(): ) agent.is_channel_admin = False - with patch("app.agent.orchestrator.UserOper") as user_oper: + with patch("app.agent.orchestrator.get_agent_user_port") as user_oper: user_oper.return_value.async_get_by_name = AsyncMock( return_value=SimpleNamespace(is_superuser=True) ) @@ -414,7 +414,7 @@ def test_channel_agent_accepts_trusted_admin_principal_without_local_user(): ) agent.is_channel_admin = True - with patch("app.agent.orchestrator.UserOper") as user_oper: + with patch("app.agent.orchestrator.get_agent_user_port") as user_oper: context = asyncio.run( agent._build_tool_context(should_dispatch_reply=True) ) diff --git a/tests/test_agent_search_subscribe.py b/tests/test_agent_search_subscribe.py index 74cd062b3..e1f381f41 100644 --- a/tests/test_agent_search_subscribe.py +++ b/tests/test_agent_search_subscribe.py @@ -47,7 +47,7 @@ def test_search_subscribe_uses_async_data_port(monkeypatch) -> None: raise AssertionError("async 工具不应调用同步订阅更新") port = _AsyncSubscribePort() - monkeypatch.setattr(search_subscribe_module, "SubscribeOper", lambda: port) + monkeypatch.setattr(search_subscribe_module, "get_agent_subscribe_port", lambda: port) async def _run_blocking(*_args, **_kwargs): await asyncio.sleep(0) @@ -79,7 +79,7 @@ def test_search_subscribe_rejects_paused_subscription_without_search(monkeypatch async_update = AsyncMock() port = _AsyncSubscribePort() - monkeypatch.setattr(search_subscribe_module, "SubscribeOper", lambda: port) + monkeypatch.setattr(search_subscribe_module, "get_agent_subscribe_port", lambda: port) run_blocking = AsyncMock() monkeypatch.setattr(SearchSubscribeTool, "run_blocking", run_blocking) diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 6ed361531..6f19f4920 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -568,6 +568,37 @@ def test_transfer_chains_use_explicit_data_port_getters(): assert violations == [] +def test_agent_consumers_use_explicit_data_port_getters(): + """Agent 生产模块不得把兼容数据端口代理重新伪装成数据库 Oper。""" + forbidden = { + "AgentChatPort", + "AgentTaskPort", + "DownloadHistoryPort", + "PluginDataPort", + "SitePort", + "SubscribeHistoryPort", + "SubscribePort", + "TransferHistoryPort", + "UserPort", + "WorkflowPort", + } + violations: list[str] = [] + for path in (APP_ROOT / "agent").rglob("*.py"): + tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + for node in ast.walk(tree): + if not isinstance(node, ast.ImportFrom): + continue + if node.module != "app.application.agentdata": + continue + for alias in node.names: + if alias.name in forbidden: + violations.append( + f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}:{alias.name}" + ) + + assert violations == [] + + def test_plugin_components_do_not_reexport_legacy_abi_names(): """新插件组件只提供 canonical 能力,不得复制旧 Helper、Manager 或 Oper 导出。""" violations: list[str] = [] diff --git a/tests/test_delete_transfer_history_tool.py b/tests/test_delete_transfer_history_tool.py index d656480af..015433fb1 100644 --- a/tests/test_delete_transfer_history_tool.py +++ b/tests/test_delete_transfer_history_tool.py @@ -41,7 +41,7 @@ def test_delete_transfer_history_tool_removes_old_dest_file_before_history(monke return True monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr( @@ -96,7 +96,7 @@ def test_delete_transfer_history_tool_keeps_history_when_old_dest_delete_fails(m return False monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr( @@ -150,7 +150,7 @@ def test_delete_transfer_history_tool_deletes_history_when_old_dest_is_missing(m return False monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr( @@ -205,7 +205,7 @@ def test_delete_transfer_history_tool_keeps_successful_move_dest_as_reorganize_s return True monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr( @@ -258,7 +258,7 @@ def test_delete_transfer_history_tool_only_treats_exact_move_as_reorganize_sourc return True monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr( @@ -313,7 +313,7 @@ def test_delete_transfer_history_storage_work_runs_outside_event_loop(monkeypatc return True monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr( @@ -371,7 +371,7 @@ def test_delete_transfer_history_cancellation_keeps_history_record(monkeypatch): return True monkeypatch.setattr( - "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port", FakeTransferHistoryOper, ) monkeypatch.setattr(