JPG 恢复
时间：2026-05-08 15:50
import os import sys import ctypes import threading import time import tkinter as tk from tkinter import ttk, scrolledtext, messagebox from pathlib import Path # 保存目录 SAVE_DIR = Path("D:/11") # JPG 特征码 JPG_START = b"\xff\xd8\xff" JPG_END = b"\xff\xd9" # 扇区大小（Windows默认512字节） SECTOR_SIZE = 512 # 扫描缓冲区大小 BUFFER_SIZE = SECTOR_SIZE * 1024 # 每次读取512KB class DiskJPGRecovery: def __init__(self, root): self.root = root self.root.title("磁盘JPG裸数据恢复工具 | 管理员运行") self.root.geometry("850x600") self.root.resizable(False, False) # 状态控制变量 self.is_running = False self.is_paused = False self.stop_flag = False self.scan_thread = None # 初始化UI self._init_ui() # 检查管理员权限 self._check_admin() # 自动获取物理磁盘 self._get_physical_disks() def _init_ui(self): # 顶部磁盘选择区 frame_top = ttk.Frame(self.root, padding=10) frame_top.pack(fill=tk.X) ttk.Label(frame_top, text="选择物理磁盘：").grid(row=0, column=0, padx=5) self.disk_var = tk.StringVar() self.disk_combo = ttk.Combobox(frame_top, textvariable=self.disk_var, state="readonly", width=20) self.disk_combo.grid(row=0, column=1, padx=5) # 按钮区 frame_btn = ttk.Frame(self.root, padding=10) frame_btn.pack(fill=tk.X) self.start_btn = ttk.Button(frame_btn, text="开始扫描", command=self.start_scan) self.start_btn.grid(row=0, column=0, padx=5) self.pause_btn = ttk.Button(frame_btn, text="暂停", command=self.pause_scan, state=tk.DISABLED) self.pause_btn.grid(row=0, column=1, padx=5) self.stop_btn = ttk.Button(frame_btn, text="停止", command=self.stop_scan, state=tk.DISABLED) self.stop_btn.grid(row=0, column=2, padx=5) ttk.Label(frame_btn, text=" 状态：").grid(row=0, column=3, padx=5) self.status_label = ttk.Label(frame_btn, text="就绪", foreground="green") self.status_label.grid(row=0, column=4, padx=5) # 日志输出区 frame_log = ttk.Frame(self.root, padding=10) frame_log.pack(fill=tk.BOTH, expand=True) ttk.Label(frame_log, text="扫描日志：").pack(anchor=tk.W) self.log_text = scrolledtext.ScrolledText(frame_log, width=100, height=25) self.log_text.pack(fill=tk.BOTH, expand=True) def _check_admin(self): """检查是否以管理员身份运行""" try: is_admin = ctypes.windll.shell32.IsUserAnAdmin() except: is_admin = False if not is_admin: messagebox.showerror("权限错误", "请【右键-以管理员身份运行】此程序！") self.root.destroy() sys.exit() def _get_physical_disks(self): """自动获取Windows物理磁盘列表""" disks = [] for i in range(10): # 最多扫描10个物理磁盘 disk_path = f"\\\\.\\PhysicalDrive{i}" try: # 测试打开磁盘（只读） handle = open(disk_path, "rb") handle.close() disks.append(disk_path) except: continue self.disk_combo["values"] = disks if disks: self.disk_combo.current(0) self.log(f"自动发现磁盘：{disks}") else: self.log("未检测到物理磁盘，请检查权限！") def log(self, msg): """日志输出（线程安全）""" def _write(): self.log_text.insert(tk.END, f"[{time.strftime('%H:%M:%S')}] {msg}\n") self.log_text.see(tk.END) self.root.after(0, _write) def update_status(self, text, color="green"): """更新状态文本""" def _update(): self.status_label.config(text=text, foreground=color) self.root.after(0, _update) def _save_jpg(self, data): """保存找到的JPG文件""" SAVE_DIR.mkdir(exist_ok=True) filename = f"recovered_{int(time.time()*1000)}.jpg" path = SAVE_DIR / filename with open(path, "wb") as f: f.write(data) return filename def _scan_disk(self, disk_path): """磁盘扫描核心函数（子线程）""" self.log(f"开始扫描磁盘：{disk_path}") self.log(f"JPG特征：开头 {JPG_START.hex()} | 结尾 {JPG_END.hex()}") self.log(f"保存目录：{SAVE_DIR}") jpg_data = None start_pos = 0 file_count = 0 try: with open(disk_path, "rb") as disk: while not self.stop_flag: # 暂停等待 while self.is_paused and not self.stop_flag: time.sleep(0.1) # 读取数据 data = disk.read(BUFFER_SIZE) if not data: break # 搜索JPG开始 if jpg_data is None: idx = data.find(JPG_START) if idx != -1: jpg_data = data[idx:] start_pos = disk.tell() - len(data) + idx self.log(f"找到JPG开头 | 偏移：0x{start_pos:X}") # 搜索JPG结束 else: jpg_data += data end_idx = jpg_data.find(JPG_END) if end_idx != -1: # 截取完整JPG full_jpg = jpg_data[:end_idx + 2] if len(full_jpg) > 1024: # 过滤过小文件 fname = self._save_jpg(full_jpg) file_count += 1 self.log(f"保存成功：{fname} | 大小：{len(full_jpg)} 字节") # 重置 jpg_data = None if self.stop_flag: self.log("用户手动停止扫描") else: self.log(f"扫描完成！共恢复 {file_count} 个JPG文件") except Exception as e: self.log(f"扫描异常：{str(e)}") finally: self._scan_finished() def start_scan(self): """开始扫描""" disk = self.disk_var.get() if not disk: messagebox.showwarning("提示", "请选择物理磁盘") return self.is_running = True self.is_paused = False self.stop_flag = False # 按钮状态 self.start_btn.config(state=tk.DISABLED) self.pause_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.NORMAL) self.update_status("扫描中", "blue") # 启动子线程 self.scan_thread = threading.Thread(target=self._scan_disk, args=(disk,), daemon=True) self.scan_thread.start() def pause_scan(self): """暂停/继续扫描""" if not self.is_running: return if not self.is_paused: self.is_paused = True self.pause_btn.config(text="继续") self.update_status("已暂停", "orange") self.log("扫描已暂停") else: self.is_paused = False self.pause_btn.config(text="暂停") self.update_status("扫描中", "blue") self.log("扫描继续") def stop_scan(self): """停止扫描""" if not self.is_running: return self.stop_flag = True self.is_paused = False self.update_status("停止中", "red") def _scan_finished(self): """扫描结束重置状态""" self.is_running = False self.is_paused = False self.start_btn.config(state=tk.NORMAL) self.pause_btn.config(state=tk.DISABLED, text="暂停") self.stop_btn.config(state=tk.DISABLED) self.update_status("就绪", "green") if __name__ == "__main__": root = tk.Tk() app = DiskJPGRecovery(root) root.mainloop()
