From e8cfe272c4cd84e341785648b45ec314eb6832f3 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 04:22:36 +0800 Subject: [PATCH] fix(async): offload file metadata checks --- app/agent/tools/impl/_plugin_tool_utils.py | 28 ++++++++++--- app/agent/tools/impl/scrape_metadata.py | 17 +++++++- app/chain/media.py | 14 ++++++- app/modules/acoustid/__init__.py | 7 +++- app/modules/discord/discord.py | 11 +++++- tests/test_acoustid_module.py | 16 ++++++++ tests/test_agent_music_tools.py | 25 ++++++++++++ tests/test_discord_file_send.py | 30 ++++++++++++++ tests/test_music_album_match.py | 46 ++++++++++++++++++++++ 9 files changed, 182 insertions(+), 12 deletions(-) create mode 100644 tests/test_discord_file_send.py diff --git a/app/agent/tools/impl/_plugin_tool_utils.py b/app/agent/tools/impl/_plugin_tool_utils.py index cca3d5909..a7ce7c67e 100644 --- a/app/agent/tools/impl/_plugin_tool_utils.py +++ b/app/agent/tools/impl/_plugin_tool_utils.py @@ -2,6 +2,7 @@ import json import shutil +from pathlib import Path from typing import Any, Optional from app.runtime.settings import RuntimeSettingsCompat @@ -25,6 +26,17 @@ DEFAULT_PLUGIN_CANDIDATE_LIMIT = 50 MAX_PLUGIN_CANDIDATE_LIMIT = 200 +def _remove_plugin_directory(path: Path) -> bool: + """删除插件目录并返回是否完成，供受控线程执行。""" + if not path.exists(): + return False + try: + shutil.rmtree(path) + except Exception: + return False + return True + + def get_plugin_snapshot(plugin_id: str) -> Optional[dict[str, Any]]: """ 获取已安装插件的基础信息快照。 @@ -394,6 +406,7 @@ async def uninstall_plugin_runtime(plugin_id: str) -> dict[str, Any]: from app.application.plugin.folders import remove_plugin_from_folders from app.application.plugin.routes import remove_plugin_api from app.application.scheduling import remove_plugin_job + from app.agent.tools.base import run_agent_blocking plugin_manager = get_plugin_manager() virtual_instance = plugin_manager.get_plugin_instance(plugin_id) @@ -423,13 +436,16 @@ async def uninstall_plugin_runtime(plugin_id: str) -> dict[str, Any]: plugin_manager.delete_plugin_config(plugin_id) plugin_manager.delete_plugin_data(plugin_id) plugin_base_dir = settings.ROOT_PATH / "app" / "plugins" / plugin_id.lower() - if plugin_base_dir.exists(): - try: - shutil.rmtree(plugin_base_dir) + try: + clone_files_removed = await run_agent_blocking( + "plugin", + _remove_plugin_directory, + plugin_base_dir, + ) + if clone_files_removed: plugin_manager.plugins.pop(plugin_id, None) - clone_files_removed = True - except Exception: - clone_files_removed = False + except Exception: + clone_files_removed = False remove_plugin_from_folders(plugin_id) plugin_manager.remove_plugin(plugin_id) diff --git a/app/agent/tools/impl/scrape_metadata.py b/app/agent/tools/impl/scrape_metadata.py index d3a2a249d..643ceb345 100644 --- a/app/agent/tools/impl/scrape_metadata.py +++ b/app/agent/tools/impl/scrape_metadata.py @@ -26,6 +26,12 @@ from app.domain.media import normalize_music_type from ._music_utils import simplify_music_info +def _inspect_local_path(path: Path) -> tuple[bool, bool]: + """返回本地路径是否存在及是否为目录。""" + exists = path.exists() + return exists and path.is_dir(), exists + + class ScrapeMetadataInput(BaseModel): """刮削媒体元数据工具的输入参数模型""" @@ -151,7 +157,14 @@ class ScrapeMetadataTool(MoviePilotTool): media_id = normalized_media_id or None local_path = Path(path) - is_local_directory = (storage or "local") == "local" and local_path.is_dir() + is_local_directory = False + path_exists = True + if (storage or "local") == "local": + is_local_directory, path_exists = await self.run_blocking( + "storage", + _inspect_local_path, + local_path, + ) file_type = "dir" if is_local_directory or not local_path.suffix else "file" fileitem = FileItem( storage=storage or "local", @@ -161,7 +174,7 @@ class ScrapeMetadataTool(MoviePilotTool): # 检查本地存储路径是否存在 if storage == "local": - if not Path(path).exists(): + if not path_exists: return json.dumps( {"success": False, "message": f"刮削路径不存在: {path}"}, ensure_ascii=False, diff --git a/app/chain/media.py b/app/chain/media.py index ea0b3d8e3..54173f3e0 100644 --- a/app/chain/media.py +++ b/app/chain/media.py @@ -43,6 +43,16 @@ from app.domain import title as title_rules recognize_lock = Lock() +def _is_regular_file(path: Path) -> bool: + """判断路径是否仍指向可读取的普通文件。""" + return path.exists() and path.is_file() + + +def _is_directory(path: Path) -> bool: + """判断路径是否仍指向目录。""" + return path.is_dir() + + class MediaChain(ChainBase, metaclass=Singleton): """ 媒体信息处理链，单例运行 @@ -1068,7 +1078,7 @@ class MediaChain(ChainBase, metaclass=Singleton): ) -> Optional[MusicInfo]: """异步查找所在目录专辑匹配中属于当前文件的结果。""" file_path = Path(path) - if not file_path.exists() or not file_path.is_file(): + if not await run_in_threadpool(_is_regular_file, file_path): return None try: matched = await self.async_recognize_music_album_directory( @@ -1212,7 +1222,7 @@ class MediaChain(ChainBase, metaclass=Singleton): ) -> dict[str, MusicInfo]: """异步按目录级线索批量识别整张专辑。""" directory = Path(path) - if not directory.is_dir(): + if not await run_in_threadpool(_is_directory, directory): return {} files = await run_in_threadpool(self._directory_audio_files, directory) if len(files) < self._album_match_min_files: diff --git a/app/modules/acoustid/__init__.py b/app/modules/acoustid/__init__.py index 44689c37e..4dc5aedb7 100644 --- a/app/modules/acoustid/__init__.py +++ b/app/modules/acoustid/__init__.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any, Optional, Tuple, Union from uuid import UUID +from fastapi.concurrency import run_in_threadpool + from app.runtime.settings import RuntimeSettingsCompat settings = RuntimeSettingsCompat() @@ -137,7 +139,10 @@ class AcoustIdModule(_ModuleBase): ) -> Optional[str]: """异步读取音频指纹并返回高置信匹配的 MusicBrainz Recording ID。""" file_path = Path(path) - if not self._fpcalc_path or not file_path.is_file(): + if not self._fpcalc_path or not await run_in_threadpool( + Path.is_file, + file_path, + ): return None cache_key = self._file_cache_key(file_path) if cache_key: diff --git a/app/modules/discord/discord.py b/app/modules/discord/discord.py index 13eaff162..7b85fcb17 100644 --- a/app/modules/discord/discord.py +++ b/app/modules/discord/discord.py @@ -8,6 +8,7 @@ from urllib.parse import quote import discord from discord import app_commands import httpx +from fastapi.concurrency import run_in_threadpool from app.runtime.settings import RuntimeSettingsCompat @@ -28,6 +29,11 @@ PARSE_FIELD_TYPES = { } +def _is_regular_file(path: Path) -> bool: + """判断路径是否仍指向可发送的普通文件。""" + return path.exists() and path.is_file() + + class Discord: """ Discord Bot 通知与交互实现（基于 discord.py 2.6.4） @@ -780,7 +786,10 @@ class Discord: return False, None local_file = Path(file_path) - if not local_file.exists() or not local_file.is_file(): + if not await run_in_threadpool( + _is_regular_file, + local_file, + ): logger.error(f"Discord发送文件失败，文件不存在: {local_file}") return False, None diff --git a/tests/test_acoustid_module.py b/tests/test_acoustid_module.py index 290f19029..a550bfcd6 100644 --- a/tests/test_acoustid_module.py +++ b/tests/test_acoustid_module.py @@ -274,3 +274,19 @@ def test_async_identify_music_by_fingerprint_uses_async_process_and_http( post_res.assert_awaited_once() assert post_res.await_args.kwargs["data"]["meta"] == "recordingids" assert response.closed is True + + +def test_async_identify_skips_missing_file_after_threaded_check(monkeypatch): + """异步指纹入口应在线程中检查文件，并保持缺失文件的跳过语义。""" + module = AcoustIdModule() + module._fpcalc_path = "/usr/bin/fpcalc" + check_file = AsyncMock(return_value=False) + monkeypatch.setattr("app.modules.acoustid.run_in_threadpool", check_file) + + result = asyncio.run( + module.async_identify_music_by_fingerprint(Path("/music/missing.flac")) + ) + + assert result is None + check_file.assert_awaited_once() + assert check_file.await_args.args[1] == Path("/music/missing.flac") diff --git a/tests/test_agent_music_tools.py b/tests/test_agent_music_tools.py index 5e53a4c9d..af9247b81 100644 --- a/tests/test_agent_music_tools.py +++ b/tests/test_agent_music_tools.py @@ -309,6 +309,31 @@ def test_scrape_metadata_rejects_invalid_media_source_before_file_access(tmp_pat assert "media_source" in payload["message"] +def test_scrape_metadata_checks_local_path_in_agent_worker(tmp_path, monkeypatch): + """Agent 刮削的本地路径检查应通过受控存储线程执行。""" + calls = [] + + async def fake_run_agent_blocking(bucket, func, *args, **kwargs): + calls.append((bucket, func, args, kwargs)) + return False, False + + monkeypatch.setattr("app.agent.tools.base.run_agent_blocking", fake_run_agent_blocking) + tool = ScrapeMetadataTool(session_id="session-1", user_id="10001") + + result = asyncio.run( + tool.run(path=str(tmp_path / "missing"), storage="local") + ) + + payload = json.loads(result) + assert payload == { + "success": False, + "message": f"刮削路径不存在: {tmp_path / 'missing'}", + } + assert len(calls) == 1 + assert calls[0][0] == "storage" + assert calls[0][2] == (tmp_path / "missing",) + + def test_query_artist_detail_marks_entity_as_non_subscribable(): """艺术家详情应明确标记为不可订阅，避免 Agent 混入获取流程。""" artist = MusicArtistInfo( diff --git a/tests/test_discord_file_send.py b/tests/test_discord_file_send.py new file mode 100644 index 000000000..c293f9705 --- /dev/null +++ b/tests/test_discord_file_send.py @@ -0,0 +1,30 @@ +import asyncio +from pathlib import Path +from unittest.mock import AsyncMock + +from app.modules.discord.discord import Discord + + +def test_send_file_checks_local_file_in_threadpool(monkeypatch): + """Discord 文件发送应把本地文件检查移出 Discord 事件循环。""" + discord_client = Discord.__new__(Discord) + channel = AsyncMock() + discord_client._resolve_channel = AsyncMock(return_value=channel) + check_file = AsyncMock(return_value=False) + monkeypatch.setattr("app.modules.discord.discord.run_in_threadpool", check_file) + + result = asyncio.run( + discord_client._send_file( + file_path="/tmp/missing.txt", + title="标题", + text=None, + userid="user-1", + file_name=None, + original_chat_id=None, + ) + ) + + assert result == (False, None) + check_file.assert_awaited_once() + assert check_file.await_args.args[1] == Path("/tmp/missing.txt") + channel.send.assert_not_awaited() diff --git a/tests/test_music_album_match.py b/tests/test_music_album_match.py index 5a9087def..76f0d617a 100644 --- a/tests/test_music_album_match.py +++ b/tests/test_music_album_match.py @@ -220,6 +220,52 @@ def test_async_recognize_album_directory_calls_async_module( run_module.assert_not_called() +def test_async_recognize_album_directory_checks_path_in_threadpool( + tmp_path, + media_chain, + monkeypatch, +): + """异步专辑识别应把目录元数据检查移出事件循环。""" + check_directory = AsyncMock(return_value=False) + monkeypatch.setattr("app.chain.media.run_in_threadpool", check_directory) + + result = asyncio.run( + media_chain.async_recognize_music_album_directory(tmp_path / "missing") + ) + + assert result == {} + check_directory.assert_awaited_once() + assert check_directory.await_args.args[1] == tmp_path / "missing" + + +def test_async_album_fallback_propagates_cancellation_during_path_check( + tmp_path, + media_chain, + monkeypatch, +): + """异步专辑兜底不得吞掉文件检查被取消的信号。""" + started = asyncio.Event() + + async def wait_for_check(*_args, **_kwargs): + """模拟慢文件系统检查，直到调用方取消。""" + started.set() + await asyncio.Event().wait() + + monkeypatch.setattr("app.chain.media.run_in_threadpool", wait_for_check) + + async def exercise_cancellation(): + """在同一事件循环中取消正在等待文件检查的调用。""" + task = asyncio.create_task( + media_chain._async_music_album_dir_fallback(tmp_path / "track.flac") + ) + await started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(exercise_cancellation()) + + def test_recognize_album_directory_skips_single_file(tmp_path, media_chain, monkeypatch): """单文件目录不走专辑匹配，交给单曲识别链路。""" album_dir = tmp_path / "单曲"