gui.py

"""Tkinter GUI entry point for FolderVisualizer."""

from __future__ import annotations

import queue
import threading
import tkinter as tk
import webbrowser
from dataclasses import dataclass
from pathlib import Path
from tkinter import filedialog, ttk

from app_paths import APP_NAME, APP_VERSION, RESOURCE_DIR
from workflow import (
    LargeServerController,
    WorkflowError,
    WorkflowEvent,
    WorkflowOptions,
    WorkflowResult,
    create_large_server_controller,
    run_workflow,
)


QueueMessage = tuple[str, object]
WINDOW_ICON_PATH = RESOURCE_DIR / "assets" / "foldervisualizer.ico"


@dataclass(frozen=True)
class CompletedWorkflow:
    """A workflow result and its optional running large-mode service."""

    result: WorkflowResult
    controller: LargeServerController | None = None
    url: str | None = None


class FolderVisualizerApp(ttk.Frame):
    """Main FolderVisualizer window and workflow-thread adapter."""

    def __init__(self, master: tk.Tk) -> None:
        super().__init__(master, padding=20)
        self.master = master

        self.path_var = tk.StringVar()
        self.status_var = tk.StringVar(value="请选择一个文件夹开始。")
        self.summary_vars = {
            "mode": tk.StringVar(value="—"),
            "files": tk.StringVar(value="—"),
            "directories": tk.StringVar(value="—"),
            "errors": tk.StringVar(value="—"),
            "size": tk.StringVar(value="—"),
            "output": tk.StringVar(value="—"),
        }
        self.workflow_result: WorkflowResult | None = None
        self.small_result_path: Path | None = None
        self.large_result: WorkflowResult | None = None
        self.large_server_controller: LargeServerController | None = None
        self.large_url: str | None = None
        self._task_running = False
        self._task_thread: threading.Thread | None = None
        self._message_queue: queue.Queue[QueueMessage] = queue.Queue()
        self._close_requested = threading.Event()
        self._closing = False
        self._destroyed = False
        self._message_after_id: str | None = None
        self._layout_after_id: str | None = None
        self._wrap_widths: tuple[int, int] | None = None

        self._configure_window()
        self._build_widgets()
        self.bind("<Configure>", self._on_layout_configure, add="+")
        self._refresh_layout_constraints()
        self.master.protocol("WM_DELETE_WINDOW", self.close)
        self._message_after_id = self.master.after(
            100, self._process_messages
        )

    def _configure_window(self) -> None:
        self.master.title(f"{APP_NAME} v{APP_VERSION}")
        self._set_window_icon()
        self.master.geometry("760x520")
        self.master.columnconfigure(0, weight=1)
        self.master.rowconfigure(0, weight=1)

        style = ttk.Style(self.master)
        style.configure("Title.TLabel", font=("Segoe UI", 20, "bold"))
        style.configure("Subtitle.TLabel", foreground="#555555")
        style.configure("Status.TLabel", padding=(4, 8))
        style.configure("SummaryValue.TLabel", font=("Segoe UI", 10, "bold"))

        self.grid(row=0, column=0, sticky="nsew")
        self.columnconfigure(0, weight=1)
        self.rowconfigure(3, weight=1)

    def _set_window_icon(self) -> None:
        """Apply the packaged Windows icon without risking GUI startup."""

        try:
            if WINDOW_ICON_PATH.is_file():
                self.master.iconbitmap(default=str(WINDOW_ICON_PATH))
        except (OSError, tk.TclError):
            pass

    def _build_widgets(self) -> None:
        header = ttk.Frame(self)
        header.grid(row=0, column=0, sticky="ew", pady=(0, 18))
        header.columnconfigure(0, weight=1)

        ttk.Label(
            header,
            text=APP_NAME,
            style="Title.TLabel",
        ).grid(row=0, column=0, sticky="w")
        ttk.Label(
            header,
            text="选择文件夹，生成更直观的本地文件结构视图",
            style="Subtitle.TLabel",
        ).grid(row=1, column=0, sticky="w", pady=(4, 0))

        selection = ttk.LabelFrame(self, text="目标文件夹", padding=12)
        selection.grid(row=1, column=0, sticky="ew", pady=(0, 14))
        selection.columnconfigure(0, weight=1)

        self.path_entry = ttk.Entry(
            selection,
            textvariable=self.path_var,
        )
        self.path_entry.grid(row=0, column=0, sticky="ew", padx=(0, 8))
        self.path_entry.bind("<Return>", self._on_path_entered)

        self.choose_button = ttk.Button(
            selection,
            text="选择文件夹",
            command=self.choose_folder,
        )
        self.choose_button.grid(row=0, column=1, padx=(0, 8))

        self.start_button = ttk.Button(
            selection,
            text="开始分析",
            command=self.start_analysis,
        )
        self.start_button.grid(row=0, column=2)

        status_frame = ttk.LabelFrame(self, text="当前状态", padding=10)
        status_frame.grid(row=2, column=0, sticky="ew", pady=(0, 14))
        status_frame.columnconfigure(0, weight=1)
        self.status_label = ttk.Label(
            status_frame,
            textvariable=self.status_var,
            style="Status.TLabel",
            anchor="w",
            justify="left",
            wraplength=680,
        )
        self.status_label.grid(row=0, column=0, sticky="ew")

        self.summary_frame = ttk.LabelFrame(
            self, text="结果摘要", padding=14
        )
        self.summary_frame.grid(
            row=3, column=0, sticky="nsew", pady=(0, 14)
        )
        self.summary_frame.columnconfigure(1, weight=1)
        self.summary_value_labels: dict[str, ttk.Label] = {}

        summary_rows = (
            ("展示模式", "mode"),
            ("文件数量", "files"),
            ("目录数量", "directories"),
            ("扫描错误", "errors"),
            ("总大小", "size"),
            ("结果位置", "output"),
        )
        for row, (label, key) in enumerate(summary_rows):
            self.summary_frame.rowconfigure(row, weight=0)
            ttk.Label(self.summary_frame, text=f"{label}：").grid(
                row=row,
                column=0,
                sticky="nw",
                padx=(0, 16),
                pady=5,
            )
            value_label = ttk.Label(
                self.summary_frame,
                textvariable=self.summary_vars[key],
                style="SummaryValue.TLabel",
                anchor="w",
                justify="left",
                wraplength=560,
            )
            value_label.grid(row=row, column=1, sticky="ew", pady=5)
            self.summary_value_labels[key] = value_label

        actions = ttk.Frame(self)
        actions.grid(row=4, column=0, sticky="ew")
        actions.columnconfigure(0, weight=1)

        self.open_result_button = ttk.Button(
            actions,
            text="打开结果",
            command=self.open_result,
            state="disabled",
        )
        self.open_result_button.grid(row=0, column=1, padx=(0, 8))

        self.open_output_button = ttk.Button(
            actions,
            text="打开输出目录",
            command=self.open_output_directory,
            state="disabled",
        )
        self.open_output_button.grid(row=0, column=2)

    def choose_folder(self) -> None:
        """Open the native folder picker and display the selected path."""

        selected_path = filedialog.askdirectory(
            parent=self.master,
            title="选择需要分析的文件夹",
            mustexist=True,
        )
        if not selected_path:
            return

        self.path_var.set(selected_path)
        self.update_status(f"已选择文件夹：{selected_path}")
        self.path_entry.icursor(tk.END)

    def start_analysis(self) -> None:
        """Validate input and start the shared workflow in a worker thread."""

        if self._task_running:
            self.update_status("分析任务正在运行，请等待当前任务完成。")
            return

        selected_path = self.path_var.get().strip().strip('"')
        if not selected_path:
            self.update_status("请先选择或输入一个文件夹路径。")
            self.path_entry.focus_set()
            return

        candidate = Path(selected_path).expanduser()
        if not candidate.exists():
            self.update_status(f"路径不存在：{candidate}")
            self.path_entry.focus_set()
            return
        if not candidate.is_dir():
            self.update_status(f"输入路径不是文件夹：{candidate}")
            self.path_entry.focus_set()
            return
        try:
            input_path = candidate.resolve(strict=True)
        except OSError as error:
            self.update_status(f"无法解析文件夹路径：{error}")
            return

        self.path_var.set(str(input_path))
        previous_controller = self.large_server_controller
        self.large_server_controller = None
        self.large_url = None
        self.workflow_result = None
        self.small_result_path = None
        self.large_result = None
        self.open_result_button.configure(state="disabled")
        self.update_summary(
            mode="—",
            files="—",
            directories="—",
            errors="—",
            size="—",
            output="正在生成……",
        )
        self.update_status("正在准备分析任务……")
        self._set_task_running(True)

        self._task_thread = threading.Thread(
            target=self._run_workflow_worker,
            args=(input_path, previous_controller),
            name="FolderVisualizerWorkflow",
            daemon=True,
        )
        self._task_thread.start()

    def _run_workflow_worker(
        self,
        input_path: Path,
        previous_controller: LargeServerController | None,
    ) -> None:
        """Run workflow code without touching any Tkinter object."""

        controller: LargeServerController | None = None
        try:
            if previous_controller is not None:
                previous_controller.stop()
            result = run_workflow(
                WorkflowOptions(input_path=input_path),
                callback=self._queue_workflow_event,
            )
            url: str | None = None
            if result.mode == "large" and not self._close_requested.is_set():
                controller = create_large_server_controller(result, port=0)
                url = controller.start()

            if self._close_requested.is_set() and controller is not None:
                controller.stop()
                controller = None
                url = None
        except WorkflowError as error:
            if controller is not None:
                controller.stop()
            self._message_queue.put(("error", error))
        except Exception as error:  # Keep unexpected worker failures visible.
            if controller is not None:
                controller.stop()
            self._message_queue.put(("error", error))
        else:
            self._message_queue.put(
                ("completed", CompletedWorkflow(result, controller, url))
            )

    def _queue_workflow_event(self, event: WorkflowEvent) -> None:
        """Receive a worker event and enqueue it for the Tk main thread."""

        self._message_queue.put(("event", event))

    def _process_messages(self) -> None:
        """Process worker messages on the Tkinter main thread."""

        self._message_after_id = None
        try:
            while True:
                message_type, payload = self._message_queue.get_nowait()
                if message_type == "event":
                    self._handle_workflow_event(payload)
                elif message_type == "completed":
                    self._handle_workflow_completed(payload)
                elif message_type == "error":
                    self._handle_workflow_error(payload)
                elif message_type == "browser_opened":
                    self.update_status("已在默认浏览器中打开结果。")
                elif message_type == "browser_error":
                    self.update_status(f"无法打开浏览器：{payload}")
                elif message_type == "shutdown_complete":
                    self._handle_shutdown_complete(payload)
        except queue.Empty:
            pass
        if not self._destroyed:
            self._message_after_id = self.master.after(
                100, self._process_messages
            )

    def _handle_workflow_event(self, payload: object) -> None:
        if not isinstance(payload, WorkflowEvent):
            return

        event = payload
        if event.stage == "scan_completed":
            file_count = int(event.data["file_count"])
            directory_count = int(event.data["directory_count"])
            error_count = int(event.data["error_count"])
            self.update_summary(
                files=f"{file_count:,}",
                directories=f"{directory_count:,}",
                errors=f"{error_count:,}",
            )
            self.update_status(
                "扫描完成："
                f"{file_count:,} 个文件，"
                f"{directory_count:,} 个目录，"
                f"{error_count:,} 个错误。"
            )
            return

        if event.stage == "mode_selected":
            decision = event.data["decision"]
            mode_label = self._mode_label(decision.mode)
            self.update_summary(mode=mode_label)
            reasons = "；".join(decision.reasons)
            self.update_status(f"已选择{mode_label}：{reasons}")
            return

        self.update_status(event.message)

    def _handle_workflow_completed(self, payload: object) -> None:
        if not isinstance(payload, CompletedWorkflow):
            self._handle_workflow_error(RuntimeError("工作流返回了无效结果"))
            return

        result = payload.result
        self.workflow_result = result
        if result.mode == "small":
            self.small_result_path = result.small_tree_path
            self.large_result = None
            self.large_server_controller = None
            self.large_url = None
        else:
            self.small_result_path = None
            self.large_result = result
            self.large_server_controller = payload.controller
            self.large_url = payload.url

        self.update_summary(
            mode=self._mode_label(result.mode),
            files=f"{result.file_count:,}",
            directories=f"{result.directory_count:,}",
            errors=f"{result.scan_error_count:,}",
            size=self._format_size(result.total_size),
            output=str(result.primary_output_path),
        )
        self._set_task_running(False)
        self._task_thread = None

        if self._closing:
            self._begin_server_shutdown()
            return

        self.update_status(
            f"分析完成，已生成{self._mode_label(result.mode)}结果。"
        )
        self.open_result_button.configure(state="normal")
        self.open_result()

    def _handle_workflow_error(self, payload: object) -> None:
        self._set_task_running(False)
        self._task_thread = None
        if self._closing:
            self._begin_server_shutdown()
            return
        self.update_status(f"分析失败：{payload}")
        self.update_summary(output="未生成结果")

    def _set_task_running(self, running: bool) -> None:
        self._task_running = running
        state = "disabled" if running or self._closing else "normal"
        self.path_entry.configure(state=state)
        self.choose_button.configure(state=state)
        self.start_button.configure(state=state)

    @staticmethod
    def _mode_label(mode: str) -> str:
        return "小目录完整树" if mode == "small" else "大目录按需浏览"

    @staticmethod
    def _format_size(size: int) -> str:
        units = ("B", "KB", "MB", "GB", "TB")
        value = float(size)
        for unit in units:
            if value < 1024 or unit == units[-1]:
                if unit == "B":
                    return f"{int(value)} {unit}"
                return f"{value:.2f} {unit}"
            value /= 1024
        return f"{size} B"

    def update_status(self, message: str) -> None:
        """Update the user-facing status text."""

        self.status_var.set(message)
        self._schedule_layout_refresh()

    def update_summary(
        self,
        *,
        mode: str | None = None,
        files: str | None = None,
        directories: str | None = None,
        errors: str | None = None,
        size: str | None = None,
        output: str | None = None,
    ) -> None:
        """Update the supplied values in the result summary."""

        values = {
            "mode": mode,
            "files": files,
            "directories": directories,
            "errors": errors,
            "size": size,
            "output": output,
        }
        for key, value in values.items():
            if value is not None:
                self.summary_vars[key].set(value)
        self._schedule_layout_refresh()

    def _on_layout_configure(self, event: tk.Event[tk.Misc]) -> None:
        """Keep text wrapping and minimum size aligned with actual DPI."""

        if event.widget is not self or self._destroyed:
            return
        status_width = max(240, event.width - 56)
        value_width = max(240, event.width - 180)
        wrap_widths = (status_width, value_width)
        if wrap_widths == self._wrap_widths:
            return
        self._wrap_widths = wrap_widths
        self.status_label.configure(wraplength=status_width)
        for label in self.summary_value_labels.values():
            label.configure(wraplength=value_width)
        self._schedule_layout_refresh()

    def _schedule_layout_refresh(self) -> None:
        if self._destroyed or self._layout_after_id is not None:
            return
        self._layout_after_id = self.master.after_idle(
            self._refresh_layout_constraints
        )

    def _refresh_layout_constraints(self) -> None:
        """Derive the minimum window size from requested widget geometry."""

        self._layout_after_id = None
        if self._destroyed:
            return

        self.master.update_idletasks()
        width = max(self.winfo_width(), 680)
        status_width = max(240, width - 56)
        value_width = max(240, width - 180)
        wrap_widths = (status_width, value_width)
        if wrap_widths != self._wrap_widths:
            self._wrap_widths = wrap_widths
            self.status_label.configure(wraplength=status_width)
            for label in self.summary_value_labels.values():
                label.configure(wraplength=value_width)
            self.master.update_idletasks()

        summary_height = self.summary_frame.winfo_reqheight()
        self.rowconfigure(3, weight=1, minsize=summary_height)
        self.master.update_idletasks()
        minimum_width = max(680, self.winfo_reqwidth())
        minimum_height = self.winfo_reqheight()
        self.master.minsize(minimum_width, minimum_height)

    def open_result(self) -> None:
        """Open the current small tree or large browser without blocking Tk."""

        target: str | None = None
        if self.workflow_result is not None:
            if self.workflow_result.mode == "small":
                if self.small_result_path is not None:
                    target = self.small_result_path.resolve().as_uri()
            else:
                target = self.large_url

        if target is None:
            self.update_status("当前尚未生成可打开的结果。")
            return

        threading.Thread(
            target=self._open_browser_worker,
            args=(target,),
            name="FolderVisualizerBrowserOpen",
            daemon=True,
        ).start()

    def _open_browser_worker(self, target: str) -> None:
        try:
            opened = webbrowser.open(target)
        except (OSError, webbrowser.Error) as error:
            self._message_queue.put(("browser_error", error))
            return
        if opened:
            self._message_queue.put(("browser_opened", target))
        else:
            self._message_queue.put(
                ("browser_error", "系统未接受打开浏览器的请求")
            )

    def open_output_directory(self) -> None:
        """Handle opening the output directory in a later stage."""

        self.update_status("当前尚未生成输出目录。")

    def _on_path_entered(self, _event: tk.Event[tk.Misc]) -> None:
        self.start_analysis()

    def close(self) -> None:
        """Stop any large-mode service before destroying the window."""

        if self._closing:
            return
        self._closing = True
        self._close_requested.set()
        self._set_task_running(self._task_running)
        self.open_result_button.configure(state="disabled")
        self.open_output_button.configure(state="disabled")

        if self._task_running:
            self.update_status("正在等待当前分析任务结束并关闭本地服务……")
            return
        self._begin_server_shutdown()

    def _begin_server_shutdown(self) -> None:
        controller = self.large_server_controller
        self.large_server_controller = None
        self.large_url = None
        if controller is None:
            self._destroy_window()
            return

        self.update_status("正在关闭大目录本地服务……")
        threading.Thread(
            target=self._stop_server_worker,
            args=(controller,),
            name="FolderVisualizerServerShutdown",
            daemon=True,
        ).start()

    def _stop_server_worker(
        self, controller: LargeServerController
    ) -> None:
        try:
            controller.stop()
            controller.wait()
        except Exception as error:
            self._message_queue.put(("shutdown_complete", error))
        else:
            self._message_queue.put(("shutdown_complete", None))

    def _handle_shutdown_complete(self, payload: object) -> None:
        if payload is not None:
            self.update_status(f"关闭本地服务时发生错误：{payload}")
        self._destroy_window()

    def _destroy_window(self) -> None:
        self._destroyed = True
        if self._layout_after_id is not None:
            try:
                self.master.after_cancel(self._layout_after_id)
            except tk.TclError:
                pass
            self._layout_after_id = None
        if self._message_after_id is not None:
            try:
                self.master.after_cancel(self._message_after_id)
            except tk.TclError:
                pass
            self._message_after_id = None
        self.master.destroy()


def main() -> int:
    root = tk.Tk()
    FolderVisualizerApp(root)
    root.mainloop()
    return 0


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