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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""服务端：加 TCP 连接总次数计数（自检要用），diagnostics 直接读 status。"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows\lanremote")
|
|
SERVER = ROOT / "server.py"
|
|
DIAG = ROOT / "diagnostics.py"
|
|
|
|
# ---- server.py: 初始化计数 ----
|
|
INIT_OLD = ''' self._accept_thread: threading.Thread | None = None'''
|
|
INIT_NEW = ''' self._accept_thread: threading.Thread | None = None
|
|
# 累计收到过多少次 TCP 连接 —— 自检用它判断"手表的包有没有到"
|
|
self._total_connections = 0'''
|
|
|
|
# ---- server.py: accept 时累加 ----
|
|
ACCEPT_OLD = ''' conn, addr = self._sock.accept()'''
|
|
ACCEPT_NEW = ''' conn, addr = self._sock.accept()
|
|
self._total_connections += 1'''
|
|
|
|
# ---- server.py: status 里带上 ----
|
|
STATUS_OLD = ''' "discovery": self._discovery.status() if self._discovery else {"running": False},
|
|
"local_ip": local_ip(),'''
|
|
STATUS_NEW = ''' "discovery": self._discovery.status() if self._discovery else {"running": False},
|
|
"total_connections": self._total_connections,
|
|
"local_ip": local_ip(),'''
|
|
|
|
# ---- diagnostics.py: 改成读 status ----
|
|
DIAG_OLD = ''' # 服务端计数器
|
|
try:
|
|
status = server.status()
|
|
data["connected"] = status.get("connected", [])
|
|
data["pairing_code"] = status.get("pairing_code", "")
|
|
data["screen_sharing"] = status.get("screen_sharing", False)
|
|
stats = status.get("stats") or {}
|
|
data["tcp_connections"] = int(stats.get("connections", 0) or 0)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
# 自动发现的计数器
|
|
try:
|
|
discovery = getattr(server, "_discovery", None)
|
|
if discovery is not None:
|
|
info = discovery.status()
|
|
data["probes"] = int(info.get("answered", 0) or 0)
|
|
data["selfcheck_requests"] = int(info.get("selfcheck_requests", 0) or 0)
|
|
data["discovery_running"] = bool(info.get("running"))
|
|
data["discovery_error"] = info.get("error", "")
|
|
else:
|
|
data["discovery_running"] = False
|
|
data["discovery_error"] = "未启动"
|
|
except Exception: # noqa: BLE001
|
|
data["discovery_running"] = False
|
|
'''
|
|
|
|
DIAG_NEW = ''' # 服务端状态（含自动发现计数、累计 TCP 连接数）
|
|
try:
|
|
status = server.status()
|
|
data["connected"] = status.get("connected", [])
|
|
data["pairing_code"] = status.get("pairing_code", "")
|
|
data["screen_sharing"] = status.get("screen_sharing", False)
|
|
data["tcp_connections"] = int(status.get("total_connections", 0) or 0)
|
|
|
|
info = status.get("discovery") or {}
|
|
data["probes"] = int(info.get("answered", 0) or 0)
|
|
data["selfcheck_requests"] = int(info.get("selfcheck_requests", 0) or 0)
|
|
data["discovery_running"] = bool(info.get("running"))
|
|
data["discovery_error"] = info.get("error", "")
|
|
except Exception as exc: # noqa: BLE001
|
|
data["discovery_running"] = False
|
|
data["discovery_error"] = str(exc)
|
|
'''
|
|
|
|
|
|
def main() -> int:
|
|
ok = True
|
|
text = SERVER.read_text(encoding="utf-8")
|
|
for old, new, tag in ((INIT_OLD, INIT_NEW, "计数初始化"),
|
|
(ACCEPT_OLD, ACCEPT_NEW, "accept 累加"),
|
|
(STATUS_OLD, STATUS_NEW, "status 带出")):
|
|
if old not in text:
|
|
print(f" [失败] server.py {tag} 锚点未匹配")
|
|
ok = False
|
|
else:
|
|
text = text.replace(old, new, 1)
|
|
print(f" server.py / {tag}: OK")
|
|
if ok:
|
|
SERVER.write_text(text, encoding="utf-8")
|
|
|
|
dtext = DIAG.read_text(encoding="utf-8")
|
|
if DIAG_OLD not in dtext:
|
|
print(" [失败] diagnostics.py 锚点未匹配")
|
|
ok = False
|
|
else:
|
|
dtext = dtext.replace(DIAG_OLD, DIAG_NEW, 1)
|
|
DIAG.write_text(dtext, encoding="utf-8")
|
|
print(" diagnostics.py: OK")
|
|
|
|
print("\n语法检查：")
|
|
import ast
|
|
for name, path in (("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())
|