GogsAdminConsole.py

#!/usr/bin/env python3
"""Gogs Admin Console - GUI safety cleanup and repair tool."""

import os
import queue
import re
import threading
import tkinter as tk
from tkinter import messagebox, simpledialog, ttk
from typing import Dict, List, Optional, Sequence, Set, Tuple
from urllib import parse

from config_store import ConfigStore
from gogs_cleanup import (
    Account,
    DEFAULT_SCAN_QUERIES,
    DEFAULT_PROTECTED,
    GogsApiError,
    GogsClient,
    Repository,
    redact_sensitive,
    safe_current_orgs,
    safe_current_user,
)
from gogs_db_repair_sql import generate_membership_repair_sql
from query_service import QueryOptions, QueryResult, QueryService


CHECKED = "[x]"
UNCHECKED = "[ ]"
APP_BG = "#f3f5f7"
SURFACE = "#ffffff"
TEXT = "#1f2937"
MUTED = "#6b7280"
ACCENT = "#2563eb"
DANGER = "#b91c1c"
PROTECTED_BG = "#f8fafc"
CHECKED_BG = "#eff6ff"
CONFIG_FILE = "GogsAdminConsole.config.json"


class GogsAdminConsole(tk.Tk):
    def __init__(self) -> None:
        super().__init__()
        self.title("Gogs Admin Console")
        self.geometry("1240x820")
        self.minsize(1060, 680)

        self.queue: "queue.Queue[Tuple[str, object]]" = queue.Queue()
        self.client: Optional[GogsClient] = None
        self.worker_running = False
        self.current_operation = ""
        self.query_cancel_event = threading.Event()
        self.query_cancel_buttons: List[ttk.Button] = []

        self.users: Dict[str, Dict[str, object]] = {}
        self.orgs: Dict[str, Dict[str, object]] = {}
        self.repos: Dict[str, Dict[str, object]] = {}
        self.protected: Set[str] = set()
        self.current_user = ""
        self.current_orgs: List[str] = []
        self.known_org_names: List[str] = []
        self.config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), CONFIG_FILE)
        self.config_store = ConfigStore(self.config_path)
        self.config_password: Optional[str] = None

        self._build_vars()
        self.load_config(silent=True)
        self._configure_style()
        self._build_ui()
        self.after(100, self._drain_queue)

    def _configure_style(self) -> None:
        self.configure(bg=APP_BG)
        style = ttk.Style(self)
        try:
            style.theme_use("clam")
        except tk.TclError:
            pass
        style.configure(".", font=("Segoe UI", 10), background=APP_BG, foreground=TEXT)
        style.configure("TFrame", background=APP_BG)
        style.configure("Surface.TFrame", background=SURFACE)
        style.configure("TLabel", background=APP_BG, foreground=TEXT)
        style.configure("Muted.TLabel", background=APP_BG, foreground=MUTED)
        style.configure("Title.TLabel", font=("Segoe UI", 18, "bold"), background=APP_BG, foreground=TEXT)
        style.configure("Subtitle.TLabel", font=("Segoe UI", 9), background=APP_BG, foreground=MUTED)
        style.configure("TLabelframe", background=APP_BG, foreground=TEXT, bordercolor="#d7dde5", relief="solid")
        style.configure("TLabelframe.Label", background=APP_BG, foreground=TEXT, font=("Segoe UI", 10, "bold"))
        style.configure("TButton", padding=(12, 7), borderwidth=1)
        style.map("TButton", background=[("active", "#e5e7eb")])
        style.configure("Accent.TButton", background=ACCENT, foreground="#ffffff", bordercolor=ACCENT)
        style.map("Accent.TButton", background=[("active", "#1d4ed8"), ("pressed", "#1e40af")], foreground=[("disabled", "#d1d5db")])
        style.configure("Danger.TButton", background=DANGER, foreground="#ffffff", bordercolor=DANGER)
        style.map("Danger.TButton", background=[("active", "#991b1b"), ("pressed", "#7f1d1d")])
        style.configure("TEntry", fieldbackground="#ffffff", bordercolor="#cbd5e1", lightcolor="#cbd5e1", darkcolor="#cbd5e1", padding=5)
        style.configure("TCheckbutton", background=APP_BG, foreground=TEXT, padding=(0, 3))
        style.configure("TNotebook", background=APP_BG, borderwidth=0)
        style.configure("TNotebook.Tab", padding=(16, 9), background="#e5e7eb", foreground=TEXT)
        style.map("TNotebook.Tab", background=[("selected", SURFACE), ("active", "#f8fafc")])
        style.configure("Treeview", background=SURFACE, fieldbackground=SURFACE, foreground=TEXT, rowheight=28, bordercolor="#d7dde5", borderwidth=1)
        style.configure("Treeview.Heading", font=("Segoe UI", 9, "bold"), background="#eef2f7", foreground=TEXT, padding=(6, 7))
        style.map("Treeview", background=[("selected", "#dbeafe")], foreground=[("selected", TEXT)])

    def _build_vars(self) -> None:
        self.base_url_var = tk.StringVar(value=os.getenv("GOGS_URL", "http://127.0.0.1:3000"))
        self.token_var = tk.StringVar(value=os.getenv("GOGS_TOKEN", ""))
        self.web_username_var = tk.StringVar(value=os.getenv("GOGS_WEB_USERNAME", ""))
        self.web_password_var = tk.StringVar(value=os.getenv("GOGS_WEB_PASSWORD", ""))
        self.query_var = tk.StringVar(value="")
        self.limit_var = tk.StringVar(value="1000")
        self.timeout_var = tk.StringVar(value="20")
        self.protect_var = tk.StringVar(value="")
        self.status_var = tk.StringVar(value="Ready")
        self.protected_var = tk.StringVar(value="Protected: admin, root, git, gogs")

        self.no_scan_var = tk.BooleanVar(value=False)
        self.insecure_var = tk.BooleanVar(value=False)
        self.include_users_var = tk.BooleanVar(value=True)
        self.include_orgs_var = tk.BooleanVar(value=True)
        self.include_repos_var = tk.BooleanVar(value=True)
        self.purge_repos_var = tk.BooleanVar(value=True)
        self.remove_org_memberships_var = tk.BooleanVar(value=True)
        self.allow_current_user_var = tk.BooleanVar(value=False)
        self.allow_current_orgs_var = tk.BooleanVar(value=False)

    def _build_ui(self) -> None:
        self.columnconfigure(0, weight=1)
        self.rowconfigure(1, weight=1)

        header = ttk.Frame(self, padding=(16, 14, 16, 8))
        header.grid(row=0, column=0, sticky="ew")
        header.columnconfigure(0, weight=1)
        ttk.Label(header, text="Gogs Admin Console", 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=(2, 0))
        ttk.Button(header, text="查询全部", style="Accent.TButton", command=self.query_all).grid(row=0, column=1, rowspan=2, sticky="e", padx=(8, 0))
        cancel_button = ttk.Button(header, text="取消查询", command=self.cancel_query, state="disabled")
        cancel_button.grid(row=0, column=2, rowspan=2, sticky="e", padx=(8, 0))
        self.query_cancel_buttons.append(cancel_button)
        ttk.Button(header, text="删除勾选项", style="Danger.TButton", command=self.delete_checked).grid(row=0, column=3, rowspan=2, sticky="e", padx=(8, 0))

        self.notebook = ttk.Notebook(self)
        self.notebook.grid(row=1, column=0, sticky="nsew", padx=16, pady=(0, 8))

        self.connection_tab = ttk.Frame(self.notebook, padding=10)
        self.users_tab = ttk.Frame(self.notebook, padding=10)
        self.orgs_tab = ttk.Frame(self.notebook, padding=10)
        self.repos_tab = ttk.Frame(self.notebook, padding=10)
        self.repair_tab = ttk.Frame(self.notebook, padding=10)
        self.logs_tab = ttk.Frame(self.notebook, padding=10)

        self.notebook.add(self.connection_tab, text="连接")
        self.notebook.add(self.users_tab, text="用户")
        self.notebook.add(self.orgs_tab, text="组织")
        self.notebook.add(self.repos_tab, text="仓库")
        self.notebook.add(self.repair_tab, text="成员关系修复")
        self.notebook.add(self.logs_tab, text="日志")

        self._build_connection_tab()
        self.user_tree = self._build_account_tab(self.users_tab, "user")
        self.org_tree = self._build_account_tab(self.orgs_tab, "org")
        self.repo_tree = self._build_repo_tab(self.repos_tab)
        self._build_repair_tab()
        self._build_log_tab()

        footer = ttk.Frame(self, padding=(16, 6, 16, 12))
        footer.grid(row=2, column=0, sticky="ew")
        footer.columnconfigure(0, weight=1)
        ttk.Label(footer, textvariable=self.protected_var).grid(row=0, column=0, sticky="ew")
        ttk.Label(footer, textvariable=self.status_var).grid(row=1, column=0, sticky="ew", pady=(2, 0))

    def _build_connection_tab(self) -> None:
        tab = self.connection_tab
        for col in (1, 3):
            tab.columnconfigure(col, weight=1)

        ttk.Label(tab, text="Gogs 地址").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=4)
        ttk.Entry(tab, textvariable=self.base_url_var).grid(row=0, column=1, sticky="ew", pady=4)
        ttk.Label(tab, text="Access Token").grid(row=0, column=2, sticky="w", padx=(16, 8), pady=4)
        ttk.Entry(tab, textvariable=self.token_var, show="*").grid(row=0, column=3, sticky="ew", pady=4)

        ttk.Label(tab, text="Web 用户").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=4)
        ttk.Entry(tab, textvariable=self.web_username_var).grid(row=1, column=1, sticky="ew", pady=4)
        ttk.Label(tab, text="Web 密码").grid(row=1, column=2, sticky="w", padx=(16, 8), pady=4)
        ttk.Entry(tab, textvariable=self.web_password_var, show="*").grid(row=1, column=3, sticky="ew", pady=4)

        ttk.Label(tab, text="搜索词").grid(row=2, column=0, sticky="w", padx=(0, 8), pady=4)
        ttk.Entry(tab, textvariable=self.query_var).grid(row=2, column=1, sticky="ew", pady=4)
        ttk.Label(tab, text="每次上限").grid(row=2, column=2, sticky="w", padx=(16, 8), pady=4)
        ttk.Entry(tab, textvariable=self.limit_var, width=10).grid(row=2, column=3, sticky="w", pady=4)

        ttk.Label(tab, text="额外保护").grid(row=3, column=0, sticky="w", padx=(0, 8), pady=4)
        ttk.Entry(tab, textvariable=self.protect_var).grid(row=3, column=1, columnspan=3, sticky="ew", pady=4)

        options = ttk.LabelFrame(tab, text="查询与安全选项", padding=10)
        options.grid(row=4, column=0, columnspan=4, sticky="ew", pady=(12, 8))
        for text, var in (
            ("用户", self.include_users_var),
            ("组织", self.include_orgs_var),
            ("仓库", self.include_repos_var),
            ("跳过常见字符扫描", self.no_scan_var),
            ("删除组织/用户前先删其仓库", self.purge_repos_var),
            ("删除用户前解除组织成员关系", self.remove_org_memberships_var),
            ("允许删除当前用户", self.allow_current_user_var),
            ("允许删除当前用户组织", self.allow_current_orgs_var),
            ("忽略 TLS 证书", self.insecure_var),
        ):
            ttk.Checkbutton(options, text=text, variable=var, onvalue=True, offvalue=False).pack(side="left", padx=(0, 12), pady=2)

        actions = ttk.Frame(tab)
        actions.grid(row=5, column=0, columnspan=4, sticky="ew", pady=(8, 0))
        ttk.Button(actions, text="连接测试", command=self.test_connection).pack(side="left")
        ttk.Button(actions, text="查询全部", style="Accent.TButton", command=self.query_all).pack(side="left", padx=(8, 0))
        cancel_button = ttk.Button(actions, text="取消查询", command=self.cancel_query, state="disabled")
        cancel_button.pack(side="left", padx=(8, 0))
        self.query_cancel_buttons.append(cancel_button)
        ttk.Button(actions, text="保存配置", command=self.save_config).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="重置配置", command=lambda: self.load_config(silent=False)).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="清空输入", command=self.clear_connection_inputs).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="删除所有勾选项", style="Danger.TButton", command=self.delete_checked).pack(side="right")

    def _build_account_tab(self, tab: ttk.Frame, kind: str) -> ttk.Treeview:
        tab.columnconfigure(0, weight=1)
        tab.rowconfigure(0, weight=1)
        columns = ("checked", "name", "id", "repos", "email", "full_name", "protected")
        tree = ttk.Treeview(tab, columns=columns, show="headings", selectmode="extended")
        tree.tag_configure("protected", background=PROTECTED_BG, foreground=MUTED)
        tree.tag_configure("checked", background=CHECKED_BG)
        headings = {
            "checked": "勾选",
            "name": "名称",
            "id": "ID",
            "repos": "仓库数",
            "email": "邮箱",
            "full_name": "显示名",
            "protected": "保护",
        }
        widths = {"checked": 58, "name": 220, "id": 70, "repos": 72, "email": 240, "full_name": 180, "protected": 260}
        for col in columns:
            tree.heading(col, text=headings[col])
            tree.column(col, width=widths[col], anchor="w", stretch=col in {"name", "email", "full_name", "protected"})
        tree.grid(row=0, column=0, sticky="nsew")
        tree.bind("<Button-1>", lambda event, t=tree, k=kind: self._tree_click(event, t, k))
        tree.bind("<space>", lambda event, t=tree, k=kind: self._toggle_selected_event(event, t, k))
        scrollbar = ttk.Scrollbar(tab, orient="vertical", command=tree.yview)
        scrollbar.grid(row=0, column=1, sticky="ns")
        tree.configure(yscrollcommand=scrollbar.set)

        actions = ttk.Frame(tab)
        actions.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(8, 0))
        ttk.Button(actions, text="全选可删项", command=lambda: self._select_deletable(kind)).pack(side="left")
        ttk.Button(actions, text="取消全选", command=lambda: self._clear_checks(kind)).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="反选可删项", command=lambda: self._invert_checks(kind)).pack(side="left", padx=(8, 0))
        if kind == "org":
            ttk.Button(actions, text="查看成员/Teams", command=self.inspect_selected_orgs).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="手动添加", command=lambda: self.add_manual_targets(kind)).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="删除勾选项", style="Danger.TButton", command=self.delete_checked).pack(side="right")
        return tree

    def _build_repo_tab(self, tab: ttk.Frame) -> ttk.Treeview:
        tab.columnconfigure(0, weight=1)
        tab.rowconfigure(0, weight=1)
        columns = ("checked", "owner", "name", "visibility", "protected")
        tree = ttk.Treeview(tab, columns=columns, show="headings", selectmode="extended")
        tree.tag_configure("protected", background=PROTECTED_BG, foreground=MUTED)
        tree.tag_configure("checked", background=CHECKED_BG)
        for col, title, width in (
            ("checked", "勾选", 58),
            ("owner", "归属", 220),
            ("name", "仓库", 360),
            ("visibility", "可见性", 90),
            ("protected", "保护", 260),
        ):
            tree.heading(col, text=title)
            tree.column(col, width=width, anchor="w", stretch=col in {"owner", "name", "protected"})
        tree.grid(row=0, column=0, sticky="nsew")
        tree.bind("<Button-1>", lambda event: self._tree_click(event, tree, "repo"))
        tree.bind("<space>", lambda event: self._toggle_selected_event(event, tree, "repo"))
        scrollbar = ttk.Scrollbar(tab, orient="vertical", command=tree.yview)
        scrollbar.grid(row=0, column=1, sticky="ns")
        tree.configure(yscrollcommand=scrollbar.set)

        actions = ttk.Frame(tab)
        actions.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(8, 0))
        ttk.Button(actions, text="全选可删项", command=lambda: self._select_deletable("repo")).pack(side="left")
        ttk.Button(actions, text="取消全选", command=lambda: self._clear_checks("repo")).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="反选可删项", command=lambda: self._invert_checks("repo")).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="手动添加仓库/URL", command=lambda: self.add_manual_targets("repo")).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="删除勾选仓库", style="Danger.TButton", command=self.delete_checked).pack(side="right")
        return tree

    def _build_repair_tab(self) -> None:
        tab = self.repair_tab
        tab.columnconfigure(0, weight=1)
        ttk.Label(tab, text="勾选用户后生成 org_user/team_user 检查 SQL。默认只查询，不会删除。").grid(row=0, column=0, sticky="w", pady=(0, 8))
        actions = ttk.Frame(tab)
        actions.grid(row=1, column=0, sticky="ew")
        ttk.Button(actions, text="为勾选用户生成检查 SQL", command=lambda: self.generate_membership_sql(False)).pack(side="left")
        ttk.Button(actions, text="为勾选用户生成删除 SQL", command=lambda: self.generate_membership_sql(True)).pack(side="left", padx=(8, 0))
        ttk.Button(actions, text="复制选中用户 UID", command=self.copy_selected_user_ids).pack(side="left", padx=(8, 0))
        self.repair_text = tk.Text(tab, height=24, wrap="none")
        self.repair_text.grid(row=2, column=0, sticky="nsew", pady=(10, 0))
        tab.rowconfigure(2, weight=1)

    def _build_log_tab(self) -> None:
        tab = self.logs_tab
        tab.columnconfigure(0, weight=1)
        tab.rowconfigure(0, weight=1)
        self.log = tk.Text(tab, wrap="word")
        self.log.configure(bg="#0f172a", fg="#e5e7eb", insertbackground="#e5e7eb", relief="flat", padx=10, pady=10)
        self.log.grid(row=0, column=0, sticky="nsew")
        scrollbar = ttk.Scrollbar(tab, orient="vertical", command=self.log.yview)
        scrollbar.grid(row=0, column=1, sticky="ns")
        self.log.configure(yscrollcommand=scrollbar.set)

    def _config_payload(self) -> Dict[str, object]:
        return {
            "base_url": self.base_url_var.get().strip(),
            "token": self.token_var.get().strip(),
            "web_username": self.web_username_var.get().strip(),
            "query": self.query_var.get(),
            "limit": self.limit_var.get(),
            "timeout": self.timeout_var.get(),
            "protect": self.protect_var.get(),
            "no_scan": self.no_scan_var.get(),
            "insecure": self.insecure_var.get(),
            "include_users": self.include_users_var.get(),
            "include_orgs": self.include_orgs_var.get(),
            "include_repos": self.include_repos_var.get(),
            "purge_repos": self.purge_repos_var.get(),
            "remove_org_memberships": self.remove_org_memberships_var.get(),
            "allow_current_user": self.allow_current_user_var.get(),
            "allow_current_orgs": self.allow_current_orgs_var.get(),
        }

    def save_config(self) -> None:
        if self.config_password is None:
            password = self._ask_new_config_password()
            if password is None:
                messagebox.showwarning("保存配置", "未设置配置密码，配置未保存。")
                return
            self.config_password = password
        if self.token_var.get().strip():
            ok = messagebox.askokcancel(
                "保存配置",
                "Access Token 会保存到加密本地配置文件。\nWeb 密码不会保存。\n\n继续保存？",
            )
            if not ok:
                return
        self.config_store.write(self._config_payload(), self.config_password)
        self.web_password_var.set("")
        self.status_var.set(f"配置已加密保存：{self.config_path}")
        self._log(self.status_var.get())

    def _save_config_auto(self) -> bool:
        if self.config_password is None:
            password = self._ask_new_config_password()
            if password is None:
                messagebox.showwarning("自动保存配置", "未设置配置密码，查询已取消。")
                return False
            self.config_password = password
        try:
            self.config_store.write(self._config_payload(), self.config_password)
            self._log(f"配置已自动加密保存：{self.config_path}")
            return True
        except Exception as exc:
            messagebox.showerror("自动保存配置失败", redact_sensitive(exc))
            return False

    def load_config(self, silent: bool) -> None:
        if not self.config_store.exists():
            reason = "未找到本地配置文件。"
            if not self._create_new_config(reason, backup_path=None, silent=silent):
                self.destroy()
                raise SystemExit(1)
            return

        while True:
            password = self.config_password if self.config_password is not None and silent else self._ask_existing_config_password()
            if password is None:
                backup_path = self.config_store.backup()
                if not self._create_new_config("已取消输入配置密码，旧配置已备份。", backup_path, silent=False):
                    self.destroy()
                    raise SystemExit(1)
                return

            try:
                data = self.config_store.read(password)
                self._apply_config(data)
                self.config_password = password
                if not silent:
                    self.status_var.set(f"配置已重置为保存值：{self.config_path}")
                    self._log(self.status_var.get())
                return
            except Exception as exc:
                self.config_password = None
                messagebox.showwarning(
                    "配置读取失败",
                    f"配置密码错误或文件无法解密：{redact_sensitive(exc)}\n\n请重新输入配置密码；取消输入将备份旧配置并生成新配置。",
                )

    def _apply_config(self, data: Dict[str, object]) -> None:
        self.base_url_var.set(str(data.get("base_url") or os.getenv("GOGS_URL", "http://127.0.0.1:3000")))
        self.token_var.set(str(data.get("token") or os.getenv("GOGS_TOKEN", "")))
        self.web_username_var.set(str(data.get("web_username") or os.getenv("GOGS_WEB_USERNAME", "")))
        self.web_password_var.set("")
        self.query_var.set(str(data.get("query") or ""))
        self.limit_var.set(str(data.get("limit") or "1000"))
        self.timeout_var.set(str(data.get("timeout") or "20"))
        self.protect_var.set(str(data.get("protect") or ""))
        self._set_bool(self.no_scan_var, data.get("no_scan"), False)
        self._set_bool(self.insecure_var, data.get("insecure"), False)
        self._set_bool(self.include_users_var, data.get("include_users"), True)
        self._set_bool(self.include_orgs_var, data.get("include_orgs"), True)
        self._set_bool(self.include_repos_var, data.get("include_repos"), True)
        self._set_bool(self.purge_repos_var, data.get("purge_repos"), True)
        self._set_bool(self.remove_org_memberships_var, data.get("remove_org_memberships"), True)
        self._set_bool(self.allow_current_user_var, data.get("allow_current_user"), False)
        self._set_bool(self.allow_current_orgs_var, data.get("allow_current_orgs"), False)

    def _ask_existing_config_password(self) -> Optional[str]:
        return simpledialog.askstring("配置密码", "请输入本地配置文件密码：", show="*", parent=self)

    def _ask_new_config_password(self) -> Optional[str]:
        while True:
            password = simpledialog.askstring("设置配置密码", "请输入新的配置密码：", show="*", parent=self)
            if password is None:
                return None
            if not password:
                messagebox.showwarning("设置配置密码", "配置密码不能为空。")
                continue
            confirm = simpledialog.askstring("确认配置密码", "请再次输入新的配置密码：", show="*", parent=self)
            if confirm is None:
                return None
            if password != confirm:
                messagebox.showwarning("设置配置密码", "两次输入的密码不一致。")
                continue
            return password

    def _create_new_config(self, reason: str, backup_path: Optional[str], silent: bool) -> bool:
        if backup_path:
            message = f"{reason}\n\n旧配置已备份到：\n{backup_path}\n\n请设置新的配置密码。"
        else:
            message = f"{reason}\n\n请设置新的配置密码。"
        if silent:
            messagebox.showinfo("初始化配置", message)
        else:
            messagebox.showwarning("初始化配置", message)

        password = self._ask_new_config_password()
        if password is None:
            messagebox.showerror("初始化配置", "未设置配置密码，程序将退出。")
            return False
        self.config_password = password
        self.clear_connection_inputs()
        self.config_store.write(self._config_payload(), password)
        self.status_var.set(f"已生成新的加密配置：{self.config_path}")
        return True

    def clear_connection_inputs(self) -> None:
        self.base_url_var.set(os.getenv("GOGS_URL", "http://127.0.0.1:3000"))
        self.token_var.set(os.getenv("GOGS_TOKEN", ""))
        self.web_username_var.set(os.getenv("GOGS_WEB_USERNAME", ""))
        self.web_password_var.set("")
        self.query_var.set("")
        self.limit_var.set("1000")
        self.timeout_var.set("20")
        self.protect_var.set("")
        self.no_scan_var.set(False)
        self.insecure_var.set(False)
        self.include_users_var.set(True)
        self.include_orgs_var.set(True)
        self.include_repos_var.set(True)
        self.purge_repos_var.set(True)
        self.remove_org_memberships_var.set(True)
        self.allow_current_user_var.set(False)
        self.allow_current_orgs_var.set(False)
        self.status_var.set("输入已清空为默认值。")

    def _set_bool(self, var: tk.BooleanVar, value: object, default: bool) -> None:
        var.set(default if value is None else bool(value))

    def _make_client(self) -> GogsClient:
        return GogsClient(
            base_url=self.base_url_var.get().strip(),
            token=self.token_var.get().strip(),
            timeout=self._timeout(),
            insecure=self.insecure_var.get(),
            web_username=self.web_username_var.get().strip(),
            web_password=self.web_password_var.get(),
        )

    def _timeout(self) -> float:
        raw = self.timeout_var.get().strip() or "20"
        try:
            timeout = float(raw)
        except ValueError as exc:
            raise ValueError("请求超时必须是数字（秒）") from exc
        if timeout <= 0:
            raise ValueError("请求超时必须大于 0 秒")
        return timeout

    def test_connection(self) -> None:
        if self.worker_running:
            return
        self._run_worker(self._test_connection_worker, "正在测试连接...")

    def _test_connection_worker(self) -> None:
        client = self._make_client()
        current = client.current_user()
        orgs = client.my_orgs()
        self.queue.put(("connected", (client, current, orgs)))

    def query_all(self) -> None:
        if self.worker_running:
            return
        if not (self.include_users_var.get() or self.include_orgs_var.get() or self.include_repos_var.get()):
            messagebox.showwarning("查询选项无效", "请至少勾选用户、组织或仓库中的一项。")
            return
        if not self._save_config_auto():
            return
        self.query_cancel_event.clear()
        self._run_worker(self._query_worker, "正在查询用户、组织和仓库...", operation="query")

    def cancel_query(self) -> None:
        if self.worker_running and self.current_operation == "query":
            self.query_cancel_event.set()
            self._set_query_cancel_enabled(False)
            self.status_var.set("正在取消查询...")
            self._log("正在取消查询...")

    def _query_worker(self) -> None:
        client = self._make_client()
        protected = self._load_protected(client, strict=False)
        self.queue.put(("query_begin", (client, protected)))
        options = QueryOptions(
            queries=self._query_terms(),
            limit=self._limit(),
            include_users=self.include_users_var.get(),
            include_orgs=self.include_orgs_var.get(),
            include_repos=self.include_repos_var.get(),
        )
        service = QueryService(client, options, self.query_cancel_event, self._publish_worker_event)
        result = service.run()
        self.queue.put(("query_done", (client, protected, result)))

    def _query_terms(self) -> List[str]:
        query_text = self.query_var.get().strip()
        queries = [query_text] if query_text else []
        if not self.no_scan_var.get():
            queries.extend(DEFAULT_SCAN_QUERIES)
        if not queries:
            queries.append("")
        return sorted(set(queries))

    def _publish_worker_event(self, kind: str, payload: object) -> None:
        self.queue.put((kind, payload))

    def _limit(self) -> int:
        limit = int(self.limit_var.get().strip() or "1000")
        if limit <= 0:
            raise ValueError("每次上限必须大于 0")
        return limit

    def _load_protected(self, client: GogsClient, strict: bool) -> Set[str]:
        protected = {item.lower() for item in DEFAULT_PROTECTED}
        protected.update(name.lower() for name in self._extra_protected_names())
        self.current_user = ""
        self.current_orgs = []
        if not self.allow_current_user_var.get():
            current = safe_current_user(client, strict)
            if current:
                self.current_user = current.name
                protected.add(current.name.lower())
        if not self.allow_current_orgs_var.get():
            orgs = safe_current_orgs(client, strict)
            self.current_orgs = [org.name for org in orgs]
            protected.update(org.name.lower() for org in orgs)
        return protected

    def _extra_protected_names(self) -> List[str]:
        raw = self.protect_var.get().replace("\n", ",")
        return [item.strip() for item in raw.split(",") if item.strip()]

    def _run_worker(self, fn, status: str, operation: str = "") -> None:
        self.worker_running = True
        self.current_operation = operation
        self.status_var.set(status)
        self.configure(cursor="watch")
        if operation == "query":
            self._set_query_cancel_enabled(True)
        threading.Thread(target=self._worker_wrapper, args=(fn,), daemon=True).start()

    def _set_query_cancel_enabled(self, enabled: bool) -> None:
        state = "normal" if enabled else "disabled"
        for button in self.query_cancel_buttons:
            button.configure(state=state)

    def _worker_wrapper(self, fn) -> None:
        try:
            fn()
        except Exception as exc:
            self.queue.put(("error", exc))

    def _drain_queue(self) -> None:
        while True:
            try:
                kind, payload = self.queue.get_nowait()
            except queue.Empty:
                break
            if kind == "connected":
                self._handle_connected(payload)  # type: ignore[arg-type]
            elif kind == "query_begin":
                self._handle_query_begin(payload)  # type: ignore[arg-type]
            elif kind == "query_accounts":
                self._handle_query_accounts(payload)  # type: ignore[arg-type]
            elif kind == "query_repos":
                self._handle_query_repos(payload)  # type: ignore[arg-type]
            elif kind == "query_done":
                self._handle_query_done(payload)  # type: ignore[arg-type]
            elif kind == "delete_done":
                self._handle_delete_done(payload)  # type: ignore[arg-type]
            elif kind == "log":
                self._log(redact_sensitive(payload))
            elif kind == "query_status":
                self.status_var.set(redact_sensitive(payload))
            elif kind == "error":
                self._finish("操作失败")
                self._log(f"ERROR: {redact_sensitive(payload)}")
                messagebox.showerror("操作失败", redact_sensitive(payload))
        self.after(100, self._drain_queue)

    def _handle_connected(self, payload: Tuple[GogsClient, Optional[Account], List[Account]]) -> None:
        client, current, orgs = payload
        self.client = client
        name = current.name if current else "(unknown)"
        self._finish(f"连接成功：当前用户 {name}，当前用户组织 {len(orgs)} 个")
        self._log(self.status_var.get())

    def _handle_query_begin(self, payload: Tuple[GogsClient, Set[str]]) -> None:
        client, protected = payload
        self.client = client
        self.protected = protected
        self.known_org_names = []

        self._clear_tree(self.user_tree)
        self._clear_tree(self.org_tree)
        self._clear_tree(self.repo_tree)
        self.users.clear()
        self.orgs.clear()
        self.repos.clear()
        self._refresh_protected_label()

    def _handle_query_accounts(self, payload: Tuple[str, List[Account]]) -> None:
        kind, accounts = payload
        for account in accounts:
            self._insert_account(kind, account, [])
        if kind == "org":
            self.known_org_names = sorted({str(row["name"]) for row in self.orgs.values()}, key=str.lower)

    def _handle_query_repos(self, payload: Tuple[str, str, List[Repository]]) -> None:
        owner_kind, owner, repos = payload
        self._update_account_repo_count(owner_kind, owner, len(repos))
        if self.include_repos_var.get():
            self._insert_repos({owner: repos})

    def _handle_query_done(self, payload: Tuple[GogsClient, Set[str], QueryResult]) -> None:
        client, protected, result = payload
        self.client = client
        self.protected = protected
        self.known_org_names = [account.name for account in result.orgs] if self.include_orgs_var.get() else []

        self._refresh_protected_label()
        prefix = "查询已取消" if result.cancelled else "查询完成"
        status = f"{prefix}：用户 {len(self.users)} 个，组织 {len(self.orgs)} 个，仓库 {len(self.repos)} 个。"
        if result.saturated:
            status += " 有搜索结果达到上限，请提高每次上限。"
        self._finish(status)
        self._log(status)

    def _insert_account(self, kind: str, account: Account, repos: Sequence[Repository]) -> None:
        tree = self.user_tree if kind == "user" else self.org_tree
        store = self.users if kind == "user" else self.orgs
        key = account.name.lower()
        reason = self._protected_reason(account.name)
        row = store.get(key, {})
        checked = bool(row.get("checked")) if row else False
        store[key] = {
            "name": account.name,
            "id": account.id,
            "repos": len(repos),
            "checked": checked,
            "protected": bool(reason),
        }
        values = (CHECKED if checked else UNCHECKED, account.name, account.id or "", len(repos), account.email, account.full_name, reason)
        tags = ("protected",) if reason else (("checked",) if checked else ())
        if tree.exists(key):
            tree.item(key, values=values, tags=tags)
        else:
            tree.insert("", "end", iid=key, values=values, tags=tags)

    def _update_account_repo_count(self, kind: str, name: str, count: int) -> None:
        tree = self.user_tree if kind == "user" else self.org_tree
        store = self.users if kind == "user" else self.orgs
        key = name.lower()
        row = store.get(key)
        if not row or not tree.exists(key):
            return
        row["repos"] = count
        values = list(tree.item(key, "values"))
        values[3] = count
        tree.item(key, values=values)

    def _insert_repos(self, repo_map: Dict[str, List[Repository]]) -> None:
        for owner in sorted(repo_map, key=str.lower):
            for repo in sorted(repo_map[owner], key=lambda item: item.name.lower()):
                key = f"{repo.owner.lower()}/{repo.name.lower()}"
                if key in self.repos:
                    continue
                reason = self._repo_protected_reason(repo.owner)
                visibility = "" if repo.private is None else ("私有" if repo.private else "公开")
                self.repos[key] = {
                    "owner": repo.owner,
                    "name": repo.name,
                    "checked": False,
                    "protected": bool(reason),
                }
                tags = ("protected",) if reason else ()
                self.repo_tree.insert("", "end", iid=key, values=(UNCHECKED, repo.owner, repo.name, visibility, reason), tags=tags)

    def _clear_tree(self, tree: ttk.Treeview) -> None:
        for item in tree.get_children():
            tree.delete(item)

    def _tree_click(self, event: tk.Event, tree: ttk.Treeview, kind: str) -> None:
        if tree.identify_region(event.x, event.y) == "cell" and tree.identify_column(event.x) == "#1":
            row_id = tree.identify_row(event.y)
            if row_id:
                self._toggle_row(kind, row_id)

    def _toggle_selected_event(self, _event: tk.Event, tree: ttk.Treeview, kind: str) -> str:
        for row_id in tree.selection():
            self._toggle_row(kind, row_id)
        return "break"

    def _toggle_row(self, kind: str, key: str) -> None:
        store, tree = self._store_tree(kind)
        row = store.get(key)
        if not row or row.get("protected"):
            return
        row["checked"] = not bool(row["checked"])
        self._update_check(tree, key, bool(row["checked"]))

    def _store_tree(self, kind: str) -> Tuple[Dict[str, Dict[str, object]], ttk.Treeview]:
        if kind == "user":
            return self.users, self.user_tree
        if kind == "org":
            return self.orgs, self.org_tree
        return self.repos, self.repo_tree

    def _update_check(self, tree: ttk.Treeview, key: str, checked: bool) -> None:
        values = list(tree.item(key, "values"))
        values[0] = CHECKED if checked else UNCHECKED
        tree.item(key, values=values)
        row = self._row_for_tree(tree, key)
        if row and row.get("protected"):
            tree.item(key, tags=("protected",))
        elif checked:
            tree.item(key, tags=("checked",))
        else:
            tree.item(key, tags=())

    def _row_for_tree(self, tree: ttk.Treeview, key: str) -> Optional[Dict[str, object]]:
        if tree == self.user_tree:
            return self.users.get(key)
        if tree == self.org_tree:
            return self.orgs.get(key)
        return self.repos.get(key)

    def _select_deletable(self, kind: str) -> None:
        store, tree = self._store_tree(kind)
        for key, row in store.items():
            row["checked"] = not bool(row.get("protected"))
            self._update_check(tree, key, bool(row["checked"]))

    def _clear_checks(self, kind: str) -> None:
        store, tree = self._store_tree(kind)
        for key, row in store.items():
            row["checked"] = False
            self._update_check(tree, key, False)

    def _invert_checks(self, kind: str) -> None:
        store, tree = self._store_tree(kind)
        for key, row in store.items():
            if not row.get("protected"):
                row["checked"] = not bool(row["checked"])
                self._update_check(tree, key, bool(row["checked"]))

    def add_manual_targets(self, default_kind: str) -> None:
        text = self._ask_manual_targets(default_kind)
        if not text:
            return
        if self.client is None:
            try:
                self.client = self._make_client()
                self.protected = self._load_protected(self.client, strict=False)
            except Exception as exc:
                self._log(f"WARN: cannot refresh protection before manual add: {exc}")
        added = 0
        for line in text.splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            try:
                kind, owner, name = self._parse_manual_target(line, default_kind)
                if kind == "repo":
                    self._add_manual_repo(owner, name)
                else:
                    self._add_manual_account(kind, name)
                added += 1
            except Exception as exc:
                self._log(f"WARN manual target skipped: {line}: {exc}")
        self._refresh_protected_label()
        self.status_var.set(f"手动添加完成：{added} 项")

    def _ask_manual_targets(self, default_kind: str) -> Optional[str]:
        dialog = tk.Toplevel(self)
        dialog.title("手动添加目标")
        dialog.geometry("760x420")
        dialog.transient(self)
        dialog.grab_set()
        dialog.columnconfigure(0, weight=1)
        dialog.rowconfigure(1, weight=1)
        ttk.Label(dialog, text="一行一个目标。支持 repo owner/repo、org name、user name、完整 URL。").grid(row=0, column=0, sticky="ew", padx=10, pady=(10, 6))
        text = tk.Text(dialog, wrap="none")
        text.grid(row=1, column=0, sticky="nsew", padx=10, pady=6)
        if default_kind == "repo":
            text.insert("1.0", "repo data/gogs/data/tmp/local-r/251/c275s1ak")
        result: Dict[str, Optional[str]] = {"value": None}
        buttons = ttk.Frame(dialog)
        buttons.grid(row=2, column=0, sticky="e", padx=10, pady=(6, 10))
        ttk.Button(buttons, text="添加", command=lambda: (result.update(value=text.get("1.0", "end").strip()), dialog.destroy())).pack(side="left", padx=(0, 8))
        ttk.Button(buttons, text="取消", command=dialog.destroy).pack(side="left")
        dialog.wait_window()
        return result["value"]

    def _parse_manual_target(self, raw: str, default_kind: str) -> Tuple[str, str, str]:
        kind = ""
        value = raw.strip()
        parts = value.split(None, 1)
        if len(parts) == 2 and parts[0].lower() in {"repo", "r", "org", "o", "user", "u"}:
            token = parts[0].lower()
            kind = "repo" if token in {"repo", "r"} else ("user" if token in {"user", "u"} else "org")
            value = parts[1].strip()
        value = self._path_from_url(value)
        if not kind:
            kind = "repo" if default_kind == "repo" or "/" in value else default_kind
        if kind == "repo":
            pieces = [piece for piece in value.strip("/").split("/") if piece]
            if len(pieces) < 2:
                raise ValueError("repo target must be owner/repo")
            return "repo", pieces[0], "/".join(pieces[1:])
        name = value.strip("/")
        if not name:
            raise ValueError(f"{kind} target name is required")
        return kind, "", name

    def _path_from_url(self, value: str) -> str:
        value = value.strip().strip("<>").strip()
        parsed = parse.urlparse(value)
        if parsed.scheme and parsed.netloc:
            value = parse.unquote(parsed.path)
            base = parse.urlparse(self.base_url_var.get().strip())
            base_path = parse.unquote(base.path or "").rstrip("/")
            if base_path and value.startswith(base_path + "/"):
                value = value[len(base_path) :]
        else:
            value = parse.unquote(value)
        return value.split("?", 1)[0].split("#", 1)[0].strip("/")

    def _add_manual_account(self, kind: str, name: str) -> None:
        account = Account(name=name)
        self._insert_account(kind, account, [])
        store, tree = self._store_tree(kind)
        key = name.lower()
        if key in store and not store[key].get("protected"):
            store[key]["checked"] = True
            self._update_check(tree, key, True)

    def _add_manual_repo(self, owner: str, name: str) -> None:
        self._insert_repos({owner: [Repository(owner=owner, name=name)]})
        key = f"{owner.lower()}/{name.lower()}"
        if key in self.repos and not self.repos[key].get("protected"):
            self.repos[key]["checked"] = True
            self._update_check(self.repo_tree, key, True)

    def inspect_selected_orgs(self) -> None:
        if self.client is None:
            messagebox.showwarning("未连接", "请先查询或连接测试。")
            return
        selected = [self.orgs[key]["name"] for key in self.org_tree.selection() if key in self.orgs]
        if not selected:
            messagebox.showwarning("未选择组织", "请选择一个或多个组织。")
            return
        self._run_worker(lambda: self._inspect_orgs_worker([str(name) for name in selected]), "正在查询组织成员...")

    def _inspect_orgs_worker(self, orgs: Sequence[str]) -> None:
        client = self._make_client()
        for org in orgs:
            self.queue.put(("log", f"ORG {org}"))
            try:
                members = client.org_members(org)
                self.queue.put(("log", "  members: " + ", ".join(member.name for member in members)))
            except Exception as exc:
                self.queue.put(("log", f"  members error: {exc}"))
            try:
                teams = client.org_teams(org)
                for team in teams:
                    try:
                        members = client.team_members(team.id)
                        self.queue.put(("log", f"  team {team.id}/{team.name}: " + ", ".join(member.name for member in members)))
                    except Exception as exc:
                        self.queue.put(("log", f"  team {team.id}/{team.name} error: {exc}"))
            except Exception as exc:
                self.queue.put(("log", f"  teams error: {exc}"))
        self.queue.put(("delete_done", (0, [], [])))

    def delete_checked(self) -> None:
        if self.worker_running:
            return
        repos = [(str(row["owner"]), str(row["name"])) for row in self.repos.values() if row.get("checked") and not row.get("protected")]
        orgs = [str(row["name"]) for row in self.orgs.values() if row.get("checked") and not row.get("protected")]
        users = [str(row["name"]) for row in self.users.values() if row.get("checked") and not row.get("protected")]
        if not repos and not orgs and not users:
            messagebox.showwarning("没有可删除项", "请至少勾选一个未受保护的用户、组织或仓库。")
            return
        preview = ", ".join([f"{o}/{r}" for o, r in repos] + orgs + users)
        if len(preview) > 400:
            preview = preview[:400] + "..."
        msg = f"将删除仓库 {len(repos)} 个、组织 {len(orgs)} 个、用户 {len(users)} 个。\n\n{preview}\n\n请输入 DELETE 二次确认。"
        if not messagebox.askokcancel("确认删除", msg):
            return
        if simpledialog.askstring("最终确认", "请输入 DELETE：", parent=self) != "DELETE":
            self._log("取消删除：确认文本不匹配")
            return
        self._run_worker(lambda: self._delete_worker(repos, users, orgs), "正在删除勾选项...")

    def _delete_worker(self, repos: Sequence[Tuple[str, str]], users: Sequence[str], orgs: Sequence[str]) -> None:
        client = self._make_client()
        protected = self._load_protected(client, strict=True)
        failures = 0
        deleted_repos: List[Tuple[str, str]] = []
        deleted_accounts: List[str] = []
        for owner, repo in repos:
            if self._delete_repo(client, owner, repo, protected):
                deleted_repos.append((owner, repo))
            else:
                failures += 1
        for org in orgs:
            ok = self._delete_account(client, org, is_org=True, protected=protected)
            deleted_accounts.extend([org] if ok else [])
            failures += 0 if ok else 1
        for user in users:
            ok = self._delete_account(client, user, is_org=False, protected=protected)
            deleted_accounts.extend([user] if ok else [])
            failures += 0 if ok else 1
        self.queue.put(("delete_done", (failures, deleted_repos, deleted_accounts)))

    def _delete_account(self, client: GogsClient, name: str, is_org: bool, protected: Set[str]) -> bool:
        label = "org" if is_org else "user"
        if name.lower() in protected:
            self.queue.put(("log", f"SKIP protected {label} {name}"))
            return True
        if self.purge_repos_var.get() and not self._purge_repos(client, name, is_org, protected):
            self.queue.put(("log", f"SKIP {label} {name}: repo purge failed"))
            return False
        if not is_org and self.remove_org_memberships_var.get():
            self._remove_user_org_memberships(client, name, protected)
        try:
            client.delete_account(name)
            self.queue.put(("log", f"deleted {label} {name}"))
            return True
        except GogsApiError as exc:
            if not is_org and self._is_org_membership_error(exc):
                self._web_remove_user_from_known_orgs(client, name, self._user_id_for(name, exc))
                self._remove_user_org_memberships(client, name, protected)
                try:
                    client.delete_account(name)
                    self.queue.put(("log", f"deleted {label} {name}"))
                    return True
                except GogsApiError as retry_exc:
                    exc = retry_exc
            self.queue.put(("log", f"ERROR deleting {label} {name}: HTTP {exc.status} {exc.safe_body}"))
            return False

    def _purge_repos(self, client: GogsClient, owner: str, is_org: bool, protected: Set[str]) -> bool:
        try:
            repos = client.org_repos(owner) if is_org else client.user_repos(owner)
        except GogsApiError as exc:
            self.queue.put(("log", f"ERROR listing repos for {owner}: HTTP {exc.status} {exc.safe_body}"))
            return False
        ok = True
        for repo in repos:
            if not self._delete_repo(client, repo.owner, repo.name, protected):
                ok = False
        return ok

    def _delete_repo(self, client: GogsClient, owner: str, repo: str, protected: Set[str]) -> bool:
        if owner.lower() in protected:
            self.queue.put(("log", f"SKIP protected repo {owner}/{repo}"))
            return True
        try:
            client.delete_repo(owner, repo)
            self.queue.put(("log", f"deleted repo {owner}/{repo}"))
            return True
        except GogsApiError as exc:
            self.queue.put(("log", f"ERROR deleting repo {owner}/{repo}: HTTP {exc.status} {exc.safe_body}"))
            return False

    def _remove_user_org_memberships(self, client: GogsClient, username: str, protected: Set[str]) -> bool:
        ok = True
        try:
            orgs = client.user_orgs(username)
        except GogsApiError as exc:
            self.queue.put(("log", f"ERROR listing org memberships for {username}: HTTP {exc.status} {exc.safe_body}"))
            return False
        for org in orgs:
            try:
                teams = client.org_teams(org.name)
                for team in teams:
                    try:
                        client.remove_team_member(team.id, username)
                        self.queue.put(("log", f"removed org membership {org.name}/{team.name or team.id}/{username}"))
                    except GogsApiError as exc:
                        ok = False
                        self.queue.put(("log", f"ERROR removing membership {org.name}/{username}: HTTP {exc.status} {exc.safe_body}"))
            except GogsApiError as exc:
                ok = False
                self.queue.put(("log", f"ERROR listing teams for {org.name}: HTTP {exc.status} {exc.safe_body}"))
        return ok

    def _web_remove_user_from_known_orgs(self, client: GogsClient, username: str, user_id: Optional[int]) -> None:
        if user_id is None:
            self.queue.put(("log", f"SKIP web cleanup for {username}: missing uid"))
            return
        if not client.web_username or not client.web_password:
            self.queue.put(("log", f"SKIP web cleanup for {username}: Web 用户/密码未填写"))
            return
        seen: Set[str] = set()
        for org in self.known_org_names + self.current_orgs:
            key = org.lower()
            if not key or key in seen:
                continue
            seen.add(key)
            try:
                client.web_remove_org_member(org, user_id)
                self.queue.put(("log", f"web removed org membership {org}/{username} uid={user_id}"))
            except Exception as exc:
                self.queue.put(("log", f"WARN web cleanup failed for {org}/{username}: {exc}"))

    def _is_org_membership_error(self, exc: GogsApiError) -> bool:
        return exc.status == 422 and "membership of organizations" in exc.body

    def _user_id_for(self, username: str, exc: GogsApiError) -> Optional[int]:
        row = self.users.get(username.lower())
        if row and isinstance(row.get("id"), int):
            return row["id"]  # type: ignore[return-value]
        match = re.search(r"uid:\s*(\d+)", exc.body)
        return int(match.group(1)) if match else None

    def generate_membership_sql(self, include_delete: bool) -> None:
        ids = [int(row["id"]) for row in self.users.values() if row.get("checked") and isinstance(row.get("id"), int)]
        if not ids:
            messagebox.showwarning("没有用户 ID", "请先查询并勾选需要修复的用户。")
            return
        sql = generate_membership_repair_sql(ids, "sqlite", include_delete=include_delete)
        output = os.path.join(os.path.dirname(os.path.abspath(__file__)), "gogs_repair_memberships.sql")
        with open(output, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(sql)
        self.clipboard_clear()
        self.clipboard_append(sql)
        self.repair_text.delete("1.0", "end")
        self.repair_text.insert("1.0", sql)
        self._log(f"generated membership SQL: {output}")
        messagebox.showinfo("SQL 已生成", f"已写入并复制到剪贴板：\n{output}")

    def copy_selected_user_ids(self) -> None:
        ids = [str(row["id"]) for row in self.users.values() if row.get("checked") and isinstance(row.get("id"), int)]
        self.clipboard_clear()
        self.clipboard_append(",".join(ids))
        self.status_var.set("已复制 UID：" + ",".join(ids))

    def _handle_delete_done(self, payload: Tuple[int, List[Tuple[str, str]], List[str]]) -> None:
        failures, deleted_repos, deleted_accounts = payload
        for owner, repo in deleted_repos:
            key = f"{owner.lower()}/{repo.lower()}"
            if key in self.repos:
                self.repo_tree.delete(key)
                del self.repos[key]
        account_names = {name.lower() for name in deleted_accounts}
        for key in list(self.users):
            if key in account_names:
                self.user_tree.delete(key)
                del self.users[key]
        for key in list(self.orgs):
            if key in account_names:
                self.org_tree.delete(key)
                del self.orgs[key]
        for key, row in list(self.repos.items()):
            if str(row["owner"]).lower() in account_names:
                self.repo_tree.delete(key)
                del self.repos[key]
        self._finish(f"删除完成，失败 {failures} 个")
        self._log(self.status_var.get())

    def _protected_reason(self, name: str) -> str:
        lower = name.lower()
        if lower not in self.protected:
            return ""
        if self.current_user and lower == self.current_user.lower():
            return "当前 token 用户"
        if any(lower == org.lower() for org in self.current_orgs):
            return "当前用户加入的组织"
        if lower in DEFAULT_PROTECTED:
            return "内置保护名"
        return "额外保护名"

    def _repo_protected_reason(self, owner: str) -> str:
        reason = self._protected_reason(owner)
        return f"归属受保护：{reason}" if reason else ""

    def _refresh_protected_label(self) -> None:
        bits = sorted(self.protected)
        extra = []
        if self.current_user:
            extra.append(f"current={self.current_user}")
        if self.current_orgs:
            extra.append("orgs=" + ",".join(sorted(self.current_orgs)))
        suffix = " | " + " | ".join(extra) if extra else ""
        self.protected_var.set("Protected: " + ", ".join(bits) + suffix)

    def _finish(self, status: str) -> None:
        if self.current_operation == "query":
            self.query_cancel_event.clear()
            self._set_query_cancel_enabled(False)
        self.current_operation = ""
        self.worker_running = False
        self.configure(cursor="")
        self.status_var.set(status)

    def _log(self, text: object) -> None:
        self.log.insert("end", redact_sensitive(text) + "\n")
        self.log.see("end")


def main() -> None:
    app = GogsAdminConsole()
    app.mainloop()


if __name__ == "__main__":
    main()