提示

返回博客列表

用 Docker 部署私人视频下载服务——粘贴链接即下,支持 Web 界面和 API

用 Docker 部署私人视频下载服务——粘贴链接即下,支持 Web 界面和 API

你有一台 NAS 或者 VPS 24 小时在线,想在上面跑一个视频下载服务——浏览器打开一个网页,粘贴链接就能下,下完自动存到指定目录。这样你在手机上也能提交下载任务,电脑关机了也不影响。听起来需要前后端开发?其实用 Docker 把 yt-dlp + 一个轻量 Web 界面打包,半小时就能跑起来。这篇文章教你用 Docker Compose 部署一套完整的私人视频下载站:Web 界面提交任务 → 后台队列下载 → 自动归档到媒体库 → 手机也能用。

TL;DR:核心架构:Flask/FastAPI Web 服务(接收下载请求) + Redis 任务队列 + yt-dlp Worker(执行下载) + Nginx 提供文件访问。Docker Compose 一键编排,环境变量配置端口和路径。推荐方案:metube(开源 YouTube 下载器,自带 Web UI)或自己用 FastAPI + Celery 搭建。部署到 NAS/VPS 后,手机浏览器打开就能用。

目录

一、方案选型:用现成的还是自己搭

方案 开发量 功能完整度 灵活性 适合人群
metube 0(直接 docker run) ⭐⭐⭐ ⭐⭐ 只要下载 YouTube,不想写代码
yt-dlp-web 0 ⭐⭐ ⭐⭐ 简单 Web 界面
自建 FastAPI 2-3 小时 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ 需要自定义功能、多平台支持

如果你只需要下载 YouTube 视频,metube 足够了。如果你需要支持多平台(B 站/抖音/快手)、自定义命名规则、下载后自动转码或上传到云存储,那就自己搭。

二、方案一:metube——开箱即用的 YouTube 下载器

2.1 Docker Compose 部署

# docker-compose.yml
version: '3.8'

services:
  metube:
    image: ghcr.io/alexta69/metube:latest
    container_name: metube
    restart: unless-stopped
    ports:
      - "8081:8081"
    volumes:
      - ./downloads:/downloads
    environment:
      - OUTPUT_TEMPLATE=%(uploader)s - %(title)s.%(ext)s
      - DOWNLOAD_DIR=/downloads
      - STATE_DIR=/downloads/.metube
# 启动
docker compose up -d

# 访问 http://你的IP:8081

metube 的界面极其简洁:一个输入框 + 一个下载按钮。粘贴 YouTube 链接 → 选择画质 → 下载。支持播放列表批量下载。

2.2 进阶配置

environment:
  # 默认下载画质
  - DEFAULT_OPTION_RES=bestvideo+bestaudio/best

  # 播放列表下载模式
  - YTDL_OPTIONS={"playlist_items":"1-10"}

  # 请求间隔(避免被限流)
  - YTDL_OPTIONS={"sleep_interval":2}

  # 自动清理(保留最近 100 个下载记录)
  - DELETE_DOWNLOADS_OLDER_THAN=7  # 删除 7 天前的文件

三、方案二:自建 FastAPI 下载服务

如果你需要多平台支持、自定义逻辑,推荐自建。

3.1 目录结构

download-service/
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── app/
│   ├── main.py          # FastAPI 应用
│   ├── worker.py        # 下载 Worker
│   ├── models.py        # 任务模型
│   └── templates/
│       └── index.html   # Web 界面
└── downloads/            # 下载目录(挂载卷)

3.2 FastAPI 应用

# app/main.py
import os
import json
import subprocess
from pathlib import Path
from datetime import datetime
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, BackgroundTasks
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.templating import Jinja2Templates
import redis


# Redis 任务队列
redis_client = redis.Redis(
    host=os.getenv('REDIS_HOST', 'redis'),
    port=int(os.getenv('REDIS_PORT', 6379)),
    db=0,
    decode_responses=True,
)

DOWNLOAD_DIR = os.getenv('DOWNLOAD_DIR', '/downloads')
TEMPLATES_DIR = os.path.join(os.path.dirname(__file__), 'templates')

templates = Jinja2Templates(directory=TEMPLATES_DIR)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """启动和关闭时的操作"""
    os.makedirs(DOWNLOAD_DIR, exist_ok=True)
    yield


app = FastAPI(title="私人视频下载服务", lifespan=lifespan)


@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    """Web 界面"""
    return templates.TemplateResponse("index.html", {"request": request})


