import sys
import os
import re
import time
from datetime import datetime
from urllib.parse import urlparse, unquote
from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QTextEdit, QFileDialog,
QMessageBox, QCheckBox, QProgressBar)
from PyQt5.QtCore import QProcess, Qt
class M3U8Downloader(QMainWindow):
def init(self):
super().init()
初始化组件
self.process = QProcess(self)
self.current_dir = os.path.dirname(os.path.abspath(__file__))
self.log_file = open('download.log', 'a', encoding='utf-8')
self.init_ui()
self.init_process()
def init_ui(self):
"""初始化用户界面"""
self.setWindowTitle('M3U8视频下载器 PRO')
self.setGeometry(200, 200, 1000, 800)
# 主控件
central_widget = QWidget()
self.setCentralWidget(central_widget)
main_layout = QVBoxLayout(central_widget)
# 输入组件
self.url_input = QLineEdit(placeholderText='输入M3U8链接（支持http/https）')
self.url_input.textChanged.connect(self.auto_generate_filename)
self.filename_input = QLineEdit(placeholderText='自动生成文件名')
self.path_input = QLineEdit()
self.path_input.setText(os.path.normpath(os.path.join(self.current_dir, 'downloads')))
# 代理设置
self.proxy_check = QCheckBox("使用代理")
self.proxy_input = QLineEdit(placeholderText='127.0.0.1:7890')
# 功能按钮
self.browse_btn = QPushButton('选择路径', clicked=self.select_output_dir)
self.download_btn = QPushButton('开始下载', clicked=self.start_download)
self.download_btn.setStyleSheet("background-color: #4CAF50; color: white;")
# 进度条
self.progress_bar = QProgressBar()
self.progress_bar.setAlignment(Qt.AlignCenter)
# 日志显示
self.log_text = QTextEdit(readOnly=True)
self.log_text.setStyleSheet("font-family: Consolas; font-size: 10pt;")
# 布局管理
input_layout = self.create_input_layout()
proxy_layout = self.create_proxy_layout()
main_layout.addLayout(input_layout)
main_layout.addLayout(proxy_layout)
main_layout.addWidget(self.progress_bar)
main_layout.addWidget(self.download_btn)
main_layout.addWidget(QLabel('下载日志:'))
main_layout.addWidget(self.log_text)
def init_process(self):
"""初始化进程信号连接"""
self.process.readyReadStandardOutput.connect(self.handle_stdout)
self.process.readyReadStandardError.connect(self.handle_stderr)
self.process.finished.connect(self.on_process_finished)
self.process.errorOccurred.connect(self.handle_process_error)
def create_input_layout(self):
"""创建输入区域布局"""
layout = QVBoxLayout()
url_layout = QHBoxLayout()
url_layout.addWidget(QLabel('视频链接:'))
url_layout.addWidget(self.url_input)
file_layout = QHBoxLayout()
file_layout.addWidget(QLabel('保存名称:'))
file_layout.addWidget(self.filename_input)
path_layout = QHBoxLayout()
path_layout.addWidget(QLabel('保存路径:'))
path_layout.addWidget(self.path_input)
path_layout.addWidget(self.browse_btn)
layout.addLayout(url_layout)
layout.addLayout(file_layout)
layout.addLayout(path_layout)
return layout
def create_proxy_layout(self):
"""创建代理设置布局"""
layout = QHBoxLayout()
layout.addWidget(self.proxy_check)
layout.addWidget(self.proxy_input)
layout.addStretch()
return layout
def auto_generate_filename(self):
"""智能生成文件名"""
url = self.url_input.text().strip()
if not url:
return
try:
parsed = urlparse(unquote(url)) # URL解码处理
base_name = os.path.basename(parsed.path).split('?')[0] # 去除查询参数
if not base_name or '.' not in base_name:
domain_part = parsed.netloc.split('.')[-2] if '.' in parsed.netloc else "video"
base_name = f"{domain_part}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
self.filename_input.setText(base_name)
except Exception as e:
self.log_message('WARN', f'自动命名失败: {str(e)}')
self.filename_input.setText(f"video_{int(time.time())}")
def validate_inputs(self):
"""输入验证"""
if not self.url_input.text().startswith(('http://', 'https://')):
QMessageBox.warning(self, '警告', '请输入有效的HTTP(S)链接')
return False
if not os.access(self.path_input.text(), os.W_OK):
QMessageBox.critical(self, '错误', '目标路径不可写，请选择其他目录')
return False
return True
def build_ffmpeg_command(self):
"""构建FFmpeg命令"""
output_path = os.path.join(
self.path_input.text(),
f"{self.filename_input.text()}.mp4"
)
return [
'ffmpeg',
'-i', self.url_input.text(),
'-c', 'copy',
'-y', # 覆盖已存在文件
output_path
]
def start_download(self):
"""启动下载进程"""
if not self.validate_inputs():
return
try:
# 确保输出目录存在
os.makedirs(self.path_input.text(), exist_ok=True)
# 配置进程参数
self.process.setProgram('ffmpeg')
self.process.setArguments(self.build_ffmpeg_command()[1:]) # 跳过程序名
# 设置代理环境变量（如果需要）
if self.proxy_check.isChecked():
proxy = self.proxy_input.text()
env = self.process.processEnvironment()
env.insert('HTTP_PROXY', proxy)
env.insert('HTTPS_PROXY', proxy)
self.process.setProcessEnvironment(env)
self.download_btn.setEnabled(False)
self.process.start()
self.log_message('INFO', '下载进程已启动...')
except Exception as e:
self.log_message('ERROR', f'进程启动失败: {str(e)}')
QMessageBox.critical(self, '错误', '无法启动下载进程')
def handle_stdout(self):
"""处理标准输出"""
data = self.process.readAllStandardOutput().data().decode()
self.log_message('INFO', data)
# 解析进度信息
if match := re.search(r'time=(\d+:\d+:\d+\.\d+)', data):
self.log_text.append(f"▶ 当前进度: {match.group(1)}")
def handle_stderr(self):
"""处理错误输出"""
data = self.process.readAllStandardError().data().decode()
self.log_message('ERROR', data)
def handle_process_error(self, error):
"""处理进程错误"""
self.log_message('CRITICAL', f'进程异常: {error.name}')
QMessageBox.critical(self, '错误', f'进程错误: {error.name}')
def on_process_finished(self, exit_code, exit_status):
"""进程结束处理"""
self.download_btn.setEnabled(True)
if exit_code == 0:
self.log_message('SUCCESS', '下载完成！')
QMessageBox.information(self, '完成', '视频下载成功')
else:
self.log_message('ERROR', f'进程异常退出，代码: {exit_code}')
def select_output_dir(self):
"""选择保存目录"""
path = QFileDialog.getExistingDirectory(
self,
"选择保存目录",
self.path_input.text(),
QFileDialog.ShowDirsOnly
)
if path:
self.path_input.setText(os.path.normpath(path))
def log_message(self, level, message):
"""分级日志记录"""
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
log_entry = f"[{timestamp}] [{level}] {message}"
# 控制台颜色显示
if level == 'ERROR':
formatted = f'<span style="color: red;">{log_entry}</span>'
elif level == 'SUCCESS':
formatted = f'<span style="color: green;">{log_entry}</span>'
else:
formatted = log_entry
self.log_text.append(formatted)
self.log_file.write(log_entry + '\n')
self.log_file.flush()
def closeEvent(self, event):
"""窗口关闭时清理资源"""
self.log_file.close()
if self.process.state() == QProcess.Running:
self.process.terminate()
event.accept()
if name == 'main':
app = QApplication(sys.argv)
window = M3U8Downloader()
window.show()
sys.exit(app.exec_())
[复制链接]
发表于 2025-3-10 09:47
