astelm
/
tg-ws-proxy
mirror of https://github.com/Flowseal/tg-ws-proxy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: xdg-open not working

This commit is contained in:
Pitonic 2026-03-19 02:15:55 +03:00
parent d660f6a270
commit 9d732a3cce
1 changed files with 231 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

334
linux.py
View File

@ -1,29 +1,27 @@
from __future__ import annotations
import asyncio as _asyncio
import json
import logging
import os
import psutil
import subprocess
import sys
import threading
import time
import webbrowser
import pyperclip
import asyncio as _asyncio
from pathlib import Path
from typing import Dict, Optional
import pystray
import customtkinter as ctk
import psutil
import pyperclip
import pystray
from PIL import Image, ImageDraw, ImageFont
import proxy.tg_ws_proxy as tg_ws_proxy
APP_NAME = "TgWsProxy"
APP_DIR = Path(os.environ.get("XDG_CONFIG_HOME",
Path.home() / ".config")) / APP_NAME
APP_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / APP_NAME
CONFIG_FILE = APP_DIR / "config.json"
LOG_FILE = APP_DIR / "proxy.log"
FIRST_RUN_MARKER = APP_DIR / ".first_run_done"
@ -120,8 +118,7 @@ def _acquire_lock() -> bool:
payload = {
"create_time": proc.create_time(),
}
lock_file.write_text(json.dumps(payload, ensure_ascii=False),
encoding="utf-8")
lock_file.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
except Exception:
lock_file.touch()
@ -160,17 +157,22 @@ def setup_logging(verbose: bool = False):
fh = logging.FileHandler(str(LOG_FILE), encoding="utf-8")
fh.setLevel(logging.DEBUG)
fh.setFormatter(logging.Formatter(
fh.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname)-5s %(name)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%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"))
ch.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname)-5s %(message)s", datefmt="%H:%M:%S"
)
)
root.addHandler(ch)
@ -181,18 +183,20 @@ def _make_icon_image(size: int = 64):
draw = ImageDraw.Draw(img)
margin = 2
draw.ellipse([margin, margin, size - margin, size - margin],
fill=(0, 136, 204, 255))
draw.ellipse(
[margin, margin, size - margin, size - margin], fill=(0, 136, 204, 255)
)
try:
font = ImageFont.truetype(
"/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
size=int(size * 0.55))
size=int(size * 0.55),
)
except Exception:
try:
font = ImageFont.truetype(
"/usr/share/fonts/TTF/DejaVuSans-Bold.ttf",
size=int(size * 0.55))
"/usr/share/fonts/TTF/DejaVuSans-Bold.ttf", size=int(size * 0.55)
)
except Exception:
font = ImageFont.load_default()
bbox = draw.textbbox((0, 0), "T", font=font)
@ -214,9 +218,9 @@ def _load_icon():
return _make_icon_image()
def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool,
host: str = '127.0.0.1'):
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)
@ -225,11 +229,14 @@ def _run_proxy_thread(port: int, dc_opt: Dict[int, str], verbose: bool,
try:
loop.run_until_complete(
tg_ws_proxy._run(port, dc_opt, stop_event=stop_ev, host=host))
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 "Address already in use" in str(exc):
_show_error("Не удалось запустить прокси:\nПорт уже используется другим приложением.\n\nЗакройте приложение, использующее этот порт, или измените порт в настройках прокси и перезапустите.")
_show_error(
"Не удалось запустить прокси:\nПорт уже используется другим приложением.\n\nЗакройте приложение, использующее этот порт, или измените порт в настройках прокси и перезапустите."
)
finally:
loop.close()
_async_stop = None
@ -258,7 +265,9 @@ def start_proxy():
_proxy_thread = threading.Thread(
target=_run_proxy_thread,
args=(port, dc_opt, verbose, host),
daemon=True, name="proxy")
daemon=True,
name="proxy",
)
_proxy_thread.start()
@ -283,6 +292,7 @@ def restart_proxy():
def _show_error(text: str, title: str = "TG WS Proxy — Ошибка"):
import tkinter as _tk
from tkinter import messagebox as _mb
root = _tk.Tk()
root.withdraw()
_mb.showerror(title, text, parent=root)
@ -292,6 +302,7 @@ def _show_error(text: str, title: str = "TG WS Proxy — Ошибка"):
def _show_info(text: str, title: str = "TG WS Proxy"):
import tkinter as _tk
from tkinter import messagebox as _mb
root = _tk.Tk()
root.withdraw()
_mb.showinfo(title, text, parent=root)
@ -301,25 +312,14 @@ def _show_info(text: str, title: str = "TG WS Proxy"):
def _on_open_in_telegram(icon=None, item=None):
port = _config.get("port", DEFAULT_CONFIG["port"])
url = f"tg://socks?server=127.0.0.1&port={port}"
log.info("Opening %s", url)
try:
result = subprocess.call(['xdg-open', url])
if result != 0:
raise RuntimeError("xdg-open failed")
except Exception:
log.info("xdg-open failed, trying webbrowser")
try:
result = webbrowser.open(url)
if not result:
raise RuntimeError("webbrowser.open returned False")
except Exception:
log.info("Browser open failed, copying to clipboard")
log.info("Copying %s", url)
try:
pyperclip.copy(url)
_show_info(
f"Не удалось открыть Telegram автоматически.\n\n"
f"Ссылка скопирована в буфер обмена, отправьте её в Telegram и нажмите по ней ЛКМ:\n{url}",
"TG WS Proxy")
"TG WS Proxy",
)
except Exception as exc:
log.error("Clipboard copy failed: %s", exc)
_show_error(f"Не удалось скопировать ссылку:\n{exc}")
@ -351,6 +351,7 @@ def _edit_config_dialog():
icon_img = _load_icon()
if icon_img:
from PIL import ImageTk
_photo = ImageTk.PhotoImage(icon_img.resize((64, 64)))
root.iconphoto(False, _photo)
@ -366,61 +367,107 @@ def _edit_config_dialog():
w, h = 420, 480
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{w}x{h}+{(sw-w)//2}+{(sh-h)//2}")
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))
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 = 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))
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 = 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=("Monospace", 12), corner_radius=10,
fg_color=FIELD_BG, border_color=FIELD_BORDER,
border_width=1, text_color=TEXT_PRIMARY)
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=("Monospace", 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),
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))
fg_color=TG_BLUE,
hover_color=TG_BLUE_HOVER,
corner_radius=6,
border_width=2,
border_color=FIELD_BORDER,
).pack(anchor="w", pady=(0, 8))
# Info label
ctk.CTkLabel(frame, text="Изменения вступят в силу после перезапуска прокси.",
font=(FONT_FAMILY, 11), text_color=TEXT_SECONDARY,
anchor="w").pack(anchor="w", pady=(0, 16))
ctk.CTkLabel(
frame,
text="Изменения вступят в силу после перезапуска прокси.",
font=(FONT_FAMILY, 11),
text_color=TEXT_SECONDARY,
anchor="w",
).pack(anchor="w", pady=(0, 16))
def on_save():
import socket as _sock
host_val = host_var.get().strip()
try:
_sock.inet_aton(host_val)
@ -436,8 +483,11 @@ def _edit_config_dialog():
_show_error("Порт должен быть числом 1-65535")
return
lines = [l.strip() for l in dc_textbox.get("1.0", "end").strip().splitlines()
if l.strip()]
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:
@ -457,10 +507,12 @@ def _edit_config_dialog():
_tray_icon.menu = _build_menu()
from tkinter import messagebox
if messagebox.askyesno("Перезапустить?",
"Настройки сохранены.\n\n"
"Перезапустить прокси сейчас?",
parent=root):
if messagebox.askyesno(
"Перезапустить?",
"Настройки сохранены.\n\nПерезапустить прокси сейчас?",
parent=root,
):
root.destroy()
restart_proxy()
else:
@ -471,17 +523,32 @@ def _edit_config_dialog():
btn_frame = ctk.CTkFrame(frame, fg_color="transparent")
btn_frame.pack(fill="x")
ctk.CTkButton(btn_frame, text="Сохранить", width=140, height=38,
font=(FONT_FAMILY, 14, "bold"), corner_radius=10,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
ctk.CTkButton(
btn_frame,
text="Сохранить",
width=140,
height=38,
font=(FONT_FAMILY, 14, "bold"),
corner_radius=10,
fg_color=TG_BLUE,
hover_color=TG_BLUE_HOVER,
text_color="#ffffff",
command=on_save).pack(side="left", padx=(0, 10))
ctk.CTkButton(btn_frame, text="Отмена", width=140, height=38,
font=(FONT_FAMILY, 14), corner_radius=10,
fg_color=FIELD_BG, hover_color=FIELD_BORDER,
text_color=TEXT_PRIMARY, border_width=1,
command=on_save,
).pack(side="left", padx=(0, 10))
ctk.CTkButton(
btn_frame,
text="Отмена",
width=140,
height=38,
font=(FONT_FAMILY, 14),
corner_radius=10,
fg_color=FIELD_BG,
hover_color=FIELD_BORDER,
text_color=TEXT_PRIMARY,
border_width=1,
border_color=FIELD_BORDER,
command=on_cancel).pack(side="left")
command=on_cancel,
).pack(side="left")
root.mainloop()
@ -489,7 +556,19 @@ def _edit_config_dialog():
def _on_open_logs(icon=None, item=None):
log.info("Opening log file: %s", LOG_FILE)
if LOG_FILE.exists():
subprocess.Popen(['xdg-open', str(LOG_FILE)])
env = os.environ.copy()
env.pop("VIRTUAL_ENV", None)
env.pop("PYTHONPATH", None)
env.pop("PYTHONHOME", None)
subprocess.Popen(
["xdg-open", str(LOG_FILE)],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
start_new_session=True,
)
else:
_show_info("Файл логов ещё не создан.", "TG WS Proxy")
@ -505,13 +584,13 @@ def _on_exit(icon=None, item=None):
def _force_exit():
time.sleep(3)
os._exit(0)
threading.Thread(target=_force_exit, daemon=True, name="force-exit").start()
if icon:
icon.stop()
def _show_first_run():
_ensure_dirs()
if FIRST_RUN_MARKER.exists():
@ -545,13 +624,14 @@ def _show_first_run():
icon_img = _load_icon()
if icon_img:
from PIL import ImageTk
_photo = ImageTk.PhotoImage(icon_img.resize((64, 64)))
root.iconphoto(False, _photo)
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.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)
@ -561,13 +641,17 @@ def _show_first_run():
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 = 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="Прокси запущен и работает в системном трее",
ctk.CTkLabel(
title_frame,
text="Прокси запущен и работает в системном трее",
font=(FONT_FAMILY, 17, "bold"),
text_color=TEXT_PRIMARY).pack(side="left")
text_color=TEXT_PRIMARY,
).pack(side="left")
# Info sections
sections = [
@ -582,26 +666,37 @@ def _show_first_run():
for text, bold in sections:
weight = "bold" if bold else "normal"
ctk.CTkLabel(frame, text=text,
ctk.CTkLabel(
frame,
text=text,
font=(FONT_FAMILY, 13, weight),
text_color=TEXT_PRIMARY,
anchor="w", justify="left").pack(anchor="w", pady=1)
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))
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),
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))
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()
@ -610,11 +705,18 @@ def _show_first_run():
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,
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))
command=on_ok,
).pack(pady=(0, 0))
root.protocol("WM_DELETE_WINDOW", on_ok)
root.mainloop()
@ -622,17 +724,18 @@ def _show_first_run():
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'):
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.bind(("::1", 0))
s.close()
return True
except Exception:
@ -662,7 +765,8 @@ def _show_ipv6_dialog():
"по IPv6. Если данная мера не помогает, попробуйте отключить IPv6 "
"в системе.\n\n"
"Это предупреждение будет показано только один раз.",
"TG WS Proxy")
"TG WS Proxy",
)
def _build_menu():
@ -672,9 +776,8 @@ def _build_menu():
port = _config.get("port", DEFAULT_CONFIG["port"])
return pystray.Menu(
pystray.MenuItem(
f"Открыть в Telegram ({host}:{port})",
_on_open_in_telegram,
default=True),
f"Открыть в Telegram ({host}:{port})", _on_open_in_telegram, default=True
),
pystray.Menu.SEPARATOR,
pystray.MenuItem("Перезапустить прокси", _on_restart),
pystray.MenuItem("Настройки...", _on_edit_config),
@ -702,8 +805,7 @@ def run_tray():
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")
log.error("pystray or Pillow not installed; running in console mode")
start_proxy()
try:
while True:
@ -718,11 +820,7 @@ def run_tray():
_check_ipv6_warning()
icon_image = _load_icon()
_tray_icon = pystray.Icon(
APP_NAME,
icon_image,
"TG WS Proxy",
menu=_build_menu())
_tray_icon = pystray.Icon(APP_NAME, icon_image, "TG WS Proxy", menu=_build_menu())
log.info("Tray icon running")
_tray_icon.run()