@app.post("/api/download")
async def submit_download(url: str, quality: str = "best"):
    """提交下载任务"""
    task_id = datetime.now().strftime("%Y%m%d%H%M%S%f")

    task = {
        'id': task_id,
        'url': url,
        'quality': quality,
        'status': 'pending',
        'created_at': datetime.now().isoformat(),
        'title': '',
        'progress': 0,
        'error': '',
    }

    # 推送到 Redis 队列
    redis_client.lpush('download_queue', json.dumps(task))
    redis_client.hset(f'task:{task_id}', mapping=task)

    return {'task_id': task_id, 'status': 'pending'}


@app.get("/api/tasks")
async def list_tasks():
    """列出所有任务"""
    keys = redis_client.keys('task:*')
    tasks = []
    for key in keys:
        task = redis_client.hgetall(key)
        tasks.append(task)

    tasks.sort(key=lambda t: t.get('created_at', ''), reverse=True)
    return {'tasks': tasks}


@app.get("/api/tasks/{task_id}")
async def get_task(task_id: str):
    """查询单个任务状态"""
    task = redis_client.hgetall(f'task:{task_id}')
    if not task:
        return {'error': '任务不存在'}
    return task


@app.get("/files")
async def list_files():
    """列出已下载的文件"""
    files = []
    for f in sorted(Path(DOWNLOAD_DIR).iterdir(),
                    key=lambda x: x.stat().st_mtime,
                    reverse=True):
        if f.is_file():
            files.append({
                'name': f.name,
                'size_mb': round(f.stat().st_size / 1024 / 1024, 1),
                'modified': datetime.fromtimestamp(
                    f.stat().st_mtime
                ).isoformat(),
            })
    return {'files': files}


@app.get("/files/{filename}")
async def download_file(filename: str):
    """下载文件"""
    filepath = os.path.join(DOWNLOAD_DIR, filename)
    if not os.path.exists(filepath):
        return {'error': '文件不存在'}
    return FileResponse(filepath, filename=filename)

3.3 下载 Worker

# app/worker.py
import os
import json
import time
import subprocess
import redis

redis_client = redis.Redis(
    host=os.getenv('REDIS_HOST', 'redis'),
    port=int(os.getenv('REDIS_PORT', 6379)),
    db=0,
    decode_responses=True,
)

DOWNLOAD_DIR = os.getenv('DOWNLOAD_DIR', '/downloads')
MAX_CONCURRENT = int(os.getenv('MAX_CONCURRENT', '3'))

import threading
semaphore = threading.Semaphore(MAX_CONCURRENT)


def process_task(task_data):
    """处理单个下载任务"""
    task = json.loads(task_data)
    task_id = task['id']

    with semaphore:
        try:
            # 更新状态
            redis_client.hset(f'task:{task_id}', 'status', 'downloading')

            # 构建 yt-dlp 命令
            output_template = os.path.join(
                DOWNLOAD_DIR,
                '%(uploader)s - %(title)s [%(resolution)s].%(ext)s'
            )

            cmd = [
                'yt-dlp',
                '--newline',                    # 逐行输出进度
                '--output', output_template,
                '--format', f'{task["quality"]}+bestaudio/best',
                '--merge-output-format', 'mp4',
                '--write-info-json',
                '--no-overwrites',
                '--retries', '3',
                task['url'],
            ]

            # 实时捕获进度
            process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
                cwd=DOWNLOAD_DIR,
            )

            for line in process.stdout:
                # 解析 yt-dlp 的进度输出
                if '[download]' in line and '%' in line:
                    try:
                        pct = line.split('%')[0].split()[-1]
                        redis_client.hset(
                            f'task:{task_id}',
                            'progress', pct
                        )
                    except (IndexError, ValueError):
                        pass

                # 提取标题
                if '[download] Destination:' in line:
                    filename = line.split('Destination:')[1].strip()
                    redis_client.hset(
                        f'task:{task_id}',
                        'title',
                        os.path.basename(filename)
                    )

            process.wait()

            if process.returncode == 0:
                redis_client.hset(f'task:{task_id}', 'status', 'completed')
                redis_client.hset(f'task:{task_id}', 'progress', '100')
            else:
                redis_client.hset(f'task:{task_id}', 'status', 'failed')
                redis_client.hset(
                    f'task:{task_id}',
                    'error',
                    f'退出码: {process.returncode}'
                )

        except Exception as e:
            redis_client.hset(f'task:{task_id}', 'status', 'failed')
            redis_client.hset(f'task:{task_id}', 'error', str(e))


