605 lines
18 KiB
Python
605 lines
18 KiB
Python
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import json
|
||
import logging
|
||
import os
|
||
import psutil
|
||
import sys
|
||
import threading
|
||
import time
|
||
import webbrowser
|
||
import asyncio as _asyncio
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
|
||
try:
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
except ImportError:
|
||
Image = ImageDraw = ImageFont = None # type: ignore
|
||
|
||
try:
|
||
import pystray
|
||
except ImportError:
|
||
pystray = None # type: ignore
|
||
|
||
try:
|
||
import customtkinter as ctk
|
||
except ImportError:
|
||
ctk = None # type: ignore
|
||
|
||
# Proxy engine
|
||
import tg_ws_proxy
|
||
|
||
|
||
APP_NAME = "TgWsProxy"
|
||
APP_DIR = Path(os.environ.get("APPDATA", Path.home())) / APP_NAME
|
||
CONFIG_FILE = APP_DIR / "config.json"
|
||
LOG_FILE = APP_DIR / "proxy.log"
|
||
FIRST_RUN_MARKER = APP_DIR / ".first_run_done"
|
||
|
||
|
||
DEFAULT_CONFIG = {
|
||
"port": 1080,
|
||
"dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
|
||
"verbose": False,
|
||
}
|
||
|
||
|
||
_proxy_thread: Optional[threading.Thread] = None
|
||
_stop_event: Optional[threading.Event] = None
|
||
_async_stop: Optional[object] = None
|
||
_tray_icon: Optional[object] = None
|
||
_config: dict = {}
|
||
_exiting: bool = False
|
||
|
||
log = logging.getLogger("tg-ws-tray")
|
||
|
||
|
||
def is_already_running():
|
||
current_proc = os.path.basename(sys.argv[0])
|
||
count = 0
|
||
for process in psutil.process_iter(['name']):
|
||
if process.info['name'] == current_proc:
|
||
count += 1
|
||
return count > 2
|
||
|
||
|
||
def _ensure_dirs():
|
||
APP_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def load_config() -> dict:
|
||
_ensure_dirs()
|
||
if CONFIG_FILE.exists():
|
||
try:
|
||
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
# Merge with defaults for missing keys
|
||
for k, v in DEFAULT_CONFIG.items():
|
||
data.setdefault(k, v)
|
||
return data
|
||
except Exception as exc:
|
||
log.warning("Failed to load config: %s", exc)
|
||
return dict(DEFAULT_CONFIG)
|
||
|
||
|
||
def save_config(cfg: dict):
|
||
_ensure_dirs()
|
||
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(cfg, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
def setup_logging(verbose: bool = False):
|
||
_ensure_dirs()
|
||
root = logging.getLogger()
|
||
root.setLevel(logging.DEBUG if verbose else logging.INFO)
|
||
|
||
fh = logging.FileHandler(str(LOG_FILE), encoding="utf-8")
|
||
fh.setLevel(logging.DEBUG)
|
||
fh.setFormatter(logging.Formatter(
|
||
"%(asctime)s %(levelname)-5s %(name)s %(message)s",
|
||
datefmt="%Y-%m-%d %H:%M:%S"))
|
||
root.addHandler(fh)
|
||
|
||
if not getattr(sys, "frozen", False):
|
||
ch = logging.StreamHandler(sys.stdout)
|
||
ch.setLevel(logging.DEBUG if verbose else logging.INFO)
|
||
ch.setFormatter(logging.Formatter(
|
||
"%(asctime)s %(levelname)-5s %(message)s",
|
||
datefmt="%H:%M:%S"))
|
||
root.addHandler(ch)
|
||
|
||
|
||
def _make_icon_image(size: int = 64):
|
||
"""Create a simple tray icon: blue circle with a white 'T' letter."""
|
||
if Image is None:
|
||
raise RuntimeError("Pillow is required for tray icon")
|
||
img = Image.new("RGBA", (size, size), (0, 0, 0, 0))
|
||
draw = ImageDraw.Draw(img)
|
||
|
||
# Blue circle
|
||
margin = 2
|
||
draw.ellipse([margin, margin, size - margin, size - margin],
|
||
fill=(0, 136, 204, 255))
|
||
|
||
# White "T"
|
||
try:
|
||
font = ImageFont.truetype("arial.ttf", size=int(size * 0.55))
|
||
except Exception:
|
||
font = ImageFont.load_default()
|
||
bbox = draw.textbbox((0, 0), "T", font=font)
|
||
tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
|
||
tx = (size - tw) // 2 - bbox[0]
|
||
ty = (size - th) // 2 - bbox[1]
|
||
draw.text((tx, ty), "T", fill=(255, 255, 255, 255), font=font)
|
||
|
||
return img
|
||
|
||
|
||
def _load_icon():
|
||
"""Load icon from file or generate one."""
|
||
icon_path = Path(__file__).parent / "icon.ico"
|
||
if icon_path.exists() and Image:
|
||
try:
|
||
return Image.open(str(icon_path))
|
||
except Exception:
|
||
pass
|
||
return _make_icon_image()
|
||
|
||
|
||
|
||
def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool):
|
||
"""Target for the proxy thread — runs asyncio event loop."""
|
||
global _async_stop
|
||
loop = _asyncio.new_event_loop()
|
||
_asyncio.set_event_loop(loop)
|
||
stop_ev = _asyncio.Event()
|
||
_async_stop = (loop, stop_ev)
|
||
|
||
try:
|
||
loop.run_until_complete(
|
||
tg_ws_proxy._run(port, dc_opt, stop_event=stop_ev))
|
||
except Exception as exc:
|
||
log.error("Proxy thread crashed: %s", exc)
|
||
finally:
|
||
loop.close()
|
||
_async_stop = None
|
||
|
||
|
||
def start_proxy():
|
||
global _proxy_thread, _config
|
||
if _proxy_thread and _proxy_thread.is_alive():
|
||
log.info("Proxy already running")
|
||
return
|
||
|
||
cfg = _config
|
||
port = cfg.get("port", DEFAULT_CONFIG["port"])
|
||
dc_ip_list = cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])
|
||
verbose = cfg.get("verbose", False)
|
||
|
||
try:
|
||
dc_opt = tg_ws_proxy.parse_dc_ip_list(dc_ip_list)
|
||
except ValueError as e:
|
||
log.error("Bad config dc_ip: %s", e)
|
||
_show_error(f"Ошибка конфигурации:\n{e}")
|
||
return
|
||
|
||
log.info("Starting proxy on port %d ...", port)
|
||
_proxy_thread = threading.Thread(
|
||
target=_run_proxy_thread,
|
||
args=(port, dc_opt, verbose),
|
||
daemon=True, name="proxy")
|
||
_proxy_thread.start()
|
||
|
||
|
||
def stop_proxy():
|
||
global _proxy_thread, _async_stop
|
||
if _async_stop:
|
||
loop, stop_ev = _async_stop
|
||
loop.call_soon_threadsafe(stop_ev.set)
|
||
if _proxy_thread:
|
||
_proxy_thread.join(timeout=2)
|
||
_proxy_thread = None
|
||
log.info("Proxy stopped")
|
||
|
||
|
||
def restart_proxy():
|
||
log.info("Restarting proxy...")
|
||
stop_proxy()
|
||
time.sleep(0.3)
|
||
start_proxy()
|
||
|
||
|
||
def _show_error(text: str, title: str = "TG WS Proxy — Ошибка"):
|
||
ctypes.windll.user32.MessageBoxW(0, text, title, 0x10)
|
||
|
||
|
||
def _show_info(text: str, title: str = "TG WS Proxy"):
|
||
ctypes.windll.user32.MessageBoxW(0, text, title, 0x40)
|
||
|
||
|
||
def _on_open_in_telegram(icon=None, item=None):
|
||
port = _config.get("port", DEFAULT_CONFIG["port"])
|
||
url = f"tg://socks?server=127.0.0.1&port={port}"
|
||
log.info("Opening %s", url)
|
||
try:
|
||
result = webbrowser.open(url)
|
||
if not result:
|
||
raise RuntimeError("webbrowser.open returned False")
|
||
except Exception:
|
||
log.info("Browser open failed, copying to clipboard")
|
||
try:
|
||
_copy_to_clipboard(url)
|
||
_show_info(
|
||
f"Не удалось открыть Telegram автоматически.\n\n"
|
||
f"Ссылка скопирована в буфер обмена, отправьте её в телеграмм и нажмите по ней ЛКМ:\n{url}",
|
||
"TG WS Proxy")
|
||
except Exception as exc:
|
||
log.error("Clipboard copy failed: %s", exc)
|
||
_show_error(f"Не удалось скопировать ссылку:\n{exc}")
|
||
|
||
|
||
def _copy_to_clipboard(text: str):
|
||
"""Copy text to Windows clipboard using ctypes."""
|
||
import ctypes.wintypes
|
||
CF_UNICODETEXT = 13
|
||
kernel32 = ctypes.windll.kernel32
|
||
user32 = ctypes.windll.user32
|
||
|
||
user32.OpenClipboard(0)
|
||
user32.EmptyClipboard()
|
||
|
||
encoded = text.encode("utf-16-le") + b"\x00\x00"
|
||
h = kernel32.GlobalAlloc(0x0042, len(encoded)) # GMEM_MOVEABLE | GMEM_ZEROINIT
|
||
p = kernel32.GlobalLock(h)
|
||
ctypes.memmove(p, encoded, len(encoded))
|
||
kernel32.GlobalUnlock(h)
|
||
user32.SetClipboardData(CF_UNICODETEXT, h)
|
||
user32.CloseClipboard()
|
||
|
||
|
||
def _on_restart(icon=None, item=None):
|
||
threading.Thread(target=restart_proxy, daemon=True).start()
|
||
|
||
|
||
def _on_edit_config(icon=None, item=None):
|
||
"""Open a simple dialog to edit config."""
|
||
threading.Thread(target=_edit_config_dialog, daemon=True).start()
|
||
|
||
|
||
def _edit_config_dialog():
|
||
if ctk is None:
|
||
_show_error("customtkinter не установлен.")
|
||
return
|
||
|
||
cfg = dict(_config)
|
||
|
||
ctk.set_appearance_mode("light")
|
||
ctk.set_default_color_theme("blue")
|
||
|
||
root = ctk.CTk()
|
||
root.title("TG WS Proxy — Настройки")
|
||
root.resizable(False, False)
|
||
root.attributes("-topmost", True)
|
||
|
||
TG_BLUE = "#3390ec"
|
||
TG_BLUE_HOVER = "#2b7cd4"
|
||
BG = "#ffffff"
|
||
FIELD_BG = "#f0f2f5"
|
||
FIELD_BORDER = "#d6d9dc"
|
||
TEXT_PRIMARY = "#000000"
|
||
TEXT_SECONDARY = "#707579"
|
||
FONT_FAMILY = "Segoe UI"
|
||
|
||
w, h = 420, 400
|
||
sw = root.winfo_screenwidth()
|
||
sh = root.winfo_screenheight()
|
||
root.geometry(f"{w}x{h}+{(sw-w)//2}+{(sh-h)//2}")
|
||
root.configure(fg_color=BG)
|
||
|
||
frame = ctk.CTkFrame(root, fg_color=BG, corner_radius=0)
|
||
frame.pack(fill="both", expand=True, padx=24, pady=20)
|
||
|
||
# Port
|
||
ctk.CTkLabel(frame, text="Порт прокси",
|
||
font=(FONT_FAMILY, 13), text_color=TEXT_PRIMARY,
|
||
anchor="w").pack(anchor="w", pady=(0, 4))
|
||
port_var = ctk.StringVar(value=str(cfg.get("port", 1080)))
|
||
port_entry = ctk.CTkEntry(frame, textvariable=port_var, width=120, height=36,
|
||
font=(FONT_FAMILY, 13), corner_radius=10,
|
||
fg_color=FIELD_BG, border_color=FIELD_BORDER,
|
||
border_width=1, text_color=TEXT_PRIMARY)
|
||
port_entry.pack(anchor="w", pady=(0, 12))
|
||
|
||
# DC-IP mappings
|
||
ctk.CTkLabel(frame, text="DC → IP маппинги (по одному на строку, формат DC:IP)",
|
||
font=(FONT_FAMILY, 13), text_color=TEXT_PRIMARY,
|
||
anchor="w").pack(anchor="w", pady=(0, 4))
|
||
dc_textbox = ctk.CTkTextbox(frame, width=370, height=120,
|
||
font=("Consolas", 12), corner_radius=10,
|
||
fg_color=FIELD_BG, border_color=FIELD_BORDER,
|
||
border_width=1, text_color=TEXT_PRIMARY)
|
||
dc_textbox.pack(anchor="w", pady=(0, 12))
|
||
dc_textbox.insert("1.0", "\n".join(cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])))
|
||
|
||
# Verbose
|
||
verbose_var = ctk.BooleanVar(value=cfg.get("verbose", False))
|
||
ctk.CTkCheckBox(frame, text="Подробное логирование (verbose)",
|
||
variable=verbose_var, font=(FONT_FAMILY, 13),
|
||
text_color=TEXT_PRIMARY,
|
||
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
|
||
corner_radius=6, border_width=2,
|
||
border_color=FIELD_BORDER).pack(anchor="w", pady=(0, 8))
|
||
|
||
# Info label
|
||
ctk.CTkLabel(frame, text="Изменения вступят в силу после перезапуска прокси.",
|
||
font=(FONT_FAMILY, 11), text_color=TEXT_SECONDARY,
|
||
anchor="w").pack(anchor="w", pady=(0, 16))
|
||
|
||
def on_save():
|
||
try:
|
||
port_val = int(port_var.get().strip())
|
||
if not (1 <= port_val <= 65535):
|
||
raise ValueError
|
||
except ValueError:
|
||
_show_error("Порт должен быть числом 1-65535")
|
||
return
|
||
|
||
lines = [l.strip() for l in dc_textbox.get("1.0", "end").strip().splitlines()
|
||
if l.strip()]
|
||
try:
|
||
tg_ws_proxy.parse_dc_ip_list(lines)
|
||
except ValueError as e:
|
||
_show_error(str(e))
|
||
return
|
||
|
||
new_cfg = {
|
||
"port": port_val,
|
||
"dc_ip": lines,
|
||
"verbose": verbose_var.get(),
|
||
}
|
||
save_config(new_cfg)
|
||
_config.update(new_cfg)
|
||
log.info("Config saved: %s", new_cfg)
|
||
|
||
from tkinter import messagebox
|
||
if messagebox.askyesno("Перезапустить?",
|
||
"Настройки сохранены.\n\n"
|
||
"Перезапустить прокси сейчас?",
|
||
parent=root):
|
||
root.destroy()
|
||
restart_proxy()
|
||
else:
|
||
root.destroy()
|
||
|
||
def on_cancel():
|
||
root.destroy()
|
||
|
||
btn_frame = ctk.CTkFrame(frame, fg_color="transparent")
|
||
btn_frame.pack(fill="x")
|
||
ctk.CTkButton(btn_frame, text="Сохранить", width=140, height=38,
|
||
font=(FONT_FAMILY, 14, "bold"), corner_radius=10,
|
||
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
|
||
text_color="#ffffff",
|
||
command=on_save).pack(side="left", padx=(0, 10))
|
||
ctk.CTkButton(btn_frame, text="Отмена", width=140, height=38,
|
||
font=(FONT_FAMILY, 14), corner_radius=10,
|
||
fg_color=FIELD_BG, hover_color=FIELD_BORDER,
|
||
text_color=TEXT_PRIMARY, border_width=1,
|
||
border_color=FIELD_BORDER,
|
||
command=on_cancel).pack(side="left")
|
||
|
||
root.mainloop()
|
||
|
||
|
||
def _on_open_logs(icon=None, item=None):
|
||
log.info("Opening log file: %s", LOG_FILE)
|
||
if LOG_FILE.exists():
|
||
os.startfile(str(LOG_FILE))
|
||
else:
|
||
_show_info("Файл логов ещё не создан.", "TG WS Proxy")
|
||
|
||
|
||
def _on_exit(icon=None, item=None):
|
||
global _exiting
|
||
if _exiting:
|
||
os._exit(0)
|
||
return
|
||
_exiting = True
|
||
log.info("User requested exit")
|
||
|
||
def _force_exit():
|
||
time.sleep(3)
|
||
os._exit(0)
|
||
threading.Thread(target=_force_exit, daemon=True, name="force-exit").start()
|
||
|
||
if icon:
|
||
icon.stop()
|
||
|
||
|
||
|
||
def _show_first_run():
|
||
_ensure_dirs()
|
||
if FIRST_RUN_MARKER.exists():
|
||
return
|
||
|
||
port = _config.get("port", DEFAULT_CONFIG["port"])
|
||
tg_url = f"tg://socks?server=127.0.0.1&port={port}"
|
||
|
||
if ctk is None:
|
||
FIRST_RUN_MARKER.touch()
|
||
return
|
||
|
||
ctk.set_appearance_mode("light")
|
||
ctk.set_default_color_theme("blue")
|
||
|
||
TG_BLUE = "#3390ec"
|
||
TG_BLUE_HOVER = "#2b7cd4"
|
||
BG = "#ffffff"
|
||
FIELD_BG = "#f0f2f5"
|
||
FIELD_BORDER = "#d6d9dc"
|
||
TEXT_PRIMARY = "#000000"
|
||
TEXT_SECONDARY = "#707579"
|
||
FONT_FAMILY = "Segoe UI"
|
||
|
||
root = ctk.CTk()
|
||
root.title("TG WS Proxy")
|
||
root.resizable(False, False)
|
||
root.attributes("-topmost", True)
|
||
|
||
w, h = 520, 440
|
||
sw = root.winfo_screenwidth()
|
||
sh = root.winfo_screenheight()
|
||
root.geometry(f"{w}x{h}+{(sw-w)//2}+{(sh-h)//2}")
|
||
root.configure(fg_color=BG)
|
||
|
||
frame = ctk.CTkFrame(root, fg_color=BG, corner_radius=0)
|
||
frame.pack(fill="both", expand=True, padx=28, pady=24)
|
||
|
||
title_frame = ctk.CTkFrame(frame, fg_color="transparent")
|
||
title_frame.pack(anchor="w", pady=(0, 16), fill="x")
|
||
|
||
# Blue accent bar
|
||
accent_bar = ctk.CTkFrame(title_frame, fg_color=TG_BLUE,
|
||
width=4, height=32, corner_radius=2)
|
||
accent_bar.pack(side="left", padx=(0, 12))
|
||
|
||
ctk.CTkLabel(title_frame, text="Прокси запущен и работает в системном трее",
|
||
font=(FONT_FAMILY, 17, "bold"),
|
||
text_color=TEXT_PRIMARY).pack(side="left")
|
||
|
||
# Info sections
|
||
sections = [
|
||
("Как подключить Telegram Desktop:", True),
|
||
(" Автоматически:", True),
|
||
(f" ПКМ по иконке в трее → «Открыть в Telegram»", False),
|
||
(f" Или ссылка: {tg_url}", False),
|
||
("\n Вручную:", True),
|
||
(" Настройки → Продвинутые → Тип подключения → Прокси", False),
|
||
(f" SOCKS5 → 127.0.0.1 : {port} (без логина/пароля)", False),
|
||
]
|
||
|
||
for text, bold in sections:
|
||
weight = "bold" if bold else "normal"
|
||
ctk.CTkLabel(frame, text=text,
|
||
font=(FONT_FAMILY, 13, weight),
|
||
text_color=TEXT_PRIMARY,
|
||
anchor="w", justify="left").pack(anchor="w", pady=1)
|
||
|
||
# Spacer
|
||
ctk.CTkFrame(frame, fg_color="transparent", height=16).pack()
|
||
|
||
# Separator
|
||
ctk.CTkFrame(frame, fg_color=FIELD_BORDER, height=1,
|
||
corner_radius=0).pack(fill="x", pady=(0, 12))
|
||
|
||
# Checkbox
|
||
auto_var = ctk.BooleanVar(value=True)
|
||
ctk.CTkCheckBox(frame, text="Открыть прокси в Telegram сейчас",
|
||
variable=auto_var, font=(FONT_FAMILY, 13),
|
||
text_color=TEXT_PRIMARY,
|
||
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
|
||
corner_radius=6, border_width=2,
|
||
border_color=FIELD_BORDER).pack(anchor="w", pady=(0, 16))
|
||
|
||
def on_ok():
|
||
FIRST_RUN_MARKER.touch()
|
||
open_tg = auto_var.get()
|
||
root.destroy()
|
||
if open_tg:
|
||
_on_open_in_telegram()
|
||
|
||
ctk.CTkButton(frame, text="Начать", width=180, height=42,
|
||
font=(FONT_FAMILY, 15, "bold"), corner_radius=10,
|
||
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
|
||
text_color="#ffffff",
|
||
command=on_ok).pack(pady=(0, 0))
|
||
|
||
root.protocol("WM_DELETE_WINDOW", on_ok)
|
||
root.mainloop()
|
||
|
||
|
||
def _build_menu():
|
||
if pystray is None:
|
||
return None
|
||
port = _config.get("port", DEFAULT_CONFIG["port"])
|
||
return pystray.Menu(
|
||
pystray.MenuItem(
|
||
f"Открыть в Telegram (:{port})",
|
||
_on_open_in_telegram,
|
||
default=True),
|
||
pystray.Menu.SEPARATOR,
|
||
pystray.MenuItem("Перезапустить прокси", _on_restart),
|
||
pystray.MenuItem("Настройки...", _on_edit_config),
|
||
pystray.MenuItem("Открыть логи", _on_open_logs),
|
||
pystray.Menu.SEPARATOR,
|
||
pystray.MenuItem("Выход", _on_exit),
|
||
)
|
||
|
||
|
||
def run_tray():
|
||
global _tray_icon, _config
|
||
|
||
_config = load_config()
|
||
save_config(_config)
|
||
|
||
if LOG_FILE.exists():
|
||
try:
|
||
LOG_FILE.unlink()
|
||
except Exception:
|
||
pass
|
||
|
||
setup_logging(_config.get("verbose", False))
|
||
log.info("TG WS Proxy tray app starting")
|
||
log.info("Config: %s", _config)
|
||
log.info("Log file: %s", LOG_FILE)
|
||
|
||
if pystray is None or Image is None:
|
||
log.error("pystray or Pillow not installed; "
|
||
"running in console mode")
|
||
start_proxy()
|
||
try:
|
||
while True:
|
||
time.sleep(1)
|
||
except KeyboardInterrupt:
|
||
stop_proxy()
|
||
return
|
||
|
||
start_proxy()
|
||
|
||
_show_first_run()
|
||
|
||
icon_image = _load_icon()
|
||
_tray_icon = pystray.Icon(
|
||
APP_NAME,
|
||
icon_image,
|
||
"TG WS Proxy",
|
||
menu=_build_menu())
|
||
|
||
log.info("Tray icon running")
|
||
_tray_icon.run()
|
||
|
||
stop_proxy()
|
||
log.info("Tray app exited")
|
||
|
||
|
||
def main():
|
||
if is_already_running():
|
||
_show_info("Приложение уже запущено.", os.path.basename(sys.argv[0]))
|
||
return
|
||
|
||
# Hide console window if running as frozen exe
|
||
if getattr(sys, "frozen", False):
|
||
try:
|
||
ctypes.windll.user32.ShowWindow(
|
||
ctypes.windll.kernel32.GetConsoleWindow(), 0)
|
||
except Exception:
|
||
pass
|
||
|
||
run_tray()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|