From ec70188385c6be5c51166c354362a9f81c7c5af4 Mon Sep 17 00:00:00 2001
From: Dark-Avery <ingv0rr@yandex.ru>
Date: Mon, 16 Mar 2026 17:22:51 +0300
Subject: [PATCH] refactor(core): extract cross-platform proxy runtime from
 windows app

---
 proxy/app_runtime.py      | 170 ++++++++++++++++++++++++++++++++++++++
 tests/test_app_runtime.py | 121 +++++++++++++++++++++++++++
 windows.py                | 134 +++++-------------------------
 3 files changed, 310 insertions(+), 115 deletions(-)
 create mode 100644 proxy/app_runtime.py
 create mode 100644 tests/test_app_runtime.py

diff --git a/proxy/app_runtime.py b/proxy/app_runtime.py
new file mode 100644
index 0000000..3710723
--- /dev/null
+++ b/proxy/app_runtime.py
@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import asyncio as _asyncio
+import json
+import logging
+import sys
+import threading
+import time
+from pathlib import Path
+from typing import Callable, Dict, Optional
+
+import proxy.tg_ws_proxy as tg_ws_proxy
+
+
+DEFAULT_CONFIG = {
+    "port": 1080,
+    "host": "127.0.0.1",
+    "dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
+    "verbose": False,
+}
+
+
+class ProxyAppRuntime:
+    def __init__(self, app_dir: Path,
+                 default_config: Optional[dict] = None,
+                 logger_name: str = "tg-ws-runtime",
+                 on_error: Optional[Callable[[str], None]] = None,
+                 parse_dc_ip_list: Optional[
+                     Callable[[list[str]], Dict[int, str]]
+                 ] = None,
+                 run_proxy: Optional[Callable[..., object]] = None,
+                 thread_factory: Optional[Callable[..., object]] = None):
+        self.app_dir = Path(app_dir)
+        self.config_file = self.app_dir / "config.json"
+        self.log_file = self.app_dir / "proxy.log"
+        self.default_config = dict(default_config or DEFAULT_CONFIG)
+        self.log = logging.getLogger(logger_name)
+        self.on_error = on_error
+        self.parse_dc_ip_list = parse_dc_ip_list or tg_ws_proxy.parse_dc_ip_list
+        self.run_proxy = run_proxy or tg_ws_proxy._run
+        self.thread_factory = thread_factory or threading.Thread
+        self.config: dict = {}
+        self._proxy_thread = None
+        self._async_stop = None
+
+    def ensure_dirs(self):
+        self.app_dir.mkdir(parents=True, exist_ok=True)
+
+    def load_config(self) -> dict:
+        self.ensure_dirs()
+        if self.config_file.exists():
+            try:
+                with open(self.config_file, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                for key, value in self.default_config.items():
+                    data.setdefault(key, value)
+                self.config = data
+                return data
+            except Exception as exc:
+                self.log.warning("Failed to load config: %s", exc)
+
+        self.config = dict(self.default_config)
+        return dict(self.config)
+
+    def save_config(self, cfg: dict):
+        self.ensure_dirs()
+        self.config = dict(cfg)
+        with open(self.config_file, "w", encoding="utf-8") as f:
+            json.dump(cfg, f, indent=2, ensure_ascii=False)
+
+    def reset_log_file(self):
+        if self.log_file.exists():
+            try:
+                self.log_file.unlink()
+            except Exception:
+                pass
+
+    def setup_logging(self, verbose: bool = False):
+        self.ensure_dirs()
+        root = logging.getLogger()
+        root.setLevel(logging.DEBUG if verbose else logging.INFO)
+
+        fh = logging.FileHandler(str(self.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 prepare(self) -> dict:
+        cfg = self.load_config()
+        self.save_config(cfg)
+        return cfg
+
+    def _emit_error(self, text: str):
+        if self.on_error:
+            self.on_error(text)
+
+    def _run_proxy_thread(self, port: int, dc_opt: Dict[int, str],
+                          host: str = "127.0.0.1"):
+        loop = _asyncio.new_event_loop()
+        _asyncio.set_event_loop(loop)
+        stop_ev = _asyncio.Event()
+        self._async_stop = (loop, stop_ev)
+
+        try:
+            loop.run_until_complete(
+                self.run_proxy(port, dc_opt, stop_event=stop_ev, host=host))
+        except Exception as exc:
+            self.log.error("Proxy thread crashed: %s", exc)
+            if ("10048" in str(exc) or
+                    "Address already in use" in str(exc)):
+                self._emit_error(
+                    "Не удалось запустить прокси:\n"
+                    "Порт уже используется другим приложением.\n\n"
+                    "Закройте приложение, использующее этот порт, "
+                    "или измените порт в настройках прокси и перезапустите.")
+        finally:
+            loop.close()
+            self._async_stop = None
+
+    def start_proxy(self, cfg: Optional[dict] = None) -> bool:
+        if self._proxy_thread and self._proxy_thread.is_alive():
+            self.log.info("Proxy already running")
+            return True
+
+        active_cfg = dict(cfg or self.config or self.default_config)
+        self.config = dict(active_cfg)
+        port = active_cfg.get("port", self.default_config["port"])
+        host = active_cfg.get("host", self.default_config["host"])
+        dc_ip_list = active_cfg.get("dc_ip", self.default_config["dc_ip"])
+
+        try:
+            dc_opt = self.parse_dc_ip_list(dc_ip_list)
+        except ValueError as exc:
+            self.log.error("Bad config dc_ip: %s", exc)
+            self._emit_error("Ошибка конфигурации:\n%s" % exc)
+            return False
+
+        self.log.info("Starting proxy on %s:%d ...", host, port)
+        self._proxy_thread = self.thread_factory(
+            target=self._run_proxy_thread,
+            args=(port, dc_opt, host),
+            daemon=True,
+            name="proxy")
+        self._proxy_thread.start()
+        return True
+
+    def stop_proxy(self):
+        if self._async_stop:
+            loop, stop_ev = self._async_stop
+            loop.call_soon_threadsafe(stop_ev.set)
+            if self._proxy_thread:
+                self._proxy_thread.join(timeout=2)
+        self._proxy_thread = None
+        self.log.info("Proxy stopped")
+
+    def restart_proxy(self, delay_seconds: float = 0.3) -> bool:
+        self.log.info("Restarting proxy...")
+        self.stop_proxy()
+        time.sleep(delay_seconds)
+        return self.start_proxy()
diff --git a/tests/test_app_runtime.py b/tests/test_app_runtime.py
new file mode 100644
index 0000000..b6026f9
--- /dev/null
+++ b/tests/test_app_runtime.py
@@ -0,0 +1,121 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from proxy.app_runtime import DEFAULT_CONFIG, ProxyAppRuntime
+
+
+class _FakeThread:
+    def __init__(self, target=None, args=(), daemon=None, name=None):
+        self.target = target
+        self.args = args
+        self.daemon = daemon
+        self.name = name
+        self.started = False
+        self.join_timeout = None
+        self._alive = False
+
+    def start(self):
+        self.started = True
+        self._alive = True
+
+    def is_alive(self):
+        return self._alive
+
+    def join(self, timeout=None):
+        self.join_timeout = timeout
+        self._alive = False
+
+
+class ProxyAppRuntimeTests(unittest.TestCase):
+    def test_load_config_returns_defaults_when_missing(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            runtime = ProxyAppRuntime(Path(tmpdir))
+
+            cfg = runtime.load_config()
+
+            self.assertEqual(cfg, DEFAULT_CONFIG)
+
+    def test_load_config_merges_defaults_into_saved_config(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app_dir = Path(tmpdir)
+            config_path = app_dir / "config.json"
+            app_dir.mkdir(parents=True, exist_ok=True)
+            config_path.write_text(
+                json.dumps({"port": 9050, "host": "127.0.0.2"}),
+                encoding="utf-8")
+            runtime = ProxyAppRuntime(app_dir)
+
+            cfg = runtime.load_config()
+
+            self.assertEqual(cfg["port"], 9050)
+            self.assertEqual(cfg["host"], "127.0.0.2")
+            self.assertEqual(cfg["dc_ip"], DEFAULT_CONFIG["dc_ip"])
+            self.assertEqual(cfg["verbose"], DEFAULT_CONFIG["verbose"])
+
+    def test_invalid_config_file_falls_back_to_defaults(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app_dir = Path(tmpdir)
+            app_dir.mkdir(parents=True, exist_ok=True)
+            (app_dir / "config.json").write_text("{broken", encoding="utf-8")
+            runtime = ProxyAppRuntime(app_dir)
+
+            cfg = runtime.load_config()
+
+            self.assertEqual(cfg, DEFAULT_CONFIG)
+
+    def test_start_proxy_starts_thread_with_parsed_dc_options(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            captured = {}
+            thread_holder = {}
+
+            def fake_parse(entries):
+                captured["dc_ip"] = list(entries)
+                return {2: "149.154.167.220"}
+
+            def fake_thread_factory(**kwargs):
+                thread = _FakeThread(**kwargs)
+                thread_holder["thread"] = thread
+                return thread
+
+            runtime = ProxyAppRuntime(
+                Path(tmpdir),
+                parse_dc_ip_list=fake_parse,
+                thread_factory=fake_thread_factory)
+
+            started = runtime.start_proxy(dict(DEFAULT_CONFIG))
+
+            self.assertTrue(started)
+            self.assertEqual(captured["dc_ip"], DEFAULT_CONFIG["dc_ip"])
+            self.assertTrue(thread_holder["thread"].started)
+            self.assertEqual(
+                thread_holder["thread"].args,
+                (DEFAULT_CONFIG["port"], {2: "149.154.167.220"},
+                 DEFAULT_CONFIG["host"]))
+
+    def test_start_proxy_reports_bad_config(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            errors = []
+
+            def fake_parse(entries):
+                raise ValueError("bad dc mapping")
+
+            runtime = ProxyAppRuntime(
+                Path(tmpdir),
+                parse_dc_ip_list=fake_parse,
+                on_error=errors.append)
+
+            started = runtime.start_proxy({
+                "host": "127.0.0.1",
+                "port": 1080,
+                "dc_ip": ["broken"],
+                "verbose": False,
+            })
+
+            self.assertFalse(started)
+            self.assertEqual(errors, ["Ошибка конфигурации:\nbad dc mapping"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/windows.py b/windows.py
index 64e581b..c96531d 100644
--- a/windows.py
+++ b/windows.py
@@ -11,39 +11,33 @@ import time
 import webbrowser
 import pystray
 import pyperclip
-import asyncio as _asyncio
 import customtkinter as ctk
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Optional
 from PIL import Image, ImageDraw, ImageFont
 
 import proxy.tg_ws_proxy as tg_ws_proxy
+from proxy.app_runtime import DEFAULT_CONFIG, ProxyAppRuntime
 
 
 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"
 IPV6_WARN_MARKER = APP_DIR / ".ipv6_warned"
-
-
-DEFAULT_CONFIG = {
-    "port": 1080,
-    "host": "127.0.0.1",
-    "dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
-    "verbose": False,
-}
-
-
-_proxy_thread: Optional[threading.Thread] = None
-_async_stop: Optional[object] = None
 _tray_icon: Optional[object] = None
 _config: dict = {}
 _exiting: bool = False
 _lock_file_path: Optional[Path] = None
 
 log = logging.getLogger("tg-ws-tray")
+_runtime = ProxyAppRuntime(
+    APP_DIR,
+    default_config=DEFAULT_CONFIG,
+    logger_name="tg-ws-tray",
+    on_error=lambda text: _show_error(text),
+)
+CONFIG_FILE = _runtime.config_file
+LOG_FILE = _runtime.log_file
 
 
 def _same_process(lock_meta: dict, proc: psutil.Process) -> bool:
@@ -120,48 +114,19 @@ def _acquire_lock() -> bool:
 
 
 def _ensure_dirs():
-    APP_DIR.mkdir(parents=True, exist_ok=True)
+    _runtime.ensure_dirs()
 
 
 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)
-            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)
+    return _runtime.load_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)
+    _runtime.save_config(cfg)
 
 
 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)
+    _runtime.setup_logging(verbose)
 
 
 def _make_icon_image(size: int = 64):
@@ -196,71 +161,16 @@ def _load_icon():
             pass
     return _make_icon_image()
 
-
-
-def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool,
-                      host: str = '127.0.0.1'):
-    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, host=host))
-    except Exception as exc:
-        log.error("Proxy thread crashed: %s", exc)
-        if "10048" in str(exc) or "Address already in use" in str(exc):
-            _show_error("Не удалось запустить прокси:\nПорт уже используется другим приложением.\n\nЗакройте приложение, использующее этот порт, или измените порт в настройках прокси и перезапустите.")
-    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"])
-    host = cfg.get("host", DEFAULT_CONFIG["host"])
-    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 %s:%d ...", host, port)
-    _proxy_thread = threading.Thread(
-        target=_run_proxy_thread,
-        args=(port, dc_opt, verbose, host),
-        daemon=True, name="proxy")
-    _proxy_thread.start()
+    _runtime.start_proxy(_config)
 
 
 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")
+    _runtime.stop_proxy()
 
 
 def restart_proxy():
-    log.info("Restarting proxy...")
-    stop_proxy()
-    time.sleep(0.3)
-    start_proxy()
+    _runtime.restart_proxy()
 
 
 def _show_error(text: str, title: str = "TG WS Proxy — Ошибка"):
@@ -642,14 +552,8 @@ def _build_menu():
 def run_tray():
     global _tray_icon, _config
 
-    _config = load_config()
-    save_config(_config)
-
-    if LOG_FILE.exists():
-        try:
-            LOG_FILE.unlink()
-        except Exception:
-            pass
+    _config = _runtime.prepare()
+    _runtime.reset_log_file()
 
     setup_logging(_config.get("verbose", False))
     log.info("TG WS Proxy tray app starting")