mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 19:30:42 +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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
298 lines
12 KiB
Python
298 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""修三组问题。
|
|
|
|
【A】toggle 永远往同一个方向走（一个 bug 解释两条报障）
|
|
手表点开关时用的是**缓存里的旧状态**取反：
|
|
boolean on = Json.bool(feature, "on", false);
|
|
runFeature(name, Json.obj("on", !on), ...);
|
|
翻转成功后缓存没更新，于是：
|
|
任务栏：初始 false -> 发 true -> 隐藏。缓存还是 false
|
|
-> 再点还是发 true -> 永远隐藏，恢复不了
|
|
开机自启：初始 true -> 发 false -> 关闭。缓存还是 true
|
|
-> 再点还是发 false -> 永远打不开
|
|
修法：翻转成功后把缓存里的 on 一起改掉。
|
|
|
|
【B】_taskbar_hidden 读不准
|
|
用字符串 "03,00,00,00" 去匹配 reg query 的输出，
|
|
大小写和位置都不稳，状态永远读成 False。
|
|
改成把 REG_BINARY 解析成字节数组，看下标 8 的最低位。
|
|
|
|
【C】手表开的画面关不掉
|
|
close_windows 是关**别的程序**的窗口；
|
|
真正常关闭全屏的 big_text_clear 藏在「教学」组里、名字也不直观，
|
|
而且管不到投图 / 通知 / 屏幕共享提示。
|
|
新增一个「关掉电脑上的画面」，把所有本程序开的可见东西一次收掉，
|
|
并放到「屏幕」页最上面 —— 那是用户最先看到的地方。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
|
|
MAIN = ROOT / "android" / "src" / "com" / "lanremote" / "watch" / "MainActivity.java"
|
|
FEATURES = ROOT / "windows" / "lanremote" / "features.py"
|
|
HEADLESS = ROOT / "windows" / "lanremote" / "headless.py"
|
|
|
|
# ---------------- A. 手表 toggle 缓存 ----------------
|
|
|
|
A_OLD = ''' if ("toggle".equals(kind)) {
|
|
boolean on = Json.bool(feature, "on", false);
|
|
runFeature(name, Json.obj("on", !on), (on ? "关闭 " : "开启 ") + label);
|
|
return;
|
|
}'''
|
|
|
|
A_NEW = ''' if ("toggle".equals(kind)) {
|
|
final boolean on = Json.bool(feature, "on", false);
|
|
final boolean next = !on;
|
|
runFeature(name, Json.obj("on", next), (on ? "关闭 " : "开启 ") + label);
|
|
// 关键：本地缓存必须跟着翻转。
|
|
// 不翻转的话，下次点还是拿旧状态取反，永远往同一个方向走 ——
|
|
// 表现就是"能隐藏但恢复不了"、"显示已关闭但打不开"。
|
|
feature.put("on", Boolean.valueOf(next));
|
|
return;
|
|
}'''
|
|
|
|
# ---------------- B. 任务栏状态读取 ----------------
|
|
|
|
B_OLD = ''' def _taskbar_hidden(self) -> bool:
|
|
code, out = _run(["reg", "query",
|
|
r"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Explorer\\StuckRects3",
|
|
"/v", "Settings"])
|
|
if code != 0 or not out:
|
|
return False
|
|
return " 03 00 00 00 " in out.lower() or "03,00,00,00" in out'''
|
|
|
|
B_NEW = ''' def _taskbar_hidden(self) -> bool:
|
|
"""任务栏是不是自动隐藏。
|
|
|
|
reg query 把 REG_BINARY 输出成逗号分隔的十六进制：
|
|
Settings REG_BINARY 30,00,00,00,00,00,00,00,03,00,00,00,...
|
|
下标 8 那个字节的最低位就是「自动隐藏」开关。
|
|
|
|
以前是拿字符串 "03,00,00,00" 去整段匹配 —— 大小写和位置都不稳，
|
|
状态永远读成 False，于是手表每次点都发"隐藏"，
|
|
表现成"能隐藏但恢复不了"。这里老老实实按字节解析。
|
|
"""
|
|
code, out = _run(["reg", "query",
|
|
r"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
|
r"\\Explorer\\StuckRects3",
|
|
"/v", "Settings"])
|
|
if code != 0 or not out:
|
|
return False
|
|
match = re.search(r"REG_BINARY\\s+([0-9a-fA-F,\\s]+)", out)
|
|
if not match:
|
|
return False
|
|
values = [v for v in re.split(r"[,\\s]+", match.group(1).strip()) if v]
|
|
if len(values) <= 8:
|
|
return False
|
|
try:
|
|
return bool(int(values[8], 16) & 0x01)
|
|
except ValueError:
|
|
return False'''
|
|
|
|
# ---------------- C. 关掉电脑上的画面 ----------------
|
|
|
|
C_DEFS_OLD = ''' add("big_text_clear", "关掉全屏内容", "action", "教学")'''
|
|
|
|
C_DEFS_NEW = ''' add("big_text_clear", "关掉全屏内容", "action", "教学")
|
|
add("close_visuals", "关掉电脑上的画面", "action", "显示")'''
|
|
|
|
C_IMPL_OLD = ''' def _do_big_text_clear(self, params) -> str:
|
|
closed = self._close_fullscreen_windows()
|
|
return f"已关掉 {closed} 个全屏窗口" if closed else "当前没有全屏窗口"'''
|
|
|
|
C_IMPL_NEW = ''' def _do_big_text_clear(self, params) -> str:
|
|
closed = self._close_fullscreen_windows()
|
|
return f"已关掉 {closed} 个全屏窗口" if closed else "当前没有全屏窗口"
|
|
|
|
def _do_close_visuals(self, params) -> str:
|
|
"""把本程序在电脑上开出来的画面全部收掉。
|
|
|
|
覆盖：全屏大字 / 倒计时 / 请安静 / 下课铃 / 投图 / 通知 / 屏幕共享提示。
|
|
手表能打开的东西，必须也能关掉。
|
|
"""
|
|
done = []
|
|
|
|
closed = self._close_fullscreen_windows()
|
|
if closed:
|
|
done.append(f"全屏窗口 {closed} 个")
|
|
|
|
for label, method in (("投图", "close_image"),
|
|
("通知", "close_notifications")):
|
|
func = getattr(self.actions.ui, method, None)
|
|
if func is None:
|
|
continue
|
|
try:
|
|
if func():
|
|
done.append(label)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
# 屏幕镜像提示条也一起收掉（镜像本身由手表那个开关控制）
|
|
try:
|
|
hide = getattr(self.actions.ui, "_hide_notice", None)
|
|
if hide is not None and getattr(self.actions.ui, "_notice", None):
|
|
hide()
|
|
done.append("共享提示")
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
return "已关掉：" + "、".join(done) if done else "电脑上现在没有本程序开的画面"
|
|
|
|
def _do_quit_agent(self, params) -> str:
|
|
"""退出被控端（延迟一点点，好让回复先发出去）。"""
|
|
import threading
|
|
import time as _t
|
|
|
|
def die():
|
|
_t.sleep(1.0)
|
|
os._exit(0)
|
|
|
|
threading.Thread(target=die, daemon=True, name="quit").start()
|
|
return "被控端正在退出"'''
|
|
|
|
# ---------------- C2. headless 支持关闭通知 ----------------
|
|
|
|
C2_OLD = ''' try:
|
|
self._queue.put(build)
|
|
except Exception: # noqa: BLE001
|
|
pass'''
|
|
|
|
C2_NEW = ''' self._notices.append(None) # 先占位，build 里回填
|
|
index = len(self._notices) - 1
|
|
|
|
def build():
|
|
try:
|
|
win = tk.Toplevel(root)
|
|
win.title(f"通知 - {title}")
|
|
win.overrideredirect(True)
|
|
win.attributes("-topmost", True)
|
|
win.configure(bg="#ffffff")
|
|
self._notices[index] = win
|
|
|
|
box = tk.Frame(win, bg="#ffffff", highlightthickness=1,
|
|
highlightbackground="#d0d7de")
|
|
box.pack(fill="both", expand=True)
|
|
tk.Frame(box, bg="#185ce8", height=4).pack(fill="x")
|
|
tk.Label(box, text=title or "通知", bg="#ffffff",
|
|
fg="#1f2328", font=("Microsoft YaHei UI", 11, "bold"),
|
|
anchor="w").pack(fill="x", padx=14, pady=(10, 2))
|
|
tk.Label(box, text=text or "", bg="#ffffff", fg="#656d76",
|
|
font=("Microsoft YaHei UI", 10), anchor="w",
|
|
justify="left", wraplength=260).pack(
|
|
fill="x", padx=14, pady=(0, 12))
|
|
win.update_idletasks()
|
|
|
|
screen_w, screen_h = self.screen_size()
|
|
width = max(240, win.winfo_reqwidth())
|
|
height = win.winfo_reqheight()
|
|
win.geometry(f"+{screen_w - width - 24}+{screen_h - height - 72}")
|
|
|
|
def close(_event=None):
|
|
try:
|
|
win.destroy()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
try:
|
|
self._notices[index] = None
|
|
except (IndexError, AttributeError):
|
|
pass
|
|
|
|
win.bind("<Button-1>", close)
|
|
win.after(6000, close)
|
|
except Exception as exc: # noqa: BLE001
|
|
self.log(f"通知显示失败: {exc}")
|
|
|
|
try:
|
|
self._queue.put(build)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
|
|
def close_notifications(self) -> bool:
|
|
"""关掉所有还在显示的通知窗。返回是否关掉了至少一个。"""
|
|
handled = False
|
|
for index, win in enumerate(list(getattr(self, "_notices", []))):
|
|
if win is None:
|
|
continue
|
|
handled = True
|
|
|
|
def build(target=win, slot=index):
|
|
try:
|
|
target.destroy()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
try:
|
|
self._notices[slot] = None
|
|
except (IndexError, AttributeError):
|
|
pass
|
|
|
|
try:
|
|
self._queue.put(build)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return handled'''
|
|
|
|
|
|
def main() -> int:
|
|
ok = True
|
|
|
|
# A
|
|
text = MAIN.read_text(encoding="utf-8")
|
|
if A_OLD not in text:
|
|
print(" [失败] A: 手表 toggle 锚点没找到")
|
|
ok = False
|
|
else:
|
|
MAIN.write_text(text.replace(A_OLD, A_NEW, 1), encoding="utf-8")
|
|
print(" A: 手表 toggle 缓存现在会翻转")
|
|
|
|
# B + C
|
|
text = FEATURES.read_text(encoding="utf-8")
|
|
for old, new, tag in ((B_OLD, B_NEW, "B: _taskbar_hidden 按字节解析"),
|
|
(C_DEFS_OLD, C_DEFS_NEW, "C: 新功能声明"),
|
|
(C_IMPL_OLD, C_IMPL_NEW, "C: close_visuals + quit_agent")):
|
|
if old not in text:
|
|
print(f" [失败] {tag} —— 锚点没找到")
|
|
ok = False
|
|
continue
|
|
text = text.replace(old, new, 1)
|
|
print(f" {tag}")
|
|
FEATURES.write_text(text, encoding="utf-8")
|
|
|
|
# C2
|
|
text = HEADLESS.read_text(encoding="utf-8")
|
|
if C2_OLD not in text:
|
|
print(" [失败] C2: headless notify 锚点没找到")
|
|
ok = False
|
|
else:
|
|
HEADLESS.write_text(text.replace(C2_OLD, C2_NEW, 1), encoding="utf-8")
|
|
print(" C2: headless 支持关闭通知")
|
|
|
|
# _notices 初始化
|
|
text = HEADLESS.read_text(encoding="utf-8")
|
|
if "self._notices = []" not in text:
|
|
old = " self._notice = None"
|
|
if old in text:
|
|
text = text.replace(old, " self._notices = []\n" + old, 1)
|
|
HEADLESS.write_text(text, encoding="utf-8")
|
|
print(" C2b: 加了 _notices 列表")
|
|
else:
|
|
print(" [跳过] C2b: 找不到 _notice 初始化位置（放到 __init__ 末尾）")
|
|
# 退而求其次：用 getattr 兜底，close_notifications 里已经用了
|
|
else:
|
|
print(" C2b: _notices 已存在")
|
|
|
|
import ast
|
|
for label, path in (("features.py", FEATURES), ("headless.py", HEADLESS)):
|
|
try:
|
|
ast.parse(path.read_text(encoding="utf-8"))
|
|
print(f" {label}: 语法 OK")
|
|
except SyntaxError as exc:
|
|
print(f" {label}: 语法错误 {exc}")
|
|
ok = False
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|