[Python] 纯文本查看 复制代码
import os.path
import time
from PyQt5.QtWidgets import QApplication, QWidget, QMessageBox,QFileDialog
import sys
from PyQt5 import QtCore, QtWidgets
from PyQt5.QtCore import QThread, pyqtSignal, Qt
from PyQt5.QtGui import QIcon
import asyncio
from qasync import QEventLoop
# 隐藏运行调试框的警告信息
import warnings
from get_film import download_film
warnings.filterwarnings("ignore", category=DeprecationWarning)
import warnings; warnings.filterwarnings("ignore", category=UserWarning, message="libpng warning: iCCP: known incorrect sRGB profile")
class Ui_Form(object):
def setupUi(self, Form):
Form.setObjectName("Form")
Form.resize(520, 608)
self.gridLayout = QtWidgets.QGridLayout(Form)
self.gridLayout.setObjectName("gridLayout")
self.label_3 = QtWidgets.QLabel(Form)
self.label_3.setObjectName("label_3")
self.gridLayout.addWidget(self.label_3, 3, 0, 1, 1)
self.listWidget_movie_list = QtWidgets.QListWidget(Form)
self.listWidget_movie_list.setVerticalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff)
self.listWidget_movie_list.setObjectName("listWidget_movie_list")
self.gridLayout.addWidget(self.listWidget_movie_list, 4, 0, 1, 5)
self.pushButton_search = QtWidgets.QPushButton(Form)
self.pushButton_search.setObjectName("pushButton_search")
self.gridLayout.addWidget(self.pushButton_search, 0, 2, 1, 2)
self.lineEdit_path = QtWidgets.QLineEdit(Form)
self.lineEdit_path.setObjectName("lineEdit_path")
self.gridLayout.addWidget(self.lineEdit_path, 3, 1, 1, 3)
self.lineEdit_film = QtWidgets.QLineEdit(Form)
self.lineEdit_film.setObjectName("lineEdit_film")
self.gridLayout.addWidget(self.lineEdit_film, 0, 1, 1, 1)
self.label_2 = QtWidgets.QLabel(Form)
self.label_2.setObjectName("label_2")
self.gridLayout.addWidget(self.label_2, 2, 0, 1, 1)
self.label = QtWidgets.QLabel(Form)
self.label.setObjectName("label")
self.gridLayout.addWidget(self.label, 0, 0, 1, 1)
self.pushButton_restlist = QtWidgets.QPushButton(Form)
self.pushButton_restlist.setObjectName("pushButton_restlist")
self.gridLayout.addWidget(self.pushButton_restlist, 0, 4, 1, 1)
self.pushButton = QtWidgets.QPushButton(Form)
self.pushButton.setObjectName("pushButton")
self.gridLayout.addWidget(self.pushButton, 3, 4, 1, 1)
self.lineEdit_numThreads = QtWidgets.QLineEdit(Form)
self.lineEdit_numThreads.setMinimumSize(QtCore.QSize(71, 0))
self.lineEdit_numThreads.setMaximumSize(QtCore.QSize(71, 16777215))
self.lineEdit_numThreads.setObjectName("lineEdit_numThreads")
self.lineEdit_numThreads.setToolTip('根据电脑的性能调整线程数 最小1，最大10000(嘿嘿卡爆你)超出范围默认使用100')
self.lineEdit_numThreads.setAlignment(Qt.AlignCenter)
self.gridLayout.addWidget(self.lineEdit_numThreads, 2, 4, 1, 1)
self.label_4 = QtWidgets.QLabel(Form)
self.label_4.setMaximumSize(QtCore.QSize(36, 16777215))
self.label_4.setObjectName("label_4")
self.gridLayout.addWidget(self.label_4, 2, 3, 1, 1)
self.lineEdit_epnum = QtWidgets.QLineEdit(Form)
self.lineEdit_epnum.setToolTip('输入"1"只下第1集\n输入"5+"从5集开始往后全下\n输入"7+9" 只下7-9集\n留空或电影全下')
self.lineEdit_epnum.setWhatsThis("")
self.lineEdit_epnum.setText("")
self.lineEdit_epnum.setObjectName("lineEdit_epnum")
self.gridLayout.addWidget(self.lineEdit_epnum, 2, 1, 1, 2)
self.retranslateUi(Form)
QtCore.QMetaObject.connectSlotsByName(Form)
def retranslateUi(self, Form):
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "视频采集 - 吾爱破解 by：我很忙！"))
self.label_3.setText(_translate("Form", "路径："))
self.pushButton_search.setText(_translate("Form", "搜索片源"))
self.label_2.setText(_translate("Form", "集数："))
self.label.setText(_translate("Form", "片名："))
self.pushButton_restlist.setText(_translate("Form", "清空"))
self.pushButton.setText(_translate("Form", "选择"))
self.label_4.setText(_translate("Form", "线程："))
# 绑定搜索按钮(btn_search函数)
self.pushButton_search.clicked.connect(Form.btn_search)
# 绑定清空按钮
self.pushButton_restlist.clicked.connect(Form.btn_restlist)
# 绑定更改路径按钮
self.pushButton.clicked.connect(Form.btn_data_path)
# 定义 WorkerThread 类，继承自 QThread 调用普通函数
class WorkerTherad(QThread):
result_signal = pyqtSignal(object)
#传入指定模块函数，并接收不定长参数
def __init__(self, target_function, *args):
super().__init__()
self.target_function = target_function
self.args = args
# 开始调用传入的目标函数和不定长参数
def run(self):
# 接收调用后返回的结果
result = self.target_function(*self.args)
# 将结果返回给 调用的的主程序事件
self.result_signal.emit(result)
class MyWindow(QWidget):
def __init__(self):
super().__init__()
# 实例化图形界面
self.ui = Ui_Form()
self.ui.setupUi(self)
self.setWindowIcon(QIcon('icon.ico'))
self.ui.listWidget_movie_list.setVerticalScrollBarPolicy(2)
# 输入集数编辑框设置占位字符串
self.ui.lineEdit_epnum.setPlaceholderText('指定规则：1，5+，6+9，留空，其他字符串无效')
self.ui.lineEdit_film.setPlaceholderText('请输入电影/电视剧名称(尽量使用全称精确搜索)')
self.ui.lineEdit_numThreads.setText('100')
# 搜索到的数据存入film_data 和选择列表后分别存入 集数列表和 url列表
self.film_data, self.film_title, self.film_urls, self.film_episodes = '' , '', [], []
#在选择集数之后，把新的链接和电影名称重新赋值，等带下载片段函数调用
self.updata_film_episodes, self.updata_film_urls = '', ''
# 记录列表框的链接次数
self.double_click_connection_count = 0
# 线程 和保存的目录
self.Threads_num, self.path = 0, ''
documents_path = os.path.expanduser('~/Documents')
# 配置文件名
config_filename = 'download_videos_path.ini'
# 完整的配置文件路径
self.config_file_path = os.path.join(documents_path, config_filename)
# 判断 配置文件是否存在，存在则读取
if os.path.exists(self.config_file_path):
try:
with open(self.config_file_path,'r') as f:
self.path = f.readline().strip() # 读取第一行
self.ui.lineEdit_path.setText(self.path)
threads = f.readline().strip() # 读取第二行
self.Threads_num = int(threads)
self.ui.lineEdit_numThreads.setText(str(threads))
except Exception as e:
self.message(f'读取配置文件失败，{e}')
def btn_search(self):
# 判断列表框的链接信号，如果=1表示要断开，因为在搜索时，点击了列表框，多个连接的话，会重复执行下载m3u8的函数
if self.double_click_connection_count == 1:
try:
self.ui.listWidget_movie_list.itemDoubleClicked.disconnect(self.list_download)
# 记录链接次数 归零
self.double_click_connection_count = 0
except Exception as e:
QMessageBox.critical(self, "错误", f"断开绑定事件发生错误：{str(e)}")
# 搜索前清空列表框和旧数据
self.ui.listWidget_movie_list.clear()
# 接收编辑框搜索片源的内容,(搜索片源按钮被按下时触发)
flim_name = self.ui.lineEdit_film.text()
flim_name = flim_name.strip()
if not flim_name:
return
#
# 调用线程类，传入指定模块download_film().并调用该模块下的函数get_m3u8_url，传入不定长参数
try:
self.worker_thread = WorkerTherad(download_film().get_m3u8_url, self.ui,flim_name)
self.worker_thread.result_signal.connect(self.handle_result)
# 开始线程
self.worker_thread.start()
except Exception as e:
self.message(f'搜索时出现错误！{e}')
# self.film_data,status = self.dl.get_m3u8_url(self.ui, flim_name)
if self.film_data == None:
return
# 在搜索到结果之后绑定列表框双击事件，每次链接，次数=1
self.ui.listWidget_movie_list.itemDoubleClicked.connect(self.list_download)
self.double_click_connection_count = 1
def handle_result(self, result):
"""
:param result: #接收模块返回的内容，字典形式。
'type': 'select_film',表示搜索电影
'isSuccess': '成功/失败'
'msg': 原因/结果}
:return: 无返回值
"""
# 如果调用函数没有返回值 则不进行取值判断
if result == None:
#解除所有组件的禁用
self.isCease(True)
return
# 判断搜索的片源是否成功，成功后将数据传入当前类的全局变量中，共双击列表框下载时使用
if result['type'] == 'select_film':
if result['isSuccess'] == '成功':
self.film_data = result['msg']
else:
self.isCease(True)
self.message(result['msg'])
# 判断 调用模块的下载m3u8文件功能，成功后，可根据每个文件开始下载ts片段
if result['type'] == 'dolad_m3u8':
if result['isSuccess'] == '成功':
obj = download_film()
asyncio.ensure_future(obj.download_all_videos(self.updata_film_urls, self.updata_film_episodes,self.film_title, self.path, self.ui, self.Threads_num, QMessageBox))
else:
self.isCease(True)
self.message(result['msg'])
def isCease(self, True_or_false):
'''
在操作时 通常情况需要禁止按钮，等待操作完成之后释放用
:param True_or_false: false = 不能使用按钮 相反可以使用
:return:
'''
if type(True_or_false) == bool:
self.ui.pushButton.setEnabled(True_or_false)
self.ui.pushButton_restlist.setEnabled(True_or_false)
self.ui.pushButton_search.setEnabled(True_or_false)
def btn_restlist(self):
# 接收清空按钮的触发事件
self.ui.lineEdit_epnum.clear()
self.ui.lineEdit_film.clear()
self.ui.listWidget_movie_list.clear()
def btn_data_path(self):
# 接收更改路径的触发事件
directory = QFileDialog.getExistingDirectory(self,'选择电影保存目录')
if directory:
self.ui.lineEdit_path.setText(directory)
with open (self.config_file_path,'w') as f:
f.write(directory)
numthreads = self.ui.lineEdit_numThreads.text()
if numthreads:
f.write(f'\n{numthreads}')
def list_download(self, items):
self.film_title, self.film_urls, self.film_episodes = '', [] ,[]
try:
# 双击列表框后，选择对应的电影数据
index = self.ui.listWidget_movie_list.indexFromItem(items).row()
self.film_title = self.film_data[index]['name']
for item in self.film_data[index]['source']['eps']:
# 取视频集数
self.film_episodes.append(item['name'])
# 取视频集数对应m3u8文件列表
self.film_urls.append(item['url'])
msg = QMessageBox.information(self, '提示', f'是否确认下载 "{self.film_title}"?\t',QMessageBox.Ok | QMessageBox.No)
if msg == QMessageBox.Ok:
# 询问是否开始下载
#开始下载m3u8文件，传入，url，集数列表和电视剧标题
# result = self.dl.download_m3u8(self.film_urls, self.film_episodes, self.film_title,'E:/电视剧')
# 读取编辑指定集数的内容
ep_data = self.ui.lineEdit_epnum.text()
# 判断 集数列表
if len(self.film_episodes) > 1:
# 如果输入5+ 那么从5往后下载
if ep_data.find('+') != -1:
if ep_data.split('+')[1]:
start_index = int(ep_data.split('+')[0]) - 1 # 取+号左边
end_index = int(ep_data.split('+')[1])
# 按指定起始位置和列表总长度取出对应集数和url
self.updata_film_episodes = self.film_episodes[start_index:end_index]
self.updata_film_urls = self.film_urls[start_index:end_index]
else:
start_index = int(ep_data.split('+')[0]) - 1 # 取+号左边
#按指定起始位置和列表总长度取出对应集数和url
self.updata_film_episodes = self.film_episodes[start_index:len(self.film_episodes)]
self.updata_film_urls = self.film_urls[start_index:len(self.film_urls)]
#如果没有输入，默认取全部集数
elif ep_data == '':
self.updata_film_episodes = self.film_episodes
self.updata_film_urls = self.film_urls
# 如果没找到+ 或者不等于空格，则取出指定一集 如输入5 取第五集
else:
self.updata_film_episodes = self.film_episodes[int(ep_data)-1:int(ep_data)]
self.updata_film_urls = self.film_urls[int(ep_data)-1:int(ep_data)]
# 如果集数列表只有一条，默认全部下载
else:
self.updata_film_episodes = self.film_episodes
self.updata_film_urls = self.film_urls
# 如果赛选出的列表存在数据，则启动协程 开始下载，反之报错提示规则错误！
if len(self.updata_film_episodes) >= 1:
# 取编辑框输入的线程数
self.Threads_num = int(self.ui.lineEdit_numThreads.text())
self.path = self.ui.lineEdit_path.text()
if not self.path:
self.message('未选中目录')
return
#判断是否为纯整数
if str(self.Threads_num).isdigit():
# 如果输入的线程小于=0或大于10000，则默认设置为100
self.Threads_num = 100 if self.Threads_num <= 0 or self.Threads_num > 10000 else self.Threads_num
self.worker_thread = WorkerTherad(download_film().download_m3u8, self.updata_film_urls, self.updata_film_episodes, self.film_title,self.path,self.ui)
self.worker_thread.result_signal.connect(self.handle_result)
# 开始线程
self.worker_thread.start()
# 取消绑定列表框,如记录绑定次数=1，执行解绑，防止误触列表框
if self.double_click_connection_count == 1:
self.ui.listWidget_movie_list.itemDoubleClicked.disconnect()
self.double_click_connection_count = 0
# 开始下载后，将禁止点击所有按钮，保护程序正常执行
self.isCease(False)
else:
self.message('没有选出集数信息，请检查集数筛选规则（不能越界选择，如电视剧共10集，只能输入包括10的范围内）电影可忽略！')
except Exception as e:
self.message(f"在选择指定电影时出现意外：{str(e)}")
def message(self, msg):
'''
传入提醒信息，可直接提示信息框
:param msg: 需要提示的内容
:return: 返回逻辑值
'''
if type(msg) == str:
msg_box = QMessageBox()
msg_box.setWindowIcon(QIcon('icon.ico'))
msg_box.setText(msg)
msg_box.setWindowTitle("提示")
msg_box.setIcon(QMessageBox.Information)
msg_box.exec_()
if __name__ == '__main__':
# 获取命令行参数
app = QApplication(sys.argv)
# 创建 QEventLoop 对象
loop = QEventLoop(app)
# 设置当前的事件循环为创建的 QEventLoop 对象
asyncio.set_event_loop(loop)
# 创建 MyWindow 类的实例，即窗口对象
win = MyWindow()
# 显示窗口
win.show()
# 启动事件循环，确保程序持续运行
with loop:
loop.run_forever()
[复制链接]
发表于 2024-9-4 11:12
收藏
淘帖
有用
分享到朋友圈
