Compare commits

...

30 Commits
v1.2.0 ... main

Author SHA1 Message Date
gogamlg3 da4b521aba
Изменение README для AUR (#485) 2026-03-30 09:55:44 +03:00
Flowseal 07facfe18c Version bump 2026-03-29 20:00:31 +03:00
Qirashi 7a886dff26
Update ctk_theme.py (#480) 2026-03-29 19:56:43 +03:00
Flowseal 17e37f9ca0 host detect in first-run window 2026-03-29 19:55:39 +03:00
Flowseal 968827445f copy link, mtproto new first run notify 2026-03-29 17:57:55 +03:00
Flowseal be8d178e5c secret validation 2026-03-29 17:30:39 +03:00
Flowseal 46426c45b0 ctk refactoring 2026-03-29 15:21:56 +03:00
Flowseal c4a044542c fixes 2026-03-29 15:21:45 +03:00
Flowseal af74009b11 icon fix 2026-03-28 16:48:59 +03:00
Flowseal 6766db9812 mtproto recode 2026-03-28 15:45:08 +03:00
Flowseal 95f99be26b old socks removed 2026-03-28 13:07:51 +03:00
deexsed 0d11062c92
fix: остановка прокси из трея, пул WS и проверка обновлений (#443) 2026-03-27 10:54:33 +03:00
Flowseal b3a9bc6a8f icon size increase 2026-03-27 09:17:15 +03:00
Flowseal c179c299bb tooltip fixes 2026-03-27 09:07:25 +03:00
Aleksandr bd4746004e
Docker image for headless proxy (#289) 2026-03-27 09:05:56 +03:00
deexsed 77a0b837d9
Общий UI трея в ui/, тултипы, исправление tg:// с реальным host, доработки windows.py (импорты, lock, IPv6, остановка прокси) (#417) 2026-03-27 08:54:36 +03:00
Kroshik the Seal 5d28a50740
Исправление зависания "Обновления..." на iOS/iPadOS (#415) 2026-03-27 07:43:49 +03:00
KG7x 7a1e2f3f5b
Miss update actions/download-artifact (#412) 2026-03-23 18:06:46 +03:00
KG7x c0183bf448
Fix warn Node24 actions update & Simplify build (#410) 2026-03-23 17:10:20 +03:00
Flowseal f95b9b7da0 Update win7 build according to #292 2026-03-23 09:12:06 +03:00
delewer f3d05f7efc
chore: pyproject optimization (#292) 2026-03-23 09:09:30 +03:00
gogamlg3 e3d4578eed
Добавление способа установки через AUR для Arch дистрибутивов (#296) 2026-03-23 08:39:29 +03:00
xdshkaaa e1004e5e73
Fix macOS settings dialog cancellation flow (#392) 2026-03-23 08:39:00 +03:00
delewer 4304c71f89
build: win7 32bit support (#298) 2026-03-23 08:38:35 +03:00
Flowseal 3cb1929dc8 removed test script 2026-03-23 04:30:38 +03:00
Flowseal afb7c5f56d revert keepalive mechanism 2026-03-22 08:00:14 +03:00
Flowseal 18a1bced83 logrotate #366; configurable pool and buffer sizes 2026-03-22 02:54:03 +03:00
Flowseal ed85e2a284 keepalive for stale mitigation 2026-03-21 09:26:34 +03:00
Flowseal c1452c23da Optimizations 2026-03-20 22:57:15 +03:00
Flowseal 6a80ca85e3 Optimizations 2026-03-19 22:07:47 +03:00
19 changed files with 2920 additions and 2363 deletions

28
.dockerignore Normal file
View File

@ -0,0 +1,28 @@
.git
.github
.gitignore
__pycache__/
*.py[cod]
*.pyo
*.egg-info/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.venv/
venv/
dist/
build/
packaging/
windows.py
icon.ico
*.spec
*.spec.bak
*.manifest
*.log
.vscode/
.idea/
*.swp
*.swo
.DS_Store
Thumbs.db
Desktop.ini

View File

@ -17,20 +17,20 @@ permissions:
contents: write
jobs:
build:
build-windows:
runs-on: windows-latest
steps:
- name: Checkout
uses: actions/checkout@v4
uses: actions/checkout@v6
- name: Setup Python
uses: actions/setup-python@v5
uses: actions/setup-python@v6
with:
python-version: "3.12"
cache: "pip"
- name: Install dependencies
run: pip install ".[win10]"
run: pip install .
- name: Install pyinstaller
run: pip install "pyinstaller==6.13.0"
@ -42,47 +42,49 @@ jobs:
run: mv dist/TgWsProxy.exe dist/TgWsProxy_windows.exe
- name: Upload artifact
uses: actions/upload-artifact@v4
uses: actions/upload-artifact@v7
with:
name: TgWsProxy
path: |
dist/TgWsProxy_windows.exe
path: dist/TgWsProxy_windows.exe
build-win7:
runs-on: windows-latest
strategy:
matrix:
include:
- arch: x64
suffix: 64bit
- arch: x86
suffix: 32bit
steps:
- name: Checkout
uses: actions/checkout@v4
- uses: actions/checkout@v6
- name: Setup Python 3.8 (last version supporting Win7)
uses: actions/setup-python@v5
- uses: actions/setup-python@v6
with:
python-version: "3.8"
architecture: ${{ matrix.arch }}
cache: "pip"
- name: Install dependencies (Win7-compatible)
run: pip install ".[win7]"
- name: Install dependencies & pyinstaller
run: pip install . "pyinstaller==5.13.2"
- name: Install pyinstaller
run: pip install "pyinstaller==5.13.2"
- name: Build EXE with PyInstaller (Win7)
- name: Build EXE with PyInstaller
run: pyinstaller packaging/windows.spec --noconfirm
- name: Rename artifact
run: mv dist/TgWsProxy.exe dist/TgWsProxy_windows_7.exe
run: mv dist/TgWsProxy.exe dist/TgWsProxy_windows_7_${{ matrix.suffix }}.exe
- name: Upload artifact
uses: actions/upload-artifact@v4
uses: actions/upload-artifact@v7
with:
name: TgWsProxy-win7
path: dist/TgWsProxy_windows_7.exe
name: TgWsProxy-win7-${{ matrix.suffix }}
path: dist/TgWsProxy_windows_7_${{ matrix.suffix }}.exe
build-macos:
runs-on: macos-latest
steps:
- name: Checkout
uses: actions/checkout@v4
uses: actions/checkout@v6
- name: Install universal2 Python
run: |
@ -142,7 +144,7 @@ jobs:
-w wheelhouse/universal2
python3.12 -m pip install --no-deps wheelhouse/universal2/*.whl
python3.12 -m pip install ".[macos]"
python3.12 -m pip install .
python3.12 -m pip install pyinstaller==6.13.0
- name: Create macOS icon from ICO
@ -217,7 +219,7 @@ jobs:
rm -rf "$DMG_TEMP"
- name: Upload artifact
uses: actions/upload-artifact@v4
uses: actions/upload-artifact@v7
with:
name: TgWsProxy-macOS
path: dist/TgWsProxy_macos_universal.dmg
@ -226,7 +228,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
uses: actions/checkout@v6
- name: Install system dependencies
run: |
@ -244,7 +246,7 @@ jobs:
- name: Install dependencies
run: |
.venv/bin/pip install --upgrade pip
.venv/bin/pip install ".[linux]"
.venv/bin/pip install .
.venv/bin/pip install "pyinstaller==6.13.0"
- name: Build binary with PyInstaller
@ -301,7 +303,7 @@ jobs:
Maintainer: Flowseal
Depends: libgtk-3-0, libayatana-appindicator3-1, python3-tk
Description: Telegram Desktop WebSocket Bridge Proxy
SOCKS5/WebSocket bridge proxy for Telegram Desktop with tray UI.
MTProto/WebSocket bridge proxy for Telegram Desktop with tray UI.
EOF
dpkg-deb --build --root-owner-group \
@ -309,7 +311,7 @@ jobs:
"dist/TgWsProxy_linux_amd64.deb"
- name: Upload artifact
uses: actions/upload-artifact@v4
uses: actions/upload-artifact@v7
with:
name: TgWsProxy-linux
path: |
@ -317,33 +319,15 @@ jobs:
dist/TgWsProxy_linux_amd64.deb
release:
needs: [build, build-win7, build-macos, build-linux]
needs: [build-windows, build-win7, build-macos, build-linux]
runs-on: ubuntu-latest
if: ${{ github.event.inputs.make_release == 'true' }}
steps:
- name: Download main build
uses: actions/download-artifact@v4
- uses: actions/download-artifact@v8
with:
name: TgWsProxy
path: dist
- name: Download Win7 build
uses: actions/download-artifact@v4
with:
name: TgWsProxy-win7
path: dist
- name: Download macOS build
uses: actions/download-artifact@v4
with:
name: TgWsProxy-macOS
path: dist
- name: Download Linux build
uses: actions/download-artifact@v4
with:
name: TgWsProxy-linux
pattern: TgWsProxy*
path: dist
merge-multiple: true
- name: Create GitHub Release
uses: softprops/action-gh-release@v2
@ -354,7 +338,8 @@ jobs:
## TG WS Proxy ${{ github.event.inputs.version }}
files: |
dist/TgWsProxy_windows.exe
dist/TgWsProxy_windows_7.exe
dist/TgWsProxy_windows_7_64bit.exe
dist/TgWsProxy_windows_7_32bit.exe
dist/TgWsProxy_macos_universal.dmg
dist/TgWsProxy_linux_amd64
dist/TgWsProxy_linux_amd64.deb

45
Dockerfile Normal file
View File

@ -0,0 +1,45 @@
# syntax=docker/dockerfile:1.7
FROM python:3.12-slim AS builder
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 \
PIP_NO_CACHE_DIR=1 \
VIRTUAL_ENV=/opt/venv
RUN apt-get update \
&& apt-get install -y --no-install-recommends build-essential cargo libffi-dev libssl-dev \
&& python -m venv "$VIRTUAL_ENV" \
&& "$VIRTUAL_ENV/bin/pip" install --upgrade pip setuptools wheel \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
RUN "$VIRTUAL_ENV/bin/pip" install cryptography==46.0.5
FROM python:3.12-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PATH=/opt/venv/bin:$PATH \
TG_WS_PROXY_HOST=0.0.0.0 \
TG_WS_PROXY_PORT=1443 \
TG_WS_PROXY_DC_IPS="2:149.154.167.220 4:149.154.167.220"
RUN apt-get update \
&& apt-get install -y --no-install-recommends tini ca-certificates \
&& rm -rf /var/lib/apt/lists/* \
&& groupadd --system app \
&& useradd --system --gid app --create-home --home-dir /home/app app
WORKDIR /app
COPY --from=builder /opt/venv /opt/venv
COPY proxy ./proxy
COPY README.md LICENSE ./
USER app
EXPOSE 1443/tcp
ENTRYPOINT ["/usr/bin/tini", "--", "/bin/sh", "-lc", "set -eu; args=\"--host ${TG_WS_PROXY_HOST} --port ${TG_WS_PROXY_PORT}\"; for dc in ${TG_WS_PROXY_DC_IPS}; do args=\"$args --dc-ip $dc\"; done; exec python -u proxy/tg_ws_proxy.py $args \"$@\"", "--"]
CMD []

View File

@ -12,17 +12,17 @@
# TG WS Proxy
**Локальный SOCKS5-прокси** для Telegram Desktop, который **ускоряет работу Telegram**, перенаправляя трафик через WebSocket-соединения. Данные передаются в том же зашифрованном виде, а для работы не нужны сторонние сервера.
**Локальный MTProto-прокси** для Telegram Desktop, который **ускоряет работу Telegram**, перенаправляя трафик через WebSocket-соединения. Данные передаются в том же зашифрованном виде, а для работы не нужны сторонние сервера.
<img width="529" height="487" alt="image" src="https://github.com/user-attachments/assets/6a4cf683-0df8-43af-86c1-0e8f08682b62" />
## Как это работает
```
Telegram Desktop → SOCKS5 (127.0.0.1:1080) → TG WS Proxy → WSS → Telegram DC
Telegram Desktop → MTProto Proxy (127.0.0.1:1443) → WebSocket → Telegram DC
```
1. Приложение поднимает локальный SOCKS5-прокси на `127.0.0.1:1080`
1. Приложение поднимает MTProto прокси на `127.0.0.1:1443`
2. Перехватывает подключения к IP-адресам Telegram
3. Извлекает DC ID из MTProto obfuscation init-пакета
4. Устанавливает WebSocket (TLS) соединение к соответствующему DC через домены Telegram
@ -38,12 +38,14 @@ Telegram Desktop → SOCKS5 (127.0.0.1:1080) → TG WS Proxy → WSS → Telegra
**Меню трея:**
- **Открыть в Telegram** — автоматически настроить прокси через `tg://socks` ссылку
- **Открыть в Telegram** — автоматически настроить прокси через `tg://proxy` ссылку
- **Перезапустить прокси** — перезапуск без выхода из приложения
- **Настройки...** — GUI-редактор конфигурации
- **Настройки...** — GUI-редактор конфигурации (в т.ч. версия приложения, опциональная проверка обновлений с GitHub)
- **Открыть логи** — открыть файл логов
- **Выход** — остановить прокси и закрыть приложение
При первом запуске после старта может появиться запрос об открытии страницы релиза, если на GitHub вышла новая версия (отключается в настройках).
### macOS
Перейдите на [страницу релизов](https://github.com/Flowseal/tg-ws-proxy/releases) и скачайте **`TgWsProxy_macos_universal.dmg`** — универсальная сборка для Apple Silicon и Intel.
@ -56,6 +58,22 @@ Telegram Desktop → SOCKS5 (127.0.0.1:1080) → TG WS Proxy → WSS → Telegra
Для Debian/Ubuntu скачайте со [страницы релизов](https://github.com/Flowseal/tg-ws-proxy/releases) пакет **`TgWsProxy_linux_amd64.deb`**.
Для Arch и Arch-Based дистрибутивов подготовлены пакеты в AUR: [tg-ws-proxy-bin](https://aur.archlinux.org/packages/tg-ws-proxy-bin), [tg-ws-proxy-git](https://aur.archlinux.org/packages/tg-ws-proxy-git), [tg-ws-proxy-cli](https://aur.archlinux.org/packages/tg-ws-proxy-cli)
```shell
# Установка без AUR-helper
git clone https://aur.archlinux.org/tg-ws-proxy-bin.git
cd tg-ws-proxy-bin
makepkg -si
# При помощи AUR-helper
paru -S tg-ws-proxy-bin
# Если вы установили -cli пакет, то запуск осуществляется через systemctl, где 8888 это номер порта,
# разделитель ":" и secret, который можно сгенерировать командой: openssl rand -hex 16
sudo systemctl start tg-ws-proxy-cli@8888:3075abe65830f0325116bb0416cadf9f
```
Для остальных дистрибутивов можно использовать **`TgWsProxy_linux_amd64`** (бинарный файл для x86_64).
```bash
@ -69,38 +87,31 @@ chmod +x TgWsProxy_linux_amd64
### Консольный proxy
Для запуска только SOCKS5/WebSocket proxy без tray-интерфейса достаточно базовой установки:
Для запуска только proxy без tray-интерфейса достаточно базовой установки:
```bash
pip install -e .
tg-ws-proxy
```
### Windows 10+
### Windows 7/10+
```bash
pip install -e ".[win10]"
tg-ws-proxy-tray-win
```
### Windows 7
```bash
pip install -e ".[win7]"
pip install -e .
tg-ws-proxy-tray-win
```
### macOS
```bash
pip install -e ".[macos]"
pip install -e .
tg-ws-proxy-tray-macos
```
### Linux
```bash
pip install -e ".[linux]"
pip install -e .
tg-ws-proxy-tray-linux
```
@ -114,9 +125,15 @@ tg-ws-proxy [--port PORT] [--host HOST] [--dc-ip DC:IP ...] [-v]
| Аргумент | По умолчанию | Описание |
|---|---|---|
| `--port` | `1080` | Порт SOCKS5-прокси |
| `--host` | `127.0.0.1` | Хост SOCKS5-прокси |
| `--port` | `1443` | Порт прокси |
| `--host` | `127.0.0.1` | Хост прокси |
| `--secret` | `random` | 32 hex chars secret для авторизации клиентов |
| `--dc-ip` | `2:149.154.167.220`, `4:149.154.167.220` | Целевой IP для DC (можно указать несколько раз) |
| `--buf-kb` | `256` | Размер буфера в КБ
| `--pool-size` | `4` | Количество заготовленных соединений на каждый DC
| `--log-file` | выкл. | Путь до файла, в который сохранять логи
| `--log-max-mb` | `5` | Максимальный размер файла логов в МБ (после идёт перезапись)
| `--log-backups` | `0` | Количество сохранений логов после перезаписи
| `-v`, `--verbose` | выкл. | Подробное логирование (DEBUG) |
**Примеры:**
@ -156,10 +173,10 @@ tg-ws-proxy-tray-linux = "linux:main"
1. Telegram → **Настройки****Продвинутые настройки****Тип подключения** → **Прокси**
2. Добавить прокси:
- **Тип:** SOCKS5
- **Сервер:** `127.0.0.1`
- **Порт:** `1080`
- **Логин/Пароль:** оставить пустыми
- **Тип:** MTProto
- **Сервер:** `127.0.0.1` (или переопределенный вами)
- **Порт:** `1443` (или переопределенный вами)
- **Secret:** из настроек или логов
## Конфигурация
@ -171,15 +188,23 @@ Tray-приложение хранит данные в:
```json
{
"port": 1080,
"host": "127.0.0.1",
"port": 1443,
"secret": "...",
"dc_ip": [
"2:149.154.167.220",
"4:149.154.167.220"
],
"verbose": false
"verbose": false,
"buf_kb": 256,
"pool_size": 4,
"log_max_mb": 5.0,
"check_updates": true
}
```
Ключ **`check_updates`** — при `true` при запросе к GitHub сравнивается версия с последним релизом (только уведомление и ссылка на страницу загрузки). На Windows в конфиге может быть **`autostart`** (автозапуск при входе в систему).
## Автоматическая сборка
Проект содержит спецификации PyInstaller ([`packaging/windows.spec`](packaging/windows.spec), [`packaging/macos.spec`](packaging/macos.spec), [`packaging/linux.spec`](packaging/linux.spec)) и GitHub Actions workflow ([`.github/workflows/build.yml`](.github/workflows/build.yml)) для автоматической сборки.
@ -187,7 +212,8 @@ Tray-приложение хранит данные в:
Минимально поддерживаемые версии ОС для текущих бинарных сборок:
- Windows 10+ для `TgWsProxy_windows.exe`
- Windows 7 для `TgWsProxy_windows_7.exe`
- Windows 7 (x64) для `TgWsProxy_windows_7_64bit.exe`
- Windows 7 (x32) для `TgWsProxy_windows_7_32bit.exe`
- Intel macOS 10.15+
- Apple Silicon macOS 11.0+
- Linux x86_64 (требуется AppIndicator для системного трея)

BIN
icon.ico

Binary file not shown.

Before

Width:  |  Height:  |  Size: 5.7 KiB

After

Width:  |  Height:  |  Size: 473 B

891
linux.py

File diff suppressed because it is too large Load Diff

617
macos.py
View File

@ -1,17 +1,13 @@
from __future__ import annotations
import json
import logging
import os
import psutil
import subprocess
import sys
import threading
import time
import webbrowser
import asyncio as _asyncio
from pathlib import Path
from typing import Dict, Optional
from typing import Optional
try:
import rumps
@ -29,234 +25,135 @@ except ImportError:
pyperclip = None
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy import __version__
from utils.tray_common import (
APP_DIR, APP_NAME, DEFAULT_CONFIG, FIRST_RUN_MARKER, IPV6_WARN_MARKER,
LOG_FILE, acquire_lock, apply_proxy_config, ensure_dirs, load_config,
log, release_lock, save_config, setup_logging, stop_proxy, tg_proxy_url,
)
APP_NAME = "TgWsProxy"
APP_DIR = Path.home() / "Library" / "Application Support" / 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"
MENUBAR_ICON_PATH = APP_DIR / "menubar_icon.png"
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
_app: Optional[object] = None
_config: dict = {}
_exiting: bool = False
_lock_file_path: Optional[Path] = None
log = logging.getLogger("tg-ws-tray")
# osascript dialogs
# Single-instance lock
def _esc(text: str) -> str:
return text.replace("\\", "\\\\").replace('"', '\\"')
def _same_process(lock_meta: dict, proc: psutil.Process) -> bool:
try:
lock_ct = float(lock_meta.get("create_time", 0.0))
proc_ct = float(proc.create_time())
if lock_ct > 0 and abs(lock_ct - proc_ct) > 1.0:
return False
except Exception:
def _osascript(script: str) -> str:
r = subprocess.run(["osascript", "-e", script], capture_output=True, text=True)
return r.stdout.strip()
def _show_error(text: str, title: str = "TG WS Proxy") -> None:
_osascript(
f'display dialog "{_esc(text)}" with title "{_esc(title)}" '
f'buttons {{"OK"}} default button "OK" with icon stop'
)
def _show_info(text: str, title: str = "TG WS Proxy") -> None:
_osascript(
f'display dialog "{_esc(text)}" with title "{_esc(title)}" '
f'buttons {{"OK"}} default button "OK" with icon note'
)
def _ask_yes_no(text: str, title: str = "TG WS Proxy") -> bool:
return _ask_yes_no_close(text, title) is True
def _ask_yes_no_close(text: str, title: str = "TG WS Proxy") -> Optional[bool]:
r = subprocess.run(
[
"osascript", "-e",
f'button returned of (display dialog "{_esc(text)}" '
f'with title "{_esc(title)}" '
f'buttons {{"Закрыть", "Нет", "Да"}} '
f'default button "Да" cancel button "Закрыть" with icon note)',
],
capture_output=True, text=True,
)
if r.returncode != 0:
return None
btn = r.stdout.strip()
if btn == "Да":
return True
if btn == "Нет":
return False
frozen = bool(getattr(sys, "frozen", False))
if frozen:
return APP_NAME.lower() in proc.name().lower()
return False
return None
def _release_lock():
global _lock_file_path
if not _lock_file_path:
return
try:
_lock_file_path.unlink(missing_ok=True)
except Exception:
pass
_lock_file_path = None
def _osascript_input(prompt: str, default: str, title: str = "TG WS Proxy") -> Optional[str]:
r = subprocess.run(
[
"osascript", "-e",
f'text returned of (display dialog "{_esc(prompt)}" '
f'default answer "{_esc(default)}" '
f'with title "{_esc(title)}" '
f'buttons {{"Закрыть", "OK"}} '
f'default button "OK" cancel button "Закрыть")',
],
capture_output=True, text=True,
)
if r.returncode != 0:
return None
return r.stdout.rstrip("\r\n")
def _acquire_lock() -> bool:
global _lock_file_path
_ensure_dirs()
lock_files = list(APP_DIR.glob("*.lock"))
# menubar icon
for f in lock_files:
pid = None
meta: dict = {}
try:
pid = int(f.stem)
except Exception:
f.unlink(missing_ok=True)
continue
try:
raw = f.read_text(encoding="utf-8").strip()
if raw:
meta = json.loads(raw)
except Exception:
meta = {}
try:
proc = psutil.Process(pid)
if _same_process(meta, proc):
return False
except Exception:
pass
f.unlink(missing_ok=True)
lock_file = APP_DIR / f"{os.getpid()}.lock"
try:
proc = psutil.Process(os.getpid())
payload = {"create_time": proc.create_time()}
lock_file.write_text(json.dumps(payload, ensure_ascii=False),
encoding="utf-8")
except Exception:
lock_file.touch()
_lock_file_path = lock_file
return True
# Filesystem helpers
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)
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)
# Menubar icon
def _make_menubar_icon(size: int = 44):
if Image is None:
return None
img = Image.new("RGBA", (size, size), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
margin = size // 11
draw.ellipse([margin, margin, size - margin, size - margin],
fill=(0, 0, 0, 255))
draw.ellipse([margin, margin, size - margin, size - margin], fill=(0, 0, 0, 255))
try:
font = ImageFont.truetype(
"/System/Library/Fonts/Helvetica.ttc",
size=int(size * 0.55))
font = ImageFont.truetype("/System/Library/Fonts/Helvetica.ttc", 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)
draw.text(
((size - tw) // 2 - bbox[0], (size - th) // 2 - bbox[1]),
"T", fill=(255, 255, 255, 255), font=font,
)
return img
# Generate menubar icon PNG if it does not exist.
def _ensure_menubar_icon():
def _ensure_menubar_icon() -> None:
if MENUBAR_ICON_PATH.exists():
return
_ensure_dirs()
ensure_dirs()
img = _make_menubar_icon(44)
if img:
img.save(str(MENUBAR_ICON_PATH), "PNG")
# Native macOS dialogs
# proxy lifecycle (macOS-local)
def _osascript(script: str) -> str:
r = subprocess.run(
['osascript', '-e', script],
capture_output=True, text=True)
return r.stdout.strip()
import asyncio as _asyncio
def _show_error(text: str, title: str = "TG WS Proxy"):
text_esc = text.replace('\\', '\\\\').replace('"', '\\"')
title_esc = title.replace('\\', '\\\\').replace('"', '\\"')
_osascript(
f'display dialog "{text_esc}" with title "{title_esc}" '
f'buttons {{"OK"}} default button "OK" with icon stop')
def _show_info(text: str, title: str = "TG WS Proxy"):
text_esc = text.replace('\\', '\\\\').replace('"', '\\"')
title_esc = title.replace('\\', '\\\\').replace('"', '\\"')
_osascript(
f'display dialog "{text_esc}" with title "{title_esc}" '
f'buttons {{"OK"}} default button "OK" with icon note')
def _ask_yes_no(text: str, title: str = "TG WS Proxy") -> bool:
text_esc = text.replace('\\', '\\\\').replace('"', '\\"')
title_esc = title.replace('\\', '\\\\').replace('"', '\\"')
result = _osascript(
f'display dialog "{text_esc}" with title "{title_esc}" '
f'buttons {{"Нет", "Да"}} default button "Да" with icon note')
return "Да" in result
# Proxy lifecycle
def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool,
host: str = '127.0.0.1'):
def _run_proxy_thread() -> None:
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))
loop.run_until_complete(tg_ws_proxy._run(stop_event=stop_ev))
except Exception as exc:
log.error("Proxy thread crashed: %s", exc)
if "Address already in use" in str(exc):
@ -264,40 +161,28 @@ def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool,
"Не удалось запустить прокси:\n"
"Порт уже используется другим приложением.\n\n"
"Закройте приложение, использующее этот порт, "
"или измените порт в настройках прокси и перезапустите.")
"или измените порт в настройках прокси и перезапустите."
)
finally:
loop.close()
_async_stop = None
def start_proxy():
global _proxy_thread, _config
def _start_proxy() -> None:
global _proxy_thread
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}")
if not apply_proxy_config(_config):
_show_error("Ошибка конфигурации DC → IP.")
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")
pc = tg_ws_proxy.proxy_config
log.info("Starting proxy on %s:%d ...", pc.host, pc.port)
_proxy_thread = threading.Thread(target=_run_proxy_thread, daemon=True, name="proxy")
_proxy_thread.start()
def stop_proxy():
def _stop_proxy() -> None:
global _proxy_thread, _async_stop
if _async_stop:
loop, stop_ev = _async_stop
@ -308,21 +193,21 @@ def stop_proxy():
log.info("Proxy stopped")
def restart_proxy():
def _restart_proxy() -> None:
log.info("Restarting proxy...")
stop_proxy()
_stop_proxy()
time.sleep(0.3)
start_proxy()
_start_proxy()
# Menu callbacks
# menu callbacks
def _on_open_in_telegram(_=None):
port = _config.get("port", DEFAULT_CONFIG["port"])
url = f"tg://socks?server=127.0.0.1&port={port}"
def _on_open_in_telegram(_=None) -> None:
url = tg_proxy_url(_config)
log.info("Opening %s", url)
try:
result = subprocess.call(['open', url])
result = subprocess.call(["open", url])
if result != 0:
raise RuntimeError("open command failed")
except Exception:
@ -336,68 +221,109 @@ def _on_open_in_telegram(_=None):
if pyperclip:
pyperclip.copy(url)
else:
subprocess.run(['pbcopy'], input=url.encode(),
check=True)
subprocess.run(["pbcopy"], input=url.encode(), check=True)
_show_info(
"Не удалось открыть Telegram автоматически.\n\n"
f"Ссылка скопирована в буфер обмена:\n{url}")
f"Ссылка скопирована в буфер обмена:\n{url}"
)
except Exception as exc:
log.error("Clipboard copy failed: %s", exc)
_show_error(f"Не удалось скопировать ссылку:\n{exc}")
def _on_restart(_=None):
def _do_restart():
def _on_copy_link(_=None) -> None:
url = tg_proxy_url(_config)
log.info("Copying link: %s", url)
try:
if pyperclip:
pyperclip.copy(url)
else:
subprocess.run(["pbcopy"], input=url.encode(), check=True)
except Exception as exc:
log.error("Clipboard copy failed: %s", exc)
_show_error(f"Не удалось скопировать ссылку:\n{exc}")
def _on_restart(_=None) -> None:
def _do():
global _config
_config = load_config()
if _app:
_app.update_menu_title()
restart_proxy()
_restart_proxy()
threading.Thread(target=_do_restart, daemon=True).start()
threading.Thread(target=_do, daemon=True).start()
def _on_open_logs(_=None):
def _on_open_logs(_=None) -> None:
log.info("Opening log file: %s", LOG_FILE)
if LOG_FILE.exists():
subprocess.call(['open', str(LOG_FILE)])
subprocess.call(["open", str(LOG_FILE)])
else:
_show_info("Файл логов ещё не создан.")
# Show a native text input dialog. Returns None if cancelled.
def _osascript_input(prompt: str, default: str,
title: str = "TG WS Proxy") -> Optional[str]:
prompt_esc = prompt.replace('\\', '\\\\').replace('"', '\\"')
default_esc = default.replace('\\', '\\\\').replace('"', '\\"')
title_esc = title.replace('\\', '\\\\').replace('"', '\\"')
r = subprocess.run(
['osascript', '-e',
f'text returned of (display dialog "{prompt_esc}" '
f'default answer "{default_esc}" '
f'with title "{title_esc}" '
f'buttons {{"Отмена", "OK"}} default button "OK")'],
capture_output=True, text=True)
if r.returncode != 0:
return None
return r.stdout.rstrip("\r\n")
def _on_edit_config(_=None):
def _on_edit_config(_=None) -> None:
threading.Thread(target=_edit_config_dialog, daemon=True).start()
# Settings via native macOS dialogs
def _edit_config_dialog():
def _check_updates_menu_title() -> str:
on = bool(_config.get("check_updates", True))
return "✓ Проверять обновления при запуске" if on else "Проверять обновления при запуске (выкл)"
def _toggle_check_updates(_=None) -> None:
global _config
_config["check_updates"] = not bool(_config.get("check_updates", True))
save_config(_config)
if _app is not None:
_app._check_updates_item.title = _check_updates_menu_title()
def _on_open_release_page(_=None) -> None:
from utils.update_check import RELEASES_PAGE_URL
webbrowser.open(RELEASES_PAGE_URL)
# update check
def _maybe_notify_update_async() -> None:
def _work():
time.sleep(1.5)
if _exiting:
return
if not _config.get("check_updates", True):
return
try:
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
run_check(__version__)
st = get_status()
if not st.get("has_update"):
return
url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
ver = st.get("latest") or "?"
if _ask_yes_no(
f"Доступна новая версия: {ver}\n\nОткрыть страницу релиза в браузере?",
"TG WS Proxy — обновление",
):
webbrowser.open(url)
except Exception as exc:
log.debug("Update check failed: %s", exc)
threading.Thread(target=_work, daemon=True, name="update-check").start()
# settings dialog
def _edit_config_dialog() -> None:
cfg = load_config()
# Host
host = _osascript_input(
"IP-адрес прокси:",
cfg.get("host", DEFAULT_CONFIG["host"]))
host = _osascript_input("IP-адрес прокси:", cfg.get("host", DEFAULT_CONFIG["host"]))
if host is None:
return
host = host.strip()
import socket as _sock
try:
_sock.inet_aton(host)
@ -405,10 +331,7 @@ def _edit_config_dialog():
_show_error("Некорректный IP-адрес.")
return
# Port
port_str = _osascript_input(
"Порт прокси:",
str(cfg.get("port", DEFAULT_CONFIG["port"])))
port_str = _osascript_input("Порт прокси:", str(cfg.get("port", DEFAULT_CONFIG["port"])))
if port_str is None:
return
try:
@ -419,30 +342,66 @@ def _edit_config_dialog():
_show_error("Порт должен быть числом 1-65535")
return
# DC-IP mappings
secret_str = _osascript_input(
"MTProto Secret (32 hex символа):", cfg.get("secret", DEFAULT_CONFIG["secret"])
)
if secret_str is None:
return
secret_str = secret_str.strip().lower()
if len(secret_str) != 32 or not all(c in "0123456789abcdef" for c in secret_str):
_show_error("Secret должен быть строкой из 32 шестнадцатеричных символов.")
return
dc_default = ", ".join(cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"]))
dc_str = _osascript_input(
"DC → IP маппинги (через запятую, формат DC:IP):\n"
"Например: 2:149.154.167.220, 4:149.154.167.220",
dc_default)
dc_default,
)
if dc_str is None:
return
dc_lines = [s.strip() for s in dc_str.replace(',', '\n').splitlines()
if s.strip()]
dc_lines = [s.strip() for s in dc_str.replace(",", "\n").splitlines() if s.strip()]
try:
tg_ws_proxy.parse_dc_ip_list(dc_lines)
except ValueError as e:
_show_error(str(e))
return
# Verbose
verbose = _ask_yes_no("Включить подробное логирование (verbose)?")
verbose = _ask_yes_no_close("Включить подробное логирование (verbose)?")
if verbose is None:
return
adv_str = _osascript_input(
"Расширенные настройки (буфер KB, WS пул, лог MB):\n"
"Формат: buf_kb,pool_size,log_max_mb",
f"{cfg.get('buf_kb', DEFAULT_CONFIG['buf_kb'])},"
f"{cfg.get('pool_size', DEFAULT_CONFIG['pool_size'])},"
f"{cfg.get('log_max_mb', DEFAULT_CONFIG['log_max_mb'])}",
)
if adv_str is None:
return
adv = {}
if adv_str:
parts = [s.strip() for s in adv_str.split(",")]
keys = [("buf_kb", int), ("pool_size", int), ("log_max_mb", float)]
for i, (k, typ) in enumerate(keys):
if i < len(parts):
try:
adv[k] = typ(parts[i])
except ValueError:
pass
new_cfg = {
"host": host,
"port": port,
"secret": secret_str,
"dc_ip": dc_lines,
"verbose": verbose,
"buf_kb": adv.get("buf_kb", cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"])),
"pool_size": adv.get("pool_size", cfg.get("pool_size", DEFAULT_CONFIG["pool_size"])),
"log_max_mb": adv.get("log_max_mb", cfg.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"])),
"check_updates": cfg.get("check_updates", True),
}
save_config(new_cfg)
log.info("Config saved: %s", new_cfg)
@ -452,20 +411,23 @@ def _edit_config_dialog():
if _app:
_app.update_menu_title()
if _ask_yes_no("Настройки сохранены.\n\nПерезапустить прокси сейчас?"):
restart_proxy()
if _ask_yes_no_close("Настройки сохранены.\n\nПерезапустить прокси сейчас?"):
_restart_proxy()
# First-run & IPv6 dialogs
# first run & ipv6
def _show_first_run():
_ensure_dirs()
def _show_first_run() -> None:
ensure_dirs()
if FIRST_RUN_MARKER.exists():
return
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
tg_url = f"tg://socks?server={host}&port={port}"
secret = _config.get("secret", DEFAULT_CONFIG["secret"])
tg_url = tg_proxy_url(_config)
link_host = tg_ws_proxy.get_link_host(host)
text = (
f"Прокси запущен и работает в строке меню.\n\n"
@ -475,54 +437,54 @@ def _show_first_run():
f" Или ссылка: {tg_url}\n\n"
f"Вручную:\n"
f" Настройки → Продвинутые → Тип подключения → Прокси\n"
f" SOCKS5 → {host} : {port} (без логина/пароля)\n\n"
f" MTProto → {link_host} : {port} \n"
f" Secret: dd{secret} \n\n"
f"Открыть прокси в Telegram сейчас?"
)
FIRST_RUN_MARKER.touch()
if _ask_yes_no(text, "TG WS Proxy"):
_on_open_in_telegram()
def _has_ipv6_enabled() -> bool:
import socket as _sock
try:
addrs = _sock.getaddrinfo(_sock.gethostname(), None, _sock.AF_INET6)
for addr in addrs:
ip = addr[4][0]
if ip and not ip.startswith('::1') and not ip.startswith('fe80::1'):
return True
except Exception:
pass
try:
s = _sock.socket(_sock.AF_INET6, _sock.SOCK_STREAM)
s.bind(('::1', 0))
s.close()
return True
except Exception:
return False
def _check_ipv6_warning():
_ensure_dirs()
def _check_ipv6_warning() -> None:
ensure_dirs()
if IPV6_WARN_MARKER.exists():
return
if not _has_ipv6_enabled():
import socket as _sock
has = False
try:
for addr in _sock.getaddrinfo(_sock.gethostname(), None, _sock.AF_INET6):
ip = addr[4][0]
if ip and not ip.startswith("::1") and not ip.startswith("fe80::1"):
has = True
break
except Exception:
pass
if not has:
try:
s = _sock.socket(_sock.AF_INET6, _sock.SOCK_STREAM)
s.bind(("::1", 0))
s.close()
has = True
except Exception:
pass
if not has:
return
IPV6_WARN_MARKER.touch()
_show_info(
"На вашем компьютере включена поддержка подключения по IPv6.\n\n"
"Telegram может пытаться подключаться через IPv6, "
"что не поддерживается и может привести к ошибкам.\n\n"
"Если прокси не работает, попробуйте отключить "
"попытку соединения по IPv6 в настройках прокси Telegram.\n\n"
"Это предупреждение будет показано только один раз.")
"Это предупреждение будет показано только один раз."
)
# rumps menubar app
# rumps app
_TgWsProxyAppBase = rumps.App if rumps else object
@ -530,24 +492,26 @@ _TgWsProxyAppBase = rumps.App if rumps else object
class TgWsProxyApp(_TgWsProxyAppBase):
def __init__(self):
_ensure_menubar_icon()
icon_path = (str(MENUBAR_ICON_PATH)
if MENUBAR_ICON_PATH.exists() else None)
icon_path = str(MENUBAR_ICON_PATH) if MENUBAR_ICON_PATH.exists() else None
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
link_host = tg_ws_proxy.get_link_host(host)
self._open_tg_item = rumps.MenuItem(
f"Открыть в Telegram ({host}:{port})",
callback=_on_open_in_telegram)
self._restart_item = rumps.MenuItem(
"Перезапустить прокси",
callback=_on_restart)
self._settings_item = rumps.MenuItem(
"Настройки...",
callback=_on_edit_config)
self._logs_item = rumps.MenuItem(
"Открыть логи",
callback=_on_open_logs)
f"Открыть в Telegram ({link_host}:{port})", callback=_on_open_in_telegram
)
self._copy_link_item = rumps.MenuItem("Скопировать ссылку", callback=_on_copy_link)
self._restart_item = rumps.MenuItem("Перезапустить прокси", callback=_on_restart)
self._settings_item = rumps.MenuItem("Настройки...", callback=_on_edit_config)
self._logs_item = rumps.MenuItem("Открыть логи", callback=_on_open_logs)
self._release_page_item = rumps.MenuItem(
"Страница релиза на GitHub…", callback=_on_open_release_page
)
self._check_updates_item = rumps.MenuItem(
_check_updates_menu_title(), callback=_toggle_check_updates
)
self._version_item = rumps.MenuItem(f"Версия {__version__}", callback=lambda _: None)
super().__init__(
"TG WS Proxy",
@ -556,20 +520,30 @@ class TgWsProxyApp(_TgWsProxyAppBase):
quit_button="Выход",
menu=[
self._open_tg_item,
self._copy_link_item,
None,
self._restart_item,
self._settings_item,
self._logs_item,
])
None,
self._release_page_item,
self._check_updates_item,
None,
self._version_item,
],
)
def update_menu_title(self):
def update_menu_title(self) -> None:
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
self._open_tg_item.title = (
f"Открыть в Telegram ({host}:{port})")
link_host = tg_ws_proxy.get_link_host(host)
self._open_tg_item.title = f"Открыть в Telegram ({link_host}:{port})"
def run_menubar():
# entry point
def run_menubar() -> None:
global _app, _config
_config = load_config()
@ -581,22 +555,26 @@ def run_menubar():
except Exception:
pass
setup_logging(_config.get("verbose", False))
log.info("TG WS Proxy menubar app starting")
setup_logging(
_config.get("verbose", False),
log_max_mb=_config.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"]),
)
log.info("TG WS Proxy версия %s, menubar app starting", __version__)
log.info("Config: %s", _config)
log.info("Log file: %s", LOG_FILE)
if rumps is None or Image is None:
log.error("rumps or Pillow not installed; running in console mode")
start_proxy()
_start_proxy()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
stop_proxy()
_stop_proxy()
return
start_proxy()
_start_proxy()
_maybe_notify_update_async()
_show_first_run()
_check_ipv6_warning()
@ -604,19 +582,18 @@ def run_menubar():
log.info("Menubar app running")
_app.run()
stop_proxy()
_stop_proxy()
log.info("Menubar app exited")
def main():
if not _acquire_lock():
def main() -> None:
if not acquire_lock("macos.py"):
_show_info("Приложение уже запущено.")
return
try:
run_menubar()
finally:
_release_lock()
release_lock()
if __name__ == "__main__":

View File

@ -1 +1 @@
__version__ = "1.1.3"
__version__ = "1.4.0"

File diff suppressed because it is too large Load Diff

View File

@ -18,59 +18,37 @@ authors = [
keywords = [
"telegram",
"tdesktop",
"proxy",
"websocket"
"bypass",
"websocket",
"mtproto",
]
classifiers = [
"Development Status :: 5 - Production/Stable",
"Environment :: Console",
"Environment :: MacOS X :: Cocoa",
"Environment :: Win32 (MS Windows)",
"Environment :: X11 Applications :: GTK",
"Intended Audience :: Customer Service",
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: MacOS :: MacOS X",
"Operating System :: Microsoft :: Windows",
"Operating System :: POSIX :: Linux",
"Operating System :: OS Independent",
"Topic :: System :: Networking :: Firewalls",
]
dependencies = [
"pyperclip==1.9.0",
"psutil==5.9.8; platform_system == 'Windows' and python_version < '3.9'",
"cryptography==41.0.7; platform_system == 'Windows' and python_version < '3.9'",
"Pillow==10.4.0; platform_system == 'Windows' and python_version < '3.9'",
"psutil==7.0.0; platform_system != 'Windows' or python_version >= '3.9'",
"cryptography==46.0.5; platform_system != 'Windows' or python_version >= '3.9'",
]
"Pillow==12.1.1; (platform_system != 'Windows' or python_version >= '3.9') and platform_system != 'Darwin'",
[project.optional-dependencies]
win7 = [
"customtkinter==5.2.2",
"Pillow==10.4.0",
"psutil==5.9.8",
"pystray==0.19.5",
"pyperclip==1.9.0",
]
win10 = [
"customtkinter==5.2.2",
"Pillow==12.1.1",
"psutil==7.0.0",
"pystray==0.19.5",
"pyperclip==1.9.0",
]
macos = [
"Pillow==12.1.0",
"psutil==7.0.0",
"pyperclip==1.9.0",
"rumps==0.4.0",
]
linux = [
"customtkinter==5.2.2",
"Pillow==12.1.1",
"psutil==7.0.0",
"pystray==0.19.5",
"pyperclip==1.9.0",
"customtkinter==5.2.2; platform_system != 'Darwin'",
"pystray==0.19.5; platform_system != 'Darwin'",
"rumps==0.4.0; platform_system == 'Darwin'",
"Pillow==12.1.0; platform_system == 'Darwin'",
]
[project.scripts]
@ -84,7 +62,7 @@ Source = "https://github.com/Flowseal/tg-ws-proxy"
Issues = "https://github.com/Flowseal/tg-ws-proxy/issues"
[tool.hatch.build.targets.wheel]
packages = ["proxy"]
packages = ["proxy", "ui", "utils"]
[tool.hatch.build.force-include]
"windows.py" = "windows.py"

4
ui/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""
Интерфейс tray (CustomTkinter): тема, диалоги настроек, подсказки.
Ядро прокси пакет `proxy`.
"""

108
ui/ctk_theme.py Normal file
View File

@ -0,0 +1,108 @@
from __future__ import annotations
import sys
import tkinter
from dataclasses import dataclass
from typing import Any, Callable, Optional, Tuple
_tk_variable_del_guard_installed = False
def install_tkinter_variable_del_guard() -> None:
global _tk_variable_del_guard_installed
if _tk_variable_del_guard_installed:
return
_orig = tkinter.Variable.__del__
def _safe_variable_del(self: Any, _orig: Any = _orig) -> None:
try:
_orig(self)
except (RuntimeError, tkinter.TclError):
pass
tkinter.Variable.__del__ = _safe_variable_del # type: ignore[assignment]
_tk_variable_del_guard_installed = True
CONFIG_DIALOG_SIZE: Tuple[int, int] = (460, 560)
CONFIG_DIALOG_FRAME_PAD: Tuple[int, int] = (20, 14)
FIRST_RUN_SIZE: Tuple[int, int] = (520, 480)
FIRST_RUN_FRAME_PAD: Tuple[int, int] = (28, 24)
@dataclass(frozen=True)
class CtkTheme:
tg_blue: tuple = ("#3390ec", "#3390ec")
tg_blue_hover: tuple = ("#2b7cd4", "#2b7cd4")
bg: tuple = ("#ffffff", "#1e1e1e")
field_bg: tuple = ("#f0f2f5", "#2b2b2b")
field_border: tuple = ("#d6d9dc", "#3a3a3a")
text_primary: tuple = ("#000000", "#ffffff")
text_secondary: tuple = ("#707579", "#aaaaaa")
ui_font_family: str = "Sans"
mono_font_family: str = "Monospace"
def ctk_theme_for_platform() -> CtkTheme:
if sys.platform == "win32":
return CtkTheme(ui_font_family="Segoe UI", mono_font_family="Consolas")
return CtkTheme()
def apply_ctk_appearance(ctk: Any) -> None:
ctk.set_appearance_mode("auto")
ctk.set_default_color_theme("blue")
def center_ctk_geometry(root: Any, width: int, height: int) -> None:
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{width}x{height}+{(sw - width) // 2}+{(sh - height) // 2}")
def create_ctk_toplevel(
ctk: Any,
*,
title: str,
width: int,
height: int,
theme: CtkTheme,
topmost: bool = True,
after_create: Optional[Callable[[Any], None]] = None,
) -> Any:
root = ctk.CTkToplevel()
root.title(title)
root.resizable(False, False)
center_ctk_geometry(root, width, height)
root.configure(fg_color=theme.bg)
if topmost:
root.attributes("-topmost", True)
root.lift()
root.focus_force()
if after_create:
_after_id = root.after(300, lambda: after_create(root))
_orig_destroy = root.destroy
def _safe_destroy():
try:
root.after_cancel(_after_id)
except Exception:
pass
_orig_destroy()
root.destroy = _safe_destroy
return root
def main_content_frame(
ctk: Any,
root: Any,
theme: CtkTheme,
*,
padx: int,
pady: int,
) -> Any:
frame = ctk.CTkFrame(root, fg_color=theme.bg, corner_radius=0)
frame.pack(fill="both", expand=True, padx=padx, pady=pady)
return frame

109
ui/ctk_tooltip.py Normal file
View File

@ -0,0 +1,109 @@
from __future__ import annotations
import tkinter as tk
from typing import Any, List, Optional
class CtkTooltip:
def __init__(
self,
widget: Any,
text: str,
*,
delay_ms: int = 450,
wraplength: int = 320,
) -> None:
self.widget = widget
self.text = text
self.delay_ms = delay_ms
self.wraplength = wraplength
self._after_id: Optional[str] = None
self._tip: Optional[tk.Toplevel] = None
widget.bind("<Enter>", self._schedule, add="+")
widget.bind("<Leave>", self._hide, add="+")
widget.bind("<Button>", self._hide, add="+")
widget.bind("<Destroy>", self._on_destroy, add="+")
def _schedule(self, _event: Any = None) -> None:
if self.widget is None:
return
self._cancel_after()
self._after_id = self.widget.after(self.delay_ms, self._show)
def _cancel_after(self) -> None:
if self._after_id is not None:
try:
self.widget.after_cancel(self._after_id)
except Exception:
pass
self._after_id = None
def _show(self) -> None:
self._after_id = None
if self._tip is not None:
return
try:
if not self.widget.winfo_exists():
return
except Exception:
return
tw = tk.Toplevel(self.widget.winfo_toplevel())
tw.wm_overrideredirect(True)
try:
tw.wm_attributes("-topmost", True)
except Exception:
pass
tw.configure(bg="#2b2b2b")
lbl = tk.Label(
tw,
text=self.text,
justify="left",
wraplength=self.wraplength,
background="#2b2b2b",
foreground="#f0f0f0",
relief="flat",
borderwidth=0,
padx=10,
pady=8,
font=("Segoe UI", 10) if _is_windows() else None,
)
lbl.pack()
x = self.widget.winfo_rootx() + 12
y = self.widget.winfo_rooty() + self.widget.winfo_height() + 4
tw.wm_geometry(f"+{x}+{y}")
self._tip = tw
def _hide(self, _event: Any = None) -> None:
self._cancel_after()
if self._tip is not None:
try:
self._tip.destroy()
except Exception:
pass
self._tip = None
def _on_destroy(self, _event: Any = None) -> None:
self._hide()
self.widget = None
def _is_windows() -> bool:
import sys
return sys.platform == "win32"
def attach_ctk_tooltip(
widget: Any,
text: str,
*,
delay_ms: int = 450,
wraplength: int = 320,
) -> None:
CtkTooltip(widget, text, delay_ms=delay_ms, wraplength=wraplength)
def attach_tooltip_to_widgets(widgets: List[Any], text: str, **kwargs: Any) -> None:
for w in widgets:
attach_ctk_tooltip(w, text, **kwargs)

516
ui/ctk_tray_ui.py Normal file
View File

@ -0,0 +1,516 @@
from __future__ import annotations
import os
import webbrowser
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy import __version__
from utils.update_check import RELEASES_PAGE_URL, get_status
from ui.ctk_theme import (
FIRST_RUN_FRAME_PAD,
CtkTheme,
main_content_frame,
)
from ui.ctk_tooltip import attach_ctk_tooltip, attach_tooltip_to_widgets
_TIP_HOST = (
"Адрес, на котором прокси принимает подключения.\n"
"Обычно 127.0.0.1 — локальная сеть, 0.0.0.0 - все интерфейсы"
)
_TIP_PORT = (
"Порт прокси. В Telegram Desktop в настройках прокси должен быть "
"указан тот же порт"
)
_TIP_SECRET = "Секретный ключ для авторизации клиентов"
_TIP_DC = (
"Соответствие номера датацентра Telegram (DC) и IP-адреса сервера.\n"
"Каждая строка: «номер:IP», например 2:149.154.167.220. "
"Прокси по этим правилам направляет трафик к нужным серверам Telegram"
)
_TIP_VERBOSE = (
"Если включено, в файл логов пишется больше подробностей — "
"необходимо при поиске неполадок"
)
_TIP_BUF_KB = (
"Размер буфера приёма/передачи в килобайтах.\n"
"Больше значение — больше выделение памяти на сокет"
)
_TIP_POOL = (
"Сколько параллельных WebSocket-сессий к одному датацентру можно держать.\n"
"Увеличение может помочь при высокой нагрузке"
)
_TIP_LOG_MB = (
"Максимальный размер файла лога; при достижении лимита файл перезаписывается"
)
_TIP_AUTOSTART = (
"Запускать TG WS Proxy при входе в Windows. "
"Если вы переместите программу в другую папку, автозапуск сбросится"
)
_TIP_CHECK_UPDATES = "При запуске проверять наличие обновлений"
_TIP_SAVE = "Сохранить настройки"
_TIP_CANCEL = "Закрыть окно без сохранения изменений"
_INNER_W = 396
def _entry(ctk, parent, theme, *, var=None, width=0, height=36, radius=10, **kw):
opts = dict(
font=(theme.ui_font_family, 13), corner_radius=radius,
fg_color=theme.bg, border_color=theme.field_border,
border_width=1, text_color=theme.text_primary,
)
if var is not None:
opts["textvariable"] = var
if width:
opts["width"] = width
opts["height"] = height
opts.update(kw)
return ctk.CTkEntry(parent, **opts)
def _checkbox(ctk, parent, theme, text, variable):
return ctk.CTkCheckBox(
parent, text=text, variable=variable,
font=(theme.ui_font_family, 13), text_color=theme.text_primary,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
corner_radius=6, border_width=2, border_color=theme.field_border,
)
def _label(ctk, parent, theme, text, *, size=12, bold=False, secondary=True, **kw):
weight = "bold" if bold else "normal"
return ctk.CTkLabel(
parent, text=text,
font=(theme.ui_font_family, size, weight),
text_color=theme.text_secondary if secondary else theme.text_primary,
anchor="w", **kw,
)
def _labeled_entry(ctk, parent, theme, label_text, value, *, tip="", width=0, pack_fill=False):
col = ctk.CTkFrame(parent, fg_color="transparent")
lbl = _label(ctk, col, theme, label_text)
lbl.pack(anchor="w", pady=(0, 2))
var = ctk.StringVar(value=str(value))
ent = _entry(ctk, col, theme, var=var, width=width)
if pack_fill:
ent.pack(fill="x")
else:
ent.pack(anchor="w")
if tip:
attach_tooltip_to_widgets([lbl, ent, col], tip)
return col, var
def tray_settings_scroll_and_footer(
ctk: Any,
content_parent: Any,
theme: CtkTheme,
) -> Tuple[Any, Any]:
footer = ctk.CTkFrame(content_parent, fg_color=theme.bg)
footer.pack(side="bottom", fill="x")
scroll = ctk.CTkScrollableFrame(
content_parent,
fg_color=theme.bg,
corner_radius=0,
scrollbar_button_color=theme.field_border,
scrollbar_button_hover_color=theme.text_secondary,
)
scroll.pack(fill="both", expand=True)
return scroll, footer
def _config_section(
ctk: Any,
parent: Any,
theme: CtkTheme,
title: str,
*,
bottom_spacer: int = 6,
) -> Any:
wrap = ctk.CTkFrame(parent, fg_color="transparent")
wrap.pack(fill="x", pady=(0, bottom_spacer))
_label(ctk, wrap, theme, title, secondary=False, bold=True).pack(anchor="w", pady=(0, 2))
card = ctk.CTkFrame(
wrap, fg_color=theme.field_bg, corner_radius=10,
border_width=1, border_color=theme.field_border,
)
card.pack(fill="x")
inner = ctk.CTkFrame(card, fg_color="transparent")
inner.pack(fill="x", padx=10, pady=8)
return inner
@dataclass
class TrayConfigFormWidgets:
host_var: Any
port_var: Any
secret_var: Any
dc_textbox: Any
verbose_var: Any
adv_entries: List[Any]
adv_keys: Tuple[str, ...]
autostart_var: Optional[Any]
check_updates_var: Optional[Any]
def install_tray_config_form(
ctk: Any,
frame: Any,
theme: CtkTheme,
cfg: dict,
default_config: dict,
*,
show_autostart: bool = False,
autostart_value: bool = False,
) -> TrayConfigFormWidgets:
header = ctk.CTkFrame(frame, fg_color="transparent")
header.pack(fill="x", pady=(0, 2))
ctk.CTkLabel(
header, text="Настройки прокси",
font=(theme.ui_font_family, 17, "bold"),
text_color=theme.text_primary, anchor="w",
).pack(side="left")
ctk.CTkLabel(
header, text=f"v{__version__}",
font=(theme.ui_font_family, 12),
text_color=theme.text_secondary, anchor="e",
).pack(side="right")
conn = _config_section(ctk, frame, theme, "Подключение MTProto")
host_row = ctk.CTkFrame(conn, fg_color="transparent")
host_row.pack(fill="x")
host_col, host_var = _labeled_entry(
ctk, host_row, theme, "IP-адрес",
cfg.get("host", default_config["host"]),
tip=_TIP_HOST, width=160, pack_fill=True,
)
host_col.pack(side="left", fill="x", expand=True, padx=(0, 10))
port_col, port_var = _labeled_entry(
ctk, host_row, theme, "Порт",
cfg.get("port", default_config["port"]),
tip=_TIP_PORT, width=100,
)
port_col.pack(side="left")
secret_row = ctk.CTkFrame(conn, fg_color="transparent")
secret_row.pack(fill="x")
secret_col, secret_var = _labeled_entry(
ctk, secret_row, theme, "Secret",
cfg.get("secret", default_config["secret"]),
tip=_TIP_SECRET, width=160, pack_fill=True,
)
secret_col.pack(side="left", fill="x", expand=True, padx=(0, 10))
regen_col = ctk.CTkFrame(secret_row, fg_color="transparent")
regen_col.pack(side="left", anchor="s")
ctk.CTkLabel(regen_col, text="", font=(theme.ui_font_family, 12)).pack(pady=(0, 2))
ctk.CTkButton(
regen_col, text="", width=36, height=36,
font=(theme.ui_font_family, 18), corner_radius=10,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
text_color="#ffffff", border_width=1, border_color=theme.field_border,
command=lambda: secret_var.set(os.urandom(16).hex()),
).pack()
dc_inner = _config_section(ctk, frame, theme, "Датацентры Telegram (DC → IP)")
dc_lbl = _label(ctk, dc_inner, theme, "По одному правилу на строку, формат: номер:IP", size=11)
dc_lbl.pack(anchor="w", pady=(0, 4))
dc_textbox = ctk.CTkTextbox(
dc_inner, width=_INNER_W, height=88,
font=(theme.mono_font_family, 12), corner_radius=10,
fg_color=theme.bg, border_color=theme.field_border,
border_width=1, text_color=theme.text_primary,
)
dc_textbox.pack(fill="x")
dc_textbox.insert("1.0", "\n".join(cfg.get("dc_ip", default_config["dc_ip"])))
attach_tooltip_to_widgets([dc_lbl, dc_textbox], _TIP_DC)
log_inner = _config_section(ctk, frame, theme, "Логи и производительность")
verbose_var = ctk.BooleanVar(value=cfg.get("verbose", False))
verbose_cb = _checkbox(ctk, log_inner, theme, "Подробное логирование (verbose)", verbose_var)
verbose_cb.pack(anchor="w", pady=(0, 6))
attach_ctk_tooltip(verbose_cb, _TIP_VERBOSE)
adv_frame = ctk.CTkFrame(log_inner, fg_color="transparent")
adv_frame.pack(fill="x")
adv_rows = [
("Буфер, КБ (по умолчанию 256)", "buf_kb", _TIP_BUF_KB),
("Пул WebSocket-сессий (по умолчанию 4)", "pool_size", _TIP_POOL),
("Макс. размер лога, МБ (по умолчанию 5)", "log_max_mb", _TIP_LOG_MB),
]
for label_text, key, tip in adv_rows:
col = ctk.CTkFrame(adv_frame, fg_color="transparent")
col.pack(fill="x", pady=(0, 0 if key == "log_max_mb" else 5))
adv_l = _label(ctk, col, theme, label_text, size=11)
adv_l.pack(anchor="w", pady=(0, 2))
adv_e = _entry(
ctk, col, theme, width=_INNER_W, height=32, radius=8,
textvariable=ctk.StringVar(value=str(cfg.get(key, default_config[key]))),
)
adv_e.pack(fill="x")
attach_tooltip_to_widgets([adv_l, adv_e, col], tip)
adv_entries = list(adv_frame.winfo_children())
adv_keys = ("buf_kb", "pool_size", "log_max_mb")
upd_inner = _config_section(ctk, frame, theme, "Обновления")
st = get_status()
check_updates_var = ctk.BooleanVar(
value=bool(cfg.get("check_updates", default_config.get("check_updates", True)))
)
upd_cb = _checkbox(ctk, upd_inner, theme, "Проверять обновления при запуске", check_updates_var)
upd_cb.pack(anchor="w", pady=(0, 6))
attach_ctk_tooltip(upd_cb, _TIP_CHECK_UPDATES)
if st.get("error"):
upd_status = "Не удалось связаться с GitHub. Проверьте сеть."
elif not st.get("checked"):
upd_status = "Статус появится после фоновой проверки при запуске."
elif st.get("has_update") and st.get("latest"):
upd_status = (
f"На GitHub доступна версия {st['latest']} "
f"(у вас {__version__})."
)
elif st.get("ahead_of_release") and st.get("latest"):
upd_status = (
f"У вас {__version__} — новее последнего релиза на GitHub "
f"({st['latest']})."
)
else:
upd_status = "Установлена последняя известная версия с GitHub."
_label(ctk, upd_inner, theme, upd_status, size=11,
justify="left", wraplength=_INNER_W).pack(anchor="w", pady=(0, 8))
rel_url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
ctk.CTkButton(
upd_inner, text="Открыть страницу релиза", height=32,
font=(theme.ui_font_family, 13), corner_radius=8,
fg_color=theme.field_bg, hover_color=theme.field_border,
text_color=theme.text_primary, border_width=1,
border_color=theme.field_border,
command=lambda u=rel_url: webbrowser.open(u),
).pack(anchor="w")
autostart_var = None
if show_autostart:
sys_inner = _config_section(ctk, frame, theme, "Запуск Windows", bottom_spacer=4)
autostart_var = ctk.BooleanVar(value=autostart_value)
as_cb = _checkbox(ctk, sys_inner, theme, "Автозапуск при включении компьютера", autostart_var)
as_cb.pack(anchor="w", pady=(0, 4))
as_hint = _label(
ctk, sys_inner, theme,
"Если переместить программу в другую папку, запись автозапуска может сброситься.",
size=11, justify="left", wraplength=_INNER_W,
)
as_hint.pack(anchor="w")
attach_tooltip_to_widgets([as_cb, as_hint], _TIP_AUTOSTART)
return TrayConfigFormWidgets(
host_var=host_var, port_var=port_var, secret_var=secret_var,
dc_textbox=dc_textbox, verbose_var=verbose_var,
adv_entries=adv_entries, adv_keys=adv_keys,
autostart_var=autostart_var, check_updates_var=check_updates_var,
)
def merge_adv_from_form(
widgets: TrayConfigFormWidgets,
base: Dict[str, Any],
default_config: dict,
) -> None:
for i, key in enumerate(widgets.adv_keys):
col_frame = widgets.adv_entries[i]
entry = col_frame.winfo_children()[1]
try:
val = float(entry.get().strip())
if key in ("buf_kb", "pool_size"):
val = int(val)
base[key] = val
except ValueError:
base[key] = default_config[key]
def validate_config_form(
widgets: TrayConfigFormWidgets,
default_config: dict,
*,
include_autostart: bool,
) -> Union[dict, str]:
import socket as _sock
host_val = widgets.host_var.get().strip()
try:
_sock.inet_aton(host_val)
except OSError:
return "Некорректный IP-адрес."
try:
port_val = int(widgets.port_var.get().strip())
if not (1 <= port_val <= 65535):
raise ValueError
except ValueError:
return "Порт должен быть числом 1-65535"
lines = [
l.strip()
for l in widgets.dc_textbox.get("1.0", "end").strip().splitlines()
if l.strip()
]
try:
tg_ws_proxy.parse_dc_ip_list(lines)
except ValueError as e:
return str(e)
secret_val = widgets.secret_var.get().strip()
if len(secret_val) != 32:
return "Secret должен содержать ровно 32 hex-символа (16 байт)."
try:
bytes.fromhex(secret_val)
except ValueError:
return "Secret должен состоять только из hex-символов (0-9, a-f)."
new_cfg: Dict[str, Any] = {
"host": host_val,
"port": port_val,
"secret": secret_val,
"dc_ip": lines,
"verbose": widgets.verbose_var.get(),
}
if include_autostart:
new_cfg["autostart"] = (
widgets.autostart_var.get()
if widgets.autostart_var is not None
else False
)
merge_adv_from_form(widgets, new_cfg, default_config)
if widgets.check_updates_var is not None:
new_cfg["check_updates"] = bool(widgets.check_updates_var.get())
return new_cfg
def install_tray_config_buttons(
ctk: Any,
frame: Any,
theme: CtkTheme,
*,
on_save: Callable[[], None],
on_cancel: Callable[[], None],
) -> None:
ctk.CTkFrame(
frame,
fg_color=theme.field_border,
height=1,
corner_radius=0,
).pack(fill="x", pady=(4, 10))
btn_frame = ctk.CTkFrame(frame, fg_color="transparent")
btn_frame.pack(fill="x", pady=(0, 0))
save_btn = ctk.CTkButton(
btn_frame, text="Сохранить", height=38,
font=(theme.ui_font_family, 14, "bold"), corner_radius=10,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
text_color="#ffffff",
command=on_save)
save_btn.pack(side="left", fill="x", expand=True, padx=(0, 8))
attach_ctk_tooltip(save_btn, _TIP_SAVE)
cancel_btn = ctk.CTkButton(
btn_frame, text="Отмена", height=38,
font=(theme.ui_font_family, 14), corner_radius=10,
fg_color=theme.field_bg, hover_color=theme.field_border,
text_color=theme.text_primary, border_width=1,
border_color=theme.field_border,
command=on_cancel)
cancel_btn.pack(side="right", fill="x", expand=True)
attach_ctk_tooltip(cancel_btn, _TIP_CANCEL)
def populate_first_run_window(
ctk: Any,
root: Any,
theme: CtkTheme,
*,
host: str,
port: int,
secret: str,
on_done: Callable[[bool], None],
) -> None:
link_host = tg_ws_proxy.get_link_host(host)
tg_url = f"tg://proxy?server={link_host}&port={port}&secret=dd{secret}"
fpx, fpy = FIRST_RUN_FRAME_PAD
frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy)
title_frame = ctk.CTkFrame(frame, fg_color="transparent")
title_frame.pack(anchor="w", pady=(0, 16), fill="x")
accent_bar = ctk.CTkFrame(title_frame, fg_color=theme.tg_blue,
width=4, height=32, corner_radius=2)
accent_bar.pack(side="left", padx=(0, 12))
ctk.CTkLabel(title_frame, text="Прокси запущен и работает в системном трее",
font=(theme.ui_font_family, 17, "bold"),
text_color=theme.text_primary).pack(side="left")
sections = [
("Как подключить Telegram Desktop:", True),
(" Автоматически:", True),
(" ПКМ по иконке в трее → «Открыть в Telegram»", False),
(f" Или скопировать ссылку, отправить её себе в TG и нажать по ней: {tg_url}", False),
("\n Вручную:", True),
(" Настройки → Продвинутые → Тип подключения → Прокси", False),
(f" MTProto → {link_host} : {port}", False),
(f" Secret: dd{secret}", False),
]
textbox = ctk.CTkTextbox(
frame,
font=(theme.ui_font_family, 13),
fg_color=theme.bg,
border_width=0,
text_color=theme.text_primary,
activate_scrollbars=False,
wrap="word",
height=275,
)
textbox._textbox.tag_configure("bold", font=(theme.ui_font_family, 13, "bold"))
textbox._textbox.configure(spacing1=1, spacing3=1)
for text, bold in sections:
if text.startswith("\n"):
textbox.insert("end", "\n")
text = text[1:]
if bold:
textbox.insert("end", text + "\n", "bold")
else:
textbox.insert("end", text + "\n")
textbox.configure(state="disabled")
textbox.pack(anchor="w", fill="x")
ctk.CTkFrame(frame, fg_color="transparent", height=16).pack()
ctk.CTkFrame(frame, fg_color=theme.field_border, height=1,
corner_radius=0).pack(fill="x", pady=(0, 12))
auto_var = ctk.BooleanVar(value=True)
_checkbox(ctk, frame, theme, "Открыть прокси в Telegram сейчас",
auto_var).pack(anchor="w", pady=(0, 16))
def on_ok():
on_done(auto_var.get())
ctk.CTkButton(frame, text="Начать", width=180, height=42,
font=(theme.ui_font_family, 15, "bold"), corner_radius=10,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
text_color="#ffffff",
command=on_ok).pack(pady=(0, 0))
root.protocol("WM_DELETE_WINDOW", on_ok)

5
utils/__init__.py Normal file
View File

@ -0,0 +1,5 @@
"""Вспомогательные утилиты (проверка релизов и т.п.)."""
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
__all__ = ["RELEASES_PAGE_URL", "get_status", "run_check"]

30
utils/default_config.py Normal file
View File

@ -0,0 +1,30 @@
"""
Общие значения по умолчанию для tray-приложений (Windows / Linux / macOS).
Единственное отличие по платформе ключ autostart только на Windows.
"""
from __future__ import annotations
import sys
import os
from typing import Any, Dict
_TRAY_DEFAULTS_COMMON: Dict[str, Any] = {
"port": 1443,
"host": "127.0.0.1",
"dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
"verbose": False,
"check_updates": True,
"log_max_mb": 5,
"buf_kb": 256,
"pool_size": 4,
}
def default_tray_config() -> Dict[str, Any]:
cfg = dict(_TRAY_DEFAULTS_COMMON)
cfg["secret"] = os.urandom(16).hex()
if sys.platform == "win32":
cfg["autostart"] = False
return cfg

460
utils/tray_common.py Normal file
View File

@ -0,0 +1,460 @@
from __future__ import annotations
import asyncio
import json
import logging
import logging.handlers
import os
import socket as _socket
import sys
import threading
import time
from pathlib import Path
from typing import Any, Callable, Dict, Optional, Tuple
import psutil
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy import __version__
from utils.default_config import default_tray_config
log = logging.getLogger("tg-ws-tray")
APP_NAME = "TgWsProxy"
def _app_dir() -> Path:
if sys.platform == "win32":
return Path(os.environ.get("APPDATA", Path.home())) / APP_NAME
if sys.platform == "darwin":
return Path.home() / "Library" / "Application Support" / APP_NAME
return Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / APP_NAME
APP_DIR = _app_dir()
CONFIG_FILE = APP_DIR / "config.json"
LOG_FILE = APP_DIR / "proxy.log"
FIRST_RUN_MARKER = APP_DIR / ".first_run_done_mtproto"
IPV6_WARN_MARKER = APP_DIR / ".ipv6_warned"
DEFAULT_CONFIG: Dict[str, Any] = default_tray_config()
IS_FROZEN = bool(getattr(sys, "frozen", False))
def ensure_dirs() -> None:
APP_DIR.mkdir(parents=True, exist_ok=True)
# single-instance lock
_lock_file_path: Optional[Path] = None
def _same_process(meta: dict, proc: psutil.Process, script_hint: str) -> bool:
try:
lock_ct = float(meta.get("create_time", 0.0))
if lock_ct > 0 and abs(lock_ct - proc.create_time()) > 1.0:
return False
except Exception:
return False
if IS_FROZEN:
return APP_NAME.lower() in proc.name().lower()
try:
for arg in proc.cmdline():
if script_hint in arg:
return True
except Exception:
pass
return False
def acquire_lock(script_hint: str = "") -> bool:
global _lock_file_path
ensure_dirs()
for f in list(APP_DIR.glob("*.lock")):
try:
pid = int(f.stem)
except Exception:
f.unlink(missing_ok=True)
continue
meta: dict = {}
try:
raw = f.read_text(encoding="utf-8").strip()
if raw:
meta = json.loads(raw)
except Exception:
pass
try:
if _same_process(meta, psutil.Process(pid), script_hint):
return False
except Exception:
pass
f.unlink(missing_ok=True)
lock_file = APP_DIR / f"{os.getpid()}.lock"
try:
proc = psutil.Process(os.getpid())
lock_file.write_text(
json.dumps({"create_time": proc.create_time()}, ensure_ascii=False),
encoding="utf-8",
)
except Exception:
lock_file.touch()
_lock_file_path = lock_file
return True
def release_lock() -> None:
global _lock_file_path
if _lock_file_path:
try:
_lock_file_path.unlink(missing_ok=True)
except Exception:
pass
_lock_file_path = None
# config
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)
def save_config(cfg: dict) -> None:
ensure_dirs()
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(cfg, f, indent=2, ensure_ascii=False)
# logging
_LOG_FMT_FILE = "%(asctime)s %(levelname)-5s %(name)s %(message)s"
_LOG_FMT_CONSOLE = "%(asctime)s %(levelname)-5s %(message)s"
def setup_logging(verbose: bool = False, log_max_mb: float = 5) -> None:
ensure_dirs()
level = logging.DEBUG if verbose else logging.INFO
root = logging.getLogger()
root.setLevel(level)
fh = logging.handlers.RotatingFileHandler(
str(LOG_FILE),
maxBytes=max(32 * 1024, int(log_max_mb * 1024 * 1024)),
backupCount=0,
encoding="utf-8",
)
fh.setLevel(logging.DEBUG)
fh.setFormatter(logging.Formatter(_LOG_FMT_FILE, datefmt="%Y-%m-%d %H:%M:%S"))
root.addHandler(fh)
if not IS_FROZEN:
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(level)
ch.setFormatter(logging.Formatter(_LOG_FMT_CONSOLE, datefmt="%H:%M:%S"))
root.addHandler(ch)
# icon
def make_icon_image(size: int = 64, *, color: Tuple[int, ...] = (0, 136, 204, 255)):
from PIL import Image, ImageDraw, ImageFont
img = Image.new("RGBA", (size, size), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
margin = 2
draw.ellipse([margin, margin, size - margin, size - margin], fill=color)
for path in _font_paths():
try:
font = ImageFont.truetype(path, size=int(size * 0.55))
break
except Exception:
continue
else:
font = ImageFont.load_default()
bbox = draw.textbbox((0, 0), "T", font=font)
tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
draw.text(
((size - tw) // 2 - bbox[0], (size - th) // 2 - bbox[1]),
"T",
fill=(255, 255, 255, 255),
font=font,
)
return img
def _font_paths():
if sys.platform == "win32":
return ["arial.ttf"]
if sys.platform == "darwin":
return ["/System/Library/Fonts/Helvetica.ttc"]
return [
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
"/usr/share/fonts/TTF/DejaVuSans-Bold.ttf",
]
def load_icon():
from PIL import Image
icon_path = Path(__file__).parents[1] / "icon.ico"
if icon_path.exists():
try:
return Image.open(str(icon_path))
except Exception:
pass
return make_icon_image(64)
# proxy lifecycle
_proxy_thread: Optional[threading.Thread] = None
_async_stop: Optional[Tuple[asyncio.AbstractEventLoop, asyncio.Event]] = None
def _run_proxy_thread(on_port_busy: Callable[[str], None]) -> None:
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(stop_event=stop_ev))
except Exception as exc:
log.error("Proxy thread crashed: %s", exc)
if "Address already in use" in str(exc) or "10048" in str(exc):
on_port_busy(
"Не удалось запустить прокси:\n"
"Порт уже используется другим приложением.\n\n"
"Закройте приложение, использующее этот порт, "
"или измените порт в настройках прокси и перезапустите."
)
finally:
loop.close()
_async_stop = None
def apply_proxy_config(cfg: dict) -> bool:
dc_ip_list = cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])
try:
dc_redirects = tg_ws_proxy.parse_dc_ip_list(dc_ip_list)
except ValueError as e:
log.error("Bad config dc_ip: %s", e)
return False
pc = tg_ws_proxy.proxy_config
pc.port = cfg.get("port", DEFAULT_CONFIG["port"])
pc.host = cfg.get("host", DEFAULT_CONFIG["host"])
pc.secret = cfg.get("secret", DEFAULT_CONFIG["secret"])
pc.dc_redirects = dc_redirects
pc.buffer_size = max(4, cfg.get("buf_kb", DEFAULT_CONFIG["buf_kb"])) * 1024
pc.pool_size = max(0, cfg.get("pool_size", DEFAULT_CONFIG["pool_size"]))
return True
def start_proxy(cfg: dict, on_error: Callable[[str], None]) -> None:
global _proxy_thread
if _proxy_thread and _proxy_thread.is_alive():
log.info("Proxy already running")
return
if not apply_proxy_config(cfg):
on_error("Ошибка конфигурации DC → IP.")
return
pc = tg_ws_proxy.proxy_config
log.info("Starting proxy on %s:%d ...", pc.host, pc.port)
_proxy_thread = threading.Thread(
target=_run_proxy_thread, args=(on_error,), daemon=True, name="proxy"
)
_proxy_thread.start()
def stop_proxy() -> None:
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=5)
_proxy_thread = None
log.info("Proxy stopped")
def restart_proxy(cfg: dict, on_error: Callable[[str], None]) -> None:
log.info("Restarting proxy...")
stop_proxy()
time.sleep(0.3)
start_proxy(cfg, on_error)
def tg_proxy_url(cfg: dict) -> str:
host = cfg.get("host", DEFAULT_CONFIG["host"])
port = cfg.get("port", DEFAULT_CONFIG["port"])
secret = cfg.get("secret", DEFAULT_CONFIG["secret"])
link_host = tg_ws_proxy.get_link_host(host)
return f"tg://proxy?server={link_host}&port={port}&secret=dd{secret}"
_IPV6_WARNING = (
"На вашем компьютере включена поддержка подключения по IPv6.\n\n"
"Telegram может пытаться подключаться через IPv6, "
"что не поддерживается и может привести к ошибкам.\n\n"
"Если прокси не работает или в логах присутствуют ошибки, "
"связанные с попытками подключения по IPv6 - "
"попробуйте отключить в настройках прокси Telegram попытку соединения "
"по IPv6. Если данная мера не помогает, попробуйте отключить IPv6 "
"в системе.\n\n"
"Это предупреждение будет показано только один раз."
)
def _has_ipv6() -> bool:
try:
for addr in _socket.getaddrinfo(_socket.gethostname(), None, _socket.AF_INET6):
ip = addr[4][0]
if ip and not ip.startswith("::1") and not ip.startswith("fe80::1"):
return True
except Exception:
pass
try:
s = _socket.socket(_socket.AF_INET6, _socket.SOCK_STREAM)
s.bind(("::1", 0))
s.close()
return True
except Exception:
return False
def check_ipv6_warning(show_info: Callable[[str, str], None]) -> None:
ensure_dirs()
if IPV6_WARN_MARKER.exists() or not _has_ipv6():
return
IPV6_WARN_MARKER.touch()
threading.Thread(
target=lambda: show_info(_IPV6_WARNING, "TG WS Proxy"),
daemon=True,
).start()
# update check
def maybe_notify_update(
cfg: dict,
is_exiting: Callable[[], bool],
ask_open: Callable[[str, str], bool],
) -> None:
if not cfg.get("check_updates", True):
return
def _work():
time.sleep(1.5)
if is_exiting():
return
try:
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
import webbrowser
run_check(__version__)
st = get_status()
if not st.get("has_update"):
return
url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
ver = st.get("latest") or "?"
if ask_open(
f"Доступна новая версия: {ver}\n\nОткрыть страницу релиза в браузере?",
"TG WS Proxy — обновление",
):
webbrowser.open(url)
except Exception as exc:
log.debug("Update check failed: %s", exc)
threading.Thread(target=_work, daemon=True, name="update-check").start()
# ctk thread (windows / linux)
_ctk_root: Any = None
_ctk_root_ready = threading.Event()
def ensure_ctk_thread(ctk: Any) -> bool:
global _ctk_root
if ctk is None:
return False
if _ctk_root_ready.is_set():
return True
def _run():
global _ctk_root
from ui.ctk_theme import apply_ctk_appearance, install_tkinter_variable_del_guard
install_tkinter_variable_del_guard()
apply_ctk_appearance(ctk)
_ctk_root = ctk.CTk()
_ctk_root.withdraw()
_ctk_root_ready.set()
_ctk_root.mainloop()
threading.Thread(target=_run, daemon=True, name="ctk-root").start()
_ctk_root_ready.wait(timeout=5.0)
return _ctk_root is not None
def ctk_run_dialog(build_fn: Callable[[threading.Event], None]) -> None:
if _ctk_root is None:
return
done = threading.Event()
def _invoke():
try:
build_fn(done)
except Exception:
log.exception("CTk dialog failed")
done.set()
_ctk_root.after(0, _invoke)
done.wait()
import gc
gc.collect()
def quit_ctk() -> None:
if _ctk_root is not None:
try:
_ctk_root.after(0, _ctk_root.quit)
except Exception:
pass
# common bootstrap
def bootstrap(cfg: dict) -> None:
save_config(cfg)
if LOG_FILE.exists():
try:
LOG_FILE.unlink()
except Exception:
pass
setup_logging(
cfg.get("verbose", False),
log_max_mb=cfg.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"]),
)
log.info("TG WS Proxy версия %s starting", __version__)
log.info("Config: %s", cfg)
log.info("Log file: %s", LOG_FILE)

223
utils/update_check.py Normal file
View File

@ -0,0 +1,223 @@
"""
Минимальная проверка новой версии через GitHub Releases API (без сторонних зависимостей).
Ограничение частоты запросов: не чаще одного раза в час на машину (кэш в каталоге
данных приложения). Поддерживается If-None-Match (ETag) для ответа 304.
"""
from __future__ import annotations
import json
import os
import sys
import time
from itertools import zip_longest
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
REPO = "Flowseal/tg-ws-proxy"
RELEASES_LATEST_API = f"https://api.github.com/repos/{REPO}/releases/latest"
RELEASES_PAGE_URL = f"https://github.com/{REPO}/releases/latest"
# Не чаще одного полного запроса к API в час (без учёта 304 с тем же ETag).
_MIN_FETCH_INTERVAL_SEC = 3600.0
_state: Dict[str, Any] = {
"checked": False,
"has_update": False,
"ahead_of_release": False,
"latest": None,
"html_url": None,
"error": None,
}
def _cache_file() -> Optional[Path]:
try:
if sys.platform == "win32":
root = Path(os.environ.get("APPDATA", str(Path.home()))) / "TgWsProxy"
elif sys.platform == "darwin":
root = Path.home() / "Library/Application Support/TgWsProxy"
else:
xdg = os.environ.get("XDG_CONFIG_HOME")
root = (Path(xdg).expanduser() if xdg else Path.home() / ".config") / "TgWsProxy"
root.mkdir(parents=True, exist_ok=True)
return root / ".update_check_cache.json"
except OSError:
return None
def _load_cache(path: Optional[Path]) -> Dict[str, Any]:
if not path or not path.is_file():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
def _save_cache(path: Optional[Path], data: Dict[str, Any]) -> None:
if not path:
return
try:
path.write_text(json.dumps(data), encoding="utf-8")
except OSError:
pass
def _parse_version_tuple(s: str) -> tuple:
s = (s or "").strip().lstrip("vV")
if not s:
return (0,)
parts = []
for seg in s.split("."):
digits = "".join(c for c in seg if c.isdigit())
if digits:
try:
parts.append(int(digits))
except ValueError:
parts.append(0)
else:
parts.append(0)
return tuple(parts) if parts else (0,)
def _version_gt(a: str, b: str) -> bool:
"""True, если версия a новее b (простое сравнение по сегментам)."""
ta = _parse_version_tuple(a)
tb = _parse_version_tuple(b)
for x, y in zip_longest(ta, tb, fillvalue=0):
if x > y:
return True
if x < y:
return False
return False
def _apply_release_tag(
tag: str, html_url: str, current_version: str,
) -> None:
global _state
if not tag:
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = html_url.strip() or RELEASES_PAGE_URL
return
latest_clean = tag.lstrip("vV")
cur = (current_version or "").strip().lstrip("vV")
_state["latest"] = latest_clean
_state["html_url"] = html_url.strip() or RELEASES_PAGE_URL
_state["has_update"] = _version_gt(latest_clean, cur)
_state["ahead_of_release"] = bool(latest_clean) and _version_gt(
cur, latest_clean
)
def fetch_latest_release(
timeout: float = 12.0,
etag: Optional[str] = None,
) -> Tuple[Optional[dict], Optional[str], int]:
"""
GET releases/latest. Возвращает (data или None при 304, etag или None, HTTP-код).
"""
headers = {
"Accept": "application/vnd.github+json",
"User-Agent": "tg-ws-proxy-update-check",
}
if etag:
headers["If-None-Match"] = etag
req = Request(
RELEASES_LATEST_API,
headers=headers,
method="GET",
)
try:
with urlopen(req, timeout=timeout) as resp:
code = getattr(resp, "status", None) or resp.getcode()
new_etag = resp.headers.get("ETag")
raw = resp.read().decode("utf-8", errors="replace")
return json.loads(raw), new_etag, int(code)
except HTTPError as e:
if e.code == 304:
hdrs = e.headers
new_etag = hdrs.get("ETag") if hdrs else None
return None, new_etag or etag, 304
raise
def run_check(current_version: str) -> None:
"""Запрашивает последний релиз и обновляет внутреннее состояние."""
global _state
_state["checked"] = True
_state["error"] = None
cache_path = _cache_file()
cache = _load_cache(cache_path)
now = time.time()
last_attempt = float(cache.get("last_attempt_at") or 0)
if last_attempt and (now - last_attempt) < _MIN_FETCH_INTERVAL_SEC:
tag = (cache.get("tag_name") or "").strip()
if tag:
_apply_release_tag(tag, cache.get("html_url") or "", current_version)
return
err = cache.get("last_error")
_state["error"] = (
err if err else "Проверка обновлений отложена (интервал между запросами)."
)
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = RELEASES_PAGE_URL
return
etag = (cache.get("etag") or "").strip() or None
try:
data, new_etag, code = fetch_latest_release(etag=etag)
cache["last_attempt_at"] = now
if code == 304:
tag = (cache.get("tag_name") or "").strip()
url = (cache.get("html_url") or "").strip() or RELEASES_PAGE_URL
_apply_release_tag(tag, url, current_version)
if new_etag:
cache["etag"] = new_etag
_save_cache(cache_path, cache)
return
assert data is not None
tag = (data.get("tag_name") or "").strip()
html_url = (data.get("html_url") or "").strip() or RELEASES_PAGE_URL
if not tag:
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = html_url
else:
_apply_release_tag(tag, html_url, current_version)
if new_etag:
cache["etag"] = new_etag
cache["tag_name"] = tag
cache["html_url"] = html_url
cache.pop("last_error", None)
_save_cache(cache_path, cache)
except (HTTPError, URLError, OSError, TimeoutError, ValueError, json.JSONDecodeError) as e:
cache["last_attempt_at"] = now
msg = str(e)
if isinstance(e, HTTPError) and e.code == 403:
msg = (
"GitHub API вернул 403 (лимит или доступ). Повторите позже."
)
cache["last_error"] = msg
_save_cache(cache_path, cache)
_state["error"] = msg
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = RELEASES_PAGE_URL
def get_status() -> Dict[str, Any]:
"""Снимок состояния после run_check (для подписей в настройках)."""
return dict(_state)

View File

@ -1,176 +1,87 @@
from __future__ import annotations
import ctypes
import json
import logging
import os
import winreg
import psutil
import sys
import threading
import time
import webbrowser
import pyperclip
import asyncio as _asyncio
import winreg
from pathlib import Path
from typing import Dict, Optional
from typing import Optional
import pystray
import customtkinter as ctk
from PIL import Image, ImageDraw, ImageFont
try:
import pyperclip
except ImportError:
pyperclip = None
try:
import pystray
except ImportError:
pystray = None
try:
import customtkinter as ctk
except ImportError:
ctk = None
try:
from PIL import Image
except ImportError:
Image = None
import proxy.tg_ws_proxy as tg_ws_proxy
from utils.tray_common import (
APP_NAME, DEFAULT_CONFIG, FIRST_RUN_MARKER, IS_FROZEN, LOG_FILE,
acquire_lock, bootstrap, check_ipv6_warning, ctk_run_dialog,
ensure_ctk_thread, ensure_dirs, load_config, load_icon, log,
maybe_notify_update, quit_ctk, release_lock, restart_proxy,
save_config, start_proxy, stop_proxy, tg_proxy_url,
)
from ui.ctk_tray_ui import (
install_tray_config_buttons, install_tray_config_form,
populate_first_run_window, tray_settings_scroll_and_footer,
validate_config_form,
)
from ui.ctk_theme import (
CONFIG_DIALOG_FRAME_PAD, CONFIG_DIALOG_SIZE, FIRST_RUN_SIZE,
create_ctk_toplevel, ctk_theme_for_platform, main_content_frame,
)
IS_FROZEN = bool(getattr(sys, "frozen", False))
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,
"autostart": 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
_exiting = False
log = logging.getLogger("tg-ws-tray")
ICON_PATH = str(Path(__file__).parent / "icon.ico")
# win32 dialogs
_u32 = ctypes.windll.user32
_u32.MessageBoxW.argtypes = [ctypes.c_void_p, ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint]
_u32.MessageBoxW.restype = ctypes.c_int
_MB_OK_ERR = 0x10
_MB_OK_INFO = 0x40
_MB_YESNO_Q = 0x24
_IDYES = 6
def _same_process(lock_meta: dict, proc: psutil.Process) -> bool:
try:
lock_ct = float(lock_meta.get("create_time", 0.0))
proc_ct = float(proc.create_time())
if lock_ct > 0 and abs(lock_ct - proc_ct) > 1.0:
return False
except Exception:
return False
frozen = bool(getattr(sys, "frozen", False))
if frozen:
return os.path.basename(sys.executable) == proc.name()
return False
def _show_error(text: str, title: str = "TG WS Proxy — Ошибка") -> None:
_u32.MessageBoxW(None, text, title, _MB_OK_ERR)
def _release_lock():
global _lock_file_path
if not _lock_file_path:
return
try:
_lock_file_path.unlink(missing_ok=True)
except Exception:
pass
_lock_file_path = None
def _show_info(text: str, title: str = "TG WS Proxy") -> None:
_u32.MessageBoxW(None, text, title, _MB_OK_INFO)
def _acquire_lock() -> bool:
global _lock_file_path
_ensure_dirs()
lock_files = list(APP_DIR.glob("*.lock"))
for f in lock_files:
pid = None
meta: dict = {}
try:
pid = int(f.stem)
except Exception:
f.unlink(missing_ok=True)
continue
try:
raw = f.read_text(encoding="utf-8").strip()
if raw:
meta = json.loads(raw)
except Exception:
meta = {}
try:
proc = psutil.Process(pid)
if _same_process(meta, proc):
return False
except Exception:
pass
f.unlink(missing_ok=True)
lock_file = APP_DIR / f"{os.getpid()}.lock"
try:
proc = psutil.Process(os.getpid())
payload = {
"create_time": proc.create_time(),
}
lock_file.write_text(json.dumps(payload, ensure_ascii=False),
encoding="utf-8")
except Exception:
lock_file.touch()
_lock_file_path = lock_file
return True
def _ask_yes_no(text: str, title: str = "TG WS Proxy") -> bool:
return _u32.MessageBoxW(None, text, title, _MB_YESNO_Q) == _IDYES
def _ensure_dirs():
APP_DIR.mkdir(parents=True, exist_ok=True)
# autostart (registry)
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)
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 _autostart_reg_name() -> str:
return APP_NAME
_RUN_KEY = r"Software\Microsoft\Windows\CurrentVersion\Run"
def _supports_autostart() -> bool:
@ -183,542 +94,213 @@ def _autostart_command() -> str:
def is_autostart_enabled() -> bool:
try:
with winreg.OpenKey(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Run",
0,
winreg.KEY_READ,
) as k:
val, _ = winreg.QueryValueEx(k, _autostart_reg_name())
stored = str(val).strip()
expected = _autostart_command().strip()
return stored == expected
except FileNotFoundError:
return False
except OSError:
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, _RUN_KEY, 0, winreg.KEY_READ) as k:
val, _ = winreg.QueryValueEx(k, APP_NAME)
return str(val).strip() == _autostart_command().strip()
except (FileNotFoundError, OSError):
return False
def set_autostart_enabled(enabled: bool) -> None:
try:
with winreg.CreateKey(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Run",
) as k:
with winreg.CreateKey(winreg.HKEY_CURRENT_USER, _RUN_KEY) as k:
if enabled:
winreg.SetValueEx(
k,
_autostart_reg_name(),
0,
winreg.REG_SZ,
_autostart_command(),
)
winreg.SetValueEx(k, APP_NAME, 0, winreg.REG_SZ, _autostart_command())
else:
try:
winreg.DeleteValue(k, _autostart_reg_name())
winreg.DeleteValue(k, APP_NAME)
except FileNotFoundError:
pass
except OSError as exc:
log.error("Failed to update autostart: %s", exc)
_show_error(
"Не удалось изменить автозапуск.\n\n"
"Попробуйте запустить приложение от имени пользователя с правами на реестр.\n\n"
f"Ошибка: {exc}"
"Попробуйте запустить приложение от имени пользователя "
f"с правами на реестр.\n\nОшибка: {exc}"
)
def _make_icon_image(size: int = 64):
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)
# tray callbacks
margin = 2
draw.ellipse([margin, margin, size - margin, size - margin],
fill=(0, 136, 204, 255))
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():
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,
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()
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}"
def _on_open_in_telegram(icon=None, item=None) -> None:
url = tg_proxy_url(_config)
log.info("Opening %s", url)
try:
result = webbrowser.open(url)
if not result:
raise RuntimeError("webbrowser.open returned False")
if not webbrowser.open(url):
raise RuntimeError
except Exception:
log.info("Browser open failed, copying to clipboard")
if pyperclip is None:
_show_error(
"Не удалось открыть Telegram автоматически.\n\n"
f"Установите пакет pyperclip для копирования в буфер или откройте вручную:\n{url}"
)
return
try:
pyperclip.copy(url)
_show_info(
f"Не удалось открыть Telegram автоматически.\n\n"
f"Ссылка скопирована в буфер обмена, отправьте её в Telegram и нажмите по ней ЛКМ:\n{url}",
"TG WS Proxy")
"Не удалось открыть Telegram автоматически.\n\n"
f"Ссылка скопирована в буфер обмена, отправьте её в Telegram и нажмите по ней ЛКМ:\n{url}"
)
except Exception as exc:
log.error("Clipboard copy failed: %s", exc)
_show_error(f"Не удалось скопировать ссылку:\n{exc}")
def _on_restart(icon=None, item=None):
threading.Thread(target=restart_proxy, daemon=True).start()
def _on_copy_link(icon=None, item=None) -> None:
url = tg_proxy_url(_config)
log.info("Copying link: %s", url)
if pyperclip is None:
_show_error(
"Установите пакет pyperclip для копирования в буфер обмена."
)
return
try:
pyperclip.copy(url)
except Exception as exc:
log.error("Clipboard copy failed: %s", exc)
_show_error(f"Не удалось скопировать ссылку:\n{exc}")
def _on_edit_config(icon=None, item=None):
def _on_restart(icon=None, item=None) -> None:
threading.Thread(
target=lambda: restart_proxy(_config, _show_error), daemon=True
).start()
def _on_edit_config(icon=None, item=None) -> None:
threading.Thread(target=_edit_config_dialog, daemon=True).start()
def _edit_config_dialog():
if ctk is None:
_show_error("customtkinter не установлен.")
return
cfg = dict(_config)
cfg["autostart"] = is_autostart_enabled()
# Make sure that the autostart key is removed if autostart
# is disabled, even if the executable file is moved.
if _supports_autostart() and not cfg["autostart"]:
set_autostart_enabled(False)
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)
icon_path = str(Path(__file__).parent / "icon.ico")
root.iconbitmap(icon_path)
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, 460
if _supports_autostart():
h += 70
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)
# Host
ctk.CTkLabel(frame, text="IP-адрес прокси",
font=(FONT_FAMILY, 13), text_color=TEXT_PRIMARY,
anchor="w").pack(anchor="w", pady=(0, 4))
host_var = ctk.StringVar(value=cfg.get("host", "127.0.0.1"))
host_entry = ctk.CTkEntry(frame, textvariable=host_var, width=200, height=36,
font=(FONT_FAMILY, 13), corner_radius=10,
fg_color=FIELD_BG, border_color=FIELD_BORDER,
border_width=1, text_color=TEXT_PRIMARY)
host_entry.pack(anchor="w", pady=(0, 12))
# 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))
autostart_var = None
if _supports_autostart():
autostart_var = ctk.BooleanVar(value=cfg["autostart"])
ctk.CTkCheckBox(frame, text="Автозапуск при включении Windows",
variable=autostart_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))
ctk.CTkLabel(frame, text="При перемещении файла или открытии из другой папки\nавтозапуск будет сброшен",
font=(FONT_FAMILY, 13), text_color=TEXT_SECONDARY,
anchor="w", justify="left").pack(anchor="w", pady=(0, 8))
def on_save():
import socket as _sock
host_val = host_var.get().strip()
try:
_sock.inet_aton(host_val)
except OSError:
_show_error("Некорректный IP-адрес.")
return
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 = {
"host": host_val,
"port": port_val,
"dc_ip": lines,
"verbose": verbose_var.get(),
"autostart": (autostart_var.get() if autostart_var is not None else False),
}
save_config(new_cfg)
_config.update(new_cfg)
log.info("Config saved: %s", new_cfg)
if _supports_autostart():
set_autostart_enabled(bool(new_cfg.get("autostart", False)))
_tray_icon.menu = _build_menu()
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", pady=(20, 0))
ctk.CTkButton(btn_frame, text="Сохранить", 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", fill="x", expand=True, padx=(0, 8))
ctk.CTkButton(btn_frame, text="Отмена", 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="right", fill="x", expand=True)
root.mainloop()
def _on_open_logs(icon=None, item=None):
def _on_open_logs(icon=None, item=None) -> None:
log.info("Opening log file: %s", LOG_FILE)
if LOG_FILE.exists():
os.startfile(str(LOG_FILE))
else:
_show_info("Файл логов ещё не создан.", "TG WS Proxy")
_show_info("Файл логов ещё не создан.")
def _on_exit(icon=None, item=None):
def _on_exit(icon=None, item=None) -> 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()
quit_ctk()
threading.Thread(target=lambda: (time.sleep(3), os._exit(0)), daemon=True, name="force-exit").start()
if icon:
icon.stop()
# settings dialog
def _show_first_run():
_ensure_dirs()
def _edit_config_dialog() -> None:
if not ensure_ctk_thread(ctk):
_show_error("customtkinter не установлен.")
return
cfg = dict(_config)
cfg["autostart"] = is_autostart_enabled()
if _supports_autostart() and not cfg["autostart"]:
set_autostart_enabled(False)
def _build(done: threading.Event) -> None:
theme = ctk_theme_for_platform()
w, h = CONFIG_DIALOG_SIZE
if _supports_autostart():
h += 100
root = create_ctk_toplevel(
ctk, title="TG WS Proxy — Настройки", width=w, height=h, theme=theme,
after_create=lambda r: r.iconbitmap(ICON_PATH),
)
fpx, fpy = CONFIG_DIALOG_FRAME_PAD
frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy)
scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme)
widgets = install_tray_config_form(
ctk, scroll, theme, cfg, DEFAULT_CONFIG,
show_autostart=_supports_autostart(),
autostart_value=cfg.get("autostart", False),
)
def _finish() -> None:
root.destroy()
done.set()
def on_save() -> None:
from tkinter import messagebox
merged = validate_config_form(widgets, DEFAULT_CONFIG, include_autostart=_supports_autostart())
if isinstance(merged, str):
messagebox.showerror("TG WS Proxy — Ошибка", merged, parent=root)
return
save_config(merged)
_config.update(merged)
log.info("Config saved: %s", merged)
if _supports_autostart():
set_autostart_enabled(bool(merged.get("autostart", False)))
_tray_icon.menu = _build_menu()
do_restart = messagebox.askyesno(
"Перезапустить?",
"Настройки сохранены.\n\nПерезапустить прокси сейчас?",
parent=root,
)
_finish()
if do_restart:
threading.Thread(target=lambda: restart_proxy(_config, _show_error), daemon=True).start()
root.protocol("WM_DELETE_WINDOW", _finish)
install_tray_config_buttons(ctk, footer, theme, on_save=on_save, on_cancel=_finish)
ctk_run_dialog(_build)
# first run
def _show_first_run() -> None:
ensure_dirs()
if FIRST_RUN_MARKER.exists():
return
if not ensure_ctk_thread(ctk):
FIRST_RUN_MARKER.touch()
return
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
tg_url = f"tg://socks?server={host}&port={port}"
secret = _config.get("secret", DEFAULT_CONFIG["secret"])
if ctk is None:
FIRST_RUN_MARKER.touch()
return
def _build(done: threading.Event) -> None:
theme = ctk_theme_for_platform()
w, h = FIRST_RUN_SIZE
root = create_ctk_toplevel(
ctk, title="TG WS Proxy", width=w, height=h, theme=theme,
after_create=lambda r: r.iconbitmap(ICON_PATH),
)
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
def on_done(open_tg: bool) -> None:
FIRST_RUN_MARKER.touch()
root.destroy()
done.set()
if open_tg:
_on_open_in_telegram()
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"
populate_first_run_window(ctk, root, theme, host=host, port=port, secret=secret, on_done=on_done)
root = ctk.CTk()
root.title("TG WS Proxy")
root.resizable(False, False)
root.attributes("-topmost", True)
icon_path = str(Path(__file__).parent / "icon.ico")
root.iconbitmap(icon_path)
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 → {host} : {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()
ctk_run_dialog(_build)
def _has_ipv6_enabled() -> bool:
import socket as _sock
try:
addrs = _sock.getaddrinfo(_sock.gethostname(), None, _sock.AF_INET6)
for addr in addrs:
ip = addr[4][0]
if ip and not ip.startswith('::1') and not ip.startswith('fe80::1'):
return True
except Exception:
pass
try:
s = _sock.socket(_sock.AF_INET6, _sock.SOCK_STREAM)
s.bind(('::1', 0))
s.close()
return True
except Exception:
return False
def _check_ipv6_warning():
_ensure_dirs()
if IPV6_WARN_MARKER.exists():
return
if not _has_ipv6_enabled():
return
IPV6_WARN_MARKER.touch()
threading.Thread(target=_show_ipv6_dialog, daemon=True).start()
def _show_ipv6_dialog():
_show_info(
"На вашем компьютере включена поддержка подключения по IPv6.\n\n"
"Telegram может пытаться подключаться через IPv6, "
"что не поддерживается и может привести к ошибкам.\n\n"
"Если прокси не работает или в логах присутствуют ошибки, "
"связанные с попытками подключения по IPv6 - "
"попробуйте отключить в настройках прокси Telegram попытку соединения "
"по IPv6. Если данная мера не помогает, попробуйте отключить IPv6 "
"в системе.\n\n"
"Это предупреждение будет показано только один раз.",
"TG WS Proxy")
# tray menu
def _build_menu():
if pystray is None:
return None
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
link_host = tg_ws_proxy.get_link_host(host)
return pystray.Menu(
pystray.MenuItem(
f"Открыть в Telegram ({host}:{port})",
_on_open_in_telegram,
default=True),
pystray.MenuItem(f"Открыть в Telegram ({link_host}:{port})", _on_open_in_telegram, default=True),
pystray.MenuItem("Скопировать ссылку", _on_copy_link),
pystray.Menu.SEPARATOR,
pystray.MenuItem("Перезапустить прокси", _on_restart),
pystray.MenuItem("Настройки...", _on_edit_config),
@ -728,27 +310,17 @@ def _build_menu():
)
def run_tray():
# entry point
def run_tray() -> None:
global _tray_icon, _config
_config = load_config()
save_config(_config)
bootstrap(_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()
if pystray is None or Image is None or ctk is None:
log.error("pystray, Pillow or customtkinter not installed; running in console mode")
start_proxy(_config, _show_error)
try:
while True:
time.sleep(1)
@ -756,18 +328,12 @@ def run_tray():
stop_proxy()
return
start_proxy()
start_proxy(_config, _show_error)
maybe_notify_update(_config, lambda: _exiting, _ask_yes_no)
_show_first_run()
_check_ipv6_warning()
icon_image = _load_icon()
_tray_icon = pystray.Icon(
APP_NAME,
icon_image,
"TG WS Proxy",
menu=_build_menu())
check_ipv6_warning(_show_info)
_tray_icon = pystray.Icon(APP_NAME, load_icon(), "TG WS Proxy", menu=_build_menu())
log.info("Tray icon running")
_tray_icon.run()
@ -775,15 +341,14 @@ def run_tray():
log.info("Tray app exited")
def main():
if not _acquire_lock():
def main() -> None:
if not acquire_lock("windows.py"):
_show_info("Приложение уже запущено.", os.path.basename(sys.argv[0]))
return
try:
run_tray()
finally:
_release_lock()
release_lock()
if __name__ == "__main__":