Files
ykxmz/windows/run_agent.py
T
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

980 lines
38 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 -*-
"""被控端启动入口。
双击 EXE 的完整流程:
第一次运行
弹出一个窗口「第一次使用」→ 输入 6 位配对码
可勾选:设置开机自启 / 顺便放行防火墙
↓ 点确定
黑终端一闪(提权跑一键修复防火墙,会先弹一次 UAC)
↓
窗口消失,程序转入后台静默运行 —— 屏幕上再没有任何东西
之后每次开机
直接后台静默运行,一个窗口都没有
手表
输那 6 位数字就连上
要重新设置:ykxmz.exe --setup
命令行参数:
--setup 重新打开设置窗口
--silent 直接静默运行(开机自启用的就是这个)
--add-device [分钟] 静默运行 + 临时允许新设备配对
--autostart on|off 开关开机自启
--status 打印状态后退出
--port 9000 指定端口
--new-code 换一组配对码
--code 123456 把配对码设成指定值(安装包用)
--name "三年二班" 设置电脑在手表上显示的名字
--reset-devices 清空已配对设备
"""
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
from lanremote import APP_NAME, __version__
from lanremote.config import Store, app_dir, data_dir
from lanremote.instance import InstanceLock
def parse_args(argv=None):
parser = argparse.ArgumentParser(
prog="lanremote-agent", description=f"{APP_NAME} v{__version__}")
parser.add_argument("--port", type=int, help="监听端口(默认 8765)")
parser.add_argument("--new-code", action="store_true", help="换一组配对码")
parser.add_argument("--code", default=None, metavar="6位数字",
help="把配对码设成指定值(必须是 6 位数字)。"
"安装包用这个把用户在安装向导里填的码写进去。")
parser.add_argument("--name", default=None, metavar="名字",
help="设置这台电脑在手表上显示的名字"
"(默认用 Windows 的计算机名)")
parser.add_argument("--reset-devices", action="store_true", help="清空已配对设备")
parser.add_argument("--silent", action="store_true",
help="全静默运行:没有任何窗口,也没有终端")
parser.add_argument("--add-device", nargs="?", type=int, const=10,
default=None, metavar="分钟",
help="静默运行,但临时允许新设备配对(默认 10 分钟)")
parser.add_argument("--autostart", choices=["on", "off"],
help="开关开机自启")
parser.add_argument("--status", action="store_true", help="打印状态后退出")
parser.add_argument("--setup", action="store_true",
help="重新打开设置窗口")
parser.add_argument("--netcheck", nargs="?", type=int, const=90,
default=None, metavar="秒",
help="网络自检(自开控制台,不需要装 Python)")
parser.add_argument("--fix-firewall", action="store_true",
help="提权修复防火墙")
parser.add_argument("--relay-url", default=None, metavar="地址",
help="只把中转地址写进配置,不连接、不退出(安装包用)")
parser.add_argument("--relay", default=None, metavar="地址",
help="设置中转服务器地址(跨网络用)。"
"可用分号分隔多个,按顺序尝试,例如:"
"http://192.168.1.10:8080;https://aaa.bbb.com"
" —— 局域网优先,断了自动切隧道。"
"传空字符串关闭(默认就是空,只走局域网)")
parser.add_argument("--relay-status", action="store_true",
help="查看中转连接状态")
parser.add_argument("--uninstall", action="store_true",
help="卸载:停进程、删开机自启、删数据目录")
parser.add_argument("--keep-dir", action="store_true",
help="配合 --uninstall:不安排自删程序目录。"
"安装包的卸载器调用时用这个 —— "
"Inno 自己会删 {app},两边一起删会抢文件")
parser.add_argument("--quiet", action="store_true",
help="配合 --uninstall:完全不碰控制台。"
"安装包的卸载器(runhidden)**必须**用这个,"
"否则 AllocConsole/input 会阻塞导致卸载卡死")
return parser.parse_args(argv)
# ======================================================================
# 命令行小工具
# ======================================================================
def cli_autostart(action: str) -> int:
from lanremote.autostart import enable, disable, status
try:
print(enable() if action == "on" else disable())
except ValueError as exc:
print(f"失败: {exc}")
return 1
info = status()
print(f"当前: {'已开启' if info['enabled'] else '未开启'}")
return 0
def cli_uninstall(keep_dir: bool = False, quiet: bool = False) -> int:
"""卸载。
keep_dir=True 不安排自删程序目录(安装包卸载器调用时用)
quiet=True 完全不碰控制台(安装包卸载器调用时**必须**用)
⚠ quiet 为什么必须:
安装包的卸载器用 runhidden 调这个命令,进程没有控制台。
这时候 attach_console() 里的 AllocConsole() + open("CONIN$")
会**永久阻塞** —— 现象就是"从控制面板卸载卡住不结束"。
(实测:卡住的进程只有 1 个线程、0.16 秒 CPU,明显在启动早期就停了。)
另外 input() 也不能调:没有 stdin 时它同样阻塞。
所以 quiet 模式下把结果写进日志文件,人要看就去看日志。
"""
from lanremote.uninstall import run as do_uninstall
if not quiet:
attach_console()
try:
report = do_uninstall(keep_dir=keep_dir)
except Exception as exc: # noqa: BLE001
report = f"卸载出错:{exc}"
code = 1
else:
code = 0
if quiet:
# 没有控制台,把结果落到文件,排查时能看。
#
# ⚠ 别写进数据目录!do_uninstall() 会把它整个删掉,
# 写到那里等于白写(实测:日志文件从来没出现过)。
# 放 %TEMP% 最稳妥。
try:
import tempfile
log = Path(tempfile.gettempdir()) / "ykxmz-uninstall.log"
log.write_text(report, encoding="utf-8")
except Exception: # noqa: BLE001
pass
return code
print(report)
print()
# 只有真有人在看的时候才等回车
if sys.stdin is not None and sys.stdin.isatty():
try:
input("按回车关闭…")
except (EOFError, KeyboardInterrupt):
pass
return code
def attach_console() -> None:
"""给 --windowed 程序自己开一个控制台窗口。
打包成 --windowed 后进程没有控制台,直接 print 什么都看不到。
AllocConsole 可以临时开一个,并把 stdout/stderr 接过去。
⚠ 调用方要小心:**在"隐藏窗口 + 没有控制台"的场景下这个函数会挂死**。
安装包的卸载器是用 runhidden 调 --uninstall 的,
AllocConsole() 之后的 open("CONIN$") 会一直阻塞,
表现就是"从控制面板卸载永远卡住不结束"
(实测:进程 1 个线程、0.16 秒 CPU,卡在启动早期)。
所以那条路径必须传 --quiet 跳过整个函数。
"""
if os.name != "nt":
return
try:
import ctypes
# 从 .bat 双击启动时已经有控制台了,再 AllocConsole 会失败并且输出丢失
if ctypes.windll.kernel32.GetConsoleWindow():
return
ctypes.windll.kernel32.AllocConsole()
sys.stdout = open("CONOUT$", "w", encoding="utf-8", buffering=1)
sys.stderr = open("CONOUT$", "w", encoding="utf-8", buffering=1)
try:
sys.stdin = open("CONIN$", "r", encoding="utf-8")
except OSError:
pass
except Exception: # noqa: BLE001
pass
def wait_key() -> None:
try:
print()
input("按回车键关闭这个窗口…")
except Exception: # noqa: BLE001
pass
class _Tee:
"""把输出同时送到控制台和文件。"""
def __init__(self, *streams):
self.streams = streams
def write(self, text):
for stream in self.streams:
try:
stream.write(text)
stream.flush()
except Exception: # noqa: BLE001
pass
def flush(self):
for stream in self.streams:
try:
stream.flush()
except Exception: # noqa: BLE001
pass
def cli_netcheck(seconds: int) -> int:
"""网络自检:监听端口,看手表的包到底有没有到这台电脑。
EXE 是 --windowed 打包的,本来没有控制台。这里自己开一个,
同时把结果写一份到「网络自检结果.txt」——
控制台万一没显示出来(被杀软拦、权限受限等),
用户还能从文件里看到结论。wlzj.bat 跑完会自动打开这个文件。
"""
attach_console()
original = sys.stdout
path = app_dir() / "网络自检结果.txt"
handle = None
try:
handle = open(path, "w", encoding="utf-8")
except OSError:
handle = None
if handle is not None:
sys.stdout = _Tee(original, handle)
code = 1
try:
from lanremote import netcheck
code = netcheck.run_check(int(seconds or 90))
except Exception as exc: # noqa: BLE001
print(f"自检运行失败: {exc}")
finally:
if handle is not None:
try:
sys.stdout = original
handle.close()
except Exception: # noqa: BLE001
pass
try:
print()
print(f"结果已保存到:{path}")
except Exception: # noqa: BLE001
pass
wait_key()
return code
def cli_fix_firewall() -> int:
"""提权修复防火墙(会弹一次 UAC,然后黑终端一闪)。"""
store = Store(data_dir())
ok, detail = fix_firewall_with_flash(store)
attach_console()
print(("已修复:" if ok else "未完成:") + str(detail))
wait_key()
return 0 if ok else 1
def normalize_relay_url(raw: str) -> str:
"""把用户填的中转地址规范化。
规则(和安装向导、设置窗口保持**完全一致**):
* 去掉首尾空白
* 没有 http:// / https:// 就补 https://
(用户基本只会填个域名,逼他打协议头没意义)
* 去掉末尾斜杠(拼接 "/v1/health" 时不会变成双斜杠)
* 空字符串 = 关掉中转,只走局域网
⚠ 为什么抽成函数:原来这段逻辑散在 cli_relay 里,安装向导
再写一份就容易两边不一致。现在统一走这里。
"""
url = (raw or "").strip()
if not url:
return ""
# 先去掉末尾斜杠,再补协议头(顺序反了会把 "https://" 的斜杠也削掉)
url = url.rstrip("/")
if not url.startswith(("http://", "https://")):
url = "https://" + url
return url
def cli_relay(url: str) -> int:
"""设置中转地址。传空字符串就关掉。"""
store = Store(data_dir())
url = normalize_relay_url(url)
store.set("relay_url", url)
attach_console()
if not url:
print("已关闭中转(以后只在局域网内可用)")
wait_key()
return 0
print(f"中转地址已设为:{url}")
print()
print("正在测试能不能连上…")
try:
from lanremote.relay import RelayLink
link = RelayLink(store)
# start() 的 url 是必填参数 —— 之前这里没传,
# 被 try 兜住后打成"✗ 测试失败:missing 1 required positional
# argument: 'url'",用户以为是自己地址填错了。
if not link.start(url):
print(" 启动失败")
wait_key()
return 1
# 等它登记成功
import time as _t
for _ in range(30):
_t.sleep(1)
status = link.status()
if status.get("online"):
print(f" ✓ 已连上中转(设备 {status.get('deviceId')},"
f"配对码 {status.get('code')})")
link.stop()
wait_key()
return 0
if status.get("error"):
print(f" … {status['error']}")
print(f" ✗ 30 秒内没能连上 最后错误:{link.status().get('error')}")
link.stop()
except Exception as exc: # noqa: BLE001
print(f" ✗ 测试失败:{exc}")
wait_key()
return 1
def cli_relay_status() -> int:
"""看现在的中转状态。"""
from lanremote.relay import RelayLink
store = Store(data_dir())
# 不给默认值:中转地址由用户自己填(--relay 或设置界面)。
# 以前这里兜底了一个写死的地址,结果是「我明明没填,它却去连一个
# 我根本不认识的服务器」,而且那个地址早过期了。
url = str(store.get("relay_url") or "")
attach_console()
if not url:
print("没有配置中转(只在局域网内可用)")
print("要跨网络请运行:ykxmz.exe --relay https://你的中转地址")
wait_key()
return 0
# start() 的 url 是必填参数。之前这里调 link.start() 不传参,
# 而且不在任何 try 里 —— 直接抛 TypeError 给用户看一堆堆栈。
try:
link = RelayLink(store)
link.start(url)
except Exception as exc: # noqa: BLE001
print("启动中转链路失败:%s" % exc)
wait_key()
return 1
print(f"中转地址:{url}")
print("正在测…")
import time as _t
for _ in range(30):
_t.sleep(1)
status = link.status()
if status.get("online"):
print(f" ✓ 在线")
print(f" 设备标识 {status.get('deviceId')}")
print(f" 配对码 {status.get('code')}")
print(f" 本次已服务 {status.get('sessions')} 个手表会话")
link.stop()
wait_key()
return 0
print(f" ✗ 不在线 最后错误:{link.status().get('error')}")
link.stop()
wait_key()
return 1
def cli_status() -> int:
from lanremote.autostart import status as auto_status
store = Store(data_dir())
auto = auto_status()
print(f"{APP_NAME} v{__version__}")
print(f" 数据目录: {data_dir()}")
print(f" 监听端口: {store.get('port', 8765)}")
print(f" 配对码: {store.pairing_code() or '(还没生成)'}")
print(f" 已配设备: {len(store.device_list())} 台")
print(f" 允许新配对: {'是' if store.get('allow_new_pairing', True) else '否'}")
print(f" 开机自启: {'已开启' if auto['enabled'] else '未开启'}")
if auto["enabled"]:
print(f" 命令行: {auto['command']}")
return 0
# 设完配对码之后留多久给手表来配对
SETUP_PAIR_MINUTES = 60
# 一台设备都没配对过时,自动开多久的配对窗口(自愈用)
FIRST_PAIR_MINUTES = 120
# ======================================================================
# 黑终端一闪:提权修防火墙
# ======================================================================
def fix_firewall_with_flash(store) -> tuple[bool, str]:
"""提权执行防火墙修复。
会先弹一次 UAC;确认之后黑色终端窗口一闪,script 跑完自动关掉。
这就是设置流程里那个"闪过的黑窗口"。
"""
from lanremote import firewall
from lanremote.discovery import DISCOVERY_PORT
tcp_port = int(store.get("port", 8765) or 8765)
udp_port = int(DISCOVERY_PORT)
try:
return firewall.apply_elevated(tcp_port, udp_port)
except Exception as exc: # noqa: BLE001
return False, f"防火墙修复失败: {exc}"
# ======================================================================
# 静默运行
# ======================================================================
def run_silent(store, allow_pair_minutes: int | None) -> int:
"""全静默运行:没有窗口,没有终端,只在后台监听。"""
import time
from lanremote.actions import Actions
from lanremote.headless import HeadlessUI
from lanremote.server import RemoteServer
log_file = data_dir() / "silent.log"
# 静默模式下默认**不接受新配对**(防止局域网里任何人接进来)。
# 需要加设备时用 --add-device 临时开一个时间窗。
#
# 自愈:如果一台设备都还没配对过,就必须把配对打开 ——
# 否则用户设完码会永远连不上(报"被控端已关闭新设备配对")。
# 配对码仍然是门槛,所以不影响安全;第一台配对成功后自动关闭。
if not store.device_list():
allow_pair_minutes = max(allow_pair_minutes or 0,
FIRST_PAIR_MINUTES)
store.set("close_pair_after_first", True)
if allow_pair_minutes:
store.set("allow_new_pairing", True)
deadline = time.time() + max(1, allow_pair_minutes) * 60
else:
store.set("allow_new_pairing", False)
deadline = 0.0
ui = HeadlessUI(store, log_file=log_file)
ui.log(f"{APP_NAME} v{__version__} 静默启动")
if not store.get("setup_done", False):
ui.log("注意:还没有设置过配对码。请先运行:ykxmz.exe --setup")
ui.log(f"当前临时配对码是 {store.pairing_code()}")
lock = InstanceLock()
if not lock.try_claim():
ui.log("已有实例在运行,退出")
return 0
server = RemoteServer(store, None, ui=ui)
server.actions = Actions(ui, store)
from lanremote.features import Features
server.features = Features(server.actions)
server.pairing_enabled = bool(store.get("allow_new_pairing", True))
ui.server = server
try:
server.start()
except (OSError, OverflowError, ValueError) as exc:
ui.log(f"启动失败:{exc}")
lock.release()
return 1
code = store.pairing_code()
if allow_pair_minutes and code:
ui.log(f"配对窗口已开 {allow_pair_minutes} 分钟,配对码 {code}")
ui.log(f"正在监听 TCP {server.port} / UDP 8766")
import threading
def watchdog():
time.sleep(1.0)
while not ui._stopping:
time.sleep(5.0)
if allow_pair_minutes and deadline and time.time() > deadline:
store.set("allow_new_pairing", False)
server.pairing_enabled = False
ui.log("配对窗口已关闭")
threading.Thread(target=watchdog, daemon=True, name="pair-window").start()
try:
ui.run_forever()
except KeyboardInterrupt:
pass
finally:
try:
server.stop()
except Exception: # noqa: BLE001
pass
ui.stop()
lock.release()
# 必须用 os._exit 结束。
# 界面线程拥有 Tk 解释器,正常走解释器退出会报
# "Tcl_AsyncDelete: async handler deleted by the wrong thread",
# 而且进程退出码会变成 -2147483645(崩溃码)。
# os._exit 跳过解释器收尾,干净退出。
sys.stdout.flush()
sys.stderr.flush()
os._exit(0)
# ======================================================================
# 首次设置
# ======================================================================
def check_code_available(store, code: str):
"""这个配对码能用吗?
能用(或没法判断)返回 None;被别的电脑占了返回一段说明文字。
检查走中转的只读接口 /v1/code/check —— 不注册,所以不会误占。
连不上中转时返回 None(放行):不能因为服务器连不上就装不了,
真有问题等转静默之后还会再提示一次。
"""
if not code or len(code) != 6 or not code.isdigit():
return "配对码必须是 6 位数字"
target = str(store.get("relay_url") or "").strip()
if not target:
# 没配过就用内置默认值(定义在 server.py)
from lanremote.server import DEFAULT_RELAY_URL
target = DEFAULT_RELAY_URL
if not target.strip():
return None # 没配中转,没法检查
try:
from lanremote.relay import RelayLink
link = RelayLink(store, log=lambda _m: None)
# start() 会去连中转并登记 —— 但我们只用它的 candidate 解析,
# 不用它登记(登记会占用配对码)。
link.candidates = []
for piece in target.replace(",", ";").split(";"):
piece = piece.strip().rstrip("/")
if not piece:
continue
if not piece.startswith(("http://", "https://")):
from lanremote.relay import _guess_scheme
piece = _guess_scheme(piece) + piece
link.candidates.append(piece)
link.url = link.candidates[0] if link.candidates else ""
reply = link._check_code(code)
except Exception: # noqa: BLE001
return None # 检查失败就当没问题
if not reply.get("ok"):
return None # 服务器没回应,放行
if not reply.get("taken"):
return None # 没人用,可以用
where = reply.get("deviceName") or "另一台电脑"
return (
f"配对码 {code} 已经被「{where}」占用了。\n\n"
f"一个配对码同时只能给一台电脑用。\n"
f"请换一个别的 6 位数字。\n\n"
f"(想继续用这个码的话,可以去中转管理页把占用它的那台设备删掉)"
)
def _tell_firewall_problem(detail: str) -> None:
"""防火墙没放行成功,说明为什么这次没生效。"""
message = (
"防火墙没有放行成功,所以这次**没有**把配对码交给服务器。\n\n"
f"原因:{detail or '未知'}\n\n"
"该怎么办:\n"
" 1. 右键点 ykxmz.exe →「以管理员身份运行」,重新走一遍设置\n"
" 2. 或者手动放行:控制面板 → Windows Defender 防火墙 →\n"
" 允许应用通过防火墙 → 找到 ykxmz → 勾上「专用网络」\n\n"
"放行之后再打开 ykxmz.exe 就行了。"
)
try:
import tkinter as tk
from tkinter import messagebox
root = tk.Tk()
root.withdraw()
root.attributes("-topmost", True)
messagebox.showerror("还没生效:防火墙没放行", message)
root.destroy()
return
except Exception: # noqa: BLE001
pass
print("=" * 56)
print("防火墙没放行,配对码未提交到服务器")
print(message)
print("=" * 56)
def _tell_code_problem(message: str) -> None:
"""把配对码的问题告诉用户。有界面就弹窗,没有就打印。"""
try:
import tkinter as tk
from tkinter import messagebox
root = tk.Tk()
root.withdraw()
root.attributes("-topmost", True)
messagebox.showerror("配对码不能用", message)
root.destroy()
return
except Exception: # noqa: BLE001
pass
print("=" * 56)
print("配对码不能用")
print(message)
print("=" * 56)
def run_setup(store) -> int:
"""弹设置窗口 → 存码 → 黑终端一闪修防火墙 → 转静默。"""
from lanremote.autostart import enable, is_enabled
from lanremote.ui import SetupWindow
# 先检查配对码能不能用,**通过了才去改防火墙**。
#
# 踩过的坑:以前是先存码、开自启、修防火墙,最后才转静默——
# 而"码被别的电脑占了"这件事要等转静默之后才发现。
# 那时防火墙已经改完了,用户在设置窗口里也没机会换码。
result = None
while True:
window = SetupWindow(store).show()
if window is None:
return 0 # 用户关掉了窗口(放弃安装)
code = window.get("code", "")
problem = check_code_available(store, code)
if problem is None:
result = window
break
# 被占了:说清楚,让他回去换一个
_tell_code_problem(problem)
store.set_pairing_code(result["code"])
store.set("setup_done", True)
# ---- 首次设置时也要保存"显示"那几项 ----
#
# 用户要求弹窗位置/大小可自定义、屏幕提示条默认可关。
# 首次设置窗口里已经有这几个控件了(见 ui.py 的"显示"卡片),
# 但**只在设置模式保存**的话,第一次装的人改了也不生效 ——
# 表现为"我在安装向导里明明改了,怎么还是老样子"。
if "screen_notice" in result:
store.set("show_screen_notice", bool(result["screen_notice"]))
if result.get("notice_position"):
store.set("notice_position", str(result["notice_position"]))
if result.get("notice_font_size"):
try:
store.set("notice_font_size",
max(8, min(32, int(result["notice_font_size"]))))
except (TypeError, ValueError):
pass
# 开机自启
if result["autostart"]:
try:
if not is_enabled():
enable()
except Exception: # noqa: BLE001
pass
# 黑终端一闪:修防火墙
firewall_ok = True
firewall_note = ""
if result["firewall"]:
firewall_ok, firewall_note = fix_firewall_with_flash(store)
# 放行防火墙之前,配对码**不加入服务器**。
#
# 为什么:注册(= 配对码挂到服务器上)如果发生在防火墙之前,
# 那么防火墙没放行成功时,服务器上会留着一个谁也连不上的码,
# 而且它会把码占住,用户重装都换不了。
# 所以这里卡一道:防火墙没成功就不注册。
if not firewall_ok:
_tell_firewall_problem(firewall_note)
return 3
# 关键:设完码必须开一个配对窗口。
# 之前这里传 None(= 不开窗口 → allow_new_pairing=False),
# 结果用户刚设完码,被控端立刻就把门锁上,手表第一次配对必然失败,
# 报"被控端已关闭新设备配对"。
# 第一台设备配对成功后服务端会自动把窗口关上。
store.set("close_pair_after_first", True)
return run_silent(store, SETUP_PAIR_MINUTES)
def run_settings(store) -> int:
"""双击 exe 时打开的设置窗口。
和 run_setup 的区别:
run_setup 第一次装:走防火墙 + 开机自启 + 最后转静默
run_settings 已装好:只改配置,改完把后台实例重启一下让设置生效
为什么"已经有一个实例在跑"时也要照样开窗口:
以前 InstanceLock 拿不到就直接 return,用户双击后**什么都看不到**,
以为程序坏了。现在开窗、改完、重启后台实例。
"""
from lanremote.instance import is_running
from lanremote.ui import SetupWindow
running = is_running()
window = SetupWindow(store, mode="settings").show()
if window is None:
return 0 # 用户关掉了,什么都不改
changed = []
# ---- 配对码 ----
if window.get("code") and window["code"] != store.pairing_code():
try:
store.set_pairing_code(window["code"])
changed.append("配对码")
except ValueError as exc:
print(f"设置配对码失败:{exc}")
return 2
# ---- 中转服务器(空 = 只走局域网)----
relay = str(window.get("relay", "") or "").strip()
if relay != str(store.get("relay_url", "") or "").strip():
store.set("relay_url", relay)
changed.append("中转地址")
# ---- 允许新设备接入 ----
allow_new = window.get("allow_new")
if allow_new is not None and bool(allow_new) != bool(
store.get("allow_new_pairing", True)):
store.set("allow_new_pairing", bool(allow_new))
changed.append("接入开关")
# ---- 显示:屏幕提示条 / 弹窗位置 / 字号 ----
#
# 用户要求:"弹文字窗口的位置和大小都能自定义"、
# "屏幕已被共享的提示以后不要弹"。
if "screen_notice" in window:
want = bool(window.get("screen_notice"))
if want != bool(store.get("show_screen_notice", False)):
store.set("show_screen_notice", want)
changed.append("屏幕提示条")
pos = window.get("notice_position")
if pos and str(pos) != str(store.get("notice_position") or ""):
store.set("notice_position", str(pos))
changed.append("弹窗位置")
font = window.get("notice_font_size")
if font is not None:
try:
font = max(8, min(32, int(font)))
except (TypeError, ValueError):
font = None
if font is not None and font != int(
store.get("notice_font_size") or 11):
store.set("notice_font_size", font)
changed.append("弹窗字号")
# ---- 开机自启 ----
try:
from lanremote.autostart import enable, disable, is_enabled
want = bool(window.get("autostart"))
if want != bool(is_enabled()):
(enable if want else disable)()
changed.append("开机自启")
except Exception: # noqa: BLE001
pass
# ---- 让后台实例重新读配置 ----
if changed and running:
_restart_background(store)
print("已保存:" + "、".join(changed) + "(后台实例已重启)")
elif changed:
print("已保存:" + "、".join(changed))
else:
print("没有改动")
return 0
def _restart_background(store) -> None:
"""把正在跑的后台实例停掉,再用新配置起一个。
为什么要重启而不是"让它自己重读":
allow_new_pairing 是启动时读进 server.pairing_enabled 的,
中转地址也是启动时连的。文件改了但进程里还是旧值,
用户会以为设置没生效。
"""
import subprocess
import time # ← 必须在这里 import:
# run_agent.py 模块级没有 import time(原来只有 run_silent 里
# 局部 import 了一个),我加这个函数时直接用了 time.sleep,
# pyflakes 立刻报 undefined name 'time'。
# 这也说明 pyflakes 那一关是有用的。
# 只杀"没有窗口"的那个(后台实例)。taskkill 按镜像名杀会连自己一起杀,
# 所以这里用 wmic/powershell 过滤命令行里带 --silent 的。
try:
subprocess.run(
["powershell", "-NoProfile", "-Command",
"Get-CimInstance Win32_Process -Filter \"Name='ykxmz.exe'\" | "
"Where-Object { $_.CommandLine -match '--silent' } | "
"ForEach-Object { Stop-Process -Id $_.ProcessId -Force }"],
capture_output=True, timeout=20)
except Exception: # noqa: BLE001
pass
time.sleep(1.2)
# 用 --silent 重新起一个,完全脱离本进程
try:
exe = sys.executable
subprocess.Popen([exe, "--silent"],
creationflags=getattr(subprocess,
"DETACHED_PROCESS", 0),
close_fds=True)
except Exception: # noqa: BLE001
pass
def main(argv=None) -> int:
try:
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[union-attr]
except Exception:
pass
args = parse_args(argv)
# --code / --name / --relay-url 要**最先**处理 —— 它们会改配置,
# 而 --status / --autostart 这些是"看一眼就退出"的命令。
# 安装包的调用链是:
# ykxmz.exe --code 789113 --name 三年二班 --relay-url https://... --autostart on --silent
# 如果先把 --autostart 处理掉就 return 了,配对码根本写不进去。
if (args.code is not None or args.name
or args.relay_url is not None):
store = Store(data_dir())
wrote = False
# ---- 中转服务器(安装向导那一页填的)----
# ⚠ 和 --relay 的区别很重要:
# --relay 连接一次、打印状态、然后 return
# (人工排查用,**不能**和 --code 连用,
# 因为它会提前退出,后面的 --silent 就不跑了)
# --relay-url 只写进配置,不连接、不退出
# (安装包用:要和 --code/--autostart/--silent
# 串在一条命令里跑完)
if args.relay_url is not None:
url = normalize_relay_url(str(args.relay_url))
store.set("relay_url", url)
print(f"中转服务器已设为 {url or '(空 —— 只走局域网,不跨网络)'}")
wrote = True
if args.code is not None:
digits = "".join(ch for ch in str(args.code) if ch.isdigit())
if len(digits) != 6:
print(f"配对码要 6 位数字,收到的是 {args.code!r}")
return 2
try:
store.set_pairing_code(digits)
except ValueError as exc:
print(f"设置配对码失败:{exc}")
return 2
# 装完就算"设置好了",不要再弹设置窗口
store.set("setup_done", True)
store.set("allow_new_pairing", True)
print(f"配对码已设为 {digits}")
wrote = True
if args.name:
store.set("name", str(args.name).strip()[:32])
print(f"电脑名已设为 {store.get('name')}")
wrote = True
# 只写了配置就结束 —— 这是一次性命令。
# 不加这个 return 的话会继续走到 run_silent(),
# 变成常驻进程永不返回,调用方(安装程序)会一直等下去。
# 想顺便启动服务,显式加 --silent。
if wrote and not (args.silent or args.add_device is not None
or args.setup):
return 0
if args.autostart:
return cli_autostart(args.autostart)
if args.status:
return cli_status()
# 这两个是给 .bat 调的:EXE 自带功能,不需要系统装 Python
if args.netcheck is not None:
return cli_netcheck(args.netcheck)
if args.fix_firewall:
return cli_fix_firewall()
if args.relay is not None:
return cli_relay(args.relay)
if args.relay_status:
return cli_relay_status()
if args.uninstall:
return cli_uninstall(keep_dir=bool(args.keep_dir),
quiet=bool(args.quiet))
store = Store(data_dir())
if args.port:
if not (1 <= int(args.port) <= 65535):
print(f"端口 {args.port} 不合法,应在 1-65535 之间")
return 2
store.set("port", int(args.port))
if args.reset_devices:
store.devices = {}
store.save_devices()
# --code:把配对码设成安装向导里填的那个。
# 严格要 6 位数字 —— 和设置窗口里的规则一致,
# 免得用户装了之后发现手表上少输一位连不上。
if args.code is not None:
digits = "".join(ch for ch in str(args.code) if ch.isdigit())
if len(digits) != 6:
print(f"配对码要 6 位数字,收到的是 {args.code!r}")
return 2
try:
store.set_pairing_code(digits)
except ValueError as exc:
print(f"设置配对码失败:{exc}")
return 2
# 装完就是"设置好了",不要再弹设置窗口
store.set("setup_done", True)
store.set("allow_new_pairing", True)
print(f"配对码已设为 {digits}")
if args.name:
store.set("name", str(args.name).strip()[:32])
if args.new_code or not store.pairing_code():
store.new_pairing_code()
# 显式要求静默 / 临时开配对窗口
if args.silent or args.add_device is not None:
return run_silent(store, args.add_device)
# --setup:重新打开设置窗口(和双击等效)
if args.setup:
return run_settings(store)
# 第一次运行:设置窗口 → 修防火墙 → 静默
if not store.get("setup_done", False):
return run_setup(store)
# 已经设置过、而且没给参数 —— 说明用户是**双击 exe**。
#
# 以前这里是 `return run_silent(store, None)`:直接后台跑,
# 屏幕上什么都不出现。用户想改配对码 / 中转地址 / 接入开关时
# 双击毫无反应,只能去翻命令行参数。
#
# 现在双击 = 打开设置窗口。后台运行走显式 --silent
# (开机自启和安装包都用的 --silent,所以不受影响)。
return run_settings(store)
if __name__ == "__main__":
sys.exit(main())