mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 10:10:43 +00:00
xmz7891
b62ac865ea
被控端 - 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%） - 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败） - 每分钟回头找局域网，自动切回快 400 倍的路径 - 画面心跳帧：静止时省流量，客户端仍知道流是活的 - 锁屏时给出可操作的提示 手表端 - 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException - 连接错误翻译成人话（证书/时间/解析/超时分别提示） - 经中转连上后用服务器报的地址刷新已配对记录 - 新增局域网连接页面、服务器延迟检测 服务器端 - 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502） - 网络调优脚本 net-tune.sh（开机自动应用） - 管理页：批量删除、网络体检、下载区 - 隧道改 HTTP/2 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
292 lines
11 KiB
Python
292 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""电脑端全部图形化：
|
|
|
|
* 自检窗口（原来是控制台 .bat，现在做进界面）
|
|
* 使用说明窗口
|
|
* 运行日志窗口
|
|
* 底部一行短按钮：接收 / 自检 / 说明 / 设置 / 退出
|
|
* 手表发来的「连不上？」请求直接打开 GUI 自检窗口，不再启动控制台
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows\lanremote")
|
|
UI = ROOT / "ui.py"
|
|
SERVER = ROOT / "server.py"
|
|
DIAG = ROOT / "diagnostics.py"
|
|
|
|
# ---- 1. 措辞修正 ----
|
|
DIAG_OLD = ''' lines.append(f" 阻止规则: {fw.get('blocked_count', 0)} 条"
|
|
f"（其他程序 {fw.get('blocked_others', 0)} 条，不影响）")'''
|
|
DIAG_NEW = ''' lines.append(f" 阻止规则: {fw.get('blocked_count', 0)} 条（针对本程序）")
|
|
if fw.get("blocked_others"):
|
|
lines.append(f" 另外还有 {fw['blocked_others']} 条是针对别的程序的，不影响")'''
|
|
|
|
# ---- 2. 底栏改成一行五个短按钮 ----
|
|
FOOT_OLD = ''' foot = tk.Frame(self.root, bg=BG)
|
|
foot.pack(fill="x", side="bottom", pady=(0, 6), **pad)
|
|
self._btn(foot, "接收目录", self.open_received, width=9)
|
|
self._btn(foot, "连不上？", self.open_selfcheck, width=9)
|
|
self._btn(foot, "设置", self.open_settings, width=6)
|
|
self._btn(foot, "退出", self.quit_app, width=5, danger=True, last=True)
|
|
'''
|
|
|
|
FOOT_NEW = ''' foot = tk.Frame(self.root, bg=BG)
|
|
foot.pack(fill="x", side="bottom", pady=(0, 6), **pad)
|
|
self._btn(foot, "接收", self.open_received, width=5)
|
|
self._btn(foot, "自检", self.open_diagnostics, width=5)
|
|
self._btn(foot, "说明", self.open_help, width=5)
|
|
self._btn(foot, "设置", self.open_settings, width=5)
|
|
self._btn(foot, "退出", self.quit_app, width=5, danger=True, last=True)
|
|
'''
|
|
|
|
# ---- 3. 用 GUI 自检替换 open_selfcheck ----
|
|
SELFCHECK_OLD = ''' def open_selfcheck(self) -> None:
|
|
"""连不上时一键启动网络自检（开一个控制台窗口显示实时结果）。"""
|
|
import subprocess
|
|
|
|
from .config import app_dir
|
|
bat = app_dir() / "网络自检.bat"
|
|
if not bat.is_file():
|
|
self.show_message("没找到 网络自检.bat")
|
|
return
|
|
try:
|
|
flags = getattr(subprocess, "CREATE_NEW_CONSOLE", 0)
|
|
subprocess.Popen(["cmd", "/c", str(bat)], creationflags=flags)
|
|
except Exception as exc: # noqa: BLE001
|
|
self.show_message(f"启动自检失败：{exc}")
|
|
return
|
|
self.show_message("已打开网络自检窗口")
|
|
'''
|
|
|
|
SELFCHECK_NEW = ''' # ---- 网络自检（图形化，不用任何 .bat） ----
|
|
|
|
def open_diagnostics(self, from_watch: str = "") -> None:
|
|
"""打开自检窗口。
|
|
|
|
被控端自己就在监听端口，所以不用再去绑端口抢资源，
|
|
直接读自己的计数器就知道手表的包有没有到。
|
|
"""
|
|
from . import diagnostics as diag
|
|
|
|
if getattr(self, "_diag_win", None) is not None:
|
|
try:
|
|
self._diag_win.destroy()
|
|
except tk.TclError:
|
|
pass
|
|
self._diag_win = None
|
|
|
|
win = tk.Toplevel(self.root)
|
|
self._diag_win = win
|
|
win.title("网络自检 · 局域网遥控")
|
|
win.configure(bg=BG)
|
|
|
|
head = tk.Frame(win, bg=BG)
|
|
head.pack(fill="x", padx=14, pady=(12, 6))
|
|
tk.Frame(head, bg=ACCENT, width=4, height=18).pack(side="left", padx=(0, 8))
|
|
tk.Label(head, text="网络自检", bg=BG, fg=FG,
|
|
font=("Microsoft YaHei UI", 12, "bold")).pack(side="left")
|
|
|
|
if from_watch:
|
|
tk.Label(head, text=f"手表 {from_watch} 请求的", bg=BG, fg=ACCENT,
|
|
font=("Microsoft YaHei UI", 9)).pack(side="right")
|
|
|
|
verdict_label = tk.Label(win, text="正在检测…", bg=BG, fg=DIM,
|
|
font=("Microsoft YaHei UI", 10, "bold"),
|
|
anchor="w", justify="left", wraplength=400)
|
|
verdict_label.pack(fill="x", padx=14, pady=(0, 8))
|
|
|
|
text = tk.Text(win, height=18, width=54, bg=CARD, fg=FG, relief="flat",
|
|
font=("Consolas", 9), highlightthickness=1,
|
|
highlightbackground=BORDER, wrap="none")
|
|
text.pack(fill="both", expand=True, padx=14)
|
|
|
|
buttons = tk.Frame(win, bg=BG)
|
|
buttons.pack(fill="x", padx=14, pady=(8, 14))
|
|
self._btn(buttons, "重新检测", lambda: refresh(), width=9)
|
|
self._btn(buttons, "复制", lambda: self._copy_diag(text), width=6)
|
|
self._btn(buttons, "修复防火墙", self.fix_firewall, width=11)
|
|
self._btn(buttons, "关闭", win.destroy, width=6, last=True)
|
|
|
|
def refresh() -> None:
|
|
data = diag.collect(self.server, self.store)
|
|
level, message = diag.verdict(data)
|
|
colors = {"ok": OK, "warn": WARN, "error": DANGER}
|
|
heads = {"ok": "● 正常", "warn": "▲ 注意", "error": "✕ 有问题"}
|
|
verdict_label.configure(text=f"{heads.get(level, '')} {message}",
|
|
fg=colors.get(level, FG))
|
|
report = diag.render(data)
|
|
text.configure(state="normal")
|
|
text.delete("1.0", "end")
|
|
text.insert("1.0", report)
|
|
text.configure(state="disabled")
|
|
self.log(f"自检：{heads.get(level, '')} {message}")
|
|
|
|
refresh()
|
|
self.center_window(win)
|
|
|
|
def _copy_diag(self, widget) -> None:
|
|
try:
|
|
content = widget.get("1.0", "end").strip()
|
|
except tk.TclError:
|
|
return
|
|
self.root.clipboard_clear()
|
|
self.root.clipboard_append(content)
|
|
self.show_message(f"已复制 {len(content)} 个字符", 2.5)
|
|
|
|
# ---- 使用说明 ----
|
|
|
|
def open_help(self) -> None:
|
|
from .config import app_dir
|
|
|
|
candidates = [app_dir() / "使用说明.txt", app_dir() / "先看这个.txt"]
|
|
content = ""
|
|
for path in candidates:
|
|
if path.is_file():
|
|
try:
|
|
content = path.read_text(encoding="utf-8")
|
|
break
|
|
except OSError:
|
|
continue
|
|
if not content:
|
|
content = ("说明文件没找到。\\n\\n"
|
|
"简单说：\\n"
|
|
" 1. 这个窗口上的 6 位数字就是配对码\\n"
|
|
" 2. 手表上点「自动搜索电脑」再输这串数字\\n"
|
|
" 3. 连不上就点「自检」看看问题出在哪一段")
|
|
|
|
win = tk.Toplevel(self.root)
|
|
win.title("使用说明 · 局域网遥控")
|
|
win.configure(bg=BG)
|
|
|
|
text = tk.Text(win, height=24, width=62, bg=CARD, fg=FG, relief="flat",
|
|
font=("Microsoft YaHei UI", 9), highlightthickness=1,
|
|
highlightbackground=BORDER, wrap="word")
|
|
text.pack(fill="both", expand=True, padx=14, pady=(14, 8))
|
|
text.insert("1.0", content)
|
|
text.configure(state="disabled")
|
|
|
|
row = tk.Frame(win, bg=BG)
|
|
row.pack(fill="x", padx=14, pady=(0, 14))
|
|
self._btn(row, "关闭", win.destroy, width=8, last=True)
|
|
self.center_window(win)
|
|
|
|
# ---- 运行日志 ----
|
|
|
|
def open_log(self) -> None:
|
|
from .config import data_dir
|
|
|
|
path = data_dir() / "运行日志.txt"
|
|
content = ""
|
|
try:
|
|
if path.is_file():
|
|
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
content = "\\n".join(lines[-400:])
|
|
except OSError as exc:
|
|
content = f"读取失败：{exc}"
|
|
if not content:
|
|
content = "还没有日志。"
|
|
|
|
win = tk.Toplevel(self.root)
|
|
win.title("运行日志 · 局域网遥控")
|
|
win.configure(bg=BG)
|
|
|
|
text = tk.Text(win, height=22, width=68, bg=CARD, fg=FG, relief="flat",
|
|
font=("Consolas", 9), highlightthickness=1,
|
|
highlightbackground=BORDER, wrap="none")
|
|
text.pack(fill="both", expand=True, padx=14, pady=(14, 8))
|
|
text.insert("1.0", content)
|
|
text.see("end")
|
|
text.configure(state="disabled")
|
|
|
|
row = tk.Frame(win, bg=BG)
|
|
row.pack(fill="x", padx=14, pady=(0, 14))
|
|
self._btn(row, "打开文件", lambda: self._open_log_file(path), width=9)
|
|
self._btn(row, "关闭", win.destroy, width=6, last=True)
|
|
self.center_window(win)
|
|
|
|
def _open_log_file(self, path) -> None:
|
|
from .actions import open_path
|
|
try:
|
|
open_path(str(path))
|
|
except Exception as exc: # noqa: BLE001
|
|
self.show_message(f"打开失败：{exc}")
|
|
'''
|
|
|
|
# ---- 4. server: 请求改为打开 GUI 窗口 ----
|
|
SERVER_OLD = ''' from .config import app_dir
|
|
bat = app_dir() / "网络自检.bat"
|
|
if not bat.is_file():
|
|
self._log("[自检] 没找到 网络自检.bat，跳过")
|
|
return
|
|
try:
|
|
flags = getattr(subprocess, "CREATE_NEW_CONSOLE", 0)
|
|
subprocess.Popen(["cmd", "/c", str(bat)],
|
|
creationflags=flags,
|
|
cwd=str(app_dir()))
|
|
except Exception as exc: # noqa: BLE001
|
|
self._log(f"[自检] 启动失败：{exc}")
|
|
'''
|
|
|
|
SERVER_NEW = ''' target = getattr(self, "ui", None)
|
|
if target is not None and hasattr(target, "open_diagnostics"):
|
|
try:
|
|
target.ui_call(lambda: target.open_diagnostics(from_ip), timeout=3.0)
|
|
return
|
|
except Exception as exc: # noqa: BLE001
|
|
self._log(f"[自检] 打开界面失败：{exc}")
|
|
return
|
|
self._log("[自检] 没有可用的界面，跳过")
|
|
'''
|
|
|
|
|
|
def main() -> int:
|
|
ok = True
|
|
|
|
dtext = DIAG.read_text(encoding="utf-8")
|
|
if DIAG_OLD not in dtext:
|
|
print(" [跳过] diagnostics 措辞（可能已改）")
|
|
else:
|
|
dtext = dtext.replace(DIAG_OLD, DIAG_NEW, 1)
|
|
DIAG.write_text(dtext, encoding="utf-8")
|
|
print(" diagnostics.py 措辞: OK")
|
|
|
|
text = UI.read_text(encoding="utf-8")
|
|
for old, new, tag in ((FOOT_OLD, FOOT_NEW, "底栏"),
|
|
(SELFCHECK_OLD, SELFCHECK_NEW, "GUI 自检")):
|
|
if old not in text:
|
|
print(f" [失败] ui.py {tag} 锚点未匹配")
|
|
ok = False
|
|
else:
|
|
text = text.replace(old, new, 1)
|
|
print(f" ui.py / {tag}: OK")
|
|
if ok:
|
|
UI.write_text(text, encoding="utf-8")
|
|
|
|
stext = SERVER.read_text(encoding="utf-8")
|
|
if SERVER_OLD not in stext:
|
|
print(" [失败] server.py 锚点未匹配")
|
|
ok = False
|
|
else:
|
|
stext = stext.replace(SERVER_OLD, SERVER_NEW, 1)
|
|
SERVER.write_text(stext, encoding="utf-8")
|
|
print(" server.py: OK")
|
|
|
|
print("\n语法检查：")
|
|
import ast
|
|
for name, path in (("ui.py", UI), ("server.py", SERVER),
|
|
("diagnostics.py", DIAG)):
|
|
try:
|
|
ast.parse(path.read_text(encoding="utf-8"))
|
|
print(f" {name}: OK")
|
|
except SyntaxError as exc:
|
|
print(f" {name}: 语法错误 {exc}")
|
|
ok = False
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|