"""Modal dialog: random batch from cached sources, multi-select.""" from __future__ import annotations import threading from typing import Any, Callable import customtkinter as ctk from .config import Settings, redact_proxy_url from .pool_ops import PICKER_DISPLAY_CAP, ProxySourceCache BG = "#0d0d1a" CARD = "#12122a" PANEL = "#1a1a3e" ACCENT = "#1e3a8a" ACCENT2 = "#2563eb" GREEN = "#00e676" RED = "#ff1744" YELLOW = "#ffab00" DIM = "#4a5568" TEXT = "#e2e8f0" TEXT2 = "#94a3b8" FONT = "Segoe UI" _UI_CHUNK = 50 # One cache per app session — avoids re-downloading 800+ URLs every open _SESSION_CACHE = ProxySourceCache() def show_proxy_picker( parent: Any, settings: Settings, prefer_elite: bool, on_done: Callable[[list[str], str], None], ) -> None: """``on_done(urls, mode)`` where mode is ``append`` or ``replace``.""" win = ctk.CTkToplevel(parent) win.title("Browse proxy lists") win.geometry("700x500") win.configure(fg_color=BG) win.transient(parent) win.grab_set() status = ctk.CTkLabel( win, text=f"Pull loads sources once, then shows {PICKER_DISPLAY_CAP} random proxies per batch.", font=(FONT, 11), text_color=TEXT2, ) status.pack(fill="x", padx=12, pady=(10, 4)) filter_var = ctk.StringVar(value="") filt_row = ctk.CTkFrame(win, fg_color="transparent") filt_row.pack(fill="x", padx=12, pady=4) ctk.CTkLabel(filt_row, text="Filter:", font=(FONT, 10), text_color=TEXT2).pack(side="left") ctk.CTkEntry( filt_row, textvariable=filter_var, width=180, height=28, fg_color=BG, border_color=ACCENT, ).pack(side="left", padx=6) proto_var = ctk.StringVar(value="all") for label, val in [("All", "all"), ("HTTP", "http"), ("SOCKS5", "socks5")]: ctk.CTkRadioButton( filt_row, text=label, variable=proto_var, value=val, font=(FONT, 10), fg_color=ACCENT2, text_color=TEXT, ).pack(side="left", padx=4) scroll = ctk.CTkScrollableFrame(win, fg_color=CARD, height=300, scrollbar_button_color=ACCENT) scroll.pack(fill="both", expand=True, padx=12, pady=6) display_proxies: list[str] = [] check_vars: dict[str, ctk.BooleanVar] = {} _build_gen = [0] def _filtered_urls() -> list[str]: q = filter_var.get().strip().lower() pv = proto_var.get() out: list[str] = [] for u in display_proxies: if pv == "http" and not u.startswith("http://"): continue if pv == "socks5" and not u.startswith("socks5://"): continue if q and q not in u.lower(): continue out.append(u) return out def _rebuild_list() -> None: _build_gen[0] += 1 gen = _build_gen[0] for w in scroll.winfo_children(): w.destroy() urls = _filtered_urls() if not urls: status.configure( text="No proxies match filter — pull a batch or change filter.", text_color=TEXT2, ) return def _chunk(start: int) -> None: if gen != _build_gen[0]: return end = min(start + _UI_CHUNK, len(urls)) for u in urls[start:end]: if u not in check_vars: check_vars[u] = ctk.BooleanVar(value=False) fr = ctk.CTkFrame(scroll, fg_color=PANEL, corner_radius=4) fr.pack(fill="x", pady=1) ctk.CTkCheckBox( fr, text=redact_proxy_url(u), variable=check_vars[u], font=("Consolas", 10), fg_color=ACCENT2, text_color=TEXT, ).pack(anchor="w", padx=8, pady=2) if end < len(urls): win.after(1, lambda: _chunk(end)) else: cached = _SESSION_CACHE.total_cached status.configure( text=f"Showing {len(urls)} of batch ({cached} cached in memory). Select → OK.", text_color=GREEN, ) _chunk(0) def _select_all(on: bool) -> None: for u in _filtered_urls(): if u in check_vars: check_vars[u].set(on) def _pull(force_reload: bool = False) -> None: pull_btn.configure(state="disabled") reload_btn.configure(state="disabled") status.configure(text="Loading sources…", text_color=YELLOW) def work() -> None: def stat(msg: str) -> None: win.after(0, lambda m=msg: status.configure(text=m, text_color=TEXT2)) if force_reload: _SESSION_CACHE.clear() if not _SESSION_CACHE.loaded: _SESSION_CACHE.load(list(settings.sources), prefer_elite, on_status=stat) batch = _SESSION_CACHE.random_batch(PICKER_DISPLAY_CAP) def finish() -> None: nonlocal display_proxies display_proxies = batch check_vars.clear() for u in batch: check_vars[u] = ctk.BooleanVar(value=False) pull_btn.configure(state="normal") reload_btn.configure(state="normal") _rebuild_list() win.after(0, finish) threading.Thread(target=work, daemon=True).start() def _selected() -> list[str]: return [u for u, v in check_vars.items() if v.get()] def _ok(mode: str) -> None: sel = _selected() if not sel: status.configure(text="Select at least one proxy.", text_color=RED) return win.grab_release() win.destroy() on_done(sel, mode) btn_row = ctk.CTkFrame(win, fg_color="transparent") btn_row.pack(fill="x", padx=12, pady=(4, 12)) pull_btn = ctk.CTkButton( btn_row, text="↻ Random batch", command=lambda: _pull(False), width=120, fg_color=ACCENT, hover_color=ACCENT2, ) pull_btn.pack(side="left", padx=2) reload_btn = ctk.CTkButton( btn_row, text="Reload sources", command=lambda: _pull(True), width=110, fg_color=DIM, hover_color=ACCENT, ) reload_btn.pack(side="left", padx=2) ctk.CTkButton(btn_row, text="Select all", command=lambda: _select_all(True), width=72, fg_color=DIM).pack(side="left", padx=2) ctk.CTkButton(btn_row, text="Clear", command=lambda: _select_all(False), width=52, fg_color=DIM).pack(side="left", padx=2) ctk.CTkButton(btn_row, text="Cancel", command=lambda: (win.grab_release(), win.destroy()), width=64, fg_color="#7f1d1d").pack(side="right", padx=2) ctk.CTkButton(btn_row, text="Add to chain", command=lambda: _ok("append"), width=100, fg_color=GREEN, hover_color=ACCENT2).pack(side="right", padx=2) ctk.CTkButton(btn_row, text="Replace chain", command=lambda: _ok("replace"), width=100, fg_color=ACCENT2).pack(side="right", padx=2) filter_var.trace_add("write", lambda *_: _rebuild_list()) proto_var.trace_add("write", lambda *_: _rebuild_list()) _pull(False)