disk_cleaner_gui.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Windows 磁盘清理工具 - 图形界面（Tkinter）。
复用 disk_cleaner.py 的核心逻辑。
"""

from __future__ import annotations

import ctypes
import datetime as _dt
import os
import queue
import re
import shutil
import subprocess
import string
import sys
import threading
import tkinter as tk
from dataclasses import dataclass
from pathlib import Path
from tkinter import messagebox, ttk

import disk_cleaner as core


APP_TITLE = "磁盘清理工具（Windows）"


def list_fixed_drives() -> list[str]:
    """返回 ['C:', 'D:', ...]，仅返回当前存在的盘符。"""
    drives: list[str] = []
    try:
        bitmask = ctypes.windll.kernel32.GetLogicalDrives()
    except Exception:
        bitmask = 0
    if bitmask:
        for i, letter in enumerate(string.ascii_uppercase):
            if bitmask & (1 << i):
                drives.append(f"{letter}:")
    if not drives:
        drives = [os.environ.get("SystemDrive", "C:")]
    return drives


def disk_free(drive: str) -> tuple[int, int]:
    """返回 (free_bytes, total_bytes)；失败为 (0, 0)。"""
    try:
        usage = __import__("shutil").disk_usage(drive + "\\")
        return usage.free, usage.total
    except Exception:
        return 0, 0


def relaunch_as_admin() -> bool:
    """尝试以管理员重启当前脚本。"""
    try:
        params = " ".join(f'"{a}"' for a in sys.argv)
        rc = ctypes.windll.shell32.ShellExecuteW(
            None, "runas", sys.executable, params, None, 1
        )
        return int(rc) > 32
    except Exception:
        return False


# 递归列出文件时的上限，避免整盘枚举卡死 UI / 内存
FILE_LIST_MAX_RECURSIVE = 300_000

# 勾选删除时：仅对「视频」文件删除其父文件夹（文件夹行则删除该文件夹本身）
VIDEO_EXTENSIONS = frozenset(
    {
        ".mp4", ".mkv", ".avi", ".mov", ".wmv", ".webm", ".flv", ".m4v",
        ".mpeg", ".mpg", ".3gp", ".ts", ".m2ts", ".rmvb", ".rm", ".asf",
    }
)


def _norm_path_key(path: Path) -> str:
    try:
        return str(path.resolve()).lower()
    except OSError:
        return str(path).lower()


def folder_name_looks_like_id_prefix(name: str) -> bool:
    """
    「以 ID 开头」的判定（满足任一即可）：
    - 不区分大小写以字母 id 开头（如 ID123、id_cache）
    - 以至少 4 位连续数字开头（如 12345678_xxx）
    - 以标准 UUID 前缀开头（xxxxxxxx-xxxx-）
    """
    n = (name or "").strip()
    if not n:
        return False
    if n.lower().startswith("id"):
        return True
    if re.match(r"^\d{4,}", n):
        return True
    if re.match(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-", n):
        return True
    return False


def _is_drive_root(path: Path) -> bool:
    try:
        p = path.resolve()
    except OSError:
        p = path
    try:
        return p.parent == p
    except OSError:
        return False


def _is_protected_system_folder(path: Path) -> bool:
    """避免误删系统关键根目录。"""
    if _is_drive_root(path):
        return True
    try:
        rp = path.resolve()
    except OSError:
        rp = path
    s = str(rp).rstrip("\\/").lower()
    windir = os.environ.get("WINDIR", r"C:\Windows").lower()
    programfiles = os.environ.get("ProgramFiles", r"C:\Program Files").lower()
    programfiles_x86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)").lower()
    if s == windir or s.startswith(windir + "\\"):
        return True
    if s == programfiles or s.startswith(programfiles + "\\"):
        return True
    if s == programfiles_x86 or s.startswith(programfiles_x86 + "\\"):
        return True
    if s.endswith(":\\windows\\system32") or "\\windows\\system32\\" in s + "\\":
        return True
    return False


@dataclass
class _FileRow:
    """磁盘文件清单中的一行（用于排序与显示）。"""

    name: str
    kind: str  # "文件夹" | "文件"
    size: int | None  # 文件夹在平铺模式下为 None
    mtime: float | None
    path: Path

    def sort_name(self) -> str:
        return self.name.lower()

    def sort_size(self) -> int:
        if self.size is None:
            return -1
        return self.size

    def sort_mtime(self) -> float:
        return self.mtime if self.mtime is not None else 0.0

    def sort_path(self) -> str:
        return str(self.path).lower()


def _collect_flat(root: Path) -> list[_FileRow]:
    rows: list[_FileRow] = []
    try:
        with os.scandir(root) as it:
            for ent in it:
                try:
                    if ent.is_dir(follow_symlinks=False):
                        st = ent.stat(follow_symlinks=False)
                        rows.append(
                            _FileRow(ent.name, "文件夹", None, st.st_mtime, Path(ent.path))
                        )
                    else:
                        st = ent.stat(follow_symlinks=False)
                        rows.append(
                            _FileRow(ent.name, "文件", st.st_size, st.st_mtime, Path(ent.path))
                        )
                except OSError:
                    continue
    except OSError:
        pass
    return rows


def _collect_recursive_files(root: Path, limit: int) -> tuple[list[_FileRow], bool]:
    rows: list[_FileRow] = []
    truncated = False
    n = 0
    for dirpath, _dirnames, filenames in os.walk(root, topdown=True, onerror=None):
        for fn in filenames:
            if n >= limit:
                truncated = True
                return rows, truncated
            p = Path(dirpath) / fn
            try:
                st = p.stat()
                rows.append(_FileRow(fn, "文件", st.st_size, st.st_mtime, p.resolve()))
                n += 1
            except OSError:
                continue
    return rows, truncated


class DiskCleanerApp:
    COLS = ("sel", "title", "size", "path")

    def __init__(self, root: tk.Tk) -> None:
        self.root = root
        self.root.title(APP_TITLE)
        self.root.geometry("980x640")
        self.root.minsize(820, 540)

        self.targets: list[core.CleanTarget] = []
        self.checked: dict[str, bool] = {}
        self.worker: threading.Thread | None = None
        self.msg_q: queue.Queue = queue.Queue()

        self._file_rows: list[_FileRow] = []
        self._file_sort_column = "name"
        self._file_sort_reverse = False
        self._browse_gen = 0
        self._file_checked_paths: set[str] = set()

        self._build_ui()
        self._refresh_admin_state()
        self._refresh_disk_label()
        self.root.after(100, self._drain_queue)
        self._start_scan()

    # ---------------- UI ----------------
    def _build_ui(self) -> None:
        try:
            ttk.Style().theme_use("vista")
        except tk.TclError:
            pass

        top = ttk.Frame(self.root, padding=(10, 8))
        top.pack(side=tk.TOP, fill=tk.X)

        ttk.Label(top, text="目标盘:").pack(side=tk.LEFT)
        self.drive_var = tk.StringVar(value=os.environ.get("SystemDrive", "C:"))
        self.drive_combo = ttk.Combobox(
            top,
            textvariable=self.drive_var,
            values=list_fixed_drives(),
            width=6,
            state="readonly",
        )
        self.drive_combo.pack(side=tk.LEFT, padx=(4, 12))
        self.drive_combo.bind("<<ComboboxSelected>>", self._on_drive_changed)

        self.scan_btn = ttk.Button(top, text="扫描", command=self._start_scan)
        self.scan_btn.pack(side=tk.LEFT)
        self.clean_btn = ttk.Button(top, text="清理选中", command=self._confirm_and_clean)
        self.clean_btn.pack(side=tk.LEFT, padx=(8, 0))

        self.dism_var = tk.BooleanVar(value=False)
        ttk.Checkbutton(
            top,
            text="清理后执行 DISM 组件清理（慢、需管理员）",
            variable=self.dism_var,
        ).pack(side=tk.LEFT, padx=(16, 0))

        self.admin_lbl = ttk.Label(top, text="")
        self.admin_lbl.pack(side=tk.RIGHT)
        self.elevate_btn = ttk.Button(
            top, text="以管理员身份重启", command=self._elevate
        )
        self.elevate_btn.pack(side=tk.RIGHT, padx=(0, 8))

        info = ttk.Frame(self.root, padding=(10, 0))
        info.pack(side=tk.TOP, fill=tk.X)
        self.disk_lbl = ttk.Label(info, text="")
        self.disk_lbl.pack(side=tk.LEFT)
        self.total_lbl = ttk.Label(info, text="预计可释放: -", font=("Segoe UI", 10, "bold"))
        self.total_lbl.pack(side=tk.RIGHT)

        nb = ttk.Notebook(self.root, padding=(10, 4))
        nb.pack(side=tk.TOP, fill=tk.BOTH, expand=True)

        tab_clean = ttk.Frame(nb)
        tab_files = ttk.Frame(nb)
        nb.add(tab_clean, text="可清理项")
        nb.add(tab_files, text="磁盘文件清单")

        clean_mid = ttk.Frame(tab_clean)
        clean_mid.pack(side=tk.TOP, fill=tk.BOTH, expand=True)

        columns = ("title", "size", "path")
        self.tree = ttk.Treeview(
            clean_mid,
            columns=columns,
            show="tree headings",
            selectmode="none",
            height=10,
        )
        self.tree.heading("#0", text="选择")
        self.tree.heading("title", text="项目")
        self.tree.heading("size", text="大小")
        self.tree.heading("path", text="路径 / 说明")
        self.tree.column("#0", width=70, anchor=tk.CENTER, stretch=False)
        self.tree.column("title", width=240, anchor=tk.W)
        self.tree.column("size", width=110, anchor=tk.E, stretch=False)
        self.tree.column("path", width=520, anchor=tk.W)

        vsb = ttk.Scrollbar(clean_mid, orient="vertical", command=self.tree.yview)
        self.tree.configure(yscrollcommand=vsb.set)
        self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        vsb.pack(side=tk.RIGHT, fill=tk.Y)
        self.tree.bind("<Button-1>", self._on_tree_click)

        sel_bar = ttk.Frame(tab_clean, padding=(0, 6, 0, 0))
        sel_bar.pack(side=tk.BOTTOM, fill=tk.X)
        ttk.Button(sel_bar, text="全选", command=lambda: self._set_all(True)).pack(side=tk.LEFT)
        ttk.Button(sel_bar, text="全不选", command=lambda: self._set_all(False)).pack(side=tk.LEFT, padx=(6, 0))
        ttk.Button(sel_bar, text="仅选安全项", command=self._select_safe).pack(side=tk.LEFT, padx=(6, 0))

        self._build_file_tab(tab_files)

        bot = ttk.Frame(self.root, padding=(10, 8))
        bot.pack(side=tk.BOTTOM, fill=tk.BOTH)

        self.progress = ttk.Progressbar(bot, mode="indeterminate")
        self.progress.pack(fill=tk.X, pady=(0, 6))

        log_frame = ttk.LabelFrame(bot, text="日志")
        log_frame.pack(fill=tk.BOTH, expand=False)
        self.log = tk.Text(log_frame, height=9, wrap="word", state=tk.DISABLED)
        log_scroll = ttk.Scrollbar(log_frame, orient="vertical", command=self.log.yview)
        self.log.configure(yscrollcommand=log_scroll.set)
        self.log.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        log_scroll.pack(side=tk.RIGHT, fill=tk.Y)

        self.status = ttk.Label(self.root, text="就绪", anchor=tk.W, padding=(10, 4))
        self.status.pack(side=tk.BOTTOM, fill=tk.X)

        self._browse_refresh()

    def _build_file_tab(self, parent: ttk.Frame) -> None:
        d0 = self.drive_var.get().rstrip("\\/")
        self.browse_path_var = tk.StringVar(value=d0 + "\\")
        self.browse_mode_var = tk.StringVar(value="flat")

        bar = ttk.Frame(parent, padding=(0, 0, 0, 6))
        bar.pack(side=tk.TOP, fill=tk.X)
        ttk.Label(bar, text="路径:").pack(side=tk.LEFT)
        self.browse_entry = ttk.Entry(bar, textvariable=self.browse_path_var, width=72)
        self.browse_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(4, 8))
        self.browse_entry.bind("<Return>", lambda _e: self._browse_go())
        self.file_go_btn = ttk.Button(bar, text="转到", command=self._browse_go)
        self.file_go_btn.pack(side=tk.LEFT)
        self.file_up_btn = ttk.Button(bar, text="上级", command=self._browse_up)
        self.file_up_btn.pack(side=tk.LEFT, padx=(4, 0))
        self.file_refresh_btn = ttk.Button(bar, text="刷新", command=self._browse_refresh)
        self.file_refresh_btn.pack(side=tk.LEFT, padx=(4, 0))

        ttk.Label(bar, text="范围:").pack(side=tk.LEFT, padx=(12, 4))
        ttk.Radiobutton(
            bar, text="当前目录", variable=self.browse_mode_var, value="flat", command=self._browse_refresh
        ).pack(side=tk.LEFT)
        ttk.Radiobutton(
            bar,
            text="递归所有文件（可能较慢）",
            variable=self.browse_mode_var,
            value="recursive",
            command=self._browse_refresh,
        ).pack(side=tk.LEFT, padx=(6, 0))

        bar2 = ttk.Frame(parent, padding=(0, 6, 0, 0))
        bar2.pack(side=tk.TOP, fill=tk.X)
        ttk.Button(bar2, text="列表全勾选", command=self._file_check_all).pack(side=tk.LEFT)
        ttk.Button(bar2, text="列表全不选", command=self._file_uncheck_all).pack(side=tk.LEFT, padx=(6, 0))
        ttk.Button(
            bar2,
            text="删除勾选所在文件夹",
            command=self._confirm_delete_checked_folders,
        ).pack(side=tk.LEFT, padx=(16, 0))
        ttk.Label(
            bar2,
            text="（仅删除名称以 ID/数字/UUID 前缀的文件夹；视频文件删其父目录）",
            foreground="#666",
        ).pack(side=tk.LEFT, padx=(12, 0))

        self.file_status_lbl = ttk.Label(parent, text="", foreground="#444")
        self.file_status_lbl.pack(side=tk.TOP, fill=tk.X, pady=(0, 4))

        ff = ttk.Frame(parent)
        ff.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
        fcols = ("name", "kind", "size", "mtime", "fullpath")
        self._file_col_id_fullpath = f"#{fcols.index('fullpath') + 1}"
        self.file_tree = ttk.Treeview(
            ff,
            columns=fcols,
            show="tree headings",
            selectmode="browse",
            height=16,
        )
        self.file_tree.heading("#0", text="勾选")
        self.file_tree.column("#0", width=52, anchor=tk.CENTER, stretch=False)
        self._file_heading_titles = {
            "name": "名称",
            "kind": "类型",
            "size": "大小",
            "mtime": "修改时间",
            "fullpath": "完整路径",
        }
        for cid, title in self._file_heading_titles.items():
            self.file_tree.heading(
                cid,
                text=title,
                command=lambda c=cid: self._on_file_sort_click(c),
            )
        self.file_tree.column("name", width=200, anchor=tk.W, stretch=True)
        self.file_tree.column("kind", width=72, anchor=tk.CENTER, stretch=False)
        self.file_tree.column("size", width=100, anchor=tk.E, stretch=False)
        self.file_tree.column("mtime", width=150, anchor=tk.CENTER, stretch=False)
        self.file_tree.column("fullpath", width=340, anchor=tk.W, stretch=True)

        f_vsb = ttk.Scrollbar(ff, orient="vertical", command=self.file_tree.yview)
        f_hsb = ttk.Scrollbar(ff, orient="horizontal", command=self.file_tree.xview)
        self.file_tree.configure(yscrollcommand=f_vsb.set, xscrollcommand=f_hsb.set)
        self.file_tree.grid(row=0, column=0, sticky="nsew")
        f_vsb.grid(row=0, column=1, sticky="ns")
        f_hsb.grid(row=1, column=0, sticky="ew")
        ff.grid_rowconfigure(0, weight=1)
        ff.grid_columnconfigure(0, weight=1)

        self.file_tree.bind("<Button-1>", self._on_file_tree_click)
        self.file_tree.bind("<Double-1>", self._on_file_tree_double_click)
        self._update_file_headings()

    # ---------------- 磁盘文件清单 ----------------
    def _on_drive_changed(self, _e: tk.Event | None = None) -> None:
        d = self.drive_var.get().rstrip("\\/")
        self.browse_path_var.set(d + "\\")
        self._start_scan()
        self._browse_refresh()

    def _browse_resolved_root(self) -> Path | None:
        raw = self.browse_path_var.get().strip().strip('"')
        if not raw:
            return None
        try:
            p = Path(raw).expanduser()
            if not p.is_absolute():
                p = Path(self.drive_var.get().rstrip("\\/")) / p
            return p.resolve()
        except OSError:
            return None

    def _browse_go(self) -> None:
        self._browse_refresh()

    def _browse_up(self) -> None:
        p = self._browse_resolved_root()
        if p is None:
            return
        try:
            rp = p.resolve()
            par = p.parent.resolve()
            if par == rp:
                return
        except OSError:
            return
        try:
            if not par.exists():
                return
        except OSError:
            return
        self.browse_path_var.set(str(par) + ("\\" if par.is_dir() else ""))
        self._browse_refresh()

    def _browse_refresh(self) -> None:
        if not hasattr(self, "browse_path_var"):
            return
        self._browse_gen += 1
        gen = self._browse_gen
        root = self._browse_resolved_root()
        mode = self.browse_mode_var.get()

        if root is None:
            self._file_rows = []
            self._apply_file_sort_and_render()
            self.file_status_lbl.config(text="请输入有效路径。")
            return
        try:
            exists = root.exists()
        except OSError:
            exists = False
        if not exists:
            self._file_rows = []
            self._apply_file_sort_and_render()
            self.file_status_lbl.config(text="路径不存在。")
            return
        if root.is_file():
            try:
                st = root.stat()
                self._file_rows = [
                    _FileRow(root.name, "文件", st.st_size, st.st_mtime, root.resolve())
                ]
            except OSError:
                self._file_rows = []
            self._apply_file_sort_and_render()
            self.file_status_lbl.config(text="当前为单个文件。")
            return

        self.file_status_lbl.config(
            text=("正在递归扫描…" if mode == "recursive" else "正在读取当前目录…")
        )
        self.file_refresh_btn.state(["disabled"])
        self.file_go_btn.state(["disabled"])
        self.file_up_btn.state(["disabled"])

        def work() -> None:
            try:
                if mode == "flat":
                    rows = _collect_flat(root)
                    self.msg_q.put(("file_list", gen, rows, None))
                else:
                    rows, truncated = _collect_recursive_files(root, FILE_LIST_MAX_RECURSIVE)
                    extra = None
                    if truncated:
                        extra = f"已达上限 {FILE_LIST_MAX_RECURSIVE} 个文件，列表已截断。"
                    self.msg_q.put(("file_list", gen, rows, extra))
            except Exception as e:
                self.msg_q.put(("file_error", gen, str(e)))

        threading.Thread(target=work, daemon=True).start()

    def _on_file_sort_click(self, col: str) -> None:
        if self._file_sort_column == col:
            self._file_sort_reverse = not self._file_sort_reverse
        else:
            self._file_sort_column = col
            self._file_sort_reverse = col in ("size", "mtime")
        self._apply_file_sort_and_render()

    def _update_file_headings(self) -> None:
        up, down = " ↑", " ↓"
        for cid, base in self._file_heading_titles.items():
            suffix = ""
            if self._file_sort_column == cid:
                suffix = down if self._file_sort_reverse else up
            self.file_tree.heading(cid, text=base + suffix)

    def _apply_file_sort_and_render(self) -> None:
        col = self._file_sort_column
        rev = self._file_sort_reverse

        def key_fn(r: _FileRow):
            if col == "name":
                return r.sort_name()
            if col == "kind":
                return r.kind
            if col == "size":
                return r.sort_size()
            if col == "mtime":
                return r.sort_mtime()
            return r.sort_path()

        self._file_rows.sort(key=key_fn, reverse=rev)
        self._update_file_headings()
        self._render_file_tree()

    def _render_file_tree(self) -> None:
        for iid in self.file_tree.get_children():
            self.file_tree.delete(iid)
        for i, row in enumerate(self._file_rows):
            sz_txt = "—" if row.size is None else core.human_bytes(row.size)
            mt_txt = "—"
            if row.mtime is not None:
                mt_txt = _dt.datetime.fromtimestamp(row.mtime).strftime("%Y-%m-%d %H:%M:%S")
            k = _norm_path_key(row.path)
            sym = "[x]" if k in self._file_checked_paths else "[ ]"
            self.file_tree.insert(
                "",
                tk.END,
                iid=str(i),
                text=sym,
                values=(row.name, row.kind, sz_txt, mt_txt, str(row.path)),
            )

    def _on_file_tree_click(self, event: tk.Event) -> str | None:
        if self.file_tree.identify_region(event.x, event.y) != "tree":
            return None
        row_id = self.file_tree.identify_row(event.y)
        if not row_id:
            return "break"
        try:
            idx = int(row_id)
        except ValueError:
            return "break"
        if idx < 0 or idx >= len(self._file_rows):
            return "break"
        row = self._file_rows[idx]
        k = _norm_path_key(row.path)
        if k in self._file_checked_paths:
            self._file_checked_paths.discard(k)
        else:
            self._file_checked_paths.add(k)
        sym = "[x]" if k in self._file_checked_paths else "[ ]"
        self.file_tree.item(row_id, text=sym)
        self.file_tree.selection_set(row_id)
        return "break"

    def _file_check_all(self) -> None:
        for row in self._file_rows:
            self._file_checked_paths.add(_norm_path_key(row.path))
        self._render_file_tree()

    def _file_uncheck_all(self) -> None:
        self._file_checked_paths.clear()
        self._render_file_tree()

    def _folder_target_for_delete_row(self, row: _FileRow) -> Path:
        """视频/文件：删除其上一级目录；勾选文件夹：删除该文件夹本身。"""
        if row.kind == "文件夹":
            return row.path.resolve()
        return row.path.resolve().parent

    def _confirm_delete_checked_folders(self) -> None:
        if not self._file_checked_paths:
            messagebox.showinfo(APP_TITLE, "请先勾选条目。")
            return
        if self.worker and self.worker.is_alive():
            messagebox.showinfo(APP_TITLE, "正在执行清理扫描/删除，请稍后再试。")
            return

        rows = [
            r
            for r in self._file_rows
            if _norm_path_key(r.path) in self._file_checked_paths
        ]
        skipped_non_video: list[str] = []
        id_mismatch: list[str] = []
        protected: list[str] = []
        folders_ok: dict[str, Path] = {}

        for row in rows:
            if row.kind == "文件":
                if row.path.suffix.lower() not in VIDEO_EXTENSIONS:
                    skipped_non_video.append(str(row.path))
                    continue
            folder = self._folder_target_for_delete_row(row)
            try:
                folder = folder.resolve()
            except OSError:
                pass
            fk = str(folder).lower()
            if _is_drive_root(folder) or _is_protected_system_folder(folder):
                protected.append(str(folder))
                continue
            if not folder_name_looks_like_id_prefix(folder.name):
                id_mismatch.append(f"{folder}\n  （文件夹名「{folder.name}」不符合以 ID 开头的规则）")
                continue
            folders_ok[fk] = folder

        if skipped_non_video:
            messagebox.showwarning(
                APP_TITLE,
                "以下条目不是视频文件，已跳过（仅视频会删除其父文件夹）：\n"
                + "\n".join(skipped_non_video[:15])
                + (f"\n…共 {len(skipped_non_video)} 项" if len(skipped_non_video) > 15 else ""),
            )
        if id_mismatch:
            messagebox.showerror(
                APP_TITLE,
                "以下目标文件夹名称不以规则要求的 ID 开头，已取消对应删除：\n"
                + "\n".join(id_mismatch[:12])
                + (f"\n…共 {len(id_mismatch)} 项" if len(id_mismatch) > 12 else ""),
            )
        if protected:
            messagebox.showerror(
                APP_TITLE,
                "以下路径受保护，禁止删除：\n" + "\n".join(protected[:12]),
            )

        to_delete = list(folders_ok.values())
        if not to_delete:
            messagebox.showinfo(APP_TITLE, "没有可删除的文件夹（请检查规则与勾选）。")
            return

        preview = "\n".join(str(p) for p in to_delete[:25])
        if len(to_delete) > 25:
            preview += f"\n…等共 {len(to_delete)} 个文件夹"
        if not messagebox.askyesno(
            APP_TITLE,
            "将永久删除以下整个文件夹（不可恢复）：\n\n"
            + preview
            + "\n\n确定继续吗？",
            icon="warning",
        ):
            return

        self.file_status_lbl.config(text="正在删除文件夹…")

        def work(paths: list[Path]) -> None:
            ok = 0
            errs: list[str] = []
            for fp in paths:
                try:
                    shutil.rmtree(fp)
                    ok += 1
                except OSError as e:
                    errs.append(f"{fp}\n  {e}")
            self.msg_q.put(("folder_deleted", ok, errs))

        threading.Thread(target=work, args=(to_delete,), daemon=True).start()

    def _explorer_exe(self) -> str:
        return os.path.join(os.environ.get("WINDIR", r"C:\Windows"), "explorer.exe")

    def _open_row_in_explorer(self, row: _FileRow) -> None:
        """在资源管理器中打开：文件夹则打开该目录；文件则 /select 选中该文件。

        注意：对 explorer 禁止使用 CREATE_NO_WINDOW，否则在部分系统上会错误打开
        「快速访问 / OneDrive」等默认位置而非目标路径。
        """
        try:
            rp = row.path.resolve()
        except OSError:
            rp = row.path
        exe = self._explorer_exe()
        try:
            if rp.is_dir():
                subprocess.Popen([exe, os.path.normpath(os.fspath(rp))])
                return
            if rp.is_file():
                norm = os.path.normpath(os.fspath(rp))
                # 路径与 /select, 分开传参，避免解析错误
                subprocess.Popen([exe, "/select,", norm])
                return
            par = rp.parent
            if par.is_dir():
                subprocess.Popen([exe, os.path.normpath(os.fspath(par))])
            else:
                messagebox.showwarning(APP_TITLE, f"路径不存在，无法打开：\n{rp}")
        except OSError as e:
            messagebox.showerror(APP_TITLE, f"无法打开资源管理器：\n{e}")

    def _double_click_hit_fullpath_column(self, event: tk.Event, row_id: str) -> bool:
        """用单元格 bbox 判断点击是否在「完整路径」列（比 identify_column 更稳）。"""
        try:
            bx = self.file_tree.bbox(row_id, "fullpath")
        except tk.TclError:
            bx = None
        if bx:
            x, y, w, h = bx
            if x <= event.x < x + w and y <= event.y < y + h:
                return True
        col = self.file_tree.identify_column(event.x)
        return col == self._file_col_id_fullpath

    def _on_file_tree_double_click(self, event: tk.Event) -> None:
        row_id = self.file_tree.identify_row(event.y)
        if not row_id:
            return
        try:
            idx = int(row_id)
        except ValueError:
            return
        if idx < 0 or idx >= len(self._file_rows):
            return
        row = self._file_rows[idx]

        if self._double_click_hit_fullpath_column(event, row_id):
            self._open_row_in_explorer(row)
            return

        if row.kind == "文件夹":
            self.browse_path_var.set(str(row.path.resolve()) + "\\")
            self.browse_mode_var.set("flat")
            self._browse_refresh()
            return
        try:
            os.startfile(str(row.path))
        except OSError as e:
            messagebox.showerror(APP_TITLE, f"无法用默认程序打开：\n{e}")

    # ---------------- 状态区 ----------------
    def _refresh_admin_state(self) -> None:
        admin = core.is_admin()
        self.admin_lbl.config(
            text=("管理员: 是" if admin else "管理员: 否（部分目录可能无法清理）"),
            foreground=("#2a7" if admin else "#a60"),
        )
        if admin:
            self.elevate_btn.state(["disabled"])
        else:
            self.elevate_btn.state(["!disabled"])

    def _refresh_disk_label(self) -> None:
        d = self.drive_var.get()
        free, total = disk_free(d)
        if total:
            self.disk_lbl.config(
                text=f"{d} 可用 {core.human_bytes(free)} / 共 {core.human_bytes(total)}"
            )
        else:
            self.disk_lbl.config(text=f"{d} 容量信息不可用")

    # ---------------- 扫描 ----------------
    def _set_busy(self, busy: bool, status: str = "") -> None:
        if busy:
            self.progress.start(80)
            self.scan_btn.state(["disabled"])
            self.clean_btn.state(["disabled"])
            self.drive_combo.state(["disabled"])
        else:
            self.progress.stop()
            self.scan_btn.state(["!disabled"])
            self.clean_btn.state(["!disabled"])
            self.drive_combo.state(["!disabled"])
        if status:
            self.status.config(text=status)

    def _start_scan(self) -> None:
        if self.worker and self.worker.is_alive():
            return
        drive = self.drive_var.get()
        try:
            self.targets = core.build_targets(drive)
        except ValueError as e:
            messagebox.showerror(APP_TITLE, str(e))
            return
        # 默认勾选可清理项；info_only 与高风险项（休眠/Prefetch）默认不勾
        default_unchecked = {"hibernation", "prefetch"}
        self.checked = {
            t.key: (not t.info_only) and (t.key not in default_unchecked)
            for t in self.targets
        }
        self._populate_tree(initial=True)
        self._refresh_disk_label()
        self._set_busy(True, f"正在扫描 {drive} …")
        self._log(f"开始扫描 {drive}")

        def work(targets: list[core.CleanTarget]) -> None:
            for t in targets:
                t._size_cache = None
                size, errs = t.size()
                self.msg_q.put(("size", t.key, size, errs))
            self.msg_q.put(("scan_done",))

        self.worker = threading.Thread(target=work, args=(self.targets,), daemon=True)
        self.worker.start()

    def _row_symbol(self, t: core.CleanTarget) -> str:
        if t.info_only:
            return "[i]"
        return "[x]" if self.checked.get(t.key) else "[ ]"

    def _row_path_text(self, t: core.CleanTarget) -> str:
        paths = t.display_paths()
        primary = str(paths[0]) if paths else str(t.resolve())
        suffix = f"  (+{len(paths) - 1} 项)" if len(paths) > 1 else ""
        return f"{primary}{suffix}  —  {t.note}"

    def _populate_tree(self, initial: bool) -> None:
        for iid in self.tree.get_children():
            self.tree.delete(iid)
        for t in self.targets:
            sym = self._row_symbol(t)
            sz = "扫描中…" if initial else core.human_bytes(t.size()[0])
            tail = " [建议管理员]" if t.needs_admin_hint and not core.is_admin() else ""
            self.tree.insert(
                "",
                tk.END,
                iid=t.key,
                text=sym,
                values=(t.title + tail, sz, self._row_path_text(t)),
            )

    def _update_row(self, key: str) -> None:
        t = next((x for x in self.targets if x.key == key), None)
        if not t:
            return
        size, _ = t.size()
        tail = " [建议管理员]" if t.needs_admin_hint and not core.is_admin() else ""
        self.tree.item(
            key,
            text=self._row_symbol(t),
            values=(t.title + tail, core.human_bytes(size), self._row_path_text(t)),
        )

    def _update_total(self) -> None:
        total = sum(
            t.size()[0]
            for t in self.targets
            if self.checked.get(t.key) and not t.info_only
        )
        self.total_lbl.config(text=f"预计可释放: {core.human_bytes(total)}")

    # ---------------- 选择 ----------------
    def _on_tree_click(self, event: tk.Event) -> None:
        region = self.tree.identify("region", event.x, event.y)
        row = self.tree.identify_row(event.y)
        if not row:
            return
        col = self.tree.identify_column(event.x)
        if region == "tree" or col == "#0":
            t = next((x for x in self.targets if x.key == row), None)
            if t is None or t.info_only:
                return
            self.checked[row] = not self.checked.get(row, False)
            self._update_row(row)
            self._update_total()

    def _set_all(self, value: bool) -> None:
        for t in self.targets:
            if t.info_only:
                continue
            self.checked[t.key] = value
            self._update_row(t.key)
        self._update_total()

    def _select_safe(self) -> None:
        safe_keys = {
            "user_temp", "user_tmp",
            "inet_cache", "thumbcache",
            "chrome_cache", "edge_cache", "firefox_cache",
            "crash_dumps",
        }
        for t in self.targets:
            if t.info_only:
                continue
            self.checked[t.key] = t.key in safe_keys
            self._update_row(t.key)
        self._update_total()

    # ---------------- 清理 ----------------
    def _confirm_and_clean(self) -> None:
        if self.worker and self.worker.is_alive():
            return
        chosen = [t for t in self.targets if self.checked.get(t.key)]
        if not chosen:
            messagebox.showinfo(APP_TITLE, "请至少选择一项要清理的内容。")
            return
        total = sum(t.size()[0] for t in chosen)
        admin_needed = any(
            t.needs_admin_hint for t in chosen
        ) and not core.is_admin()
        msg = (
            f"将清理 {len(chosen)} 项，预计释放约 {core.human_bytes(total)}。\n"
            f"目标盘: {self.drive_var.get()}\n\n"
        )
        if admin_needed:
            msg += "注意：所选包含需要管理员权限的项，未提权可能失败。\n\n"
        if self.dism_var.get():
            msg += "并将在清理结束后执行 DISM 组件清理（耗时较长，且不可撤销）。\n\n"
        msg += "确定继续吗？"
        if not messagebox.askyesno(APP_TITLE, msg, icon="warning"):
            return

        self._set_busy(True, "正在清理…")
        self._log(f"开始清理 {len(chosen)} 项 …")
        run_dism = self.dism_var.get()

        def work(items: list[core.CleanTarget]) -> None:
            freed_total = 0
            failures_total = 0
            for t in items:
                self.msg_q.put(("status", f"清理 {t.title} …"))
                try:
                    b, f = t.clean(dry_run=False)
                except Exception as e:  # 防御
                    self.msg_q.put(("log", f"  {t.key}: 异常 {e}"))
                    b, f = 0, 1
                t._size_cache = None
                freed_total += b
                failures_total += f
                self.msg_q.put(("log", f"  {t.key}: 处理约 {core.human_bytes(b)}，失败项 {f}"))
                self.msg_q.put(("rescan", t.key))
            if run_dism:
                self.msg_q.put(("status", "执行 DISM 组件清理（可能需要几分钟）…"))
                self.msg_q.put(("log", "执行: DISM /Online /Cleanup-Image /StartComponentCleanup /ResetBase"))
                if core.is_admin():
                    try:
                        import subprocess
                        r = subprocess.run(
                            [
                                "DISM.exe",
                                "/Online",
                                "/Cleanup-Image",
                                "/StartComponentCleanup",
                                "/ResetBase",
                            ],
                            capture_output=True,
                            text=True,
                        )
                        self.msg_q.put(("log", f"DISM 退出码 {r.returncode}"))
                        if r.stdout:
                            self.msg_q.put(("log", r.stdout.strip()))
                        if r.stderr:
                            self.msg_q.put(("log", r.stderr.strip()))
                    except Exception as e:
                        self.msg_q.put(("log", f"DISM 失败: {e}"))
                else:
                    self.msg_q.put(("log", "未以管理员运行，跳过 DISM。"))
            self.msg_q.put(("clean_done", freed_total, failures_total))

        self.worker = threading.Thread(target=work, args=(chosen,), daemon=True)
        self.worker.start()

    # ---------------- 日志/消息泵 ----------------
    def _log(self, text: str) -> None:
        self.log.configure(state=tk.NORMAL)
        self.log.insert(tk.END, text + "\n")
        self.log.see(tk.END)
        self.log.configure(state=tk.DISABLED)

    def _drain_queue(self) -> None:
        try:
            while True:
                msg = self.msg_q.get_nowait()
                tag = msg[0]
                if tag == "size":
                    _, key, size, errs = msg
                    t = next((x for x in self.targets if x.key == key), None)
                    if t is not None:
                        t._size_cache = (size, errs)
                        self._update_row(key)
                        self._update_total()
                elif tag == "scan_done":
                    self._set_busy(False, "扫描完成")
                    self._update_total()
                    self._log(
                        "扫描完成，预计可释放: "
                        + core.human_bytes(
                            sum(t.size()[0] for t in self.targets if self.checked.get(t.key))
                        )
                    )
                elif tag == "rescan":
                    self._update_row(msg[1])
                elif tag == "status":
                    self.status.config(text=msg[1])
                elif tag == "log":
                    self._log(msg[1])
                elif tag == "clean_done":
                    _, freed, fails = msg
                    self._set_busy(False, f"清理完成，释放约 {core.human_bytes(freed)}，失败 {fails}")
                    self._refresh_disk_label()
                    self._update_total()
                    messagebox.showinfo(
                        APP_TITLE,
                        f"清理完成。\n粗略释放/处理: {core.human_bytes(freed)}\n失败项: {fails}",
                    )
                elif tag == "file_list":
                    _, gen, rows, extra = msg
                    if gen != self._browse_gen:
                        continue
                    self._file_checked_paths.clear()
                    self._file_rows = rows
                    self._apply_file_sort_and_render()
                    if hasattr(self, "file_refresh_btn"):
                        self.file_refresh_btn.state(["!disabled"])
                        self.file_go_btn.state(["!disabled"])
                        self.file_up_btn.state(["!disabled"])
                    n = len(rows)
                    txt = f"共 {n} 项。"
                    if extra:
                        txt += " " + extra
                    self.file_status_lbl.config(text=txt)
                elif tag == "file_error":
                    _, gen, err = msg
                    if gen != self._browse_gen:
                        continue
                    self._file_rows = []
                    self._apply_file_sort_and_render()
                    if hasattr(self, "file_refresh_btn"):
                        self.file_refresh_btn.state(["!disabled"])
                        self.file_go_btn.state(["!disabled"])
                        self.file_up_btn.state(["!disabled"])
                    self.file_status_lbl.config(text=f"读取失败: {err}")
                elif tag == "folder_deleted":
                    ok, errs = msg[1], msg[2]
                    self._file_checked_paths.clear()
                    txt = f"已删除 {ok} 个文件夹。"
                    if errs:
                        txt += "\n\n失败：\n" + "\n".join(errs[:15])
                        if len(errs) > 15:
                            txt += f"\n…共 {len(errs)} 条"
                        messagebox.showwarning(APP_TITLE, txt)
                    else:
                        messagebox.showinfo(APP_TITLE, txt)
                    if hasattr(self, "file_status_lbl"):
                        self.file_status_lbl.config(text="就绪")
                    self._browse_refresh()
        except queue.Empty:
            pass
        self.root.after(120, self._drain_queue)

    def _elevate(self) -> None:
        if core.is_admin():
            return
        if messagebox.askyesno(
            APP_TITLE,
            "将以管理员身份重新启动本程序，当前窗口会关闭。继续吗？",
        ):
            if relaunch_as_admin():
                self.root.destroy()
            else:
                messagebox.showerror(APP_TITLE, "提权失败。")


def main() -> int:
    if os.name != "nt":
        print("本程序仅适用于 Windows。", file=sys.stderr)
        return 1
    root = tk.Tk()
    DiskCleanerApp(root)
    root.mainloop()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())