＜Project-20 YT-DLP＞ 给视频网站下载工具 yt-dlp/yt-dlp 加个页面 python web

介绍 yt-dlp

Github 项目：https://github.com/yt-dlp/yt-dlp

A feature-rich command-line audio/video downloader

一个功能丰富的视频与音频命令行下载器

原因与功能

之前我用的 cobalt 因为它不再提供Client Web功能，只能去它的官网使用。 翻 reddit 找到这个 YT-DLP，但它是个命令行工具，考虑参数大多很少用到，给它加个web 壳子，又可以放到docker里面运行。

在网页填入url，只列出含有视频+音频的文件。点下载后，文件可以保存在本地。命令的运行输出也在页面上显示。占用端口: 9012

YT-DLP 程序

代码在 Claude AI 帮助下完成，前端全靠它，Nice~ 

界面

目录结构

20.YT-DLP/
├── Dockerfile
├── app.py
├── static/
│   ├── css/
│   │   └── style.css
│   └── js/
│       └── script.js
├── templates/
│   └── index.html
└── temp_downloads/

完整代码

1. app.py

# app.py
from flask import Flask, render_template, request, jsonify, send_file
import yt_dlp
import os
import shutil
from werkzeug.utils import secure_filename
import time
import logging
import queue
from datetime import datetime
import sysapp = Flask(__name__)# 创建固定的临时目录
TEMP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'temp_downloads')
if not os.path.exists(TEMP_DIR):os.makedirs(TEMP_DIR)# 存储下载信息的字典
DOWNLOADS = {}# 创建日志队列
log_queue = queue.Queue(maxsize=1000)class QueueHandler(logging.Handler):def __init__(self, log_queue):super().__init__()self.log_queue = log_queuedef emit(self, record):try:# 过滤掉 Werkzeug 的常规访问日志if record.name == 'werkzeug' and any(x in record.getMessage() for x in ['127.0.0.1','GET /api/logs','GET /static/','"GET / HTTP/1.1"']):return# 清理消息格式msg = self.format(record)if record.name == 'app':# 移除 "INFO:app:" 等前缀msg = msg.split(' - ')[-1]log_entry = {'timestamp': datetime.fromtimestamp(record.created).isoformat(),'message': msg,'level': record.levelname.lower(),'logger': record.name}# 如果队列满了，移除最旧的日志if self.log_queue.full():try:self.log_queue.get_nowait()except queue.Empty:passself.log_queue.put(log_entry)except Exception as e:print(f"Error in QueueHandler: {e}")# 配置日志格式
log_formatter = logging.Formatter('%(message)s')# 配置队列处理器
queue_handler = QueueHandler(log_queue)
queue_handler.setFormatter(log_formatter)# 配置控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(log_formatter)# 配置 Flask 日志
app.logger.handlers = []
app.logger.addHandler(queue_handler)
app.logger.addHandler(console_handler)
app.logger.setLevel(logging.INFO)# Werkzeug 日志只输出错误
werkzeug_logger = logging.getLogger('werkzeug')
werkzeug_logger.handlers = []
werkzeug_logger.addHandler(console_handler)
werkzeug_logger.setLevel(logging.WARNING)def cleanup_old_files():"""清理超过10分钟的临时文件"""current_time = time.time()for token, info in list(DOWNLOADS.items()):if current_time - info['timestamp'] > 600:  # 10分钟try:file_path = info['file_path']if os.path.exists(file_path):os.remove(file_path)del DOWNLOADS[token]except Exception as e:app.logger.error(f"清理文件失败: {str(e)}")def get_video_info(url):"""获取视频信息，包括可用的格式"""ydl_opts = {'quiet': True,'no_warnings': True,'format': None,'youtube_include_dash_manifest': True,'format_sort': ['res:2160',  # 4K'res:1440',  # 2K'res:1080',  # 1080p'res:720',   # 720p'res:480',   # 480p'fps:60',    # 优先60fps'fps',       # 然后是其他fps'vcodec:h264',  # 优先H.264编码'vcodec:vp9',   # 然后是VP9'acodec'      # 最后是音频编码]}with yt_dlp.YoutubeDL(ydl_opts) as ydl:try:info = ydl.extract_info(url, download=False)formats = []def safe_number(value, default=0):try:return float(value or default)except (TypeError, ValueError):return default# 处理视频格式for f in info.get('formats', []):vcodec = f.get('vcodec', 'none')acodec = f.get('acodec', 'none')has_video = vcodec != 'none'has_audio = acodec != 'none'height = safe_number(f.get('height', 0))width = safe_number(f.get('width', 0))fps = safe_number(f.get('fps', 0))tbr = safe_number(f.get('tbr', 0))if has_video:  # 只处理包含视频的格式format_notes = []# 添加分辨率标签if height >= 2160:format_notes.append("4K")elif height >= 1440:format_notes.append("2K")# 详细的分辨率信息if height and width:format_notes.append(f"{width:.0f}x{height:.0f}p")# FPS信息if fps > 0:format_notes.append(f"{fps:.0f}fps")# 编码信息if vcodec != 'none':codec_name = {'avc1': 'H.264','vp9': 'VP9','av01': 'AV1'}.get(vcodec.split('.')[0], vcodec)format_notes.append(f"Video: {codec_name}")# 比特率信息if tbr > 0:format_notes.append(f"{tbr:.0f}kbps")# 音频信息if has_audio and acodec != 'none':format_notes.append(f"Audio: {acodec}")format_data = {'format_id': f.get('format_id', ''),'ext': f.get('ext', ''),'filesize': f.get('filesize', 0),'format_note': ' - '.join(format_notes),'vcodec': vcodec,'acodec': acodec,'height': height,'width': width,'fps': fps,'resolution_sort': height * 1000 + fps}if format_data['format_id']:formats.append(format_data)# 按分辨率和FPS排序formats.sort(key=lambda x: x['resolution_sort'], reverse=True)# 移除重复的格式seen_resolutions = set()unique_formats = []for fmt in formats:res_key = f"{fmt['height']:.0f}p-{fmt['fps']:.0f}fps"if res_key not in seen_resolutions:seen_resolutions.add(res_key)unique_formats.append(fmt)return {'title': info.get('title', 'Unknown'),'duration': info.get('duration', 0),'thumbnail': info.get('thumbnail', ''),'formats': unique_formats,'description': info.get('description', ''),'channel': info.get('channel', 'Unknown'),'view_count': info.get('view_count', 0),}except Exception as e:app.logger.error(f"获取视频信息失败: {str(e)}")return {'error': str(e)}def log_progress(d):if d['status'] == 'downloading':try:percent = d.get('_percent_str', 'N/A').strip()speed = d.get('_speed_str', 'N/A').strip()eta = d.get('_eta_str', 'N/A').strip()# 每5%记录一次进度if percent != 'N/A' and float(percent.rstrip('%')) % 5 < 1:app.logger.info(f"下载进度: {percent} | 速度: {speed} | 剩余时间: {eta}")except Exception:passelif d['status'] == 'finished':app.logger.info("下载完成，开始处理文件...")@app.route('/')
def index():"""渲染主页"""return render_template('index.html')@app.route('/api/info', methods=['POST'])
def get_info():"""获取视频信息的API端点"""url = request.json.get('url')if not url:return jsonify({'error': 'URL is required'}), 400info = get_video_info(url)return jsonify(info)@app.route('/api/download', methods=['POST'])
def download_video():"""下载视频的API端点"""url = request.json.get('url')format_id = request.json.get('format_id')if not url or not format_id:app.logger.error('缺少URL或格式ID')return jsonify({'error': 'URL and format_id are required'}), 400try:cleanup_old_files()temp_file = os.path.join(TEMP_DIR, f'download_{time.time_ns()}')app.logger.info(f"创建临时文件: {os.path.basename(temp_file)}")ydl_opts = {'format': f'{format_id}+bestaudio/best','outtmpl': temp_file + '.%(ext)s','quiet': True,'merge_output_format': 'mp4','postprocessors': [{'key': 'FFmpegVideoConvertor','preferedformat': 'mp4',}],'prefer_ffmpeg': True,'keepvideo': False,'progress_hooks': [log_progress],}app.logger.info("开始下载视频...")with yt_dlp.YoutubeDL(ydl_opts) as ydl:info = ydl.extract_info(url, download=True)final_file = ydl.prepare_filename(info)filename = secure_filename(info['title'] + '.mp4')filesize = os.path.getsize(final_file)filesize_mb = filesize / (1024 * 1024)app.logger.info(f"下载完成: {filename} ({filesize_mb:.1f}MB)")download_token = os.urandom(16).hex()DOWNLOADS[download_token] = {'file_path': final_file,'filename': filename,'timestamp': time.time()}return jsonify({'status': 'success','download_token': download_token,'filename': filename})except Exception as e:app.logger.error(f"下载失败: {str(e)}")return jsonify({'error': str(e)}), [e-mail]('/api/get_file/<token>')
def get_file(token):"""获取下载文件的API端点"""if token not in DOWNLOADS:app.logger.error("无效的下载令牌")return 'Invalid or expired download token', 400download_info = DOWNLOADS[token]file_path = download_info['file_path']filename = download_info['filename']if not os.path.exists(file_path):app.logger.error(f"文件未找到: {filename}")return 'File not found', 404try:filesize = os.path.getsize(file_path)filesize_mb = filesize / (1024 * 1024)app.logger.info(f"开始发送: {filename} ({filesize_mb:.1f}MB)")return send_file(file_path,as_attachment=True,download_name=filename,mimetype='video/mp4')except Exception as e:app.logger.error(f"发送文件失败: {str(e)}")return str(e), 500finally:def cleanup():try:if token in DOWNLOADS:os.remove(file_path)del DOWNLOADS[token]app.logger.info(f"临时文件已清理: {filename}")except Exception as e:app.logger.error(f"清理文件失败: {str(e)}")import threadingthreading.Timer(60, cleanup).start()@app.route('/api/logs')
def get_logs():"""获取日志的API端点"""logs = []temp_queue = queue.Queue()try:while not log_queue.empty():log = log_queue.get_nowait()logs.append(log)temp_queue.put(log)while not temp_queue.empty():log_queue.put(temp_queue.get_nowait())return jsonify(sorted(logs, key=lambda x: x['timestamp'], reverse=True))except Exception as e:app.logger.error(f"获取日志失败: {str(e)}")return jsonify([])if __name__ == '__main__':# 确保临时目录存在os.makedirs(TEMP_DIR, exist_ok=True)# 启动时清理旧文件cleanup_old_files()# 运行应用app.run(host='0.0.0.0', port=9012, debug=True)