def main():
    """Worker 主循环"""
    print("Worker 已启动,等待任务...")
    while True:
        # 阻塞式从队列取任务
        task_data = redis_client.brpop('download_queue', timeout=5)
        if task_data:
            _, data = task_data
            # 在新线程中处理(控制并发数)
            threading.Thread(
                target=process_task,
                args=(data,),
                daemon=True
            ).start()
        else:
            time.sleep(1)


if __name__ == '__main__':
    main()

3.4 Docker Compose

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    container_name: download-web
    restart: unless-stopped
    ports:
      - "8080:8000"
    volumes:
      - ./downloads:/downloads
    environment:
      - REDIS_HOST=redis
      - DOWNLOAD_DIR=/downloads
    depends_on:
      - redis

  worker:
    build: .
    container_name: download-worker
    restart: unless-stopped
    command: python app/worker.py
    volumes:
      - ./downloads:/downloads
    environment:
      - REDIS_HOST=redis
      - DOWNLOAD_DIR=/downloads
      - MAX_CONCURRENT=3
    depends_on:
      - redis

  redis:
    image: redis:7-alpine
    container_name: download-redis
    restart: unless-stopped
    volumes:
      - ./redis-data:/data

3.5 Dockerfile

FROM python:3.11-slim

RUN apt-get update && \
    apt-get install -y ffmpeg && \
    rm -rf /var/lib/apt/lists/*

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app/ ./app/

# 默认运行 Web 服务,Worker 通过 command 覆盖
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

3.6 Web 界面(极简版)

<!-- app/templates/index.html -->
<!DOCTYPE html>
<html lang="zh">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>私人视频下载</title>
  <style>
    * { box-sizing: border-box; margin: 0; padding: 0; }
    body { font-family: system-ui; max-width: 800px; margin: 40px auto;
           padding: 20px; background: #f8f9fa; }
    .card { background: #fff; border-radius: 8px; padding: 20px;
            margin-bottom: 16px; box-shadow: 0 1px 3px rgba(0,0,0,.1); }
    input[type="text"] { width: 100%; padding: 10px; border: 1px solid #ddd;
                         border-radius: 4px; font-size: 14px; }
    button { padding: 10px 24px; background: #2563eb; color: #fff;
             border: none; border-radius: 4px; cursor: pointer; font-size: 14px; }
    button:hover { background: #1d4ed8; }
    .task { display: flex; justify-content: space-between; align-items: center;
            padding: 10px 0; border-bottom: 1px solid #eee; }
    .status { font-size: 12px; padding: 2px 8px; border-radius: 10px; }
    .pending { background: #fef3c7; color: #92400e; }
    .downloading { background: #dbeafe; color: #1e40af; }
    .completed { background: #d1fae5; color: #065f46; }
    .failed { background: #fee2e2; color: #991b1b; }
    .file-link { color: #2563eb; text-decoration: none; }
  </style>
</head>
<body>
  <h1 style="margin-bottom:20px;">📥 私人视频下载</h1>

  <div class="card">
    <form id="downloadForm">
      <input type="text" id="urlInput"
             placeholder="粘贴视频链接(支持 YouTube/B站/抖音等)" required>
      <div style="margin-top:10px;display:flex;gap:10px;align-items:center;">
        <select id="quality">
          <option value="bestvideo+bestaudio">最佳画质</option>
          <option value="bestvideo[height<=1080]+bestaudio">1080P</option>
          <option value="bestvideo[height<=720]+bestaudio">720P</option>
          <option value="bestaudio">仅音频</option>
        </select>
        <button type="submit">开始下载</button>
      </div>
    </form>
  </div>

  <div class="card">
    <h3>下载任务</h3>
    <div id="taskList">加载中...</div>
  </div>

  <script>
    const form = document.getElementById('downloadForm');
    const taskList = document.getElementById('taskList');

    form.addEventListener('submit', async (e) => {
      e.preventDefault();
      const url = document.getElementById('urlInput').value;
      const quality = document.getElementById('quality').value;

      const resp = await fetch(
        `/api/download?url=${encodeURIComponent(url)}&quality=${quality}`,
        { method: 'POST' }
      );
      const data = await resp.json();
      document.getElementById('urlInput').value = '';
      loadTasks();
    });

    async function loadTasks() {
      const resp = await fetch('/api/tasks');
      const data = await resp.json();

      if (!data.tasks.length) {
        taskList.innerHTML = '<p style="color:#999;">暂无任务</p>';
        return;
      }

      taskList.innerHTML = data.tasks.map(t => `
        <div class="task">
          <div style="flex:1;">
            <div style="font-weight:500;">${t.title || t.url.substring(0,60)}</div>
            <div style="font-size:12px;color:#999;">${t.created_at}</div>
          </div>
          <span class="status ${t.status}">${t.status}
            ${t.status === 'downloading' ? ` ${t.progress}%` : ''}
          </span>
        </div>
      `).join('');
    }

    // 每 3 秒刷新任务列表
    setInterval(loadTasks, 3000);
    loadTasks();
  </script>
</body>
</html>

四、部署到 NAS / VPS

4.1 群晖 NAS

# SSH 到群晖
ssh admin@your-nas-ip

# 安装 Docker(套件中心 → Docker → 安装)

# 创建目录
mkdir -p /volume1/docker/download-service/downloads
cd /volume1/docker/download-service

# 上传 docker-compose.yml
# docker compose up -d

# 访问 http://nas-ip:8080

4.2 Linux VPS

# 克隆或上传项目文件
cd /opt
mkdir download-service && cd download-service

# 启动
docker compose up -d

# 查看日志
docker compose logs -f

五、外网访问与安全加固

5.1 添加密码保护

# 在 main.py 中添加简单的密码验证
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

ADMIN_USER = os.getenv('ADMIN_USER', 'admin')
ADMIN_PASS = os.getenv('ADMIN_PASS', 'changeme')


def verify(credentials: HTTPBasicCredentials = Depends(security)):
    if (credentials.username != ADMIN_USER or
        credentials.password != ADMIN_PASS):
        raise HTTPException(status_code=401, detail="未授权")


@app.post("/api/download")
async def submit_download(
    url: str,
    quality: str = "best",
    credentials: HTTPBasicCredentials = Depends(security)
):
    verify(credentials)
    # ... 原有逻辑

5.2 Nginx 反向代理 + HTTPS

server {
    listen 443 ssl;
    server_name download.your-domain.com;

    ssl_certificate /etc/ssl/certs/your-domain.pem;
    ssl_certificate_key /etc/ssl/private/your-domain.key;

    # 限制上传速率(防止滥用)
    client_max_body_size 10M;
    limit_req_zone $binary_remote_addr zone=download:10m rate=10r/m;

    location / {
        limit_req zone=download burst=5;
        proxy_pass http://127.0.0.1:8080;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

六、监控与维护

6.1 磁盘空间监控

# 在 main.py 添加健康检查
@app.get("/health")
async def health_check():
    import shutil
    usage = shutil.disk_usage(DOWNLOAD_DIR)
    free_gb = usage.free / 1024 / 1024 / 1024

    return {
        'status': 'ok' if free_gb > 5 else 'warning',
        'disk_free_gb': round(free_gb, 1),
        'disk_total_gb': round(usage.total / 1024 / 1024 / 1024, 1),
    }

6.2 定期清理

# 添加到 worker.py
import schedule

def cleanup_old_files(days=30):
    """清理 N 天前的下载文件"""
    import time
    cutoff = time.time() - days * 86400
    for f in Path(DOWNLOAD_DIR).iterdir():
        if f.is_file() and f.stat().st_mtime < cutoff:
            f.unlink()
            print(f"已清理: {f.name}")

# 每天凌晨 3 点运行
schedule.every().day.at("03:00").do(cleanup_old_files, days=30)

七、合规与温馨提示


用 Docker 部署私人下载站这件事,我最大的感受是:部署花半小时,受益好几年。 现在我在外面用手机刷到一个好视频,打开自己部署的网页,粘贴链接,回家打开 Jellyfin 就能看。整套流程全自动,不需要电脑开机、不需要传文件、不需要任何手动操作。如果你的 NAS 或 VPS 已经在跑 Docker,强烈建议顺手搭一个。

本文由 VidDown 技术博客原创发布。如果你不想自己部署,VidDown 在线版提供了一样的体验——浏览器打开就能下载,支持 30+ 平台。访问 VidDown 了解更多。

想亲手试试?用 VidDown 一键解析下载

粘贴视频链接即可解析,多平台支持、网页端即用;下载桌面客户端解锁海外平台本地解析,开通会员更享不限次下载。

顶部