Files
xmz7891 f0c05079f7 2.1.7:手表快捷栏 + 鼠标触摸板 + 弹窗可配 + 后台远程更新
用户这次提了一整套升级,这是最后一批。

== 一、手表端顶部快捷栏 ==

用户原话:"手表端不管干什么,上面都有一栏,栏里的按钮可以在设置页
自定义,最多可以加 4 个,其中有一个按钮不能改,功能就是一键返回桌面"。

("返回桌面"经确认 = **让被控电脑显示桌面**,即 Win+D。)

  * 一条常驻的栏,贴在屏幕顶部,切页面时不动
  * **4 个格子**:3 个可自定义 + **1 个固定的「显示桌面」(改不了)**
  * 7 个可选动作:传文件 / 屏幕镜像 / 鼠标控制 / 音效 / 关机 /
    语音播报 / 投图到电脑
  * 设置页里点一下换;也能把某一格留空
  * 没连上电脑时按钮置灰(不是藏起来 —— "一直在但暂时不能用"
    比"时有时无"更好理解)

实现上的几个要点:

  * 快捷栏 addView 在 5 个面板**之后** —— FrameLayout 里后加的在上层,
    加在前面会被面板(MATCH_PARENT)整个盖住
  * 面板加了 topMargin 给栏让位,集中在 applyPanelTopMargin 里改
    (每个面板各写一遍很容易漏一个,表现是"就某一页顶部被切掉")
  * 第 4 格在代码里**写死** desktop,而且 QUICK_ACTIONS 里**没有**它 ——
    这样它既不会出现在可选列表里,也不可能被改成别的
  * 存储用 SharedPreferences。读取时**过滤不认识的 id + 补齐到 3 个**:
    以后删了某个动作,老配置也不会让栏变空或下标越界

== 二、远程鼠标控制(手表上的触摸板)==

  * 拖动 = 移动光标,单击 = 左键,双击 = 右键
  * 下面还有专门的 左键 / 右键 / 滚轮上 / 滚轮下 按钮
  * 电脑端用 SendInput 注入(见下)

== 三、弹文字窗口的位置和大小可配 ==

用户原话:"弹文字窗口 …… 位置和大小都能自定义"。

  * 位置:右下(默认,保持老习惯)/ 右上 / 左上 / 左下 /
    屏幕正中间 / 自定义坐标
  * 字号:8~32(夹住,免得设成 200 一个通知占满屏幕)
  * 宽度:可指定,0 = 按内容自适应
  * 三处窗口定位统一到一个 _notice_geometry() —— 以前三处各写死一遍
    右下角,改一次要改三个地方,漏一个用户就看得见

== 四、屏幕被观看时的提示条改成默认不显示 ==

用户原话:"以后不需要弹,啥窗户都没有,就可以像 UU 远程一样"。

  * 默认**连窗口都不建**
  * **但保留成开关**,设置里可以打开

  关于这条:那条提示原来是**故意加**的(有人在看屏幕时,电脑前面的人
  应该知道)。用户要求去掉,这是他自己的电脑、自己的教室,
  属于正当的远程管理需求。所以能力保留、默认关闭 ——
  这软件应该始终有一个"让本机用户看得见正在被控制"的选项可查。

== 五、中转后台能看到版本并远程点更新 ==

  * 设备表新增「版本」列
  * 每行一个「更新」按钮:点一下让那台电脑自己静默升级
  * 离线设备的按钮**置灰并说明原因**(命令只会挂在队列里等它上线,
    那时用户早忘了自己点过,突然重启很莫名其妙)
  * 和手表那边走**同一条命令**(cmd=update, action=start),
    不给"后台触发"单独写一套逻辑 —— 少一套就少一处不一致

== 六、手表端可开关「允许新设备连接」 ==

  * 关掉后只有配对过的电脑能连,教室里防别人乱连
  * 电脑端新增 _do_allow_new_pairing 这个 feature
  * 只影响**配对新设备**,不踢现有会话
    (否则老师一点"禁止",自己正连着的那台也被踢掉,成了"自杀开关")

== 修掉的几个自己造的坑 ==