2. index.html

<!-- templates/index.html -->
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>YouTube Video Downloader</title><link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
</head>
<body><div class="container"><h1>YouTube Video Downloader</h1><div class="input-group"><input type="text" id="url-input" placeholder="Enter YouTube URL"><button id="fetch-info">Get Video Info</button></div><div id="video-info" class="hidden"><div class="info-container"><img id="thumbnail" src="" alt="Video thumbnail"><div class="video-details"><h2 id="video-title"></h2><p id="video-duration"></p></div></div><div class="formats-container"><h3>Available Formats</h3><div id="format-list"></div></div></div><div id="status" class="hidden"></div><!-- 日志显示区域 --><div class="log-container"><div class="log-header"><h3>Operation Logs</h3><button id="clear-logs" title="Clear logs">Clear</button><label class="auto-scroll"><input type="checkbox" id="auto-scroll" checked>Auto-scroll</label></div><div id="log-display"></div></div></div><script src="{{ url_for('static', filename='js/script.js') }}"></script>
</body>
</html>

3. style.css

有了 AI 后, style 产生得太简单

/* static/css/style.css */
body {font-family: Arial, sans-serif;margin: 0;padding: 20px;background-color: #f5f5f5;
}.container {max-width: 800px;margin: 0 auto;background-color: white;padding: 20px;border-radius: 8px;box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);
}h1 {text-align: center;color: #333;margin-bottom: 20px;
}.input-group {display: flex;gap: 10px;margin-bottom: 20px;
}input[type="text"] {flex: 1;padding: 10px;border: 1px solid #ddd;border-radius: 4px;font-size: 16px;
}button {padding: 10px 20px;background-color: #007bff;color: white;border: none;border-radius: 4px;cursor: pointer;font-size: 16px;
}button:hover {background-color: #0056b3;
}.hidden {display: none;
}.info-container {display: flex;gap: 20px;margin-bottom: 20px;padding: 15px;background-color: #f8f9fa;border-radius: 4px;
}#thumbnail {max-width: 200px;border-radius: 4px;
}.video-details {flex: 1;
}.video-details h2 {margin: 0 0 10px 0;color: #333;
}.formats-container {border-top: 1px solid #ddd;padding-top: 20px;
}#format-list {display: grid;gap: 10px;
}.format-item {padding: 10px;background-color: #f8f9fa;border-radius: 4px;display: flex;justify-content: space-between;align-items: center;
}#status {margin: 20px 0;padding: 10px;border-radius: 4px;text-align: center;
}#status.success {background-color: #d4edda;color: #155724;
}#status.error {background-color: #f8d7da;color: #721c24;
}/* 日志容器样式 */
.log-container {margin-top: 20px;border: 1px solid #ddd;border-radius: 4px;background-color: #1e1e1e;
}.log-header {padding: 10px;background-color: #2d2d2d;border-bottom: 1px solid #444;display: flex;align-items: center;gap: 10px;
}.log-header h3 {margin: 0;flex-grow: 1;color: #fff;
}.auto-scroll {display: flex;align-items: center;gap: 5px;font-size: 14px;color: #fff;
}#clear-logs {padding: 5px 10px;background-color: #6c757d;color: white;border: none;border-radius: 4px;cursor: pointer;
}#clear-logs:hover {background-color: #5a6268;
}#log-display {height: 300px;overflow-y: auto;padding: 10px;font-family: 'Consolas', 'Monaco', monospace;font-size: 13px;line-height: 1.4;background-color: #1e1e1e;color: #d4d4d4;
}.log-entry {margin: 2px 0;padding: 2px 5px;border-radius: 2px;white-space: pre-wrap;word-wrap: break-word;
}.log-timestamp {color: #888;margin-right: 8px;font-size: 0.9em;
}.log-info {color: #89d4ff;
}.log-error {color: #ff8989;
}.log-warning {color: #ffd700;
}/* 滚动条样式 */
#log-display::-webkit-scrollbar {width: 8px;
}#log-display::-webkit-scrollbar-track {background: #2d2d2d;
}#log-display::-webkit-scrollbar-thumb {background: #888;border-radius: 4px;
}#log-display::-webkit-scrollbar-thumb:hover {background: #555;
}

