ykxmz 1.5.0:修复手表端 TLS 证书校验 + MTU 修复网络稳定性

被控端
- 修复 MTU 问题:夹 MSS + 路径 MTU 探测(公网可用率 48% -> 100%)
- 复用 HTTP 连接(免掉每次新建 TLS 的 30% 握手失败)
- 每分钟回头找局域网,自动切回快 400 倍的路径
- 画面心跳帧:静止时省流量,客户端仍知道流是活的
- 锁屏时给出可操作的提示

手表端
- 内置 GTS Root R4 / GlobalSign Root CA,修 SSLHandshakeException
- 连接错误翻译成人话(证书/时间/解析/超时分别提示)
- 经中转连上后用服务器报的地址刷新已配对记录
- 新增局域网连接页面、服务器延迟检测

服务器端
- 支持 chunked 请求体(修每 16 秒一次的连接错乱与大面积 502)
- 网络调优脚本 net-tune.sh(开机自动应用)
- 管理页:批量删除、网络体检、下载区
- 隧道改 HTTP/2 + 双连接

文档
- 许可协议 / 安装前必读 / 安装后须知(作者:小麦子)
This commit is contained in:
xmz7891
2026-10-06 19:01:05 +08:00
commit b62ac865ea
3195 changed files with 203292 additions and 0 deletions
+643
View File
@@ -0,0 +1,643 @@
# -*- coding: utf-8 -*-
"""再加 50 个功能。
分组分布:
教学 10 教室 12 系统 8 查看 8 效率 6 媒体 3 网络 3
全部只用 Windows 自带能力(注册表 / tasklist / netsh / Tk),不引依赖。
"""
from __future__ import annotations
import sys
from pathlib import Path
FEATURES = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows"
r"\lanremote\features.py")
DEFS_ANCHOR = ' # ---- 娱乐 ----'
DEFS_NEW = ''' # ---- 教学(续)----
add("roll_group", "随机分组", "input", "教学",
hint="几个人一组", key="size")
add("roll_call_no_repeat", "点名(不重复)", "action", "教学")
add("roll_call_reset", "重置点名记录", "action", "教学")
add("class_schedule", "显示课程表", "input", "教学",
hint="用逗号隔开,如 语文,数学,英语", key="text")
add("hand_count", "举手计数", "action", "教学")
add("hand_reset", "举手清零", "action", "教学")
add("answer_poll", "答题统计", "action", "教学")
add("magnifier", "屏幕放大镜", "action", "教学")
add("projector_switch", "切换投影模式", "action", "教学")
add("lesson_timer", "上课计时", "action", "教学")
# ---- 教室(续)----
add("lock_now", "立刻锁屏", "action", "教室")
add("game_watch_start", "开始盯游戏进程", "action", "教室")
add("game_watch_stop", "停止盯游戏进程", "action", "教室")
add("usb_check", "检查有没有插U盘", "action", "教室")
add("login_history", "最近登录记录", "action", "教室")
add("boot_history", "开关机记录", "action", "教室")
add("clean_desktop", "整理桌面(收进文件夹)", "action", "教室")
add("block_new_app", "禁止新程序启动", "toggle", "教室", on=False)
add("show_desktop_now", "立刻显示桌面", "action", "教室")
add("count_windows", "看目前开了几个窗口", "action", "教室")
add("class_end_cleanup", "下课清理(关游戏+恢复)", "action", "教室")
add("who_is_playing", "谁在玩游戏", "action", "教室")
# ---- 系统(续)----
add("open_taskmgr2", "打开任务管理器", "action", "系统")
add("open_resmon", "打开资源监视器", "action", "系统")
add("power_plan", "切换电源计划", "action", "系统")
add("window_top", "当前窗口置顶", "action", "系统")
add("screen_off_lock", "关屏并锁定", "action", "系统")
add("clear_all_clip", "清空剪贴板历史", "action", "系统")
add("restart_audio", "重启音频服务", "action", "系统")
add("system_sounds", "打开声音设置", "action", "系统")
# ---- 查看(续)----
add("cpu_model", "CPU 型号", "action", "查看")
add("memory_sticks", "内存条信息", "action", "查看")
add("disk_serial", "硬盘序列号", "action", "查看")
add("gpu_temp", "显卡温度", "action", "查看")
add("net_total", "网络累计流量", "action", "查看")
add("running_games", "正在运行的游戏", "action", "查看")
add("activation", "系统激活状态", "action", "查看")
add("disk_health", "硬盘健康", "action", "查看")
# ---- 效率(续)----
add("sticky_note", "便签", "input", "效率", hint="写点什么", key="text")
add("search_note", "搜索笔记", "input", "效率", hint="关键词", key="word")
add("translator", "打开翻译", "input", "效率", hint="要翻译的词", key="text")
add("calc_open", "打开计算器2", "action", "效率")
add("unit_convert", "单位换算", "input", "效率",
hint="如 100cm", key="text")
add("countdown_day", "倒数日", "input", "效率",
hint="几月几号,如 06-20", key="date")
# ---- 媒体(续)----
add("loop_play", "循环播放", "toggle", "媒体", on=False)
add("shuffle_play", "随机播放开关", "toggle", "媒体", on=False)
add("stop_all_sound", "停掉所有声音", "action", "媒体")
# ---- 网络(续)----
add("net_traffic", "实时网速", "action", "网络")
add("traceroute", "路由追踪", "input", "网络",
hint="如 8.8.8.8", key="host")
add("net_reset", "重置网络", "action", "网络")
# ---- 娱乐 ----'''
IMPL_ANCHOR = " # ---------------- 持久化 ----------------"
IMPL_NEW = ''' # ==================================================================
# 教学(续)
# ==================================================================
def _roll_names(self) -> list:
names = self.store.get("roll_call_names") or []
return [str(n) for n in names if str(n).strip()]
def _do_roll_group(self, params) -> str:
import random
names = self._roll_names()
if not names:
raise ValueError("还没有名单,先点「设置点名名单」")
try:
size = max(2, min(20, int(params.get("size") or 4)))
except (TypeError, ValueError):
size = 4
pool = names[:]
random.shuffle(pool)
groups = [pool[i:i + size] for i in range(0, len(pool), size)]
lines = [f"第 {i + 1} 组:" + "、".join(g)
for i, g in enumerate(groups)]
self._do_big_text({"text": f"分 {len(groups)} 组"})
return "\\n".join(lines)
def _do_roll_call_no_repeat(self, params) -> str:
import random
names = self._roll_names()
if not names:
raise ValueError("还没有名单,先点「设置点名名单」")
used = self.store.get("roll_call_used") or []
left = [n for n in names if n not in used]
if not left:
return "全班都点过了(点「重置点名记录」重来)"
picked = random.choice(left)
used.append(picked)
self.store.set("roll_call_used", used)
self._do_big_text({"text": picked})
return f"点到:{picked}(还剩 {len(left) - 1} 人没点过)"
def _do_roll_call_reset(self, params) -> str:
self.store.set("roll_call_used", [])
return "点名记录已重置"
def _do_class_schedule(self, params) -> str:
raw = str(params.get("text") or "").strip()
if not raw:
self.store.set("class_schedule", "")
return "课程表已清空"
for sep in (",", "、", ";", ";"):
raw = raw.replace(sep, ",")
items = [x.strip() for x in raw.split(",") if x.strip()]
self.store.set("class_schedule", items)
self._do_big_text({"text": "\\n".join(
f"{i + 1}. {x}" for i, x in enumerate(items))})
return f"课程表已保存 {len(items)} 节"
def _do_hand_count(self, params) -> str:
count = int(self.store.get("hand_count") or 0) + 1
self.store.set("hand_count", count)
self._do_big_text({"text": str(count)})
return f"举手计数:{count}"
def _do_hand_reset(self, params) -> str:
self.store.set("hand_count", 0)
return "举手计数已清零"
def _do_answer_poll(self, params) -> str:
counts = self.store.get("answer_poll") or {}
if not isinstance(counts, dict):
counts = {}
lines = [f"{k}:{v} 人" for k, v in sorted(counts.items())] or ["还没记录"]
lines.append("")
lines.append("(在电脑上点画面对应的选项还可以继续加)")
return "\\n".join(lines)
def _do_magnifier(self, params) -> str:
subprocess.Popen("magnify.exe", shell=True, creationflags=_NO_WINDOW)
return "已打开屏幕放大镜"
def _do_projector_switch(self, params) -> str:
# Win + P
self._key_combo(0x5B, 0x50)
return "已弹出投影模式选择(Win+P)"
def _do_lesson_timer(self, params) -> str:
started = self.store.get("lesson_started")
if started:
import time as _t
mins = int((_t.time() - float(started)) / 60)
self.store.set("lesson_started", 0)
return f"本节课上了 {mins} 分钟"
import time as _t
self.store.set("lesson_started", _t.time())
return "开始计时(再点一次看上了多久)"
# ==================================================================
# 教室(续)
# ==================================================================
def _do_lock_now(self, params) -> str:
ctypes.windll.user32.LockWorkStation()
return "已锁屏"
def _game_watch_loop(self) -> None:
import time as _t
while self.store.get("game_watch"):
try:
for proc in self.actions.list_processes(limit=200):
name = (proc.get("name") or "").lower()
if any(w in name for w in self.GAME_WORDS):
self._do_big_text({"text": f"别玩游戏\\n{proc['name']}"})
break
except Exception: # noqa: BLE001
pass
_t.sleep(15)
def _do_game_watch_start(self, params) -> str:
if self.store.get("game_watch"):
return "已经在盯着了"
self.store.set("game_watch", True)
import threading
threading.Thread(target=self._game_watch_loop, daemon=True,
name="game-watch").start()
return "开始盯游戏进程(每 15 秒查一次,发现就全屏提醒)"
def _do_game_watch_stop(self, params) -> str:
self.store.set("game_watch", False)
return "已停止盯游戏进程"
def _do_usb_check(self, params) -> str:
script = ("Get-CimInstance Win32_DiskDrive | Where-Object "
"{$_.InterfaceType -eq 'USB'} | ForEach-Object { "
"\\"$($_.Model) $([math]::Round($_.Size/1GB,1))GB\\" }")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
lines = [l for l in out.splitlines() if l.strip()]
return ("发现 U 盘:" + " | ".join(lines)) if lines else "没有插 U 盘"
def _do_login_history(self, params) -> str:
script = ("Get-WinEvent -LogName Security -MaxEvents 200 "
"-FilterXPath '*[System[EventID=4624]]' "
"-ErrorAction SilentlyContinue | Select-Object -First 10 | "
"ForEach-Object { $_.TimeCreated.ToString('MM-dd HH:mm') }")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=30)
lines = [l for l in out.splitlines() if l.strip()]
if not lines:
return "读不到登录记录(需要管理员权限)"
return "最近登录:\\n" + "\\n".join(lines)
def _do_boot_history(self, params) -> str:
script = ("Get-WinEvent -LogName System -MaxEvents 300 "
"-FilterXPath '*[System[Provider[@Name=\"Microsoft-Windows-Kernel-Boot\"]"
"]]' -ErrorAction SilentlyContinue | Select-Object -First 8 | "
"ForEach-Object { $_.TimeCreated.ToString('MM-dd HH:mm') }")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=30)
lines = [l for l in out.splitlines() if l.strip()]
if not lines:
return f"本次开机时间:{self.actions._uptime()}前"
return "最近开关机:\\n" + "\\n".join(lines)
def _do_clean_desktop(self, params) -> str:
desktop = Path.home() / "Desktop"
if not desktop.is_dir():
raise ValueError("找不到桌面目录")
box = desktop / "_桌面整理"
box.mkdir(exist_ok=True)
moved = 0
for item in desktop.iterdir():
if item == box or item.is_dir():
continue
try:
item.rename(box / item.name)
moved += 1
except OSError:
continue
return f"把 {moved} 个文件收进了「_桌面整理」文件夹"
def _do_block_new_app(self, params) -> str:
on = self._truthy(params.get("on"))
# 用软件限制策略(HKCU 下不需要管理员)
path = r"Software\\Microsoft\\Windows\\CurrentVersion\\Policies\\Explorer"
ok = self._reg_set(f"HKCU\\\\{path}",
"DisallowRun", 1 if on else 0)
if not ok:
raise ValueError("设置失败,可能被组策略锁住")
return f"禁止新程序启动:{'已开启' if on else '已关闭'}"
def _do_show_desktop_now(self, params) -> str:
self._key_combo(0x5B, 0x44)
return "已显示桌面"
def _do_count_windows(self, params) -> str:
user32 = ctypes.windll.user32
EnumProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p,
ctypes.c_void_p)
own = os.getpid()
titles = []
def callback(hwnd, _lparam):
if not user32.IsWindowVisible(hwnd):
return True
length = user32.GetWindowTextLengthW(hwnd)
if length == 0:
return True
pid = ctypes.c_ulong()
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
if pid.value == own:
return True
buf = ctypes.create_unicode_buffer(length + 1)
user32.GetWindowTextW(hwnd, buf, length + 1)
titles.append(buf.value)
return True
user32.EnumWindows(EnumProc(callback), 0)
if not titles:
return "没有打开的窗口"
return f"开着 {len(titles)} 个窗口:\\n" + "\\n".join(titles[:15])
def _do_class_end_cleanup(self, params) -> str:
parts = []
try:
parts.append(self._do_kill_games({}))
except Exception: # noqa: BLE001
pass
try:
self.store.set("game_watch", False)
parts.append("已停止盯进程")
except Exception: # noqa: BLE001
pass
try:
parts.append(self._do_restore_all({}))
except Exception: # noqa: BLE001
pass
return "下课清理完成:\\n" + "\\n".join(parts)
def _do_who_is_playing(self, params) -> str:
found = []
for proc in self.actions.list_processes(limit=200):
name = (proc.get("name") or "").lower()
if any(w in name for w in self.GAME_WORDS):
found.append(f"{proc['name']} (PID {proc['pid']})")
if not found:
return "没有发现游戏进程"
return f"发现 {len(found)} 个:\\n" + "\\n".join(found[:15])
# ==================================================================
# 系统(续)
# ==================================================================
def _do_open_taskmgr2(self, params) -> str:
return self._open_msc("taskmgr.exe", "任务管理器")
def _do_open_resmon(self, params) -> str:
return self._open_msc("resmon.exe", "资源监视器")
def _do_power_plan(self, params) -> str:
_code, out = _run(["powercfg", "/list"])
plans = re.findall(r"([0-9a-fA-F-]{36})\s*\\(([^)]+)\\)", out or "")
if not plans:
return "读不到电源计划"
lines = [f"{name}" for _guid, name in plans]
return "可用的电源计划:\\n" + "\\n".join(lines)
def _do_window_top(self, params) -> str:
"""把当前前台窗口切换置顶。"""
user32 = ctypes.windll.user32
hwnd = user32.GetForegroundWindow()
if not hwnd:
raise ValueError("没有前台窗口")
HWND_TOPMOST = -1
HWND_NOTOPMOST = -2
SWP_NOMOVE = 0x0002
SWP_NOSIZE = 0x0001
# 先试着取消置顶,成功就说明本来是置顶的
user32.SetWindowPos(hwnd, HWND_NOTOPMOST, 0, 0, 0, 0,
SWP_NOMOVE | SWP_NOSIZE)
user32.SetWindowPos(hwnd, HWND_TOPMOST, 0, 0, 0, 0,
SWP_NOMOVE | SWP_NOSIZE)
return "已把当前窗口置顶"
def _do_screen_off_lock(self, params) -> str:
ctypes.windll.user32.LockWorkStation()
import time as _t
_t.sleep(0.4)
self.actions.monitor_power(False)
return "已关屏并锁定"
def _do_clear_all_clip(self, params) -> str:
self._clip_history = []
self.store.set("feat_clip_fav", [])
return "剪贴板历史和收藏夹已清空"
def _do_restart_audio(self, params) -> str:
_run(["net", "stop", "Audiosrv", "/y"])
_run(["net", "start", "Audiosrv"])
return "音频服务已重启(可能需要管理员权限)"
def _do_system_sounds(self, params) -> str:
subprocess.Popen(["cmd", "/c", "start", "ms-settings:sound"],
creationflags=_NO_WINDOW)
return "已打开声音设置"
# ==================================================================
# 查看(续)
# ==================================================================
def _do_cpu_model(self, params) -> str:
script = ("$c=Get-CimInstance Win32_Processor|Select-Object -First 1;"
"\\"$($c.Name)\\";\\"$($c.NumberOfCores) 核 "
"$($c.NumberOfLogicalProcessors) 线程\\";"
"\\"主频 $($c.MaxClockSpeed) MHz\\"")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
lines = [l for l in out.splitlines() if l.strip()]
return "\\n".join(lines) or "读不到 CPU 信息"
def _do_memory_sticks(self, params) -> str:
script = ("Get-CimInstance Win32_PhysicalMemory | ForEach-Object { "
"\\"$([math]::Round($_.Capacity/1GB,0))GB "
"$($_.Speed)MHz $($_.Manufacturer)\\" }")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
lines = [l for l in out.splitlines() if l.strip()]
return "\\n".join(lines) or "读不到内存条信息"
def _do_disk_serial(self, params) -> str:
script = ("Get-CimInstance Win32_DiskDrive | ForEach-Object { "
"\\"$($_.Model) SN=$($_.SerialNumber)\\" }")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
lines = [l for l in out.splitlines() if l.strip()]
return "\\n".join(lines) or "读不到硬盘序列号"
def _do_gpu_temp(self, params) -> str:
script = ("$t=Get-CimInstance -Namespace root/WMI -ClassName "
"MSAcpi_ThermalZoneTemperature -ErrorAction SilentlyContinue | "
"Select-Object -First 1;"
"if($t){[math]::Round(($t.CurrentTemperature/10)-273.15,1)}"
"else{'n/a'}")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
temp = (out or "").strip().splitlines()
gpu = self._do_gpu_info({})
return f"温度:{temp[-1] if temp else 'n/a'}°C\\n{gpu}"
def _do_net_total(self, params) -> str:
script = ("$s=Get-NetAdapterStatistics | Measure-Object -Property "
"ReceivedBytes,SentBytes -Sum;"
"\\"收 $([math]::Round($s[0].Sum/1GB,2)) GB 发 "
"$([math]::Round($s[1].Sum/1GB,2)) GB\\"")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
return (out or "").strip().splitlines()[-1] if out.strip() else "读不到"
def _do_running_games(self, params) -> str:
return self._do_who_is_playing(params)
def _do_activation(self, params) -> str:
script = ("$a=Get-CimInstance SoftwareLicensingProduct "
"-Filter \\"PartialProductKey is not null\\" "
"-ErrorAction SilentlyContinue | Select-Object -First 1;"
"if($a){\\"$($a.Name)\\";\\"状态码 $($a.LicenseStatus)\\"}"
"else{'读不到'}")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=30)
lines = [l for l in out.splitlines() if l.strip()]
if not lines:
return "读不到激活信息"
# LicenseStatus 1 = 已激活
text = "\\n".join(lines)
if "状态码 1" in text:
text += "\\n(1 = 已激活)"
return text
def _do_disk_health(self, params) -> str:
script = ("Get-PhysicalDisk | ForEach-Object { "
"\\"$($_.FriendlyName) 健康=$($_.HealthStatus) "
"状态=$($_.OperationalStatus)\\" }")
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
timeout=25)
lines = [l for l in out.splitlines() if l.strip()]
return "\\n".join(lines) or "读不到硬盘健康信息"
# ==================================================================
# 效率(续)
# ==================================================================
def _do_sticky_note(self, params) -> str:
text = str(params.get("text") or "").strip()
if not text:
raise ValueError("内容为空")
notes = self._load("sticky", dict)
notes.insert(0, {"text": text[:500],
"at": time.strftime("%m-%d %H:%M")})
self._persist("sticky", notes[:100])
self._do_big_text({"text": text[:40]})
return f"便签已贴(共 {len(notes[:100])} 条)"
def _do_search_note(self, params) -> str:
word = str(params.get("word") or "").strip().lower()
if not word:
raise ValueError("关键词为空")
hits = []
for key, label in (("notes", "笔记"), ("sticky", "便签"),
("todos", "待办")):
for item in self._load(key, dict):
if word in str(item.get("text", "")).lower():
hits.append(f"[{label}] {item.get('text', '')[:60]}")
if not hits:
return f"没有含「{word}」的记录"
return f"找到 {len(hits)} 条:\\n" + "\\n".join(hits[:30])
def _do_translator(self, params) -> str:
text = str(params.get("text") or "").strip()
if not text:
raise ValueError("内容为空")
from urllib.parse import quote
url = f"https://translate.google.com/?text={quote(text)}"
subprocess.Popen(["cmd", "/c", "start", url], creationflags=_NO_WINDOW)
return f"已在浏览器打开「{text[:20]}」的翻译"
def _do_calc_open(self, params) -> str:
subprocess.Popen("calc.exe", shell=True, creationflags=_NO_WINDOW)
return "已打开计算器"
def _do_unit_convert(self, params) -> str:
text = str(params.get("text") or "").strip().lower().replace(" ", "")
match = re.match(r"^(-?\\d+(?:\\.\\d+)?)([a-z厘米寸尺斤公斤克磅度]*)$", text)
if not match:
raise ValueError("格式如 100cm / 5kg / 30度")
value = float(match.group(1))
unit = match.group(2)
table = {
"cm": ("英寸", value / 2.54), "厘米": ("英寸", value / 2.54),
"m": ("英尺", value * 3.2808), "米": ("英尺", value * 3.2808),
"km": ("英里", value * 0.6214), "公里": ("英里", value * 0.6214),
"kg": ("磅", value * 2.2046), "公斤": ("磅", value * 2.2046),
"克": ("盎司", value * 0.0353),
"斤": ("公斤", value * 0.5),
"度": ("华氏度", value * 9 / 5 + 32),
"c": ("华氏度", value * 9 / 5 + 32),
}
item = table.get(unit)
if item is None:
raise ValueError(f"不认识的单位:{unit}")
return f"{value}{unit} = {item[1]:.2f}{item[0]}"
def _do_countdown_day(self, params) -> str:
date_text = str(params.get("date") or "").strip()
match = re.match(r"^(\\d{1,2})[-/月](\\d{1,2})", date_text)
if not match:
raise ValueError("格式如 06-20")
import datetime as _dt
now = _dt.date.today()
try:
target = _dt.date(now.year, int(match.group(1)), int(match.group(2)))
except ValueError as exc:
raise ValueError(f"日期不对:{exc}") from exc
if target < now:
target = _dt.date(now.year + 1, target.month, target.day)
days = (target - now).days
self._do_big_text({"text": f"还有 {days} 天"})
return f"距离 {target} 还有 {days} 天"
# ==================================================================
# 媒体(续)
# ==================================================================
def _do_loop_play(self, params) -> str:
on = self._truthy(params.get("on"))
with self._lock:
if self._audio_alias:
self._mci(f"set {self._audio_alias} time format milliseconds")
self.store.set("loop_play", bool(on))
return f"循环播放:{'已开启' if on else '已关闭'}"
def _do_shuffle_play(self, params) -> str:
on = self._truthy(params.get("on"))
self.store.set("shuffle_play", bool(on))
return f"随机播放:{'已开启' if on else '已关闭'}"
def _do_stop_all_sound(self, params) -> str:
self.actions.stop_audio()
self._volume_key(0xAD, 1)
return "已停掉被控端播放的音频,并切换系统静音"
# ==================================================================
# 网络(续)
# ==================================================================
def _do_net_traffic(self, params) -> str:
return self._do_network_speed(params)
def _do_traceroute(self, params) -> str:
host = str(params.get("host") or "").strip()
if not host:
raise ValueError("地址为空")
if not re.match(r"^[A-Za-z0-9_.:\\-]+$", host):
raise ValueError("地址含有不合法字符")
_code, out = _run(["tracert", "-d", "-h", "12", host], timeout=60)
lines = [l for l in out.splitlines() if l.strip()]
return "\\n".join(lines[-14:]) or "没有输出"
def _do_net_reset(self, params) -> str:
_run(["ipconfig", "/flushdns"])
_run(["netsh", "winsock", "reset"])
_run(["netsh", "int", "ip", "reset"])
return "网络已重置(部分项目需要重启才生效)"
''' + IMPL_ANCHOR
def main() -> int:
text = FEATURES.read_text(encoding="utf-8")
changed = 0
if '"roll_group"' in text:
print(" defs 已存在,跳过")
elif DEFS_ANCHOR in text:
text = text.replace(DEFS_ANCHOR, DEFS_NEW, 1)
changed += 1
print(" defs: 已插入 50 个功能")
else:
print(" [失败] defs 锚点没找到")
if "_do_roll_group" in text:
print(" 实现已存在,跳过")
elif IMPL_ANCHOR in text:
text = text.replace(IMPL_ANCHOR, IMPL_NEW, 1)
changed += 1
print(" 实现: 已插入")
else:
print(" [失败] 实现锚点没找到")
if changed:
FEATURES.write_text(text, encoding="utf-8")
print(f" 已写入({changed} 处)")
import ast
try:
ast.parse(FEATURES.read_text(encoding="utf-8"))
print(" features.py: 语法 OK")
except SyntaxError as exc:
print(f" features.py: 语法错误 {exc}")
return 1
return 0
if __name__ == "__main__":
sys.exit(main())