1. **管理页 JS 裸换行** —— 最严重的一个。
   ADMIN_JS 是 Python 三引号字符串,我在里面写 `\n` 拼多行文案,
   结果是**真的换行**插进 JS,字符串字面量跨行 -> 语法非法 ->
   **整个 script 块解析失败 -> 管理页所有按钮全死**(改码/删除/踢下线/全选)。
   改成 String.fromCharCode(10),不依赖任何转义层。

2. **路由落进了 do_GET** —— 我拿 "没有这个管理接口" 当锚点,
   但那个字符串在文件里有**两处**(do_GET 和 _route_post 各一个),
   于是路由加到了 GET 块里,管理页的 POST 打不到它。
   表现是"点了更新没反应"。

3. **isConnected() 根本不存在** —— 写快捷栏时凭直觉调了个方法名,
   编译才发现。顺手包了一层并加上说明。

4. **两个不存在的方法名** —— onCreate 里写了
   buildMousePanelIfNeeded / mousePanelHolder,编译报 cannot find symbol。

5. **未调用方法扫描的假阳性** —— GestureDetector 的
   onSingleTapConfirmed / onDoubleTap / onScroll 是框架回调,
   扫描把"名字只出现 1 次"当成死代码。改成**跳过 @Override 的方法**。
   这种假阳性最坏的结果是有人为了让它变绿把回调删了,手势就失灵了。

6. **测试断言太严导致误报** —— "不能有写死的 240",
   但 helper 里 `auto = max(240, ...)` 是自适应的兜底,故意的。

7. **屏幕共享提示那个测试断言的是旧行为** —— 没有简单删掉,
   改成验两个方向:开关打开时提示条正常(位置/尺寸/出现/消失),
   关闭时默认值必须是 False 且代码路径上不建窗口。

== 验证 ==

    全量回归(19 关)全部 PASS,零失败
    新增关卡:3a10 静默更新 / 3a11 鼠标控制 / 3a12 后台更新+弹窗 /
              3a13 手表快捷栏
    Java 编译通过(android-34, --release 8)
    APK / 安装包 / 服务端包 全部重建,Win7 兼容校验通过
    623 个 Python 文件语法通过
2026-10-09 22:55:28 +08:00

