feat: expose download save paths in API

Return configured download directories as API-ready save_path values so external integrations can choose download destinations without guessing local or remote path syntax.

Fixes #5737
This commit is contained in:
jxxghp
2026-05-10 12:02:22 +08:00
parent adb7aa6aa9
commit 1862a7ab4b
3 changed files with 129 additions and 0 deletions

View File

@@ -11,6 +11,7 @@ from app.core.security import verify_token
from app.db.models.user import User
from app.db.systemconfig_oper import SystemConfigOper
from app.db.user_oper import get_current_active_user
from app.helper.directory import DirectoryHelper
from app.schemas.types import SystemConfigKey
router = APIRouter()
@@ -140,6 +141,29 @@ async def clients(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
return []
@router.get("/paths", summary="查询可用下载路径", response_model=List[schemas.DownloadDirectory])
def paths(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
"""
查询可直接用于下载接口 save_path 参数的下载路径
"""
return [
schemas.DownloadDirectory(
name=dir_info.name,
storage=dir_info.storage or "local",
download_path=dir_info.download_path,
save_path=schemas.FileURI(
storage=dir_info.storage or "local",
path=dir_info.download_path,
).uri,
priority=dir_info.priority,
media_type=dir_info.media_type,
media_category=dir_info.media_category,
)
for dir_info in DirectoryHelper().get_download_dirs()
if dir_info.download_path
]
@router.delete("/{hashString}", summary="删除下载任务", response_model=schemas.Response)
def delete(hashString: str, name: Optional[str] = None,
_: schemas.TokenPayload = Depends(verify_token)) -> Any:

View File

@@ -11,3 +11,17 @@ class DownloadTask(BaseModel):
downloader: Optional[str] = Field(default=None, description="下载器")
path: Optional[str] = Field(default=None, description="下载路径")
completed: Optional[bool] = Field(default=False, description="是否完成")
class DownloadDirectory(BaseModel):
"""
下载目录
"""
name: Optional[str] = Field(default=None, description="目录名称")
storage: Optional[str] = Field(default="local", description="存储类型")
download_path: Optional[str] = Field(default=None, description="配置的下载目录")
save_path: Optional[str] = Field(default=None, description="可直接传给下载接口 save_path 的路径")
priority: Optional[int] = Field(default=0, description="目录优先级")
media_type: Optional[str] = Field(default=None, description="适用媒体类型")
media_category: Optional[str] = Field(default=None, description="适用媒体分类")

View File

@@ -0,0 +1,91 @@
import sys
import unittest
from types import ModuleType
from unittest.mock import patch
def _stub_module(name: str, **attrs):
module = sys.modules.get(name)
if module is None:
module = ModuleType(name)
sys.modules[name] = module
for key, value in attrs.items():
setattr(module, key, value)
return module
class _Dummy:
def __init__(self, *args, **kwargs):
pass
def __getattr__(self, _name):
return lambda *args, **kwargs: None
for _module_name in ("pillow_avif", "aiofiles", "psutil"):
_stub_module(_module_name)
_stub_module("app.chain.download", DownloadChain=_Dummy)
_stub_module("app.chain.media", MediaChain=_Dummy)
_stub_module("app.core.context", MediaInfo=_Dummy, Context=_Dummy, TorrentInfo=_Dummy)
_stub_module("app.core.metainfo", MetaInfo=_Dummy)
_stub_module("app.core.security", verify_token=_Dummy)
_stub_module("app.db.models.user", User=_Dummy)
_stub_module("app.db.systemconfig_oper", SystemConfigOper=_Dummy)
_stub_module("app.db.user_oper", get_current_active_user=_Dummy)
_stub_module(
"app.log",
logger=_Dummy(),
log_settings=_Dummy(),
LogConfigModel=type("LogConfigModel", (), {}),
)
_stub_module("version", APP_VERSION="test")
from app.api.endpoints import download as download_endpoint
class DownloadPathsEndpointTest(unittest.TestCase):
def test_paths_returns_api_ready_save_paths(self):
mocked_dirs = [
download_endpoint.schemas.TransferDirectoryConf(
name="电影目录",
priority=1,
storage="local",
download_path="/downloads/movies",
media_type="movie",
),
download_endpoint.schemas.TransferDirectoryConf(
name="动漫远程目录",
priority=2,
storage="rclone",
download_path="/media/anime",
media_type="tv",
media_category="动漫",
),
]
with patch.object(download_endpoint.DirectoryHelper, "get_download_dirs", return_value=mocked_dirs):
ret = download_endpoint.paths(_=download_endpoint.schemas.TokenPayload())
self.assertEqual(len(ret), 2)
self.assertEqual(ret[0].name, "电影目录")
self.assertEqual(ret[0].storage, "local")
self.assertEqual(ret[0].download_path, "/downloads/movies")
self.assertEqual(ret[0].save_path, "/downloads/movies")
self.assertEqual(ret[0].priority, 1)
self.assertEqual(ret[0].media_type, "movie")
self.assertIsNone(ret[0].media_category)
self.assertEqual(ret[1].name, "动漫远程目录")
self.assertEqual(ret[1].storage, "rclone")
self.assertEqual(ret[1].download_path, "/media/anime")
self.assertEqual(ret[1].save_path, "rclone:/media/anime")
self.assertEqual(ret[1].priority, 2)
self.assertEqual(ret[1].media_type, "tv")
self.assertEqual(ret[1].media_category, "动漫")
def test_paths_returns_empty_list_when_unconfigured(self):
with patch.object(download_endpoint.DirectoryHelper, "get_download_dirs", return_value=[]):
ret = download_endpoint.paths(_=download_endpoint.schemas.TokenPayload())
self.assertEqual(ret, [])