mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-05 07:27:15 +08:00
fix(async): offload file metadata checks
This commit is contained in:
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
import shutil
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any, Optional
|
from typing import Any, Optional
|
||||||
|
|
||||||
from app.runtime.settings import RuntimeSettingsCompat
|
from app.runtime.settings import RuntimeSettingsCompat
|
||||||
@@ -25,6 +26,17 @@ DEFAULT_PLUGIN_CANDIDATE_LIMIT = 50
|
|||||||
MAX_PLUGIN_CANDIDATE_LIMIT = 200
|
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]]:
|
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.folders import remove_plugin_from_folders
|
||||||
from app.application.plugin.routes import remove_plugin_api
|
from app.application.plugin.routes import remove_plugin_api
|
||||||
from app.application.scheduling import remove_plugin_job
|
from app.application.scheduling import remove_plugin_job
|
||||||
|
from app.agent.tools.base import run_agent_blocking
|
||||||
|
|
||||||
plugin_manager = get_plugin_manager()
|
plugin_manager = get_plugin_manager()
|
||||||
virtual_instance = plugin_manager.get_plugin_instance(plugin_id)
|
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_config(plugin_id)
|
||||||
plugin_manager.delete_plugin_data(plugin_id)
|
plugin_manager.delete_plugin_data(plugin_id)
|
||||||
plugin_base_dir = settings.ROOT_PATH / "app" / "plugins" / plugin_id.lower()
|
plugin_base_dir = settings.ROOT_PATH / "app" / "plugins" / plugin_id.lower()
|
||||||
if plugin_base_dir.exists():
|
try:
|
||||||
try:
|
clone_files_removed = await run_agent_blocking(
|
||||||
shutil.rmtree(plugin_base_dir)
|
"plugin",
|
||||||
|
_remove_plugin_directory,
|
||||||
|
plugin_base_dir,
|
||||||
|
)
|
||||||
|
if clone_files_removed:
|
||||||
plugin_manager.plugins.pop(plugin_id, None)
|
plugin_manager.plugins.pop(plugin_id, None)
|
||||||
clone_files_removed = True
|
except Exception:
|
||||||
except Exception:
|
clone_files_removed = False
|
||||||
clone_files_removed = False
|
|
||||||
|
|
||||||
remove_plugin_from_folders(plugin_id)
|
remove_plugin_from_folders(plugin_id)
|
||||||
plugin_manager.remove_plugin(plugin_id)
|
plugin_manager.remove_plugin(plugin_id)
|
||||||
|
|||||||
@@ -26,6 +26,12 @@ from app.domain.media import normalize_music_type
|
|||||||
from ._music_utils import simplify_music_info
|
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):
|
class ScrapeMetadataInput(BaseModel):
|
||||||
"""刮削媒体元数据工具的输入参数模型"""
|
"""刮削媒体元数据工具的输入参数模型"""
|
||||||
|
|
||||||
@@ -151,7 +157,14 @@ class ScrapeMetadataTool(MoviePilotTool):
|
|||||||
media_id = normalized_media_id or None
|
media_id = normalized_media_id or None
|
||||||
|
|
||||||
local_path = Path(path)
|
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"
|
file_type = "dir" if is_local_directory or not local_path.suffix else "file"
|
||||||
fileitem = FileItem(
|
fileitem = FileItem(
|
||||||
storage=storage or "local",
|
storage=storage or "local",
|
||||||
@@ -161,7 +174,7 @@ class ScrapeMetadataTool(MoviePilotTool):
|
|||||||
|
|
||||||
# 检查本地存储路径是否存在
|
# 检查本地存储路径是否存在
|
||||||
if storage == "local":
|
if storage == "local":
|
||||||
if not Path(path).exists():
|
if not path_exists:
|
||||||
return json.dumps(
|
return json.dumps(
|
||||||
{"success": False, "message": f"刮削路径不存在: {path}"},
|
{"success": False, "message": f"刮削路径不存在: {path}"},
|
||||||
ensure_ascii=False,
|
ensure_ascii=False,
|
||||||
|
|||||||
+12
-2
@@ -43,6 +43,16 @@ from app.domain import title as title_rules
|
|||||||
recognize_lock = Lock()
|
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):
|
class MediaChain(ChainBase, metaclass=Singleton):
|
||||||
"""
|
"""
|
||||||
媒体信息处理链,单例运行
|
媒体信息处理链,单例运行
|
||||||
@@ -1068,7 +1078,7 @@ class MediaChain(ChainBase, metaclass=Singleton):
|
|||||||
) -> Optional[MusicInfo]:
|
) -> Optional[MusicInfo]:
|
||||||
"""异步查找所在目录专辑匹配中属于当前文件的结果。"""
|
"""异步查找所在目录专辑匹配中属于当前文件的结果。"""
|
||||||
file_path = Path(path)
|
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
|
return None
|
||||||
try:
|
try:
|
||||||
matched = await self.async_recognize_music_album_directory(
|
matched = await self.async_recognize_music_album_directory(
|
||||||
@@ -1212,7 +1222,7 @@ class MediaChain(ChainBase, metaclass=Singleton):
|
|||||||
) -> dict[str, MusicInfo]:
|
) -> dict[str, MusicInfo]:
|
||||||
"""异步按目录级线索批量识别整张专辑。"""
|
"""异步按目录级线索批量识别整张专辑。"""
|
||||||
directory = Path(path)
|
directory = Path(path)
|
||||||
if not directory.is_dir():
|
if not await run_in_threadpool(_is_directory, directory):
|
||||||
return {}
|
return {}
|
||||||
files = await run_in_threadpool(self._directory_audio_files, directory)
|
files = await run_in_threadpool(self._directory_audio_files, directory)
|
||||||
if len(files) < self._album_match_min_files:
|
if len(files) < self._album_match_min_files:
|
||||||
|
|||||||
@@ -9,6 +9,8 @@ from pathlib import Path
|
|||||||
from typing import Any, Optional, Tuple, Union
|
from typing import Any, Optional, Tuple, Union
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
|
from fastapi.concurrency import run_in_threadpool
|
||||||
|
|
||||||
from app.runtime.settings import RuntimeSettingsCompat
|
from app.runtime.settings import RuntimeSettingsCompat
|
||||||
|
|
||||||
settings = RuntimeSettingsCompat()
|
settings = RuntimeSettingsCompat()
|
||||||
@@ -137,7 +139,10 @@ class AcoustIdModule(_ModuleBase):
|
|||||||
) -> Optional[str]:
|
) -> Optional[str]:
|
||||||
"""异步读取音频指纹并返回高置信匹配的 MusicBrainz Recording ID。"""
|
"""异步读取音频指纹并返回高置信匹配的 MusicBrainz Recording ID。"""
|
||||||
file_path = Path(path)
|
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
|
return None
|
||||||
cache_key = self._file_cache_key(file_path)
|
cache_key = self._file_cache_key(file_path)
|
||||||
if cache_key:
|
if cache_key:
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ from urllib.parse import quote
|
|||||||
import discord
|
import discord
|
||||||
from discord import app_commands
|
from discord import app_commands
|
||||||
import httpx
|
import httpx
|
||||||
|
from fastapi.concurrency import run_in_threadpool
|
||||||
|
|
||||||
from app.runtime.settings import RuntimeSettingsCompat
|
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:
|
class Discord:
|
||||||
"""
|
"""
|
||||||
Discord Bot 通知与交互实现(基于 discord.py 2.6.4)
|
Discord Bot 通知与交互实现(基于 discord.py 2.6.4)
|
||||||
@@ -780,7 +786,10 @@ class Discord:
|
|||||||
return False, None
|
return False, None
|
||||||
|
|
||||||
local_file = Path(file_path)
|
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}")
|
logger.error(f"Discord发送文件失败,文件不存在: {local_file}")
|
||||||
return False, None
|
return False, None
|
||||||
|
|
||||||
|
|||||||
@@ -274,3 +274,19 @@ def test_async_identify_music_by_fingerprint_uses_async_process_and_http(
|
|||||||
post_res.assert_awaited_once()
|
post_res.assert_awaited_once()
|
||||||
assert post_res.await_args.kwargs["data"]["meta"] == "recordingids"
|
assert post_res.await_args.kwargs["data"]["meta"] == "recordingids"
|
||||||
assert response.closed is True
|
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")
|
||||||
|
|||||||
@@ -309,6 +309,31 @@ def test_scrape_metadata_rejects_invalid_media_source_before_file_access(tmp_pat
|
|||||||
assert "media_source" in payload["message"]
|
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():
|
def test_query_artist_detail_marks_entity_as_non_subscribable():
|
||||||
"""艺术家详情应明确标记为不可订阅,避免 Agent 混入获取流程。"""
|
"""艺术家详情应明确标记为不可订阅,避免 Agent 混入获取流程。"""
|
||||||
artist = MusicArtistInfo(
|
artist = MusicArtistInfo(
|
||||||
|
|||||||
@@ -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()
|
||||||
@@ -220,6 +220,52 @@ def test_async_recognize_album_directory_calls_async_module(
|
|||||||
run_module.assert_not_called()
|
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):
|
def test_recognize_album_directory_skips_single_file(tmp_path, media_chain, monkeypatch):
|
||||||
"""单文件目录不走专辑匹配,交给单曲识别链路。"""
|
"""单文件目录不走专辑匹配,交给单曲识别链路。"""
|
||||||
album_dir = tmp_path / "单曲"
|
album_dir = tmp_path / "单曲"
|
||||||
|
|||||||
Reference in New Issue
Block a user