perf(transfer): bound queue snapshots for batch imports

2026-10-31 20:34:48 +08:00 · 2026-10-01 12:45:31 +08:00
parent 1ec003bd56
commit 56cf6fefe0
12 changed files with 334 additions and 17 deletions
--- 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
--- a/app/api/endpoints/transferqueue.py
+++ 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,
+    )
--- 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",)),
--- 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]:
--- 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)
--- 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:
        """
        获取下载器中的种子列表，并执行整理
--- 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'),
--- 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):
    """
    文件整理结果
--- 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",
--- 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 | 人工判定整理步骤的外部执行结果 |
--- 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(
--- a/tests/test_transfer_queue_paging.py
+++ 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