feat: 外部下载任务队列（网页一键走家用 worker）

backend/app/__init__.py

@@ -22,7 +22,7 @@ async def verify_web_access_password(
     return True
 
 def create_app(lifespan) -> FastAPI:
-    from .routers import note, notification, provider, model, config, chat, flashcard, hot_videos, article, trend_subscription, feishu
+    from .routers import note, notification, provider, model, config, chat, flashcard, hot_videos, article, trend_subscription, feishu, worker
     from .utils.response import ResponseWrapper as R
 
     app = FastAPI(title="VideoMemo",lifespan=lifespan)
@@ -43,5 +43,6 @@ def create_app(lifespan) -> FastAPI:
     app.include_router(trend_subscription.router, prefix="/api", dependencies=protected)
     app.include_router(notification.router, prefix="/api", dependencies=protected)
     app.include_router(feishu.router, prefix="/api", dependencies=protected)
+    app.include_router(worker.router, prefix="/api", dependencies=protected)
 
     return app

backend/app/db/download_job_dao.py

@@ -0,0 +1,77 @@
+"""外部下载任务队列的读写。"""
+import uuid
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from app.db.engine import SessionLocal
+from app.db.models.download_job import DownloadJob
+
+
+def create_job(task_id: str, url: str, platform: str, want_video: bool,
+               params: Dict[str, Any]) -> str:
+    jid = uuid.uuid4().hex
+    with SessionLocal() as db:
+        db.add(DownloadJob(
+            id=jid, task_id=task_id, url=url, platform=platform,
+            want_video=want_video, params=params, status="pending",
+        ))
+        db.commit()
+    return jid
+
+
+def claim_next() -> Optional[Dict[str, Any]]:
+    """认领最早的一个 pending 任务，标记 claimed 并返回其信息；没有则 None。
+
+    个人部署通常只有一个 worker，简单的 select-then-update 足够。
+    """
+    with SessionLocal() as db:
+        job = (db.query(DownloadJob)
+                 .filter(DownloadJob.status == "pending")
+                 .order_by(DownloadJob.created_at.asc())
+                 .first())
+        if job is None:
+            return None
+        data = {
+            "job_id": job.id,
+            "task_id": job.task_id,
+            "url": job.url,
+            "want_video": bool(job.want_video),
+        }
+        job.status = "claimed"
+        job.claimed_at = datetime.now()
+        db.commit()
+        return data
+
+
+def get_job(job_id: str) -> Optional[Dict[str, Any]]:
+    with SessionLocal() as db:
+        job = db.get(DownloadJob, job_id)
+        if job is None:
+            return None
+        return {
+            "id": job.id,
+            "task_id": job.task_id,
+            "url": job.url,
+            "platform": job.platform,
+            "want_video": bool(job.want_video),
+            "params": job.params or {},
+            "status": job.status,
+        }
+
+
+def complete_job(job_id: str, file_url: str) -> None:
+    with SessionLocal() as db:
+        job = db.get(DownloadJob, job_id)
+        if job is not None:
+            job.status = "done"
+            job.file_url = file_url
+            db.commit()
+
+
+def fail_job(job_id: str, error: str) -> None:
+    with SessionLocal() as db:
+        job = db.get(DownloadJob, job_id)
+        if job is not None:
+            job.status = "failed"
+            job.error = (error or "")[:1000]
+            db.commit()

backend/app/db/init_db.py

@@ -4,6 +4,7 @@ import os
 
 from app.db.models.app_config import AppConfig
 from app.db.models.articles import ArticleItem, ArticleSubscription, ArticleSubscriptionItem
+from app.db.models.download_job import DownloadJob
 from app.db.models.models import Model
 from app.db.models.note import Note
 from app.db.models.providers import Provider

backend/app/db/models/download_job.py

@@ -0,0 +1,25 @@
+from sqlalchemy import Column, String, JSON, Boolean, DateTime, func
+
+from app.db.engine import Base
+
+
+class DownloadJob(Base):
+    """外部下载任务队列：网页发起的某些平台（默认 YouTube）下载，交给跑在住宅 IP 上的
+    家用 worker 拉取执行，绕开机房 IP 风控。
+
+    家里 worker 轮询认领 pending 任务 → 下载 → 回传文件 → 服务器用原 task_id 跑本地管线。
+    """
+
+    __tablename__ = "download_jobs"
+
+    id = Column(String, primary_key=True)            # job uuid
+    task_id = Column(String, nullable=False)         # 关联的笔记任务（沿用它跑后续管线）
+    url = Column(String, nullable=False)
+    platform = Column(String, nullable=False)
+    want_video = Column(Boolean, default=False)      # True=要截图→下视频；False=只下音频
+    params = Column(JSON, nullable=True)             # run_note_task 的其余参数，回传后原样重放
+    status = Column(String, default="pending")       # pending / claimed / done / failed
+    file_url = Column(String, nullable=True)         # 回传后的 /uploads/xxx
+    error = Column(String, nullable=True)
+    created_at = Column(DateTime, server_default=func.now())
+    claimed_at = Column(DateTime, nullable=True)

backend/app/routers/note.py

@@ -12,6 +12,7 @@ from dataclasses import asdict
 
 from app.db.note_dao import save_note, load_note, get_status
 from app.db.video_task_dao import get_task_by_video
+from app.services import external_download
 from app.enmus.exception import NoteErrorEnum
 from app.enmus.note_enums import DownloadQuality
 from app.exceptions.note import NoteError
@@ -494,6 +495,39 @@ def generate_note(data: VideoRequest, background_tasks: BackgroundTasks):
             except Exception as e:
                 logger.warning(f"写入预取字幕失败 (task_id={task_id}): {e}")
 
+        # 外部下载：某些平台（默认 YouTube）机房 IP 被风控，改交给跑在住宅 IP 上的家用 worker。
+        # 没有预取字幕（确实要下载）且命中外部下载平台时，入队等 worker 拉取，不在服务器本地下。
+        if not data.prefetched_transcript and external_download.should_external(data.platform):
+            if not external_download.worker_alive():
+                return R.error(
+                    msg="本地下载器未运行：请在你自己的机器上启动 downloader-worker 守护进程后重试"
+                        "（见 downloader-worker/README）。",
+                    code=300103,
+                )
+            from app.db import download_job_dao
+            download_job_dao.create_job(
+                task_id=task_id,
+                url=data.video_url,
+                platform=data.platform,
+                want_video=bool(data.screenshot),
+                params={
+                    "quality": data.quality.value if hasattr(data.quality, "value") else data.quality,
+                    "link": data.link,
+                    "screenshot": data.screenshot,
+                    "model_name": data.model_name,
+                    "provider_id": data.provider_id,
+                    "format": data.format,
+                    "style": data.style,
+                    "extras": data.extras,
+                    "video_understanding": data.video_understanding,
+                    "video_interval": data.video_interval,
+                    "grid_size": data.grid_size,
+                },
+            )
+            NoteGenerator()._update_status(task_id, TaskStatus.DOWNLOADING, message="等待本地下载器获取…")
+            logger.info(f"任务 {task_id} 走外部下载，已入队等待家用 worker")
+            return R.success({"task_id": task_id})
+
         background_tasks.add_task(run_note_task, task_id, data.video_url, data.platform, data.quality, data.link,
                                   data.screenshot, data.model_name, data.provider_id, data.format, data.style,
                                   data.extras, data.video_understanding, data.video_interval, data.grid_size)

backend/app/routers/worker.py

@@ -0,0 +1,98 @@
+"""家用下载 worker 的对接接口（都在 /api 下，复用全局访问密码鉴权）。
+
+  GET  /worker/next      worker 轮询认领一个 pending 下载任务（顺带上报心跳）
+  POST /worker/complete  worker 下载完回传文件 → 服务器用原 task_id 跑本地管线
+  POST /worker/fail      worker 下载失败上报 → 任务标记失败
+  GET  /worker/config    查看外部下载配置（前端可选用）
+  POST /worker/config    改外部下载配置（enabled / platforms）
+"""
+import os
+
+from fastapi import APIRouter, BackgroundTasks, File, Form, UploadFile
+from pydantic import BaseModel
+
+from app.db import download_job_dao as jobs
+from app.enmus.note_enums import DownloadQuality
+from app.enmus.task_status_enums import TaskStatus
+from app.services import external_download
+from app.utils.response import ResponseWrapper as R
+
+router = APIRouter()
+
+
+@router.get("/worker/next")
+def worker_next():
+    external_download.mark_worker_seen()
+    job = jobs.claim_next()
+    return R.success(job or {})
+
+
+@router.post("/worker/complete")
+async def worker_complete(
+    background_tasks: BackgroundTasks,
+    job_id: str = Form(...),
+    file: UploadFile = File(...),
+):
+    from app.routers.note import UPLOAD_DIR, run_note_task
+
+    external_download.mark_worker_seen()
+    job = jobs.get_job(job_id)
+    if job is None:
+        return R.error(msg="下载任务不存在", code=404)
+
+    os.makedirs(UPLOAD_DIR, exist_ok=True)
+    dest = os.path.join(UPLOAD_DIR, file.filename)
+    with open(dest, "wb+") as f:
+        f.write(await file.read())
+    file_url = f"/uploads/{file.filename}"
+    jobs.complete_job(job_id, file_url)
+
+    # 用原 task_id 跑本地管线（platform=local），把网页发起时的参数原样重放
+    p = job["params"] or {}
+    try:
+        quality = DownloadQuality(p.get("quality", "fast"))
+    except Exception:
+        quality = DownloadQuality.fast
+    background_tasks.add_task(
+        run_note_task,
+        job["task_id"], file_url, "local", quality,
+        p.get("link", False), p.get("screenshot", False),
+        p.get("model_name"), p.get("provider_id"), p.get("format", []),
+        p.get("style"), p.get("extras"), p.get("video_understanding", False),
+        p.get("video_interval", 0), p.get("grid_size", []),
+    )
+    return R.success({"task_id": job["task_id"]})
+
+
+class WorkerFailRequest(BaseModel):
+    job_id: str
+    error: str = "下载失败"
+
+
+@router.post("/worker/fail")
+def worker_fail(data: WorkerFailRequest):
+    from app.services.note import NoteGenerator
+
+    external_download.mark_worker_seen()
+    job = jobs.get_job(data.job_id)
+    jobs.fail_job(data.job_id, data.error)
+    if job is not None:
+        NoteGenerator()._update_status(job["task_id"], TaskStatus.FAILED, message=data.error)
+    return R.success()
+
+
+@router.get("/worker/config")
+def get_worker_config():
+    cfg = external_download.get_config()
+    cfg["worker_online"] = external_download.worker_alive()
+    return R.success(cfg)
+
+
+class WorkerConfigRequest(BaseModel):
+    enabled: bool
+    platforms: list = ["youtube"]
+
+
+@router.post("/worker/config")
+def set_worker_config(data: WorkerConfigRequest):
+    return R.success(external_download.set_config(data.enabled, data.platforms))

backend/app/services/external_download.py

@@ -0,0 +1,44 @@
+"""外部下载（家用 worker）路由策略 + worker 心跳。
+
+某些平台（默认 YouTube）在机房 IP 上会被风控，改由跑在住宅 IP 上的家用 worker
+拉取下载。这里集中：是否该走外部下载、worker 是否在线。
+
+配置存 app_config（key="external_downloader"）：{enabled: bool, platforms: [str]}。
+默认开启、只对 youtube 生效——启动了 worker 就能用；没启动则入口会提示先开 worker。
+"""
+import time
+from typing import Dict, Any
+
+from app.db import app_config_dao
+
+_KEY = "external_downloader"
+_DEFAULT = {"enabled": True, "platforms": ["youtube"]}
+
+# worker 最近一次轮询的时间戳（进程内存；HF 重启清零，worker 几秒一轮很快回填）
+_last_seen = {"ts": 0.0}
+
+
+def get_config() -> Dict[str, Any]:
+    cfg = app_config_dao.get_value(_KEY) or {}
+    return {
+        "enabled": bool(cfg.get("enabled", _DEFAULT["enabled"])),
+        "platforms": cfg.get("platforms") or _DEFAULT["platforms"],
+    }
+
+
+def set_config(enabled: bool, platforms) -> Dict[str, Any]:
+    app_config_dao.set_value(_KEY, {"enabled": bool(enabled), "platforms": list(platforms or [])})
+    return get_config()
+
+
+def should_external(platform: str) -> bool:
+    cfg = get_config()
+    return cfg["enabled"] and platform in cfg["platforms"]
+
+
+def mark_worker_seen() -> None:
+    _last_seen["ts"] = time.time()
+
+
+def worker_alive(within: float = 90.0) -> bool:
+    return (time.time() - _last_seen["ts"]) < within