import os
import tkinter as tk
from tkinter import filedialog, messagebox
import requests
from queue import Queue, Empty
from threading import Thread
定义全局变量
txt_path = ""
download_path = "C:\ProgramData\Topaz Labs LLC\Topaz Photo AI\models\"
urls = []
download_queue = Queue()
thread_count = 5 # 默认线程数
root = tk.Tk()
确保下载目录存在
def ensure_download_path():
global download_path
if not os.path.exists(download_path):
os.makedirs(download_path)
更新状态栏
def update_status(text):
status_label.config(text=text)
选择TXT文档
def select_txt_file():
global txt_path, urls
file_path = filedialog.askopenfilename()
if file_path:
with open(file_path, 'r') as file:
urls = [line.strip() for line in file if line.strip()]
txt_path = file_path
txt_txt_path.delete(0, tk.END)
txt_txt_path.insert(0, txt_path)
update_status(f"已载入 {len(urls)} 个URL。")
浏览保存文件夹
def select_download_path():
global download_path
folder_path = filedialog.askdirectory()
if folder_path:
download_path = folder_path
txt_download_path.delete(0, tk.END)
txt_download_path.insert(0, download_path)
update_status(f"已选择保存路径: {download_path}")
下载文件
def download_file(url, file_name):
file_path = os.path.join(download_path, file_name)
try:
response = requests.get(url, stream=True)
if response.status_code == 200:
with open(file_path, 'wb') as file:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
file.write(chunk)
update_status(f'下载完成: {file_name}')
except Exception as e:
update_status(f'下载错误: {file_name} (错误信息: {e})')
下载文件的线程函数
def thread_download_files():
while True:
try:
url, file_name = download_queue.get(timeout=1)
except Empty:
break # 队列为空，退出线程
try:
download_file(url, file_name)
except Exception as e:
update_status(f'下载错误: {file_name} (错误信息: {e})')
finally:
download_queue.task_done()
在线程中执行下载任务
def download_in_thread():
global download_queue
if not os.path.exists(download_path):
ensure_download_path()
# 清空队列并添加新的下载任务
download_queue = Queue()
for url in urls:
file_name = url.split("/")[-1]
download_queue.put((url, file_name))
update_status(f"添加下载任务: {file_name}")
# 创建并启动下载线程
threads = []
for _ in range(thread_count): # 创建线程处理所有任务
thread = Thread(target=thread_download_files)
thread.start()
threads.append(thread)
# 等待所有下载线程完成
for thread in threads:
thread.join()
# 所有任务完成后更新状态栏
root.after(0, lambda: update_status("所有下载任务完成！"))
开始下载
def start_download():
global urls, download_queue
if not urls:
messagebox.showerror("错误", "没有可下载的URL。")
return
Thread(target=download_in_thread).start()
创建GUI窗口
root.title("Topaz模型辅助下载器")
创建状态栏
status_frame = tk.Frame(root, bd=2, relief=tk.SUNKEN)
status_label = tk.Label(status_frame, text="就绪。", anchor="w")
status_label.pack(side="left", fill="x", padx=5, pady=5)
status_frame.pack(side="bottom", fill="x")
创建TXT路径相关控件
txt_path_frame = tk.Frame(root)
lbl_txt_path = tk.Label(txt_path_frame, text="TXT路径:")
lbl_txt_path.pack(side="left", padx=5, pady=5, anchor="w")
txt_txt_path = tk.Entry(txt_path_frame, width=50)
txt_txt_path.insert(0, txt_path)
txt_txt_path.pack(side="left", padx=5, pady=5, fill="x", expand=True)
btn_select_txt = tk.Button(txt_path_frame, text="浏览TXT", command=select_txt_file)
btn_select_txt.pack(side="left", padx=5, pady=5)
创建保存路径相关控件
download_path_frame = tk.Frame(root)
lbl_download_path = tk.Label(download_path_frame, text="保存路径:")
lbl_download_path.pack(side="left", padx=5, pady=5, anchor="w")
txt_download_path = tk.Entry(download_path_frame, width=50)
txt_download_path.insert(0, download_path)
txt_download_path.pack(side="left", padx=5, pady=5, fill="x", expand=True)
btn_select_download_path = tk.Button(download_path_frame, text="选择路径", command=select_download_path)
btn_select_download_path.pack(side="left", padx=5, pady=5)
btn_start_download = tk.Button(download_path_frame, text="开始下载", command=start_download)
btn_start_download.pack(side="left", padx=5, pady=5)
打包TXT路径相关控件
txt_path_frame.pack(side="top", fill="x", padx=5, pady=5)
打包保存路径相关控件
download_path_frame.pack(side="top", fill="x", padx=5, pady=5)
运行GUI
root.mainloop()
发表于 2024-3-27 22:58
发表于 2024-3-28 05:43
|
发表于 2024-3-28 09:49
，或者是在添加一个时间监控下当前运行了多长时间，如果超出时间限制就报错退出 
