From 56cf6fefe0e846c03de2d608b74447db19ed152f Mon Sep 17 00:00:00 2001 From: ranzhigg <126594517+ranzhigg@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:45:31 +0800 Subject: [PATCH] perf(transfer): bound queue snapshots for batch imports --- app/api/endpoints/transfer.py | 6 +- app/api/endpoints/transferqueue.py | 29 +++++ app/api/routers.py | 2 + app/application/transfer/jobs.py | 102 ++++++++++++++++-- app/application/transfer/workflow.py | 20 ++++ app/chain/transfer/queue.py | 7 +- app/schemas/exports.py | 5 + app/schemas/transfer.py | 47 +++++++++ docs/refactor/agent-api-surface-audit.json | 16 ++- docs/refactor/agent-api-surface-audit.md | 5 +- scripts/generate_agent_api_surface_audit.py | 1 + tests/test_transfer_queue_paging.py | 111 ++++++++++++++++++++ 12 files changed, 334 insertions(+), 17 deletions(-) create mode 100644 app/api/endpoints/transferqueue.py create mode 100644 tests/test_transfer_queue_paging.py diff --git a/app/api/endpoints/transfer.py b/app/api/endpoints/transfer.py index afe96a07d..779bac329 100644 --- a/app/api/endpoints/transfer.py +++ b/app/api/endpoints/transfer.py @@ -11,11 +11,7 @@ from app.api.endpoints.transferhistory import ( restore_manual_transfer_history_batch, restore_manual_transfer_history_metadata, ) -from app.api.response import ( - CompatibleCountParam, - CompatiblePageParam, - ResponseAPIRouter, -) +from app.api.response import CompatibleCountParam, CompatiblePageParam, ResponseAPIRouter from app.application.configuration import get_api_runtime_config_snapshot from app.application.directory import DirectoryHelper from app.application.history import TransferHistoryLookupService diff --git a/app/api/endpoints/transferqueue.py b/app/api/endpoints/transferqueue.py new file mode 100644 index 000000000..397aa4740 --- /dev/null +++ b/app/api/endpoints/transferqueue.py @@ -0,0 +1,29 @@ +from fastapi import Depends, Query + +from app.adapters.web.security.access import verify_token +from app.api.response import COLLECTION_MAX_PAGE_SIZE, ResponseAPIRouter +from app.chain.transfer.facade import TransferChain +from app.schemas.token import TokenPayload as _SchemaTokenPayload +from app.schemas.transfer import TransferQueuePageData as _SchemaTransferQueuePageData + +router = ResponseAPIRouter() + + +@router.get( # type: ignore[misc] + "/queue/page", + summary="查询受限整理队列快照", + response_model=_SchemaTransferQueuePageData, +) +async def query_queue_page( + _: _SchemaTokenPayload = Depends(verify_token), + page: int = Query(1, ge=1), + count: int = Query(100, ge=1, le=COLLECTION_MAX_PAGE_SIZE), +) -> _SchemaTransferQueuePageData: + """返回前端所需的有限整理队列窗口，避免全量队列阻塞接口序列化。""" + items, total = TransferChain().get_queue_tasks_page(page, count) + return _SchemaTransferQueuePageData( + items=items, + total=total, + page=page, + count=count, + ) diff --git a/app/api/routers.py b/app/api/routers.py index d1a154cb8..a2e915820 100644 --- a/app/api/routers.py +++ b/app/api/routers.py @@ -41,6 +41,7 @@ from app.api.endpoints import ( tmdb, torrent, transfer, + transferqueue, user, webhook, workflow, @@ -88,6 +89,7 @@ API_V1_ROUTER_SPECS: tuple[RouterSpec, ...] = ( RouterSpec(dashboard.router, "/dashboard", ("dashboard",)), RouterSpec(storage.router, "/storage", ("storage",)), RouterSpec(transfer.router, "/transfer", ("transfer",)), + RouterSpec(transferqueue.router, "/transfer", ("transfer",)), RouterSpec(mediaserver.router, "/mediaserver", ("mediaserver",)), RouterSpec(bangumi.router, "/bangumi", ("bangumi",)), RouterSpec(anilist.router, "/anilist", ("anilist",)), diff --git a/app/application/transfer/jobs.py b/app/application/transfer/jobs.py index b8d6de88f..8b5094aaa 100644 --- a/app/application/transfer/jobs.py +++ b/app/application/transfer/jobs.py @@ -26,7 +26,14 @@ from app.schemas.file import FileItem from app.schemas.media import resolve_media_identity from app.schemas.music import MusicInfo as _SchemaMusicInfo from app.schemas.music import MusicMeta as _SchemaMusicMeta -from app.schemas.transfer import TransferJob, TransferJobTask +from app.schemas.transfer import ( + TransferJob, + TransferJobTask, + TransferQueueFileItemData, + TransferQueueItemData, + TransferQueueMediaData, + TransferQueueTaskData, +) from app.schemas.types import MUSIC_ENTITY_ALBUM, MUSIC_ENTITY_RECORDING, MediaType if TYPE_CHECKING: @@ -68,6 +75,44 @@ def _job_task_fileitem(task: TransferJobTask) -> FileItem: return cast(FileItem, task.fileitem) +def _queue_media_snapshot(media: Optional[Union[MediaInfo, MusicInfo]]) -> Optional[TransferQueueMediaData]: + """只复制整理队列页面会展示的媒体字段，避免重复输出完整刮削结果。""" + if not media: + return TransferQueueMediaData() + media_id = getattr(media, "media_id", None) + year = getattr(media, "year", None) + media_source = getattr(media, "media_source", None) + return TransferQueueMediaData( + media_source=getattr(media_source, "value", media_source), + media_id=str(media_id) if media_id is not None else None, + title=getattr(media, "title", None), + title_year=getattr(media, "title_year", None), + year=str(year) if year is not None else None, + poster_path=getattr(media, "poster_path", None), + episode_run_time=getattr(media, "episode_run_time", None) or [], + origin_country=getattr(media, "origin_country", None) or [], + ) + + +def _queue_fileitem_snapshot(fileitem: FileItem) -> TransferQueueFileItemData: + """保留 path/storage 以支持取消任务，其余只保留官方队列 UI 所需字段。""" + return TransferQueueFileItemData( + path=fileitem.path, + storage=fileitem.storage, + type=fileitem.type, + name=fileitem.name, + size=fileitem.size, + ) + + +def _queue_task_snapshot(task: TransferJobTask) -> TransferQueueTaskData: + """构造不含识别元数据和云盘目录树的轻量任务投影。""" + return TransferQueueTaskData( + fileitem=_queue_fileitem_snapshot(_job_task_fileitem(task)), + state=task.state, + ) + + def _job_task_size(task: TransferJobTask) -> int: """按既有本地目录回退规则返回已完成任务的文件大小。""" fileitem = _job_task_fileitem(task) @@ -80,6 +125,49 @@ def _job_task_size(task: TransferJobTask) -> int: return 0 +def _list_jobs_page( + jobs: Dict[JobId, TransferJob], + page: int, + count: int, + total_hint: Optional[int] = None, +) -> Tuple[List[TransferQueueItemData], int]: + """按扁平任务窗口构造轻量队列视图。""" + if page < 1 or count < 1: + raise ValueError("队列分页参数必须为正数") + + start = (page - 1) * count + end = start + count + selected: List[TransferQueueItemData] = [] + total = 0 + + for job in jobs.values(): + tasks = _job_tasks(job) + job_start = total + total += len(tasks) + if job_start >= end: + if total_hint is not None: + break + continue + if total <= start: + continue + + task_start = max(0, start - job_start) + task_end = min(len(tasks), end - job_start) + if task_start < task_end: + selected.append( + TransferQueueItemData( + media=_queue_media_snapshot(job.media), + season=job.season, + tasks=[ + _queue_task_snapshot(task) + for task in tasks[task_start:task_end] + ], + ) + ) + + return selected, total_hint if total_hint is not None else total + + job_lock = threading.Lock() @@ -99,7 +187,6 @@ class JobManager: _task_state_changed_at: Dict[FileKey, float] = {} # 记录仍由主程序整理线程直接执行的任务，避免把阻塞中的本地任务误判为失活 _active_executions: set[FileKey] = set() - def __init__(self) -> None: """初始化当前进程内的整理作业状态。""" self._job_view = {} @@ -247,11 +334,7 @@ class JobManager: with job_lock: __mediaid__ = self.__get_id(task) # 同一个源文件可能在识别前后落入不同作业，必须跨作业去重。 - if any( - self.__get_file_key(_job_task_fileitem(t)) == file_key - for job in self._job_view.values() - for t in _job_tasks(job) - ): + if file_key in self._task_state_changed_at: logger.debug(f"任务 {task.fileitem.name} 已存在，跳过重复添加") return False if __mediaid__ not in self._job_view: @@ -828,6 +911,11 @@ class JobManager: with job_lock: return list(self._job_view.values()) + def list_jobs_page(self, page: int, count: int) -> Tuple[List[TransferQueueItemData], int]: + """返回轻量队列窗口，避免复制云盘目录树和识别元数据。""" + with job_lock: + return _list_jobs_page(self._job_view, page, count, len(self._task_state_changed_at) or None) + def season_episodes( self, media: Union[MediaInfo, MusicInfo], season: Optional[int] = None ) -> List[int]: diff --git a/app/application/transfer/workflow.py b/app/application/transfer/workflow.py index ec6cf29c8..d467847f2 100644 --- a/app/application/transfer/workflow.py +++ b/app/application/transfer/workflow.py @@ -14,8 +14,11 @@ TransferJob / TransferJobTask，那两个用 app.schemas 的同名 DTO——一 视图，分开表达之后两边都不必再迁就对方。 """ from typing import ( + Any, Callable, List, + Optional, + Tuple, ) from app.application.transfer.feedback import TransferFailureNotification @@ -102,6 +105,9 @@ class TransferQueueService: remove_task: Callable[[FileItem], None], list_tasks: Callable[[], List[TransferJob]], expire_tasks: Callable[[], None], + list_tasks_page: Optional[ + Callable[[int, int], Tuple[List[Any], int]] + ] = None, ) -> None: """保存队列用例依赖，避免 Application 服务绑定具体线程队列实现。""" self._register_task = register_task @@ -112,6 +118,7 @@ class TransferQueueService: self._remove_task = remove_task self._list_tasks = list_tasks self._expire_tasks = expire_tasks + self._list_tasks_page = list_tasks_page def put(self, task: TransferTask, callback: TransferCallback) -> bool: """先持久化准入事实再入队；任何前置失败都撤销内存作业视图。""" @@ -143,3 +150,16 @@ class TransferQueueService: """先处理失活任务，再返回当前整理作业视图。""" self._expire_tasks() return self._list_tasks() + + def list_page(self, page: int, count: int) -> Tuple[List[Any], int]: + """返回受限队列快照，避免前端请求触发全量 DTO 序列化。""" + if page < 1 or count < 1: + raise ValueError("队列分页参数必须为正数") + + self._expire_tasks() + if self._list_tasks_page is not None: + return self._list_tasks_page(page, count) + + tasks = self._list_tasks() + start = (page - 1) * count + return tasks[start:start + count], len(tasks) diff --git a/app/chain/transfer/queue.py b/app/chain/transfer/queue.py index e6b24b732..b4532a417 100644 --- a/app/chain/transfer/queue.py +++ b/app/chain/transfer/queue.py @@ -42,7 +42,7 @@ from app.runtime.progress import ProgressHelper from app.runtime.stop import runtime_stop_state from app.schemas.system import TransferDirectoryConf from app.schemas.tmdb import TmdbEpisode -from app.schemas.transfer import TransferInfo, TransferJob +from app.schemas.transfer import TransferInfo, TransferJob, TransferQueueItemData from app.schemas.types import ( MediaSource, MediaType, @@ -391,6 +391,7 @@ class TransferQueueOwner(_TransferOwnerBase): remove_task=self.jobview.remove_task, list_tasks=self.jobview.list_jobs, expire_tasks=self._TransferChain__expire_stale_transfer_tasks, + list_tasks_page=self.jobview.list_jobs_page, ) def replay_pending(self) -> None: @@ -1473,6 +1474,10 @@ class TransferQueueOwner(_TransferOwnerBase): """ return self._transfer_queue_service().list() + def get_queue_tasks_page(self, page: int, count: int) -> Tuple[List[TransferQueueItemData], int]: + """获取供前端展示的受限整理队列快照。""" + return self._transfer_queue_service().list_page(page, count) + def process(self, progress_callback: Optional[Callable[..., None]] = None) -> bool: """ 获取下载器中的种子列表，并执行整理 diff --git a/app/schemas/exports.py b/app/schemas/exports.py index a81b124a8..4692de9db 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -532,6 +532,11 @@ SCHEMA_EXPORTS = { 'TransferManualReviewStepData': ('app.schemas.transfer', 'TransferManualReviewStepData'), 'TransferManualReviewTaskData': ('app.schemas.transfer', 'TransferManualReviewTaskData'), 'TransferOverwriteCheckEventData': ('app.schemas.event', 'TransferOverwriteCheckEventData'), + 'TransferQueueFileItemData': ('app.schemas.transfer', 'TransferQueueFileItemData'), + 'TransferQueueItemData': ('app.schemas.transfer', 'TransferQueueItemData'), + 'TransferQueueMediaData': ('app.schemas.transfer', 'TransferQueueMediaData'), + 'TransferQueuePageData': ('app.schemas.transfer', 'TransferQueuePageData'), + 'TransferQueueTaskData': ('app.schemas.transfer', 'TransferQueueTaskData'), 'TransferRenameBuildEventData': ('app.schemas.event', 'TransferRenameBuildEventData'), 'TransferRenameEventData': ('app.schemas.event', 'TransferRenameEventData'), 'TransferResultContractData': ('app.schemas.event', 'TransferResultContractData'), diff --git a/app/schemas/transfer.py b/app/schemas/transfer.py index 38d49604c..185fd6747 100644 --- a/app/schemas/transfer.py +++ b/app/schemas/transfer.py @@ -113,6 +113,53 @@ class TransferJob(BaseModel): tasks: Optional[List[TransferJobTask]] = Field(default_factory=list) +class TransferQueueFileItemData(BaseModel): # type: ignore[misc] + """整理队列快照只需要的源文件字段，不携带云盘目录树和 provider 扩展字段。""" + + path: Optional[str] = None + storage: Optional[str] = "local" + type: Optional[str] = None + name: Optional[str] = None + size: Optional[int] = None + + +class TransferQueueMediaData(BaseModel): # type: ignore[misc] + """整理队列 UI 所需的媒体身份与海报摘要。""" + + media_source: Optional[str] = None + media_id: Optional[str] = None + title: Optional[str] = None + title_year: Optional[str] = None + year: Optional[str] = None + poster_path: Optional[str] = None + episode_run_time: Optional[List[int]] = Field(default_factory=list) + origin_country: Optional[List[str]] = Field(default_factory=list) + + +class TransferQueueTaskData(BaseModel): # type: ignore[misc] + """整理队列 UI 所需的任务状态与源文件摘要。""" + + fileitem: TransferQueueFileItemData + state: Optional[str] = None + + +class TransferQueueItemData(BaseModel): # type: ignore[misc] + """单个媒体分组的轻量整理队列投影。""" + + media: Optional[TransferQueueMediaData] = None + season: Optional[int] = None + tasks: List[TransferQueueTaskData] = Field(default_factory=list) + + +class TransferQueuePageData(BaseModel): # type: ignore[misc] + """前端整理队列的受限快照。""" + + items: List[TransferQueueItemData] = Field(default_factory=list) + total: int = 0 + page: int = 1 + count: int = 100 + + class TransferInfo(BaseModel): """ 文件整理结果 diff --git a/docs/refactor/agent-api-surface-audit.json b/docs/refactor/agent-api-surface-audit.json index 5c3e1025b..e86eef7d1 100644 --- a/docs/refactor/agent-api-surface-audit.json +++ b/docs/refactor/agent-api-surface-audit.json @@ -6,7 +6,7 @@ "provider-skill": 12, "stream_or_binary": 10, "transport_or_identity": 76, - "ui_presentation": 13 + "ui_presentation": 14 }, "dynamic_gateway_routes": [ { @@ -21,7 +21,7 @@ "gateway_http_route_count": 229, "gateway_operation_count": 231, "matched_gateway_http_route_count": 228, - "openapi_operation_count": 421, + "openapi_operation_count": 422, "operations": [ { "disposition": "consolidated", @@ -5145,6 +5145,18 @@ "transfer" ] }, + { + "disposition": "ui_presentation", + "method": "GET", + "operation_ids": [], + "owner": "host-ui", + "path": "/api/v1/transfer/queue/page", + "reason": "Plugin-rendered page, dashboard, or navigation metadata owned by the frontend presentation contract rather than an Agent business action.", + "summary": "查询受限整理队列快照", + "tags": [ + "transfer" + ] + }, { "disposition": "gateway", "method": "GET", diff --git a/docs/refactor/agent-api-surface-audit.md b/docs/refactor/agent-api-surface-audit.md index 463e0ac78..baf951cca 100644 --- a/docs/refactor/agent-api-surface-audit.md +++ b/docs/refactor/agent-api-surface-audit.md @@ -5,7 +5,7 @@ ## Result -- OpenAPI HTTP operations: **421** +- OpenAPI HTTP operations: **422** - Stable `moviepilot_api` operations: **231** - Exact HTTP routes used by the gateway: **229** - OpenAPI routes matched directly by the gateway: **228** @@ -23,7 +23,7 @@ | `provider-skill` | 12 | Low-level downloader or media-server capability owned by a provider Skill. | | `stream_or_binary` | 10 | Streaming or binary response owned by a direct client transport. | | `transport_or_identity` | 76 | Authentication, protocol, callback, account, or conversation transport boundary. | -| `ui_presentation` | 13 | Frontend or plugin-rendered presentation contract. | +| `ui_presentation` | 14 | Frontend or plugin-rendered presentation contract. | ## Bounded Dynamic Routes @@ -423,6 +423,7 @@ | `GET` | `/api/v1/transfer/now` | transfer | `consolidated` | scheduler.run | 立即执行下载器文件整理 | | `DELETE` | `/api/v1/transfer/queue` | transfer | `gateway` | transfer.queue.delete | 从整理队列中删除任务 | | `GET` | `/api/v1/transfer/queue` | transfer | `gateway` | transfer.queue | 查询整理队列 | +| `GET` | `/api/v1/transfer/queue/page` | transfer | `ui_presentation` | host-ui | 查询受限整理队列快照 | | `GET` | `/api/v1/transfer/tasks/manual-reviews` | transfer | `gateway` | transfer.manual_reviews | 分页查询 durable 整理人工复核任务 | | `GET` | `/api/v1/transfer/tasks/{task_id}/manual-review` | transfer | `gateway` | transfer.manual_review | 查询 durable 整理人工复核详情 | | `POST` | `/api/v1/transfer/tasks/{task_id}/manual-review` | transfer | `gateway` | transfer.manual_review.resolve | 人工判定整理步骤的外部执行结果 | diff --git a/scripts/generate_agent_api_surface_audit.py b/scripts/generate_agent_api_surface_audit.py index c8a8e31b8..f405c741b 100644 --- a/scripts/generate_agent_api_surface_audit.py +++ b/scripts/generate_agent_api_surface_audit.py @@ -104,6 +104,7 @@ UI_PRESENTATION_PATHS = frozenset( "/api/v1/rule/groups/reorder", "/api/v1/storage/catalog", "/api/v1/storage/options", + "/api/v1/transfer/queue/page", } ) EXPLICIT_TRANSPORT_PATHS = frozenset( diff --git a/tests/test_transfer_queue_paging.py b/tests/test_transfer_queue_paging.py new file mode 100644 index 000000000..f8528e81e --- /dev/null +++ b/tests/test_transfer_queue_paging.py @@ -0,0 +1,111 @@ +from unittest.mock import Mock + +import pytest + +from app.application.transfer.jobs import JobManager +from app.application.transfer.workflow import TransferQueueService +from app.schemas.file import FileItem +from app.schemas.transfer import TransferJob, TransferJobTask +from tests.test_transfer_job_manager import make_task + + +def _job(*paths: str) -> TransferJob: + """构造只包含队列分页测试所需字段的整理作业。""" + return TransferJob( + season=1, + tasks=[ + TransferJobTask( + fileitem=FileItem( + name=path.rsplit("/", 1)[-1], + path=path, + size=1, + storage="local", + type="file", + children=[ + FileItem( + name="cloud-child.mkv", + path=f"{path}.child", + size=1, + storage="local", + type="file", + ) + ], + ), + state="waiting", + ) + for path in paths + ], + ) + + +def test_job_manager_page_keeps_flat_order_and_only_projects_the_requested_window(): + """大队列分页必须保持任务顺序，并按窗口复制作业投影。""" + manager = JobManager() + manager._job_view = { + ("first", 1): _job("/downloads/first-1.mkv", "/downloads/first-2.mkv"), + ("second", 1): _job( + "/downloads/second-1.mkv", + "/downloads/second-2.mkv", + "/downloads/second-3.mkv", + ), + } + + first_page, total = manager.list_jobs_page(page=1, count=3) + second_page, second_total = manager.list_jobs_page(page=2, count=3) + + assert total == second_total == 5 + assert [len(job.tasks or []) for job in first_page] == [2, 1] + assert not hasattr(first_page[0].tasks[0].fileitem, "children") + assert [task.fileitem.path for job in first_page for task in job.tasks or []] == [ + "/downloads/first-1.mkv", + "/downloads/first-2.mkv", + "/downloads/second-1.mkv", + ] + assert [task.fileitem.path for job in second_page for task in job.tasks or []] == [ + "/downloads/second-2.mkv", + "/downloads/second-3.mkv", + ] + + +@pytest.mark.parametrize("page,count", [(0, 1), (1, 0), (-1, 10)]) +def test_job_manager_page_rejects_invalid_windows(page: int, count: int): + """分页窗口不接受零或负数，避免异常切片产生隐性全量查询。""" + with pytest.raises(ValueError, match="队列分页参数必须为正数"): + JobManager().list_jobs_page(page, count) + + +def test_transfer_queue_service_uses_bounded_projection_after_expiring_tasks(): + """应用服务先清理失活任务，再调用可选的受限投影依赖。""" + expire_tasks = Mock() + list_tasks_page = Mock(return_value=([_job("/downloads/page.mkv")], 101)) + service = TransferQueueService( + register_task=lambda _task: True, + admit_task=lambda _task: None, + enqueue=lambda _queue: None, + before_enqueue=lambda _task: None, + enqueue_failed=lambda _task, _error: None, + remove_task=lambda _fileitem: None, + list_tasks=lambda: [], + expire_tasks=expire_tasks, + list_tasks_page=list_tasks_page, + ) + + items, total = service.list_page(page=3, count=50) + + assert len(items) == 1 + assert total == 101 + expire_tasks.assert_called_once_with() + list_tasks_page.assert_called_once_with(3, 50) + + +def test_job_manager_duplicate_index_is_released_when_a_cloud_task_is_removed(): + """批量云盘任务的 O(1) 去重索引不能阻止同一源文件后续重新入队。""" + manager = JobManager() + task = make_task(1) + + assert manager.add_task(task) is True + assert manager.add_task(make_task(1)) is False + + manager.remove_task(task.fileitem) + + assert manager.add_task(make_task(1)) is True