[Python] 纯文本查看 复制代码
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
from tkinter import *
from tkinter.filedialog import askopenfilename, asksaveasfilename
from tkinter.ttk import *
from tkinter import messagebox
import sqlite3
import os
import threading
import time
class BaiduPanApp:
def __init__(self, root):
self.root = root
self.root.title('百度云文件列表生成工具')
self.root.geometry('700x200')
self.db_path = StringVar()
self.save_path = StringVar()
self.status_text = StringVar()
self.status_text.set("准备就绪")
self.progress_value = DoubleVar()
self.progress_value.set(0)
self.total_files = 0
self.processed_files = 0
self.create_widgets()
def create_widgets(self):
# 数据库文件选择
db_label = Label(self.root, text="数据库文件:")
db_label.grid(row=0, column=0, sticky=W, padx=5, pady=5)
db_entry = Entry(self.root, width=60, textvariable=self.db_path)
db_entry['state'] = 'readonly'
db_entry.grid(row=0, column=1, padx=5, pady=5, sticky=W+E)
db_btn = Button(self.root, text="浏览...", command=self.select_db_file)
db_btn.grid(row=0, column=2, padx=5, pady=5)
# 保存文件选择
save_label = Label(self.root, text="保存位置:")
save_label.grid(row=1, column=0, sticky=W, padx=5, pady=5)
save_entry = Entry(self.root, width=60, textvariable=self.save_path)
save_entry['state'] = 'readonly'
save_entry.grid(row=1, column=1, padx=5, pady=5, sticky=W+E)
save_btn = Button(self.root, text="浏览...", command=self.select_save_file)
save_btn.grid(row=1, column=2, padx=5, pady=5)
# 状态信息显示区域
status_frame = Frame(self.root)
status_frame.grid(row=2, column=0, columnspan=3, sticky=W+E, padx=5, pady=5)
status_label = Label(status_frame, textvariable=self.status_text, anchor=W)
status_label.pack(side=LEFT, fill=X, expand=True)
# 进度条
self.progress = Progressbar(self.root, orient="horizontal", length=100,
mode="determinate", variable=self.progress_value)
self.progress.grid(row=3, column=0, columnspan=3, sticky=W+E, padx=5, pady=5)
# 按钮
button_frame = Frame(self.root)
button_frame.grid(row=4, column=0, columnspan=3, padx=5, pady=5)
create_btn = Button(button_frame, text="生成文件列表", command=self.start_processing)
create_btn.pack(side=LEFT, padx=5)
# 窗口自适应
self.root.columnconfigure(1, weight=1)
def select_db_file(self):
db_file = askopenfilename(
title="请选择BaiduYunCacheFileV0.db文件",
filetypes=[('数据库文件', '*.db')]
)
if db_file:
self.db_path.set(db_file)
def select_save_file(self):
save_file = asksaveasfilename(
title="选择保存位置",
defaultextension=".txt",
filetypes=[('文本文件', '*.txt')]
)
if save_file:
if not save_file.endswith('.txt'):
save_file += '.txt'
self.save_path.set(save_file)
def start_processing(self):
if not self.db_path.get():
messagebox.showerror("错误", "请先选择数据库文件！")
return
if not self.save_path.get():
messagebox.showerror("错误", "请先选择保存位置！")
return
# 启动线程处理，避免界面卡住
self.progress_value.set(0)
self.status_text.set("正在处理数据...")
threading.Thread(target=self.create_baiduyun_filelist, daemon=True).start()
def update_progress(self):
if self.total_files > 0:
progress = (self.processed_files / self.total_files) * 100
self.progress_value.set(progress)
self.status_text.set(f"正在处理: {self.processed_files}/{self.total_files} ({int(progress)}%)")
self.root.update_idletasks()
def check_table_structure(self, conn):
"""检查数据库表结构，确定正确的列名"""
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(cache_file)")
columns = cursor.fetchall()
column_names = [col[1] for col in columns]
path_column = None
name_column = None
# 寻找可能的路径列和名称列
path_candidates = ['path', 'server_path', 'server_filename', 'dir_path']
name_candidates = ['file_name', 'name', 'server_filename', 'filename']
for col in path_candidates:
if col in column_names:
path_column = col
break
for col in name_candidates:
if col in column_names:
name_column = col
break
# 如果没有找到合适的列，尝试获取前几行数据来分析
if not path_column or not name_column:
cursor.execute("SELECT * FROM cache_file LIMIT 1")
row = cursor.fetchone()
if row:
# 通过分析数据内容推测哪些列可能是路径和文件名
for i, value in enumerate(row):
if isinstance(value, str):
if '/' in value and path_column is None:
path_column = column_names[i]
elif name_column is None and '.' in value:
name_column = column_names[i]
# 如果还是无法确定，使用索引位置
if not path_column:
path_column = 2 # 假设第3列是路径
if not name_column:
name_column = 3 # 假设第4列是文件名
return path_column, name_column
def create_baiduyun_filelist(self):
try:
file_dict = {}
conn = sqlite3.connect(self.db_path.get())
try:
# 检查表结构并获取正确的列名
path_column, name_column = self.check_table_structure(conn)
cursor = conn.cursor()
# 先检查表是否存在
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='cache_file'")
if not cursor.fetchone():
raise Exception("数据库中没有找到'cache_file'表，请确认选择了正确的百度云缓存数据库文件")
# 获取数据总量
cursor.execute("SELECT COUNT(*) FROM cache_file")
self.total_files = cursor.fetchone()[0]
self.processed_files = 0
if isinstance(path_column, int) and isinstance(name_column, int):
# 使用索引位置
cursor.execute("SELECT * FROM cache_file")
for row in cursor.fetchall():
if len(row) > max(path_column, name_column):
path = str(row[path_column]) if row[path_column] else "/"
name = str(row[name_column]) if row[name_column] else "未知文件"
if path not in file_dict:
file_dict[path] = []
file_dict[path].append(name)
self.processed_files += 1
if self.processed_files % 100 == 0:
self.update_progress()
else:
# 使用列名
query = f"SELECT {path_column}, {name_column} FROM cache_file"
cursor.execute(query)
for row in cursor.fetchall():
path = str(row[0]) if row[0] else "/"
name = str(row[1]) if row[1] else "未知文件"
if path not in file_dict:
file_dict[path] = []
file_dict[path].append(name)
self.processed_files += 1
if self.processed_files % 100 == 0:
self.update_progress()
except sqlite3.OperationalError as sqle:
# 处理特定的SQLite错误
if "no such collation sequence" in str(sqle):
# 尝试使用不带排序的简单查询
cursor = conn.cursor()
cursor.execute("SELECT * FROM cache_file")
rows = cursor.fetchall()
self.total_files = len(rows)
self.processed_files = 0
for row in rows:
# 假设路径和文件名在特定位置
if len(row) > 3: # 确保至少有足够的列
path = str(row[2]) if row[2] else "/"
name = str(row[3]) if row[3] else "未知文件"
if path not in file_dict:
file_dict[path] = []
file_dict[path].append(name)
self.processed_files += 1
if self.processed_files % 100 == 0:
self.update_progress()
else:
raise # 如果是其他SQLite错误，继续抛出
finally:
conn.close()
# 如果没有获取到任何数据，抛出错误
if not file_dict:
raise Exception("未能从数据库中提取到任何文件信息，请确认数据库格式正确")
# 写入文件
self.status_text.set("正在生成文件树...")
self.root.update_idletasks()
with open(self.save_path.get(), "w", encoding='utf-8') as fp:
self.write_file(file_dict, fp, "/")
self.status_text.set(f"处理完成! 文件已保存至: {self.save_path.get()}")
self.progress_value.set(100)
messagebox.showinfo("完成", f"文件列表已成功生成!\n保存路径: {self.save_path.get()}")
except Exception as e:
self.status_text.set(f"处理出错: {str(e)}")
messagebox.showerror("错误", f"生成文件列表时发生错误:\n{str(e)}")
def write_file(self, file_dict, f, item, gap=""):
if item == "/":
f.write("━" + "/" + "\n")
if "/" in file_dict:
# 不使用内置排序函数，避免编码问题
files = file_dict["/"]
files.sort() # 简单排序
for i in files:
f.write("┣" + "━" + i + "\n")
i = item + i + "/"
if i in file_dict:
self.write_file(file_dict, f, i, gap="┣━")
else:
gap = "┃ " + gap
if item in file_dict:
# 不使用内置排序函数，避免编码问题
files = file_dict[item]
files.sort() # 简单排序
for i in files:
f.write(gap + i + "\n")
i = item + i + "/"
if i in file_dict:
self.write_file(file_dict, f, i, gap)
if __name__ == "__main__":
root = Tk()
app = BaiduPanApp(root)
root.mainloop()
发表于 2025-3-31 15:25
