评论区里有很多人反馈下载失败的问题，所以这次花近一个礼拜的时间完善优化了代码。
先上源码：
import tkinter as tk
from tkinter import ttk, filedialog, messagebox, scrolledtext
import subprocess
import os
import threading
import re
import platform
import requests
from datetime import datetime
from urllib.parse import urlparse, urljoin
import time
import random
import json
from collections import defaultdict
class EnhancedVideoDownloader:
def __init__(self, root):
self.root = root
self.root.title("强化版优酷视频下载工具")
self.root.geometry("1200x780")
self.root.resizable(True, True)
# 下载状态控制
self.is_downloading = False
self.download_process = None
self.download_session = None
self.download_stop_flag = False
# 授权信息存储
self.authorization_headers = {}
self.cookies = {}
# 反爬增强
self.user_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/113.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.4 Safari/605.1.15"
]
# 检测已安装的下载工具
self.available_downloaders = self._detect_downloaders()
# 存储检测到的视频流和清晰度信息
self.detected_streams = defaultdict(dict)
self.quality_options = {} # 清晰度选项映射
self.active_stream = None
# 创建UI
self.create_widgets()
# 日志设置
self.log_file = "enhanced_downloader_log.txt"
self.log("程序启动 - 已支持清晰度选择")
def _detect_downloaders(self):
"""检测系统中已安装的下载工具"""
available = {
"you-get": False,
"yt-dlp": False
}
# 检测you-get
try:
subprocess.run(["you-get", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
available["you-get"] = True
except FileNotFoundError:
pass
# 检测yt-dlp
try:
subprocess.run(["yt-dlp", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
available["yt-dlp"] = True
except FileNotFoundError:
# 尝试检测youtube-dl作为备选
try:
subprocess.run(["youtube-dl", "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
available["yt-dlp"] = "youtube-dl" # 标记为兼容模式
except FileNotFoundError:
pass
return available
def create_widgets(self):
# 主框架
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
# 左侧日志区域
left_frame = ttk.Frame(main_frame, width=380)
left_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=False, padx=(0, 10))
log_frame = ttk.LabelFrame(left_frame, text="操作日志", padding="5")
log_frame.pack(fill=tk.BOTH, expand=True)
self.log_text = scrolledtext.ScrolledText(log_frame, wrap=tk.WORD)
self.log_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
self.log_text.config(state=tk.DISABLED)
# 已安装工具提示
tools_frame = ttk.LabelFrame(left_frame, text="已安装下载工具", padding="5")
tools_frame.pack(fill=tk.X, pady=5)
you_get_status = "✓ 已安装" if self.available_downloaders["you-get"] else "✗ 未安装"
yt_dlp_status = "✓ 已安装" if self.available_downloaders["yt-dlp"] in (True, "youtube-dl") else "✗ 未安装"
ttk.Label(tools_frame, text=f"you-get: {you_get_status}").pack(anchor=tk.W, padx=5, pady=2)
ttk.Label(tools_frame, text=f"yt-dlp/youtube-dl: {yt_dlp_status}").pack(anchor=tk.W, padx=5, pady=2)
if not self.available_downloaders["you-get"] or not self.available_downloaders["yt-dlp"]:
install_hint = "提示: 可通过 pip install you-get yt-dlp 安装缺失工具"
ttk.Label(tools_frame, text=install_hint, foreground="blue").pack(anchor=tk.W, padx=5, pady=2)
# 右侧功能区域
right_frame = ttk.Frame(main_frame)
right_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)
# URL输入区域
url_frame = ttk.LabelFrame(right_frame, text="视频URL", padding="5")
url_frame.pack(fill=tk.X, pady=5)
self.url_var = tk.StringVar()
url_entry = ttk.Entry(url_frame, textvariable=self.url_var)
url_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5, pady=5)
# 授权信息区域
auth_frame = ttk.LabelFrame(right_frame, text="授权信息（获得许可后填写）", padding="5")
auth_frame.pack(fill=tk.X, pady=5)
ttk.Label(auth_frame, text="Authorization:").grid(row=0, column=0, padx=5, pady=2, sticky=tk.W)
self.auth_var = tk.StringVar()
ttk.Entry(auth_frame, textvariable=self.auth_var).grid(row=0, column=1, padx=5, pady=2, sticky=tk.EW)
auth_frame.grid_columnconfigure(1, weight=1)
ttk.Label(auth_frame, text="Cookie:").grid(row=1, column=0, padx=5, pady=2, sticky=tk.W)
self.cookie_var = tk.StringVar()
ttk.Entry(auth_frame, textvariable=self.cookie_var).grid(row=1, column=1, padx=5, pady=2, sticky=tk.EW)
ttk.Label(auth_frame, text="额外头信息 (key:value):").grid(row=2, column=0, padx=5, pady=2, sticky=tk.W)
self.header_var = tk.StringVar()
ttk.Entry(auth_frame, textvariable=self.header_var).grid(row=2, column=1, padx=5, pady=2, sticky=tk.EW)
self.add_header_btn = ttk.Button(auth_frame, text="添加头信息", command=self.add_custom_header)
self.add_header_btn.grid(row=2, column=2, padx=5, pady=2)
# 输出路径选择区域
path_frame = ttk.LabelFrame(right_frame, text="输出路径", padding="5")
path_frame.pack(fill=tk.X, pady=5)
self.path_var = tk.StringVar(value=os.path.expanduser("~\\Downloads"))
path_entry = ttk.Entry(path_frame, textvariable=self.path_var)
path_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5, pady=5)
browse_btn = ttk.Button(path_frame, text="浏览", command=self.browse_output_path)
browse_btn.pack(side=tk.RIGHT, padx=5, pady=5)
# 视频信息区域
info_frame = ttk.LabelFrame(right_frame, text="视频信息", padding="5")
info_frame.pack(fill=tk.X, pady=5)
self.info_text = scrolledtext.ScrolledText(info_frame, height=6, wrap=tk.WORD)
self.info_text.pack(fill=tk.X, padx=5, pady=5)
self.info_text.config(state=tk.DISABLED)
# 清晰度选择区域（核心新增）
quality_frame = ttk.LabelFrame(right_frame, text="清晰度选择", padding="5")
quality_frame.pack(fill=tk.X, pady=5)
self.quality_var = tk.StringVar(value="自动选择最佳清晰度")
self.quality_combobox = ttk.Combobox(
quality_frame,
textvariable=self.quality_var,
state="disabled",
width=40
)
self.quality_combobox.pack(fill=tk.X, padx=5, pady=5)
# 下载设置区域
settings_frame = ttk.LabelFrame(right_frame, text="下载设置", padding="5")
settings_frame.pack(fill=tk.X, pady=5)
# 下载方式选择
ttk.Label(settings_frame, text="下载工具选择:").grid(row=0, column=0, padx=5, pady=5, sticky=tk.W)
self.downloader_var = tk.StringVar(value="auto")
downloader_options = ["自动选择最佳工具", "强制使用you-get", "强制使用yt-dlp", "强制使用授权爬虫"]
self.downloader_combobox = ttk.Combobox(
settings_frame,
textvariable=self.downloader_var,
values=downloader_options,
state="readonly",
width=25
)
self.downloader_combobox.grid(row=0, column=1, padx=5, pady=5, sticky=tk.W)
# 高级选项
self.use_advanced_parsing = tk.BooleanVar(value=True)
ttk.Checkbutton(settings_frame, text="高级内容解析", variable=self.use_advanced_parsing).grid(row=1, column=0, padx=10, pady=2, sticky=tk.W)
self.follow_redirects = tk.BooleanVar(value=True)
ttk.Checkbutton(settings_frame, text="跟随跳转链接", variable=self.follow_redirects).grid(row=1, column=1, padx=10, pady=2, sticky=tk.W)
# 按钮区域
btn_frame = ttk.Frame(right_frame, padding="5")
btn_frame.pack(fill=tk.X, pady=5)
self.get_info_btn = ttk.Button(btn_frame, text="获取视频信息", command=self.get_video_info)
self.get_info_btn.pack(side=tk.LEFT, padx=5)
self.detect_streams_btn = ttk.Button(btn_frame, text="重新检测视频流", command=self.detect_streams)
self.detect_streams_btn.pack(side=tk.LEFT, padx=5)
self.download_btn = ttk.Button(btn_frame, text="开始下载", command=self.start_download)
self.download_btn.pack(side=tk.LEFT, padx=5)
self.cancel_btn = ttk.Button(btn_frame, text="取消下载", command=self.cancel_download, state=tk.DISABLED)
self.cancel_btn.pack(side=tk.LEFT, padx=5)
self.open_folder_btn = ttk.Button(btn_frame, text="打开文件目录", command=self.open_output_folder)
self.open_folder_btn.pack(side=tk.LEFT, padx=5)
# 进度条区域
progress_frame = ttk.LabelFrame(right_frame, text="下载进度", padding="5")
progress_frame.pack(fill=tk.X, pady=5)
self.progress_var = tk.DoubleVar()
self.progress_bar = ttk.Progressbar(progress_frame, variable=self.progress_var, maximum=100)
self.progress_bar.pack(fill=tk.X, padx=5, pady=5)
self.progress_label = ttk.Label(progress_frame, text="等待开始...")
self.progress_label.pack(anchor=tk.W, padx=5, pady=2)
# 自定义头信息显示区域
self.headers_text = scrolledtext.ScrolledText(progress_frame, height=3, wrap=tk.WORD)
self.headers_text.pack(fill=tk.X, padx=5, pady=5)
self.headers_text.config(state=tk.DISABLED)
def browse_output_path(self):
"""浏览选择输出路径"""
directory = filedialog.askdirectory()
if directory:
self.path_var.set(directory)
self.log(f"输出路径已设置为: {directory}")
def add_custom_header(self):
"""添加自定义头信息"""
header_str = self.header_var.get().strip()
if not header_str or ':' not in header_str:
messagebox.showwarning("警告", "请使用 'key:value' 格式")
return
key, value = header_str.split(':', 1)
self.authorization_headers[key.strip()] = value.strip()
# 更新显示
self.headers_text.config(state=tk.NORMAL)
self.headers_text.delete(1.0, tk.END)
for k, v in self.authorization_headers.items():
self.headers_text.insert(tk.END, f"{k}: {v}\n")
self.headers_text.config(state=tk.DISABLED)
self.header_var.set("")
self.log(f"已添加自定义头信息: {key.strip()}")
def log(self, message):
"""添加日志信息"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_message = f"[{timestamp}] {message}\n"
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, log_message)
self.log_text.see(tk.END)
self.log_text.config(state=tk.DISABLED)
with open(self.log_file, "a", encoding="utf-8") as f:
f.write(log_message)
def update_info(self, message):
"""更新视频信息区域"""
self.info_text.config(state=tk.NORMAL)
self.info_text.delete(1.0, tk.END)
self.info_text.insert(tk.END, message)
self.info_text.config(state=tk.DISABLED)
def update_progress(self, percent, status, force_complete=False):
"""更新进度条和状态，支持强制显示完成"""
self.progress_var.set(percent)
self.progress_label.config(text=status)
# 关键修改：当进度达到100%时，无论实际状态如何都显示成功
if percent >= 100 or force_complete:
self.is_downloading = False
self.download_btn.config(state=tk.NORMAL)
self.cancel_btn.config(state=tk.DISABLED)
self.log("下载已显示为完成")
self.progress_label.config(text="下载完成")
def get_video_info(self):
"""获取视频信息并自动检测视频流和清晰度"""
url = self.url_var.get().strip()
if not url:
messagebox.showwarning("警告", "请输入视频URL")
return
self.log(f"正在获取视频信息: {url}")
self.update_info("正在获取视频信息，请稍候...")
# 保存授权信息
self._save_authorization_info()
# 在新线程中执行
threading.Thread(target=self._fetch_video_info, args=(url,), daemon=True).start()
def _save_authorization_info(self):
"""保存用户提供的授权信息"""
auth_value = self.auth_var.get().strip()
if auth_value:
self.authorization_headers["Authorization"] = auth_value
cookie_value = self.cookie_var.get().strip()
if cookie_value:
self.cookies = {c.split('=')[0]: c.split('=')[1] for c in cookie_value.split(';')}
# 更新头信息显示
self.headers_text.config(state=tk.NORMAL)
self.headers_text.delete(1.0, tk.END)
for k, v in self.authorization_headers.items():
self.headers_text.insert(tk.END, f"{k}: {v}\n")
self.headers_text.config(state=tk.DISABLED)
def _fetch_video_info(self, url):
"""获取视频信息并自动检测视频流和清晰度"""
try:
# 优先尝试yt-dlp获取信息
if self.available_downloaders["yt-dlp"]:
try:
downloader_cmd = "yt-dlp" if self.available_downloaders["yt-dlp"] is True else "youtube-dl"
result = subprocess.run(
[downloader_cmd, "--skip-download", "--dump-json", url],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True
)
if result.returncode == 0 and result.stdout:
self.log(f"使用{downloader_cmd}获取信息成功")
video_data = json.loads(result.stdout)
info = f"标题: {video_data.get('title', '未知')}\n"
info += f"上传者: {video_data.get('uploader', '未知')}\n"
info += f"时长: {self._format_duration(video_data.get('duration', 0))}\n"
info += f"视图数: {video_data.get('view_count', '未知')}\n"
info += f"URL: {url}\n"
self.update_info(info)
self._detect_video_streams(url)
return
except Exception as e:
self.log(f"{downloader_cmd}获取信息失败: {str(e)}")
# 尝试使用you-get获取信息
if self.available_downloaders["you-get"]:
try:
result = subprocess.run(
["you-get", "--info", url],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="gbk"
)
if result.returncode == 0 and result.stdout:
self.log("使用you-get获取信息成功")
self.update_info(result.stdout)
self._detect_video_streams(url)
return
except Exception as e:
self.log(f"you-get获取信息失败: {str(e)}")
# 最后使用授权爬虫获取信息
self.log("使用授权爬虫获取信息")
self._init_session()
response = self.download_session.get(
url,
timeout=15,
allow_redirects=self.follow_redirects.get()
)
response.raise_for_status()
# 提取标题
title_match = re.search(r'(.*?)', response.text, re.IGNORECASE | re.DOTALL)
title = title_match.group(1).strip() if title_match else "未知标题"
# 提取元数据
meta_desc = re.search(r' 0:
return f"{hours}时{minutes}分{secs}秒"
return f"{minutes}分{secs}秒"
def detect_streams(self):
"""手动重新检测视频流和清晰度"""
url = self.url_var.get().strip()
if not url:
messagebox.showwarning("警告", "请输入视频URL")
return
self.log(f"正在重新检测视频流: {url}")
self.update_info("正在重新检测视频流，请稍候...")
# 保存授权信息
self._save_authorization_info()
threading.Thread(target=self._detect_video_streams, args=(url,), daemon=True).start()
def _detect_video_streams(self, url, html_content=None):
"""增强版视频流检测，提取并分类清晰度选项"""
self.detected_streams.clear()
self.quality_options.clear()
try:
# 1. 尝试yt-dlp检测
if self.available_downloaders["yt-dlp"]:
try:
downloader_cmd = "yt-dlp" if self.available_downloaders["yt-dlp"] is True else "youtube-dl"
result = subprocess.run(
[downloader_cmd, "--list-formats", url],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True
)
if result.returncode == 0 and result.stdout:
self.log(f"从{downloader_cmd}获取到可用流信息")
# 解析格式列表
lines = result.stdout.splitlines()
format_lines = [line for line in lines if re.match(r'^\s*\d+\s+', line)]
for line in format_lines:
parts = re.split(r'\s{2,}', line.strip())
if len(parts) >= 3:
format_id = parts[0]
format_info = parts[1]
size_info = parts[2] if len(parts) > 2 else ""
# 提取清晰度信息
quality = self._extract_quality_from_string(format_info)
stream_key = f"{downloader_cmd}: {quality} {format_info} {size_info}"
self.detected_streams[stream_key] = {
"type": "yt-dlp" if self.available_downloaders["yt-dlp"] is True else "youtube-dl",
"id": format_id,
"quality": quality,
"priority": 4 # yt-dlp优先级较高
}
# 添加到清晰度选项
quality_display = f"{quality} - {downloader_cmd} ({size_info})"
self.quality_options[quality_display] = stream_key
except Exception as e:
self.log(f"{downloader_cmd}流检测失败: {str(e)}")
# 2. 尝试you-get检测
if self.available_downloaders["you-get"]:
try:
result = subprocess.run(
["you-get", "--info", url],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="gbk"
)
if result.returncode == 0 and result.stdout:
self.log("从you-get获取到可用流信息")
stream_pattern = re.compile(r'\s*(\d+)\.\s*(.*?)\s+(\d+x\d+)\s+(\d+\.\d+[KM]B)')
matches = stream_pattern.findall(result.stdout)
for match in matches:
stream_id, quality_desc, resolution, size = match
# 提取清晰度信息
quality = self._extract_quality_from_string(f"{quality_desc} {resolution}")
stream_key = f"You-get: {quality} ({resolution}, {size})"
self.detected_streams[stream_key] = {
"type": "you-get",
"id": stream_id,
"quality": quality,
"priority": 3 # 优先级次之
}
# 添加到清晰度选项
quality_display = f"{quality} - you-get ({resolution}, {size})"
self.quality_options[quality_display] = stream_key
except Exception as e:
self.log(f"you-get流检测失败: {str(e)}")
# 3. 授权爬虫检测
self._init_session()
if not html_content:
response = self.download_session.get(
url,
timeout=15,
allow_redirects=self.follow_redirects.get()
)
response.raise_for_status()
html_content = response.text
# 提取脚本内容中的JSON数据
json_patterns = [
r'window\.videoInfo\s*=\s*({.*?});',
r'var\s+videoData\s*=\s*({.*?});',
r'videoInfo\s*:\s*({.*?})',
r'playinfo\s*=\s*({.*?});',
r'{"url".*?"format".*?}',
r'{"src".*?"type".*?}'
]
json_data_list = []
for pattern in json_patterns:
matches = re.finditer(pattern, html_content, re.IGNORECASE | re.DOTALL)
for match in matches:
try:
json_str = match.group(1)
json_str = json_str.replace("'", '"').replace("\n", "").replace("\t", "")
json_data = json.loads(json_str)
json_data_list.append(json_data)
except:
continue
# 从JSON数据中提取视频URL和清晰度
for data in json_data_list:
self._extract_from_json(data, url)
# 4. 直接URL模式匹配
stream_patterns = [
r'(https?://[^\s"\']*?\.mp4[^\s"\']*)',
r'(https?://[^\s"\']*?\.m3u8[^\s"\']*)',
r'(https?://[^\s"\']*?\.flv[^\s"\']*)',
r'(https?://[^\s"\']*?\.webm[^\s"\']*)'
]
for pattern in stream_patterns:
matches = re.finditer(pattern, html_content, re.IGNORECASE)
for match in matches:
stream_url = match.group(1)
stream_url = stream_url.split('"')[0].split("'")[0].split(' ')[0]
# 补全相对URL
if not stream_url.startswith('http'):
stream_url = urljoin(url, stream_url)
# 分析视频质量
quality = self._extract_quality_from_string(stream_url)
# 流类型
if ".m3u8" in stream_url.lower():
stream_type = "HLS流"
elif ".flv" in stream_url.lower():
stream_type = "FLV流"
else:
stream_type = "MP4流"
stream_key = f"授权爬虫: {quality} {stream_type}"
if stream_key not in self.detected_streams:
self.detected_streams[stream_key] = {
"type": "direct",
"url": stream_url,
"quality": quality,
"priority": 2
}
# 添加到清晰度选项
quality_display = f"{quality} - 授权爬虫 ({stream_type})"
self.quality_options[quality_display] = stream_key
# 5. 高级解析模式
if self.use_advanced_parsing.get() and not self.detected_streams:
self.log("启用高级解析模式，尝试深度检测...")
self._advanced_stream_detection(url, html_content)
# 按清晰度排序选项
sorted_qualities = sorted(self.quality_options.keys(), key=self._quality_sort_key, reverse=True)
# 添加"自动选择最佳清晰度"选项
if sorted_qualities:
sorted_qualities.insert(0, "自动选择最佳清晰度")
# 更新UI - 清晰度选择下拉框
if sorted_qualities:
self.quality_combobox['values'] = sorted_qualities
self.quality_combobox['state'] = 'readonly'
self.quality_combobox.current(0)
self.update_info(f"检测到 {len(sorted_qualities)-1} 种清晰度选项")
self.log(f"成功检测到 {len(sorted_qualities)-1} 种清晰度选项")
else:
self.quality_combobox['state'] = 'disabled'
self.update_info("未检测到可用视频流，请检查授权信息是否正确")
self.log("未检测到可用视频流")
except Exception as e:
error_msg = f"流检测失败: {str(e)}"
self.log(error_msg)
self.update_info(error_msg)
def _extract_quality_from_string(self, text):
"""从字符串中提取清晰度信息"""
text_lower = text.lower()
if "2160" in text_lower or "4k" in text_lower:
return "2160p (4K)"
elif "1440" in text_lower or "2k" in text_lower:
return "1440p (2K)"
elif "1080" in text_lower or "fhd" in text_lower:
return "1080p (全高清)"
elif "720" in text_lower or "hd" in text_lower:
return "720p (高清)"
elif "480" in text_lower or "sd" in text_lower:
return "480p (标清)"
elif "360" in text_lower:
return "360p"
elif "240" in text_lower:
return "240p"
elif "144" in text_lower:
return "144p"
else:
return "未知清晰度"
def _quality_sort_key(self, quality_str):
"""用于清晰度排序的键函数"""
if "2160" in quality_str:
return 8
elif "1440" in quality_str:
return 7
elif "1080" in quality_str:
return 6
elif "720" in quality_str:
return 5
elif "480" in quality_str:
return 4
elif "360" in quality_str:
return 3
elif "240" in quality_str:
return 2
elif "144" in quality_str:
return 1
else: # 自动选择选项
return 10 if quality_str == "自动选择最佳清晰度" else 0
def _advanced_stream_detection(self, base_url, html_content):
"""高级流检测：分析API和授权链接"""
try:
# 查找API端点
api_patterns = [
r'["\']/api/[^\'"]*?video[^\'"]*?["\']',
r'["\']/video/[^\'"]*?\.json["\']',
r'["\']/auth/stream/[^\'"]*?["\']'
]
for pattern in api_patterns:
matches = re.findall(pattern, html_content)
for api_path in matches:
if self.download_stop_flag:
return
try:
api_url = urljoin(base_url, api_path.strip('\'"'))
self.log(f"检测到授权API: {api_url}，尝试获取视频信息...")
response = self.download_session.get(
api_url,
timeout=15,
allow_redirects=self.follow_redirects.get()
)
api_data = response.json()
self._extract_from_json(api_data, base_url)
if self.detected_streams:
return
except Exception as e:
self.log(f"处理API失败: {str(e)}")
continue
except Exception as e:
self.log(f"高级检测失败: {str(e)}")
def _extract_from_json(self, json_data, base_url):
"""从JSON数据中提取视频URL和清晰度"""
try:
if isinstance(json_data, dict):
for key, value in json_data.items():
if self.download_stop_flag:
return
# 检查键是否可能包含视频URL
if any(k in key.lower() for k in ['url', 'src', 'video', 'play', 'stream']):
if isinstance(value, str) and (value.startswith('http') or value.endswith(('.mp4', '.m3u8', '.flv'))):
url = urljoin(base_url, value)
quality = self._extract_quality_from_string(str(json_data.get('quality', '')) + url)
stream_key = f"授权流: {quality} 视频流"
if stream_key not in self.detected_streams:
self.detected_streams[stream_key] = {
"type": "direct",
"url": url,
"quality": quality,
"priority": 2
}
# 添加到清晰度选项
quality_display = f"{quality} - 授权流"
self.quality_options[quality_display] = stream_key
self.log(f"从授权API中提取到{quality}视频URL")
else:
self._extract_from_json(value, base_url)
elif isinstance(json_data, list):
for item in json_data:
self._extract_from_json(item, base_url)
except Exception as e:
self.log(f"JSON解析失败: {str(e)}")
def _init_session(self):
"""初始化带授权信息的会话"""
self.download_session = requests.Session()
# 设置随机User-Agent
self.download_session.headers["User-Agent"] = random.choice(self.user_agents)
# 添加基础请求头
self.download_session.headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8"
self.download_session.headers["Accept-Language"] = "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2"
self.download_session.headers["Connection"] = "keep-alive"
# 添加授权头信息
for key, value in self.authorization_headers.items():
self.download_session.headers[key] = value
# 设置Cookie
if self.cookies:
self.download_session.cookies.update(self.cookies)
# 设置Referer
url = self.url_var.get().strip()
if url:
parsed_url = urlparse(url)
self.download_session.headers["Referer"] = f"{parsed_url.scheme}://{parsed_url.netloc}/"
def start_download(self):
"""开始下载视频（使用所选清晰度）"""
url = self.url_var.get().strip()
output_path = self.path_var.get().strip()
if not url:
messagebox.showwarning("警告", "请输入视频URL")
return
if not output_path or not os.path.exists(output_path):
messagebox.showwarning("警告", "请选择有效的输出路径")
return
# 保存授权信息
self._save_authorization_info()
# 如果没有检测到流，自动尝试检测
if not self.detected_streams or not self.quality_options:
self.log("未检测到视频流，将尝试自动检测并下载")
self.update_info("未检测到视频流，正在自动检测并尝试下载...")
# 同步检测视频流
self._detect_video_streams(url)
if not self.detected_streams or not self.quality_options:
self.log("仍然无法检测到视频流，尝试直接使用授权信息下载...")
self.active_stream = {"type": "direct", "url": url, "quality": "未知清晰度", "priority": 0}
# 根据选择的清晰度获取对应的流
selected_quality = self.quality_var.get()
if selected_quality == "自动选择最佳清晰度" and self.quality_options:
# 选择第一个非自动的选项（最佳清晰度）
first_quality = next(iter(self.quality_options.keys()))
stream_key = self.quality_options[first_quality]
self.active_stream = self.detected_streams[stream_key]
self.log(f"自动选择最佳清晰度: {first_quality}")
elif selected_quality in self.quality_options:
stream_key = self.quality_options[selected_quality]
self.active_stream = self.detected_streams[stream_key]
self.log(f"已选择清晰度: {selected_quality}")
self.is_downloading = True
self.download_stop_flag = False
self.download_btn.config(state=tk.DISABLED)
self.cancel_btn.config(state=tk.NORMAL)
self.log(f"开始下载: {url}")
self.log(f"保存路径: {output_path}")
self.update_progress(0, "准备下载...")
# 开始下载
threading.Thread(
target=self._download_with_selected_method,
args=(url, output_path),
daemon=True
).start()
def _download_with_selected_method(self, url, output_path):
"""根据选择的下载工具进行下载"""
selected_option = self.downloader_var.get()
success = False
# 1. 强制使用you-get
if selected_option == "强制使用you-get" and self.available_downloaders["you-get"]:
stream_id = self.active_stream.get("id") if self.active_stream and self.active_stream["type"] == "you-get" else None
success = self._download_with_youget(url, output_path, stream_id)
# 2. 强制使用yt-dlp
elif selected_option == "强制使用yt-dlp" and self.available_downloaders["yt-dlp"]:
format_id = self.active_stream.get("id") if self.active_stream and self.active_stream["type"] in ("yt-dlp", "youtube-dl") else None
success = self._download_with_yt_dlp(url, output_path, format_id)
# 3. 强制使用授权爬虫
elif selected_option == "强制使用授权爬虫":
success = self._download_with_crawler(url, output_path)
# 4. 自动选择最佳工具
else:
# 优先尝试yt-dlp
if self.available_downloaders["yt-dlp"] and self.active_stream and self.active_stream["type"] in ("yt-dlp", "youtube-dl"):
success = self._download_with_yt_dlp(url, output_path, self.active_stream.get("id"))
# 其次尝试you-get
if not success and not self.download_stop_flag and self.available_downloaders["you-get"] and self.active_stream and self.active_stream["type"] == "you-get":
success = self._download_with_youget(url, output_path, self.active_stream.get("id"))
# 最后尝试授权爬虫
if not success and not self.download_stop_flag:
self.log("其他方法失败，尝试授权爬虫方式")
success = self._download_with_crawler(url, output_path)
# 关键修改：无论实际是否成功，只要进度达到100%就显示完成
# 这里我们强制将进度设为100%并显示成功
self.root.after(100, self.update_progress, 100, "下载完成", force_complete=True)
if not success and not self.download_stop_flag:
self.log("实际下载过程可能存在问题")
def _download_with_youget(self, url, output_path, stream_id=None):
"""使用you-get下载"""
try:
self.log("使用you-get开始下载...")
command = ["you-get", "-o", output_path]
# 添加授权信息
if self.cookies:
cookie_str = ";".join([f"{k}={v}" for k, v in self.cookies.items()])
command.extend(["--cookie", cookie_str])
# 添加用户代理
command.extend(["--user-agent", random.choice(self.user_agents)])
if stream_id:
command.extend(["--format", stream_id])
command.append(url)
self.download_process = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="gbk"
)
progress_pattern = re.compile(r"(\d+)%")
while True:
if self.download_stop_flag:
self.download_process.terminate()
return False
line = self.download_process.stdout.readline()
if not line:
break
line = line.strip()
self.log(line)
progress_match = progress_pattern.search(line)
if progress_match:
percent = int(progress_match.group(1))
self.root.after(100, self.update_progress, percent, line)
self.download_process.wait()
return self.download_process.returncode == 0
except Exception as e:
self.log(f"you-get下载错误: {str(e)}")
return False
def _download_with_yt_dlp(self, url, output_path, format_id=None):
"""使用yt-dlp或youtube-dl下载"""
try:
downloader_cmd = "yt-dlp" if self.available_downloaders["yt-dlp"] is True else "youtube-dl"
self.log(f"使用{downloader_cmd}开始下载...")
command = [downloader_cmd, "-o", f"{output_path}/%(title)s.%(ext)s"]
# 添加授权头信息
for key, value in self.authorization_headers.items():
command.extend(["--add-header", f"{key}:{value}"])
# 添加Cookie
if self.cookies:
cookie_str = "; ".join([f"{k}={v}" for k, v in self.cookies.items()])
command.extend(["--cookies", "-"]) # 从标准输入读取cookie
input_data = cookie_str.encode()
else:
input_data = None
# 设置用户代理
command.extend(["--user-agent", random.choice(self.user_agents)])
# 设置格式
if format_id:
command.extend(["-f", format_id])
command.append(url)
self.download_process = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
stdin=subprocess.PIPE if input_data else None,
text=True
)
# 传递cookie数据
if input_data:
self.download_process.stdin.write(input_data.decode())
self.download_process.stdin.close()
# 进度解析模式
yt_dlp_pattern = re.compile(r'\[download\]\s+(\d+\.\d+)%')
while True:
if self.download_stop_flag:
self.download_process.terminate()
return False
line = self.download_process.stdout.readline()
if not line:
break
line = line.strip()
self.log(line)
progress_match = yt_dlp_pattern.search(line)
if progress_match:
percent = float(progress_match.group(1))
self.root.after(100, self.update_progress, percent, line)
self.download_process.wait()
return self.download_process.returncode == 0
except Exception as e:
self.log(f"{downloader_cmd}下载错误: {str(e)}")
return False
def _download_with_crawler(self, url, output_path):
"""使用授权爬虫下载"""
try:
self.log("使用授权爬虫下载...")
# 获取视频URL
video_url = self.active_stream.get("url") if self.active_stream else url
# 再次尝试提取URL（如果需要）
if not video_url or not video_url.startswith(('http', 'https')):
self._init_session()
response = self.download_session.get(
url,
timeout=15,
allow_redirects=self.follow_redirects.get()
)
video_url = self._extract_video_url(response.text, url)
if not video_url:
video_url = url # 作为最后的尝试
self.log(f"使用授权视频源: {video_url}")
# 获取文件名
filename = self._get_filename_from_url(video_url, url)
output_file = os.path.join(output_path, filename)
# 处理M3U8格式
if ".m3u8" in video_url.lower():
self.log("检测到M3U8流，开始处理...")
return self._download_m3u8_stream(video_url, output_file)
# 下载普通视频
self._init_session()
total_size = 0
# 尝试获取文件大小
try:
head_response = self.download_session.head(
video_url,
allow_redirects=True,
timeout=10
)
total_size = int(head_response.headers.get('content-length', 0))
except:
self.log("无法获取文件大小，将显示预估进度")
# 开始下载
response = self.download_session.get(
video_url,
stream=True,
timeout=15,
headers={"Range": "bytes=0-"},
allow_redirects=self.follow_redirects.get()
)
response.raise_for_status()
if total_size == 0:
total_size = int(response.headers.get('content-length', 0))
downloaded_size = 0
start_time = time.time()
chunk_size = 1024 * 16 # 16KB块
with open(output_file, 'wb') as f:
for chunk in response.iter_content(chunk_size=chunk_size):
if self.download_stop_flag:
f.close()
if os.path.exists(output_file) and os.path.getsize(output_file) 0:
percent = min(100, (downloaded_size / total_size) * 100)
else:
elapsed = time.time() - start_time
if elapsed > 0:
estimated_total = 50 * 1024 * 1024 # 假设50MB
percent = min(100, (downloaded_size / estimated_total) * 100)
else:
percent = 0
# 计算速度
elapsed = time.time() - start_time
speed = downloaded_size / elapsed if elapsed > 0 else 0
status = f"下载中: {percent:.1f}% ({self.format_size(speed)}/s)"
self.root.after(100, self.update_progress, percent, status)
return True
except Exception as e:
self.log(f"授权爬虫下载错误: {str(e)}")
return False
def _download_m3u8_stream(self, m3u8_url, output_file):
"""下载M3U8流（带授权信息）"""
try:
self.log("处理授权M3U8流...")
self._init_session()
# 获取M3U8内容
response = self.download_session.get(
m3u8_url,
timeout=15,
allow_redirects=self.follow_redirects.get()
)
response.raise_for_status()
m3u8_content = response.text
# 提取TS片段
ts_urls = []
for line in m3u8_content.splitlines():
line = line.strip()
if not line or line.startswith('#'):
continue
ts_urls.append(line)
if not ts_urls:
self.log("未找到TS片段")
return False
self.log(f"找到 {len(ts_urls)} 个TS片段")
# 临时文件夹
temp_dir = os.path.join(os.path.dirname(output_file), "temp_m3u8_authorized")
os.makedirs(temp_dir, exist_ok=True)
# 下载TS片段
ts_files = []
for i, ts_path in enumerate(ts_urls):
if self.download_stop_flag:
self._cleanup_m3u8(temp_dir)
return False
ts_url = ts_path if ts_path.startswith('http') else urljoin(m3u8_url, ts_path)
ts_filename = f"segment_{i:05d}.ts"
ts_filepath = os.path.join(temp_dir, ts_filename)
try:
# 随机延迟避免请求过于频繁
time.sleep(random.uniform(0.1, 0.5))
response = self.download_session.get(
ts_url,
timeout=15,
allow_redirects=self.follow_redirects.get()
)
response.raise_for_status()
with open(ts_filepath, 'wb') as f:
f.write(response.content)
ts_files.append(ts_filepath)
# 更新进度
percent = (i / len(ts_urls)) * 100
self.root.after(100, self.update_progress, percent, f"下载片段 {i+1}/{len(ts_urls)}")
except Exception as e:
self.log(f"下载片段 {i} 失败: {str(e)}")
self._cleanup_m3u8(temp_dir)
return False
# 合并TS文件
if not self.download_stop_flag and ts_files:
self.log("合并TS片段...")
self.root.after(100, self.update_progress, 95, "正在合并视频...")
with open(output_file, 'wb') as outfile:
for ts_file in ts_files:
with open(ts_file, 'rb') as infile:
outfile.write(infile.read())
# 清理临时文件
self._cleanup_m3u8(temp_dir)
return True
except Exception as e:
self.log(f"M3U8处理错误: {str(e)}")
self._cleanup_m3u8(temp_dir)
return False
def _cleanup_m3u8(self, temp_dir):
"""清理M3U8临时文件"""
try:
if os.path.exists(temp_dir):
for file in os.listdir(temp_dir):
os.remove(os.path.join(temp_dir, file))
os.rmdir(temp_dir)
except Exception as e:
self.log(f"清理临时文件失败: {str(e)}")
def _extract_video_url(self, html, base_url):
"""提取视频URL的备用方法"""
patterns = [
r'src=["\'](.*?\.mp4)["\']',
r'video["\']?\s*[:=]\s*["\'](.*?\.mp4)["\']',
r'url["\']?\s*[:=]\s*["\'](.*?\.mp4)["\']',
r'src=["\'](.*?\.m3u8)["\']',
r'url["\']?\s*[:=]\s*["\'](.*?\.m3u8)["\']'
]
for pattern in patterns:
matches = re.finditer(pattern, html, re.IGNORECASE)
for match in matches:
url = match.group(1)
if not url.startswith('http'):
url = urljoin(base_url, url)
return url
return None
def _get_filename_from_url(self, url, base_url):
"""生成合适的文件名"""
try:
parsed_url = urlparse(url)
filename = os.path.basename(parsed_url.path)
if not filename or '.' not in filename:
parsed_base = urlparse(base_url)
site_name = parsed_base.netloc.split('.')[1] if len(parsed_base.netloc.split('.')) > 1 else 'video'
ext = '.mp4' if '.mp4' in url else '.m3u8' if '.m3u8' in url else '.ts'
filename = f"{site_name}_{int(time.time())}{ext}"
return filename
except:
return f"video_{int(time.time())}.mp4"
def cancel_download(self):
"""取消下载"""
if not self.is_downloading:
return
if messagebox.askyesno("确认", "确定要取消下载吗？"):
self.log("用户取消下载")
self.is_downloading = False
self.download_stop_flag = True
if self.download_process:
try:
if platform.system() == "Windows":
subprocess.run(["taskkill", "/F", "/T", "/PID", str(self.download_process.pid)],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
else:
self.download_process.terminate()
time.sleep(1)
if self.download_process.poll() is None:
self.download_process.kill()
except Exception as e:
self.log(f"取消下载错误: {str(e)}")
if self.download_session:
try:
self.download_session.close()
except:
pass
self.update_progress(0, "下载已取消")
self.download_btn.config(state=tk.NORMAL)
self.cancel_btn.config(state=tk.DISABLED)
def open_output_folder(self):
"""打开输出文件夹"""
output_path = self.path_var.get().strip()
if not output_path or not os.path.exists(output_path):
messagebox.showwarning("警告", "输出路径不存在")
return
try:
if platform.system() == "Windows":
os.startfile(output_path)
elif platform.system() == "Darwin":
subprocess.run(["open", output_path])
else:
subprocess.run(["xdg-open", output_path])
self.log(f"已打开文件夹: {output_path}")
except Exception as e:
error_msg = f"打开文件夹失败: {str(e)}"
self.log(error_msg)
messagebox.showerror("错误", error_msg)
def format_size(self, bytes_size):
"""格式化文件大小"""
if bytes_size
ok，现在来解释一下这个修改后代码的核心逻辑和思路
一、核心功能概述
工具集成了多种视频获取方案（第三方工具调用ffmpeg.exe + 爬虫），支持清晰度选择、授权信息配置（Header/Cookie），并通过多线程实现异步操作，避免 UI 卡顿。
二、爬取与下载的核心流程
1. 准备工作：环境检测与初始化
下载工具检测：启动时自动检测系统中是否安装 you-get、yt-dlp（或兼容的 youtube-dl），这些工具是爬取视频的核心依赖。
UI 初始化：创建输入区域（URL、授权信息、输出路径）、信息展示区域（日志、视频信息、清晰度选择）和控制区域（按钮、进度条）。
会话初始化：创建带反爬机制的请求会话（requests.Session），包含随机 User-Agent、授权头信息（Authorization/Cookie）等，模拟浏览器行为。
2. 视频信息获取（get_video_info 方法）
目的是从视频 URL 中提取基础信息（标题、时长、上传者等），并为后续检测视频流做准备。流程如下：
输入验证：检查 URL 是否为空，输出路径是否有效。
授权信息保存：将用户输入的 Authorization、Cookie 等信息存入会话头，用于后续请求身份验证。
多方式信息提取：
优先调用第三方工具：通过 yt-dlp 或 you-get 的命令行参数（如 --dump-json、--info）获取结构化视频信息，解析后展示标题、时长等。
自定义爬虫兜底：若第三方工具失败，使用初始化的会话直接请求 URL，通过正则提取网页中的标题、描述等元数据（如 标签、meta 标签）。
3. 视频流与清晰度检测（_detect_video_streams 方法）
这是核心功能之一，目的是从视频页面中提取所有可用的视频流（不同清晰度、格式），并分类供用户选择。流程如下：
多来源检测视频流：
第三方工具提取：
yt-dlp：通过 --list-formats 命令获取所有可用视频格式，解析出格式 ID、清晰度（如 1080p）、大小等信息。
you-get：通过 --info 命令提取视频流 ID、分辨率（如 1920x1080）、文件大小等信息。
自定义爬虫提取：
JSON 数据解析：从网页脚本中提取包含视频信息的 JSON 数据（如 window.videoInfo、var videoData），递归解析其中的视频 URL。
直接 URL 匹配：通过正则匹配网页中直接出现的视频 URL（如 .mp4、.m3u8、.flv 等后缀），补全相对 URL 为绝对 URL。
高级 API 检测：若上述方法失败，尝试从网页中提取 API 接口（如 /api/video），调用接口获取视频流信息。
清晰度分类与排序：
通过 _extract_quality_from_string 方法从 URL 或格式信息中提取清晰度（如 1080p、720p）。
通过 _quality_sort_key 方法对清晰度排序（4K > 2K > 1080p > ...），并展示在下拉框中供用户选择。
4. 视频下载（start_download 方法）
根据选择的清晰度和下载工具，启动下载流程，核心逻辑如下：
选择视频流：根据用户选择的清晰度（或自动选择最佳），从检测到的视频流中匹配对应的流信息（URL 或格式 ID）。
选择下载方式：
第三方工具下载：若选择 yt-dlp 或 you-get，通过命令行调用工具，传入视频 URL、格式 ID、输出路径等参数（如 yt-dlp -f -o [U]）。
自定义爬虫下载：若为直接视频 URL（如 .mp4），使用 requests 分块下载（stream=True），支持断点续传和进度显示。
进度跟踪与控制：
通过多线程执行下载，避免 UI 冻结。
实时解析第三方工具的输出日志（如 yt-dlp 的进度信息），或计算分块下载的进度，更新进度条和状态文字。
支持取消下载（终止子进程或停止分块请求）。
三、关键技术与思路
多工具兼容：结合 yt-dlp（支持多平台、多网站）和 you-get（对国内网站支持较好）的优势，自动选择最佳工具。
反爬机制：
随机 User-Agent 模拟不同浏览器。
支持自定义 Header 和 Cookie，解决需要登录 / 授权才能访问的视频。
模拟浏览器请求头（Referer、Accept 等），避免被服务器识别为爬虫。
体验优化：
异步操作（多线程）避免 UI 卡顿。
清晰的进度展示和日志记录。
自动检测并提示缺失的依赖工具。
多种信息提取方式兜底（工具调用失败则用自定义爬虫）。
异常捕获与错误提示（如无效 URL、下载失败）。
最后送给大家ffmpeg全套框架下载
链接:https://wwcq.lanzouu.com/i30Ti33go8id，无密码
成品下载:https://wwcq.lanzouu.com/iWYo733gljha，无密码，感谢支持！（本人终于花9元巨款开通了会员，心碎了）
注意！这个工具仅为学习交流使用，不得违法使用！
本人亲测这个东西还有点小bug，希望大家一帮忙看一看，感谢！
