mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 16:00: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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
636 lines
26 KiB
Python
636 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""修 features.py 里审计出来的 bug。
|
||
|
||
安全（严重）
|
||
1. 删除到回收站用 PowerShell 拼路径 —— 文件名带单引号就能注入任意命令
|
||
2. 播放视频走 cmd /c start —— 路径带 & 或 | 就能执行任意命令
|
||
|
||
数据丢失（严重）
|
||
3. 记笔记前没有先 load —— 进程重启后第一次记笔记会把历史笔记全部抹掉
|
||
|
||
崩溃（应抛 ValueError 却抛别的）
|
||
4-9. rename / make_dir / copy / move / list_dir / zip_dir / folder_size
|
||
的 OSError 家族全部直接冒泡给用户
|
||
|
||
行为不对
|
||
10. 全屏功能在界面关闭时抛 RuntimeError；「关掉全屏内容」其实没关
|
||
11. 数字雨的打标签失效，画布条目无限增长
|
||
12. 网络速度显示的是开机以来累计流量，不是速度
|
||
13. 世界时钟写死时区偏移，忽略夏令时
|
||
14. on 参数传字符串 "false" 会被当成 True
|
||
15. 设置文件被改坏时，笔记/待办/收藏夹列表会崩
|
||
16. 同名复制/移动静默覆盖目标
|
||
17. 下载文件到手表返回 "READY:路径"，但没人解析 → 功能名不符实
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
FEATURES = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows"
|
||
r"\lanremote\features.py")
|
||
|
||
FIXES = []
|
||
|
||
|
||
def fix(old, new, tag):
|
||
FIXES.append((old, new, tag))
|
||
|
||
|
||
# ======================================================================
|
||
# 1+2. 安全：不再把路径拼进 shell
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' def _do_delete_file(self, params) -> str:
|
||
"""删除到回收站，不做永久删除。"""
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.exists():
|
||
raise ValueError("文件不存在")
|
||
if IS_WINDOWS:
|
||
script = (f"Add-Type -AssemblyName Microsoft.VisualBasic;"
|
||
f"[Microsoft.VisualBasic.FileIO.FileSystem]::DeleteFile("
|
||
f"'{target}', 'OnlyErrorDialogs', 'SendToRecycleBin')")
|
||
if target.is_dir():
|
||
script = (f"Add-Type -AssemblyName Microsoft.VisualBasic;"
|
||
f"[Microsoft.VisualBasic.FileIO.FileSystem]::DeleteDirectory("
|
||
f"'{target}', 'OnlyErrorDialogs', 'SendToRecycleBin')")
|
||
code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
if code != 0:
|
||
raise ValueError(f"删除失败: {out[:80]}")
|
||
else:
|
||
target.unlink()
|
||
return f"已删除到回收站：{target.name}"''',
|
||
''' def _recycle(self, target: Path) -> None:
|
||
"""删除到回收站。
|
||
|
||
之前是把路径用单引号拼进 PowerShell 脚本，文件名里带一个单引号
|
||
就能闭合字符串、注入任意命令（新建文件夹的清洗正则不删单引号，
|
||
所以远端两步就能造出这种文件名）。现在直接调 Windows 自己的
|
||
SHFileOperationW，不经过任何 shell，没有注入面。
|
||
"""
|
||
class SHFILEOPSTRUCTW(ctypes.Structure):
|
||
_fields_ = [("hwnd", ctypes.c_void_p),
|
||
("wFunc", ctypes.c_uint),
|
||
("pFrom", ctypes.c_wchar_p),
|
||
("pTo", ctypes.c_wchar_p),
|
||
("fFlags", ctypes.c_ushort),
|
||
("fAnyOperationsAborted", ctypes.c_int),
|
||
("hNameMappings", ctypes.c_void_p),
|
||
("lpszProgressTitle", ctypes.c_wchar_p)]
|
||
|
||
FO_DELETE = 3
|
||
FOF_ALLOWUNDO = 0x0040
|
||
FOF_NOCONFIRMATION = 0x0010
|
||
FOF_SILENT = 0x0004
|
||
FOF_NOERRORUI = 0x0400
|
||
|
||
# pFrom 要双 \0 结尾
|
||
raw = str(target) + "\\0"
|
||
buffer = ctypes.create_unicode_buffer(len(raw) + 1)
|
||
buffer[:len(raw)] = raw
|
||
|
||
op = SHFILEOPSTRUCTW()
|
||
op.wFunc = FO_DELETE
|
||
op.pFrom = ctypes.cast(buffer, ctypes.c_wchar_p)
|
||
op.fFlags = (FOF_ALLOWUNDO | FOF_NOCONFIRMATION
|
||
| FOF_SILENT | FOF_NOERRORUI)
|
||
result = ctypes.windll.shell32.SHFileOperationW(ctypes.byref(op))
|
||
if result != 0:
|
||
raise ValueError(f"删除失败（系统错误码 {result}）")
|
||
|
||
def _do_delete_file(self, params) -> str:
|
||
"""删除到回收站，不做永久删除。"""
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.exists():
|
||
raise ValueError("文件不存在")
|
||
try:
|
||
if IS_WINDOWS:
|
||
self._recycle(target)
|
||
else:
|
||
raise ValueError("仅支持 Windows")
|
||
except ValueError:
|
||
raise
|
||
except OSError as exc:
|
||
raise ValueError(f"删除失败: {exc}") from exc
|
||
return f"已删除到回收站：{target.name}"''',
|
||
"删除到回收站：改用 Windows API，不再拼 shell",
|
||
)
|
||
|
||
fix(
|
||
''' def _do_play_video(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.is_file():
|
||
raise ValueError("视频文件不存在")
|
||
subprocess.Popen(["cmd", "/c", "start", "", str(target)],
|
||
creationflags=_NO_WINDOW)
|
||
return f"已用播放器打开 {target.name}"''',
|
||
''' def _do_play_video(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.is_file():
|
||
raise ValueError("视频文件不存在")
|
||
# 之前走 cmd /c start，路径里的 & | ^ 会被 cmd 当成命令分隔符，
|
||
# 一个文件名就能变成任意命令执行。os.startfile 不经过 shell。
|
||
try:
|
||
os.startfile(str(target)) # noqa: S606
|
||
except OSError as exc:
|
||
raise ValueError(f"打不开: {exc}") from exc
|
||
return f"已用播放器打开 {target.name}"''',
|
||
"播放视频：改用 os.startfile，不再走 cmd",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 3. 笔记：先 load 再写（否则会抹掉历史）
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' self._notes.insert(0, {"text": text[:2000],
|
||
"at": time.strftime("%m-%d %H:%M")})
|
||
del self._notes[200:]
|
||
self._persist("notes", self._notes)''',
|
||
''' # 必须先 load。__init__ 里 self._notes 是空的，
|
||
# 直接 insert 再 persist 会把历史笔记全抹掉（重启后第一次记笔记就中招）。
|
||
self._notes = self._load("notes")
|
||
self._notes.insert(0, {"text": text[:2000],
|
||
"at": time.strftime("%m-%d %H:%M")})
|
||
del self._notes[200:]
|
||
self._persist("notes", self._notes)''',
|
||
"记笔记：先读回来再写，不再抹掉历史",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 4-6. 文件操作：OSError 一律转 ValueError + 边界检查
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' target = parent / name
|
||
target.mkdir(parents=True, exist_ok=True)
|
||
return f"已新建 {target}"''',
|
||
''' if not parent.is_dir():
|
||
raise ValueError("父目录不是文件夹")
|
||
target = parent / name
|
||
if target.exists():
|
||
raise ValueError("同名文件或文件夹已存在")
|
||
try:
|
||
target.mkdir(parents=True, exist_ok=False)
|
||
except OSError as exc:
|
||
raise ValueError(f"新建失败: {exc}") from exc
|
||
return f"已新建 {target}"''',
|
||
"新建文件夹：父目录校验 + 错误包装",
|
||
)
|
||
|
||
fix(
|
||
''' target = source.parent / name
|
||
if target.exists():
|
||
raise ValueError("同名文件已存在")
|
||
source.rename(target)
|
||
return f"已重命名为 {target.name}"''',
|
||
''' if not source.exists():
|
||
raise ValueError("文件不存在")
|
||
target = source.parent / name
|
||
if target.exists():
|
||
raise ValueError("同名文件已存在")
|
||
try:
|
||
source.rename(target)
|
||
except OSError as exc:
|
||
raise ValueError(f"重命名失败: {exc}") from exc
|
||
return f"已重命名为 {target.name}"''',
|
||
"重命名：源不存在检查 + 错误包装",
|
||
)
|
||
|
||
fix(
|
||
''' if not source.exists():
|
||
raise ValueError("源文件不存在")
|
||
if source.is_dir():
|
||
_shutil.copytree(source, target, dirs_exist_ok=True)
|
||
else:
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
_shutil.copy2(source, target)
|
||
return f"已复制到 {target}"''',
|
||
''' if not source.exists():
|
||
raise ValueError("源文件不存在")
|
||
if source == target:
|
||
raise ValueError("源和目标是同一个位置")
|
||
if target.exists():
|
||
raise ValueError("目标已存在，不覆盖（先改名或删掉）")
|
||
try:
|
||
if source.is_dir():
|
||
_shutil.copytree(source, target)
|
||
else:
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
_shutil.copy2(source, target)
|
||
except OSError as exc:
|
||
raise ValueError(f"复制失败: {exc}") from exc
|
||
return f"已复制到 {target}"''',
|
||
"复制文件：不静默覆盖 + 错误包装",
|
||
)
|
||
|
||
fix(
|
||
''' if not source.exists():
|
||
raise ValueError("源文件不存在")
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
_shutil.move(str(source), str(target))
|
||
return f"已移动到 {target}"''',
|
||
''' if not source.exists():
|
||
raise ValueError("源文件不存在")
|
||
if source == target:
|
||
raise ValueError("源和目标是同一个位置")
|
||
if target.exists():
|
||
raise ValueError("目标已存在，不覆盖（先改名或删掉）")
|
||
try:
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
_shutil.move(str(source), str(target))
|
||
except OSError as exc:
|
||
raise ValueError(f"移动失败: {exc}") from exc
|
||
return f"已移动到 {target}"''',
|
||
"移动文件：不静默覆盖 + 错误包装",
|
||
)
|
||
|
||
fix(
|
||
''' entries = sorted(target.iterdir(), key=lambda p: (p.is_file(), p.name.lower()))''',
|
||
''' try:
|
||
entries = sorted(target.iterdir(),
|
||
key=lambda p: (p.is_file(), p.name.lower()))
|
||
except OSError as exc:
|
||
raise ValueError(f"读不了这个文件夹: {exc}") from exc''',
|
||
"浏览文件夹：权限错误包装",
|
||
)
|
||
|
||
fix(
|
||
''' out = source.parent / f"{source.name}.zip"
|
||
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for item in source.rglob("*"):
|
||
if item.is_file():
|
||
zf.write(item, item.relative_to(source.parent))
|
||
return f"已压缩为 {out.name}（{out.stat().st_size // 1024} KB）"''',
|
||
''' out = source.parent / f"{source.name}.zip"
|
||
if out.exists():
|
||
raise ValueError("同名的 zip 已存在，先改名或删掉")
|
||
try:
|
||
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for item in source.rglob("*"):
|
||
if item.is_file():
|
||
zf.write(item, item.relative_to(source.parent))
|
||
size_kb = out.stat().st_size // 1024
|
||
except OSError as exc:
|
||
raise ValueError(f"压缩失败: {exc}") from exc
|
||
return f"已压缩为 {out.name}（{size_kb} KB）"''',
|
||
"压缩：不覆盖 + 错误包装",
|
||
)
|
||
|
||
fix(
|
||
''' total = 0
|
||
count = 0
|
||
for item in target.rglob("*"):
|
||
try:
|
||
if item.is_file():
|
||
total += item.stat().st_size
|
||
count += 1
|
||
except OSError:
|
||
continue
|
||
return f"{target.name}：{count} 个文件，共 {total / 1048576:.1f} MB"''',
|
||
''' total = 0
|
||
count = 0
|
||
deadline = time.time() + 15 # 大目录别把请求线程占死
|
||
try:
|
||
for item in target.rglob("*"):
|
||
if time.time() > deadline:
|
||
break
|
||
try:
|
||
if item.is_file():
|
||
total += item.stat().st_size
|
||
count += 1
|
||
except OSError:
|
||
continue
|
||
except OSError as exc:
|
||
raise ValueError(f"算不了: {exc}") from exc
|
||
return f"{target.name}：{count} 个文件，共 {total / 1048576:.1f} MB"''',
|
||
"文件夹大小：加超时 + 错误包装",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 10. 全屏窗口：错误包装 + 能真的关掉
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' def _fullscreen_window(self, builder, seconds: float = 0.0):
|
||
"""在电脑上开一个全屏窗口。builder(win, close) 负责放内容。"""
|
||
def build():
|
||
import tkinter as tk
|
||
root = getattr(self.actions.ui, "root", None)
|
||
if root is None:
|
||
return
|
||
win = tk.Toplevel(root)
|
||
win.attributes("-fullscreen", True)
|
||
win.attributes("-topmost", True)
|
||
|
||
def close(_event=None):
|
||
try:
|
||
win.destroy()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
builder(win, close)
|
||
win.bind("<Escape>", close)
|
||
win.bind("<Button-1>", close)
|
||
win.protocol("WM_DELETE_WINDOW", close)
|
||
if seconds > 0:
|
||
win.after(int(seconds * 1000), close)
|
||
|
||
self.actions.ui.ui_call(build)''',
|
||
''' def _close_fullscreen_windows(self) -> int:
|
||
"""关掉所有由本程序开的全屏窗口。"""
|
||
closed = 0
|
||
for closer in list(getattr(self, "_fullscreen_closers", [])):
|
||
try:
|
||
closer()
|
||
closed += 1
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self._fullscreen_closers = []
|
||
return closed
|
||
|
||
def _fullscreen_window(self, builder, seconds: float = 0.0):
|
||
"""在电脑上开一个全屏窗口。builder(win, close) 负责放内容。
|
||
|
||
先关掉上一个：不然连着点两次「全屏大字」，窗口会一层层叠上去，
|
||
Esc 只能关最上面那个，下面的再也关不掉。
|
||
"""
|
||
self._close_fullscreen_windows()
|
||
|
||
def build():
|
||
import tkinter as tk
|
||
root = getattr(self.actions.ui, "root", None)
|
||
if root is None:
|
||
raise RuntimeError("界面不可用")
|
||
win = tk.Toplevel(root)
|
||
win.attributes("-fullscreen", True)
|
||
win.attributes("-topmost", True)
|
||
|
||
def close(_event=None):
|
||
try:
|
||
win.destroy()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
self._fullscreen_closers.remove(close)
|
||
except (ValueError, AttributeError):
|
||
pass
|
||
|
||
builder(win, close)
|
||
win.bind("<Escape>", close)
|
||
win.bind("<Button-1>", close)
|
||
win.protocol("WM_DELETE_WINDOW", close)
|
||
if seconds > 0:
|
||
win.after(int(seconds * 1000), close)
|
||
self._fullscreen_closers.append(close)
|
||
|
||
try:
|
||
self.actions.ui.ui_call(build)
|
||
except (RuntimeError, TimeoutError, AttributeError) as exc:
|
||
# 界面关了 / 忙超时 —— 转成用户看得懂的话，别把栈信息甩出去
|
||
raise ValueError(f"电脑上的界面不可用（{exc}）") from exc''',
|
||
"全屏窗口：先关旧的 + 错误转成人话",
|
||
)
|
||
|
||
fix(
|
||
''' def _do_big_text_clear(self, params) -> str:
|
||
return "按 Esc 或点一下就能关掉全屏窗口"''',
|
||
''' def _do_big_text_clear(self, params) -> str:
|
||
closed = self._close_fullscreen_windows()
|
||
return f"已关掉 {closed} 个全屏窗口" if closed else "当前没有全屏窗口"''',
|
||
"关掉全屏内容：真的关掉",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 11. 数字雨：打上 tag，旧帧才会被擦掉
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' canvas.delete("fade")
|
||
for i, y in enumerate(drops):
|
||
char = random.choice(chars)
|
||
canvas.create_text(i * font_size + font_size // 2,
|
||
y * font_size, text=char,
|
||
fill="#00FF66",
|
||
font=("Consolas", font_size, "bold"))''',
|
||
''' canvas.delete("fade")
|
||
for i, y in enumerate(drops):
|
||
char = random.choice(chars)
|
||
canvas.create_text(i * font_size + font_size // 2,
|
||
y * font_size, text=char,
|
||
fill="#00FF66",
|
||
font=("Consolas", font_size, "bold"),
|
||
tags="fade")''',
|
||
"数字雨：补上 fade 标签（否则画布条目无限堆积）",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 12. 网络速度：算差值，不再显示累计流量
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' def _do_network_speed(self, params) -> str:
|
||
script = ("Get-NetAdapterStatistics | Where-Object {$_.ReceivedBytes -gt 0} | "
|
||
"ForEach-Object { \\"$($_.Name): 收 $([math]::Round($_.ReceivedBytes/1MB,1))MB"
|
||
" 发 $([math]::Round($_.SentBytes/1MB,1))MB\\" }")
|
||
code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
return out.replace("\\n", " | ")[:300] or "读不到"''',
|
||
''' def _do_network_speed(self, params) -> str:
|
||
"""真实速率：采两次求差，不是开机以来的累计流量。"""
|
||
script = ("(Get-NetAdapterStatistics | Measure-Object -Property "
|
||
"ReceivedBytes -Sum).Sum")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
digits = re.findall(r"\\d+", out or "")
|
||
now_bytes = int(digits[0]) if digits else 0
|
||
now = time.time()
|
||
last_bytes = getattr(self, "_net_last_bytes", None)
|
||
last_time = getattr(self, "_net_last_time", None)
|
||
self._net_last_bytes = now_bytes
|
||
self._net_last_time = now
|
||
|
||
if last_bytes is None or last_time is None or now <= last_time:
|
||
time.sleep(1.0)
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
digits = re.findall(r"\\d+", out or "")
|
||
again = int(digits[0]) if digits else now_bytes
|
||
self._net_last_bytes = again
|
||
self._net_last_time = time.time()
|
||
return f"当前约 {(again - now_bytes) / 1048576:.2f} MB/s（采样 1 秒）"
|
||
|
||
elapsed = now - last_time
|
||
delta = max(0, now_bytes - last_bytes)
|
||
mbps = delta / elapsed / 1048576
|
||
return (f"下载 {mbps:.2f} MB/s（{mbps * 8:.1f} Mbps）\\n"
|
||
f"这段时间共收了 {delta / 1048576:.1f} MB")''',
|
||
"网络速度：改成真实速率（两次采样求差）",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 13. 世界时钟：用 zoneinfo，自动处理夏令时
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' def _do_world_clock(self, params) -> str:
|
||
now = time.time()
|
||
zones = [("北京", 8), ("伦敦", 0), ("纽约", -5), ("洛杉矶", -8),
|
||
("东京", 9), ("悉尼", 10)]
|
||
return "\\n".join(
|
||
f"{name} {time.strftime('%H:%M', time.gmtime(now + off * 3600))}"
|
||
for name, off in zones)''',
|
||
''' def _do_world_clock(self, params) -> str:
|
||
"""写死偏移会忽略夏令时（现在纽约/伦敦/悉尼都差 1 小时），改用 zoneinfo。"""
|
||
zones = [("北京", "Asia/Shanghai"), ("伦敦", "Europe/London"),
|
||
("纽约", "America/New_York"), ("洛杉矶", "America/Los_Angeles"),
|
||
("东京", "Asia/Tokyo"), ("悉尼", "Australia/Sydney")]
|
||
try:
|
||
from zoneinfo import ZoneInfo
|
||
import datetime as _dt
|
||
now = _dt.datetime.now(_dt.timezone.utc)
|
||
lines = []
|
||
for name, zone in zones:
|
||
local = now.astimezone(ZoneInfo(zone))
|
||
lines.append(f"{name} {local.strftime('%H:%M')}"
|
||
f"{'（夏令时）' if local.dst() else ''}")
|
||
return "\\n".join(lines)
|
||
except Exception: # noqa: BLE001
|
||
# 没有时区数据库时退回 UTC 偏移（可能差 1 小时）
|
||
now = time.time()
|
||
fallback = [("北京", 8), ("伦敦", 0), ("纽约", -5), ("洛杉矶", -8),
|
||
("东京", 9), ("悉尼", 10)]
|
||
return "\\n".join(
|
||
f"{n} {time.strftime('%H:%M', time.gmtime(now + o * 3600))}"
|
||
for n, o in fallback)''',
|
||
"世界时钟：用 zoneinfo 处理夏令时",
|
||
)
|
||
|
||
# ======================================================================
|
||
# 14. on 参数：字符串 "false" 不该被当成 True
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' def run(self, name: str, params: dict) -> str:''',
|
||
''' @staticmethod
|
||
def _truthy(value) -> bool:
|
||
"""字符串 "false"/"0"/"off" 不能被当成真 —— bool("false") 是 True。"""
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, (int, float)):
|
||
return value != 0
|
||
if isinstance(value, str):
|
||
return value.strip().lower() in ("1", "true", "on", "yes", "是", "开")
|
||
return bool(value)
|
||
|
||
def run(self, name: str, params: dict) -> str:''',
|
||
"新增 _truthy 辅助",
|
||
)
|
||
|
||
for _old, _new, _tag in [
|
||
("want_on = bool(params.get(\"on\"))", "want_on = self._truthy(params.get(\"on\"))",
|
||
"夜间模式用 _truthy"),
|
||
("want = bool(params.get(\"on\"))", "want = self._truthy(params.get(\"on\"))",
|
||
"任务栏隐藏用 _truthy"),
|
||
("want = 1 if params.get(\"on\") else 0", "want = 1 if self._truthy(params.get(\"on\")) else 0",
|
||
"时钟秒数用 _truthy"),
|
||
("want = 1 if params.get(\"on\") else 2", "want = 2 if not self._truthy(params.get(\"on\")) else 1",
|
||
"隐藏文件用 _truthy"),
|
||
]:
|
||
fix(_old, _new, _tag)
|
||
|
||
# ======================================================================
|
||
# 15. _load：元素类型校验（设置文件被改坏时不崩）
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' def _load(self, key: str):
|
||
try:
|
||
value = self.store.get(f"feat_{key}")
|
||
return value if isinstance(value, list) else []
|
||
except Exception: # noqa: BLE001
|
||
return []''',
|
||
''' def _load(self, key: str, item_type=None):
|
||
"""读持久化列表。
|
||
|
||
设置文件可能被手改坏、或是旧版本残留，所以元素类型也要校验 ——
|
||
否则 notes/todos/clip_fav 的列表渲染会抛 TypeError 崩掉。
|
||
"""
|
||
try:
|
||
value = self.store.get(f"feat_{key}")
|
||
except Exception: # noqa: BLE001
|
||
return []
|
||
if not isinstance(value, list):
|
||
return []
|
||
if item_type is None:
|
||
return value
|
||
return [item for item in value if isinstance(item, item_type)]''',
|
||
"_load 元素类型校验",
|
||
)
|
||
|
||
fix("self._notes = self._load(\"notes\")",
|
||
"self._notes = self._load(\"notes\", dict)",
|
||
"笔记读回来校验 dict")
|
||
fix("todos = self._load(\"todos\")",
|
||
"todos = self._load(\"todos\", dict)",
|
||
"待办读回来校验 dict")
|
||
fix("favs = self._load(\"clip_fav\")",
|
||
"favs = self._load(\"clip_fav\", str)",
|
||
"收藏夹读回来校验 str")
|
||
|
||
# ======================================================================
|
||
# 17. 下载文件到手表：改成手表端真正能拉的
|
||
# ======================================================================
|
||
|
||
fix(
|
||
''' add("download_file", "下载文件到手表", "input", "文件",
|
||
hint="完整路径", key="path")''',
|
||
''' add("download_file", "下载文件到手表", "file", "文件",
|
||
hint="完整路径", key="path")''',
|
||
"下载文件：标记成 file 类型",
|
||
)
|
||
|
||
fix(
|
||
''' if target.stat().st_size > 30 * 1024 * 1024:
|
||
raise ValueError("文件超过 30MB，太大")
|
||
return f"READY:{target}"''',
|
||
''' size = target.stat().st_size
|
||
if size > 30 * 1024 * 1024:
|
||
raise ValueError("文件超过 30MB，太大")
|
||
readable = (f"{size / 1048576:.1f} MB" if size > 1048576
|
||
else f"{max(1, size // 1024)} KB")
|
||
return f"{target.name}（{readable}），正在传到手表…"'''
|
||
,
|
||
"下载文件：返回可读文案（拉取由手表端发起）",
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
text = FEATURES.read_text(encoding="utf-8")
|
||
applied, skipped = 0, []
|
||
for old, new, tag in FIXES:
|
||
if old not in text:
|
||
skipped.append(tag)
|
||
print(f" [跳过] {tag} —— 锚点没找到")
|
||
continue
|
||
if text.count(old) > 1:
|
||
print(f" [注意] {tag} 匹配 {text.count(old)} 处，只改第一处")
|
||
text = text.replace(old, new, 1)
|
||
applied += 1
|
||
print(f" {tag}: OK")
|
||
|
||
FEATURES.write_text(text, encoding="utf-8")
|
||
print(f"\n 应用 {applied} / {len(FIXES)}")
|
||
|
||
print("\n语法检查：")
|
||
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())
|