589 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""静默运行模式:启动时没有任何窗口。
和 AgentUI 的区别:
没有主控制窗口、没有托盘、没有终端。
但 **屏幕共享提示保留** —— 有人在看你的屏幕时,屏幕右下角会出现一条提示。
这在有配对认证的前提下依然保留,理由:让电脑前面的人知道"此刻有人在看"。
内容窗口(投图、全屏大字、白板)是**手表明确要求显示**的,所以照常工作 ——
它们不是"程序自己弹出来的窗口"。Tk 根窗口按需惰性创建并保持隐藏。
没有 Tk 也能用的能力(剪贴板、屏幕分辨率)直接用 Win32 API 实现,
不依赖界面。
"""
from __future__ import annotations
import ctypes
import queue
import threading
import time
from pathlib import Path
IS_WINDOWS = False
try:
import tkinter as tk
IS_WINDOWS = True
except Exception: # noqa: BLE001
tk = None
# ======================================================================
# 不依赖 Tk 的系统能力
# ======================================================================
class _Clipboard:
"""直接读写 Windows 剪贴板,不需要 Tk。"""
CF_UNICODETEXT = 13
GMEM_MOVEABLE = 0x0002
@classmethod
def get(cls) -> str:
if not IS_WINDOWS:
return ""
user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32
# 64 位下必须声明,否则句柄会被截断
user32.GetClipboardData.restype = ctypes.c_void_p
kernel32.GlobalLock.restype = ctypes.c_void_p
kernel32.GlobalLock.argtypes = [ctypes.c_void_p]
kernel32.GlobalUnlock.argtypes = [ctypes.c_void_p]
if not user32.OpenClipboard(None):
return ""
try:
handle = user32.GetClipboardData(cls.CF_UNICODETEXT)
if not handle:
return ""
pointer = kernel32.GlobalLock(handle)
if not pointer:
return ""
try:
return ctypes.c_wchar_p(pointer).value or ""
finally:
kernel32.GlobalUnlock(handle)
finally:
user32.CloseClipboard()
@classmethod
def set(cls, text: str) -> bool:
if not IS_WINDOWS:
return False
user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32
kernel32.GlobalAlloc.restype = ctypes.c_void_p
kernel32.GlobalAlloc.argtypes = [ctypes.c_uint, ctypes.c_size_t]
kernel32.GlobalLock.restype = ctypes.c_void_p
kernel32.GlobalLock.argtypes = [ctypes.c_void_p]
kernel32.GlobalUnlock.argtypes = [ctypes.c_void_p]
user32.SetClipboardData.restype = ctypes.c_void_p
user32.SetClipboardData.argtypes = [ctypes.c_uint, ctypes.c_void_p]
data = (text + "\0").encode("utf-16-le")
handle = kernel32.GlobalAlloc(cls.GMEM_MOVEABLE, len(data))
if not handle:
return False
pointer = kernel32.GlobalLock(handle)
if not pointer:
return False
ctypes.memmove(pointer, data, len(data))
kernel32.GlobalUnlock(handle)
if not user32.OpenClipboard(None):
return False
try:
user32.EmptyClipboard()
return bool(user32.SetClipboardData(cls.CF_UNICODETEXT, handle))
finally:
user32.CloseClipboard()
# ======================================================================
# 静默 UI
# ======================================================================
class HeadlessUI:
"""给静默模式用的界面替身。
接口和 AgentUI 对齐,这样 Actions / RemoteServer 一行都不用改。
"""
def __init__(self, store, log_file: Path | None = None,
allow_content_windows: bool = True):
self.store = store
self.log_file = Path(log_file) if log_file else None
self.allow_content_windows = allow_content_windows
self._root = None # 惰性创建,见下面的 root 属性
self._tk_ready = False
self._tk_lock = threading.RLock()
self._ui_lock = threading.RLock()
self._queue: queue.Queue = queue.Queue()
self.server = None
self.state = "stopped"
self.port = 0
self._screen_on = False
self._notices = []
self._notice = None
self._notice_inner = None
self.mirroring = False
self._pump_thread: threading.Thread | None = None
self._ready = threading.Event()
self._stopping = False
# ---------------- 日志 ----------------
def log(self, message: str) -> None:
line = f"[{time.strftime('%H:%M:%S')}] {message}"
if self.log_file:
try:
with self.log_file.open("a", encoding="utf-8") as handle:
handle.write(line + "\n")
except OSError:
pass
# 别名,有些地方用 set_status
def set_status(self, text: str) -> None:
self.log(text)
def notify(self, title: str, text: str) -> None:
"""弹一个小通知窗(屏幕右下角,几秒后自动消失)。
之前这里只写日志,所以手表点「弹通知」电脑上毫无反应。
静默模式的意思是"开机不弹控制窗口",而不是
"手表明确要求显示的东西也不显示" —— 投图、全屏大字、通知
都属于"明确要求",照常显示。
"""
self.log(f"[通知] {title}: {text}")
root = self._ensure_root()
if root is None:
return
def build():
try:
win = tk.Toplevel(root)
win.title(f"通知 - {title}")
win.overrideredirect(True)
win.attributes("-topmost", True)
win.configure(bg="#ffffff")
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", self._notice_font(), "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", self._notice_font() - 1), 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 = self._notice_width(win)
height = win.winfo_reqheight()
win.geometry(self._notice_geometry(width, height))
def close(_event=None):
try:
win.destroy()
except Exception: # noqa: BLE001
pass
win.bind("<Button-1>", close)
win.after(6000, close)
except Exception as exc: # noqa: BLE001
self.log(f"通知显示失败: {exc}")
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", self._notice_font(), "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", self._notice_font() - 1), 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 = self._notice_width(win)
height = win.winfo_reqheight()
win.geometry(self._notice_geometry(width, height))
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
# ---------------- Tk 根窗口(惰性、隐藏)----------------
@property
def root(self):
"""Tk 根窗口,按需创建且**始终隐藏**。
AgentUI 已经删掉了,但 Actions 里投图/全屏大字这些还要一个父窗口。
这里惰性建一个 withdrew 的根窗口 —— 它本身永远不显示,
所以启动时屏幕上依然什么都没有。
"""
return self._ensure_root()
@root.setter
def root(self, value):
self._root = value
def _ensure_root(self):
"""按需启动界面线程,并等它把**隐藏的** Tk 根窗口建好。
关键:Tk 解释器必须在**同一个线程**里创建和使用。
所以根窗口由泵线程自己建 —— 如果在主线程建、在泵线程 update,
Tkinter 会因为线程亲和性直接抛异常,泵线程当场就死了。
"""
if not IS_WINDOWS or tk is None:
return None
if threading.current_thread() is self._pump_thread:
return self._root # 已经在界面线程里
with self._tk_lock:
if self._root is not None:
return self._root
if self._pump_thread is None:
self._ready.clear()
self._pump_thread = threading.Thread(
target=self._pump, daemon=True, name="headless-ui")
self._pump_thread.start()
self._ready.wait(8.0)
return self._root
def _pump(self) -> None:
"""界面线程:Tk 在这里建,也只有这里能碰它。"""
try:
root = tk.Tk()
root.withdraw()
root.title("")
except Exception as exc: # noqa: BLE001
self.log(f"创建界面根窗口失败: {exc}")
self._ready.set()
return
self._root = root
self._ready.set()
while not self._stopping:
try:
task = self._queue.get(timeout=0.05)
except queue.Empty:
task = None
if task is not None:
try:
task()
except Exception as exc: # noqa: BLE001
self.log(f"界面任务出错: {exc}")
try:
root.update()
except Exception: # noqa: BLE001
break
try:
root.destroy()
except Exception: # noqa: BLE001
pass
self._root = None
def ui_call(self, fn, timeout: float = 8.0):
"""在界面线程里执行 fn。
Tk 只允许在拥有它的那个线程里操作,所以这里一律丢给泵线程;
调用方本来就在泵线程里的话就地执行(否则会自己等自己)。
"""
if threading.current_thread() is self._pump_thread:
return fn()
root = self._ensure_root()
if root is None:
return fn() # 没有 Tk 环境(非 Windows),就地执行
result: list = []
error: list = []
done = threading.Event()
def task():
try:
result.append(fn())
except Exception as exc: # noqa: BLE001
error.append(exc)
finally:
done.set()
self._queue.put(task)
if not done.wait(timeout):
raise TimeoutError("界面响应超时")
if error:
raise error[0]
return result[0] if result else None
def _notice_width(self, win) -> int:
"""弹窗宽度。用户要求"大小可以自定义"。
设成 0 就按内容自适应(原来是 240 起),
设了具体值就用那个值,夹在 180~1200 之间 ——
太窄文字挤成一列,太宽一个通知横穿屏幕。
"""
try:
want = int(self.store.get("notice_width") or 0)
except (TypeError, ValueError):
want = 0
auto = max(240, win.winfo_reqwidth())
if want <= 0:
return auto
return max(180, min(1200, want))
def _notice_font(self) -> int:
"""弹窗字号。用户要求"大小可以自定义",所以从设置里读。
夹在 8~32 之间:太小看不清,太大一个通知就把屏幕占满了。
"""
try:
size = int(self.store.get("notice_font_size") or 11)
except (TypeError, ValueError):
size = 11
return max(8, min(32, size))
def _notice_geometry(self, width: int, height: int) -> str:
"""算出弹窗该放哪。返回 Tk 的 geometry 字符串 `+x+y`。
用户要求"位置和大小都能自定义",所以支持 6 种位置:
bottom-right(默认,保持老习惯) / top-right /
top-left / bottom-left / center(屏幕正中间) / custom
⚠ 三处弹窗(通知 ×2、屏幕提示 ×1)都调这个函数 ——
以前三处各写死一遍右下角,改一次得改三个地方。
"""
screen_w, screen_h = self.screen_size()
pos = str(self.store.get("notice_position") or "bottom-right")
pad = 24 # 离屏幕边缘留一点空
pad_bottom = 72 # 底下多留些,避开任务栏
if pos == "custom":
try:
x = int(self.store.get("notice_x") or 0)
y = int(self.store.get("notice_y") or 0)
except (TypeError, ValueError):
x = y = 0
# 夹到屏幕内 —— 设了个离谱的值也不至于让窗口飞到看不见的地方
x = max(0, min(max(0, screen_w - width), x))
y = max(0, min(max(0, screen_h - height), y))
return f"+{x}+{y}"
if pos == "center":
return (f"+{max(0, (screen_w - width) // 2)}"
f"+{max(0, (screen_h - height) // 2)}")
left = pos in ("top-left", "bottom-left")
top = pos in ("top-left", "top-right")
x = pad if left else screen_w - width - pad
y = pad if top else screen_h - height - pad_bottom
return f"+{max(0, x)}+{max(0, y)}"
def screen_size(self) -> tuple:
if IS_WINDOWS:
try:
user32 = ctypes.windll.user32
return (user32.GetSystemMetrics(0), user32.GetSystemMetrics(1))
except Exception: # noqa: BLE001
pass
return (1920, 1080)
def center_window(self, win, width: int = 0, height: int = 0) -> None:
pass # 静默模式不做居中
# ---------------- 服务端会用到的回调 ----------------
def on_state_changed(self, state: str) -> None:
self.state = state
def flash_screen_indicator(self, seconds: float = 2.5) -> None:
self.log("(截屏)")
def set_screen_sharing(self, on: bool) -> None:
"""屏幕共享提示条 —— **默认不显示**。
用户要求:"以后不需要弹,啥窗户都没有,就可以像 UU 远程一样"。
所以先看设置项 `show_screen_notice`:
* 关着(默认)-> 什么都不做,连窗口都不建
* 开着 -> 行为跟以前一样,在用户设的位置显示一条提示
保留这个开关而不是删掉代码,是因为这软件应该始终有一个
"让本机用户看得见正在被控制"的选项可查 —— 需要时在设置里打开。
"""
if on == self._screen_on:
return
self._screen_on = on
try:
if not self.store.get("show_screen_notice"):
# 设置关着:顺手把可能残留的提示条收掉
# (比如正被观看时用户把开关关了),然后什么都不做
self._hide_notice()
return
if on:
self._show_notice("屏幕正在被共享")
else:
self._hide_notice()
except Exception as exc: # noqa: BLE001
self.log(f"共享提示失败: {exc}")
def refresh_sharing_notice(self, status: dict, sharing: bool) -> None:
self.set_screen_sharing(sharing)
def refresh(self) -> None:
pass
def open_diagnostics(self, from_watch: str = "") -> None:
"""手表请求自检:静默模式下不弹窗口,只记日志。"""
self.log(f"手表 {from_watch or '?'} 请求了网络自检(静默模式不弹窗口)")
def _show_notice(self, text: str) -> None:
root = self._ensure_root()
if root is None:
return
def build():
win = tk.Toplevel(root)
# 标题要设 —— 无边框窗口在外观上看不到,
# 但系统里能查到这个标题,是"确实显示了提示"的可验证证据。
win.title(text)
win.overrideredirect(True)
win.attributes("-topmost", True)
win.configure(bg="#cf222e")
label = tk.Label(win, text=text, bg="#cf222e", fg="#ffffff",
font=("Microsoft YaHei UI", self._notice_font(), "bold"),
padx=16, pady=8)
label.pack()
screen_w, screen_h = self.screen_size()
win.update_idletasks()
width = win.winfo_reqwidth()
height = win.winfo_reqheight()
win.geometry(self._notice_geometry(width, height))
self._notice = win
try:
self._queue.put(build)
except Exception: # noqa: BLE001
pass
def _hide_notice(self) -> None:
win = self._notice
self._notice = None
if win is None:
return
def destroy():
try:
win.destroy()
except Exception: # noqa: BLE001
pass
try:
self._queue.put(destroy)
except Exception: # noqa: BLE001
pass
# ---------------- 设置窗口等:静默模式下没有 ----------------
def open_settings(self) -> None:
self.log("静默模式没有设置窗口,请改用带界面的模式或直接编辑 data/settings.json")
def open_help(self) -> None:
self.log("静默模式没有说明窗口")
def open_log(self) -> None:
self.log("静默模式没有日志窗口;日志写在 data/silent.log")
def show_message(self, text: str) -> None:
self.log(text)
def show_text_window(self, title: str, text: str) -> None:
self.log(f"[{title}] {text}")
def quit_app(self) -> None:
self.stop()
def stop(self) -> None:
self._stopping = True
# 不要在非界面线程里 destroy 根窗口 —— Tcl 会报
# "async handler deleted by the wrong thread"。
# 让泵线程自己收尾(它的循环退出时会 destroy)。
self._hide_notice()
thread = self._pump_thread
if thread is not None and thread.is_alive():
thread.join(timeout=2.0)
self._root = None
# ---------------- 静默模式自己的控制 ----------------
def run_forever(self) -> None:
"""一直跑着,直到收到停止信号。"""
while not self._stopping:
time.sleep(0.5)