proxy-god/proxy_chain_manager/proxy_picker.py

"""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)