4. script.js

// static/js/script.js
document.addEventListener('DOMContentLoaded', function() {const urlInput = document.getElementById('url-input');const fetchButton = document.getElementById('fetch-info');const videoInfo = document.getElementById('video-info');const thumbnail = document.getElementById('thumbnail');const videoTitle = document.getElementById('video-title');const videoDuration = document.getElementById('video-duration');const formatList = document.getElementById('format-list');const status = document.getElementById('status');// 日志系统class Logger {constructor() {this.logDisplay = document.getElementById('log-display');this.autoScrollCheckbox = document.getElementById('auto-scroll');this.clearLogsButton = document.getElementById('clear-logs');this.lastLogTimestamp = null;this.setupEventListeners();}setupEventListeners() {this.clearLogsButton.addEventListener('click', () => this.clearLogs());this.startLogPolling();}formatTimestamp(isoString) {const date = new Date(isoString);return date.toLocaleTimeString('en-US', { hour12: false,hour: '2-digit',minute: '2-digit',second: '2-digit',fractionalSecondDigits: 3});}addLogEntry(entry) {const logEntry = document.createElement('div');logEntry.classList.add('log-entry');if (entry.level === 'error') {logEntry.classList.add('log-error');} else if (entry.level === 'warning') {logEntry.classList.add('log-warning');} else {logEntry.classList.add('log-info');}const timestamp = document.createElement('span');timestamp.classList.add('log-timestamp');timestamp.textContent = this.formatTimestamp(entry.timestamp);const message = document.createElement('span');message.classList.add('log-message');message.textContent = entry.message;logEntry.appendChild(timestamp);logEntry.appendChild(message);this.logDisplay.appendChild(logEntry);if (this.autoScrollCheckbox.checked) {this.scrollToBottom();}}clearLogs() {this.logDisplay.innerHTML = '';this.lastLogTimestamp = null;}scrollToBottom() {this.logDisplay.scrollTop = this.logDisplay.scrollHeight;}async fetchLogs() {try {const response = await fetch('/api/logs');const logs = await response.json();const newLogs = this.lastLogTimestamp ? logs.filter(log => log.timestamp > this.lastLogTimestamp): logs;if (newLogs.length > 0) {newLogs.forEach(log => this.addLogEntry(log));this.lastLogTimestamp = logs[0].timestamp;}} catch (error) {console.error('Failed to fetch logs:', error);}}startLogPolling() {setInterval(() => this.fetchLogs(), 500);}}// 初始化日志系统const logger = new Logger();function formatDuration(seconds) {const hours = Math.floor(seconds / 3600);const minutes = Math.floor((seconds % 3600) / 60);const remainingSeconds = seconds % 60;if (hours > 0) {return `${hours}:${minutes.toString().padStart(2, '0')}:${remainingSeconds.toString().padStart(2, '0')}`;}return `${minutes}:${remainingSeconds.toString().padStart(2, '0')}`;}function formatFileSize(bytes) {if (!bytes) return 'Unknown size';const sizes = ['Bytes', 'KB', 'MB', 'GB'];const i = Math.floor(Math.log(bytes) / Math.log(1024));return `${(bytes / Math.pow(1024, i)).toFixed(2)} ${sizes[i]}`;}function showStatus(message, isError = false) {status.textContent = message;status.className = isError ? 'error' : 'success';status.classList.remove('hidden');}async function downloadVideo(url, formatId) {try {logger.addLogEntry({timestamp: new Date().toISOString(),level: 'info',message: `Starting download preparation for format: ${formatId}`});showStatus('Preparing download...');const response = await fetch('/api/download', {method: 'POST',headers: {'Content-Type': 'application/json',},body: JSON.stringify({ url, format_id: formatId })});const data = await response.json();if (response.ok && data.download_token) {logger.addLogEntry({timestamp: new Date().toISOString(),level: 'success',message: `Download token received: ${data.download_token}`});showStatus('Starting download...');const iframe = document.createElement('iframe');iframe.style.display = 'none';iframe.src = `/api/get_file/${data.download_token}`;iframe.onload = () => {logger.addLogEntry({timestamp: new Date().toISOString(),level: 'success',message: `Download started for: ${data.filename}`});showStatus('Download started! Check your browser downloads.');setTimeout(() => document.body.removeChild(iframe), 5000);};iframe.onerror = () => {logger.addLogEntry({timestamp: new Date().toISOString(),level: 'error',message: 'Download failed to start'});showStatus('Download failed. Please try again.', true);document.body.removeChild(iframe);};document.body.appendChild(iframe);} else {const errorMessage = data.error || 'Download failed';logger.addLogEntry({timestamp: new Date().toISOString(),level: 'error',message: `Download failed: ${errorMessage}`});showStatus(errorMessage, true);}} catch (error) {logger.addLogEntry({timestamp: new Date().toISOString(),level: 'error',message: `Network error: ${error.message}`});showStatus('Network error occurred', true);console.error(error);}}fetchButton.addEventListener('click', async () => {const url = urlInput.value.trim();if (!url) {showStatus('Please enter a valid URL', true);return;}showStatus('Fetching video information...');try {const response = await fetch('/api/info', {method: 'POST',headers: {'Content-Type': 'application/json',},body: JSON.stringify({ url })});const data = await response.json();if (response.ok) {thumbnail.src = data.thumbnail;videoTitle.textContent = data.title;videoDuration.textContent = formatDuration(data.duration);formatList.innerHTML = data.formats.filter(format => format.format_id && format.ext).map(format => `<div class="format-item"><span>${format.format_note} (${format.ext}) - ${formatFileSize(format.filesize)}</span><button onclick="downloadVideo('${url}', '${format.format_id}')">Download</button></div>`).join('');videoInfo.classList.remove('hidden');status.classList.add('hidden');logger.addLogEntry({timestamp: new Date().toISOString(),level: 'info',message: `Video information retrieved: ${data.title}`});} else {showStatus(data.error || 'Failed to fetch video info', true);logger.addLogEntry({timestamp: new Date().toISOString(),level: 'error',message: `Failed to fetch video info: ${data.error || 'Unknown error'}`});}} catch (error) {showStatus('Network error occurred', true);logger.addLogEntry({timestamp: new Date().toISOString(),level: 'error',message: `Network error: ${error.message}`});}});window.downloadVideo = downloadVideo;// 支持回车键触发获取视频信息urlInput.addEventListener('keypress', (e) => {if (e.key === 'Enter') {fetchButton.click();}});});

以上文件放到相应目录，库文件参考 requirements.txt 即可。

Docker 部署

1. Dockerfile

FROM python:3.11-slimWORKDIR /appRUN apt-get update && \apt-get install -y --no-install-recommends \ffmpeg \&& rm -rf /var/lib/apt/lists/*COPY app.py ./
COPY static/css/style.css ./static/css/
COPY static/js/script.js ./static/js/
COPY templates/index.html ./templates/RUN pip install --no-cache-dir \flask \yt-dlp \werkzeugRUN mkdir -p /app/temp_downloads && \chmod 777 /app/temp_downloadsENV FLASK_APP=app.py
ENV PYTHONUNBUFFERED=1
ENV FLASK_RUN_HOST=0.0.0.0
ENV FLASK_RUN_PORT=9012EXPOSE 9012CMD ["python", "-c", "from app import app; app.run(host='0.0.0.0', port=9012)"]

2. requirements.txt

flask
Werkzeug==3.0.1
yt-dlp==2024.3.10
gunicorn==21.2.0

如果你使用这个 .txt， 可以去掉版本号。我指定版本号，是因我 NAS 的 wheel files 存有多个版本

3. 创建 Image 与 Container

# docker build -t yt-dlp .
# docker run -d -p 9012:9012 --name yt-dlp_container yt-dlp我使用了与 Github 上面项目的相同名字，只是为了方便，字少。

注：在 docker 命令中没有 加入 --restart always, 要编辑一下容器自己添加。

总结：

yt-dlp 是一个功能超强的工具，可以用 cookie file获取身份认证来下载视频，或通过 Mozila 浏览器直接获得 cookie 内容（只是说明上这么说，我没试过）。 Douyin 有 bug 不能下载 , 其它网站没有试。

我有订阅 youtube ，这个工具只是娱乐，或下载 民国 及以前的，其版权已经放弃的影像内容。

请尊重版权