aria2 反向拉卷端点 /api/transfer/{id}/pull (并发拉取+receiver校验+失败回退)

api.py

@@ -19,6 +19,7 @@ import base64
 import json
 import logging
 import shutil
+import subprocess
 from pathlib import Path
 
 from fastapi import Depends, FastAPI, Header, HTTPException, Request
@@ -238,6 +239,66 @@ async def put_chunk(transfer_id: str, idx: int, request: Request, username: str
     return {"ok": True}
 
 
+@app.post("/api/transfer/{transfer_id}/pull")
+async def pull_chunks(transfer_id: str, request: Request, username: str = Depends(verify_token)):
+    """aria2 反向拉卷: 客户端起临时 HTTP 服务, 服务端 aria2c 并发拉取 (打满带宽)
+
+    body: {base_url, token} —— 客户端临时服务地址 + 一次性拉取令牌
+    卷文件拉取后逐卷走 receiver.receive 校验 (大小+SHA-256) 落盘。
+    """
+    task = tasks.get(transfer_id)
+    if task is None:
+        return JSONResponse(status_code=404, content={"error": "任务不存在"})
+    try:
+        body = await request.json()
+        base_url = str(body["base_url"]).rstrip("/")
+        pull_token = str(body["token"])
+    except (KeyError, TypeError, ValueError):
+        return JSONResponse(status_code=400, content={"error": "缺少 base_url/token"})
+    aria2 = shutil.which("aria2c")
+    if aria2 is None:
+        return JSONResponse(status_code=409, content={"error": "服务端未安装 aria2 (apt install aria2)"})
+    chunk_count = task.get("chunk_count", 0)
+    chunk_dir = TMP_ROOT / transfer_id
+    chunk_dir.mkdir(parents=True, exist_ok=True)
+    urls_file = chunk_dir / "urls.txt"
+    with open(urls_file, "w", encoding="utf-8") as f:
+        for idx in range(1, chunk_count + 1):
+            f.write(f"{base_url}/pull/{transfer_id}/{idx}?token={pull_token}\n")
+    try:
+        r = subprocess.run(
+            [aria2, "-i", str(urls_file), "-d", str(chunk_dir),
+             "--max-concurrent-downloads=8", "--split=4", "--max-connection-per-server=4",
+             "--min-split-size=1M", "--file-allocation=none", "--allow-overwrite=true",
+             "--auto-file-renaming=false", "--console-log-level=warn",
+             "--summary-interval=1", "--quiet=true", "--no-conf=true"],
+            timeout=3600, capture_output=True, text=True,
+        )
+    except OSError as e:
+        return JSONResponse(status_code=500, content={"error": f"aria2 启动失败: {e}"})
+    if r.returncode != 0:
+        return JSONResponse(status_code=409, content={"error": f"aria2 拉取失败: {r.stdout[-200:]}"})
+    # 逐卷校验落盘 (大小 + SHA-256); aria2 落盘文件名 = URL 末段 (卷序号)
+    pulled = 0
+    for p in sorted(chunk_dir.iterdir()):
+        if p.name == "urls.txt" or p.suffix in (".aria2", ".tmp"):
+            continue
+        try:
+            idx = int(p.name)
+        except ValueError:
+            continue
+        try:
+            receiver.receive(transfer_id, idx, p.read_bytes())
+            p.unlink(missing_ok=True)
+            pulled += 1
+        except ReceiverError:
+            pass  # 校验失败的卷忽略, 客户端会补传
+    urls_file.unlink(missing_ok=True)
+    log.info("pull %s %s %d/%d 卷 (aria2)", request.client.host if request.client else "?",
+             transfer_id, pulled, chunk_count)
+    return {"ok": True, "pulled": pulled, "total": chunk_count}
+
+
 @app.post("/api/transfer/{transfer_id}/complete")
 async def complete(transfer_id: str, request: Request, username: str = Depends(verify_token)):
     if tasks.get(transfer_id) is None: