mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 21:50:44 +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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
222 lines
8.1 KiB
Python
222 lines
8.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""配置与设备凭据持久化。
|
||
|
||
数据放在程序目录下的 data/，绿色便携，不写注册表。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import secrets
|
||
import socket
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
DEFAULT_SETTINGS = {
|
||
"port": 8765,
|
||
"server_name": "",
|
||
"allow_run": True, # 允许自定义命令（可在界面关闭）
|
||
"allow_power": True, # 允许关机/重启（可在界面关闭）
|
||
"allow_screen": True, # 允许屏幕镜像/截屏（可在界面关闭）
|
||
"start_minimized": False, # 启动后直接最小化到托盘
|
||
"allow_new_pairing": True, # 是否允许新设备配对
|
||
"image_timeout": 0, # 图片显示秒数，0=手动关闭
|
||
"image_display": "window", # 投图方式: window=屏幕中间普通窗口 / fullscreen=全屏 / custom=自定义大小
|
||
"image_custom_width": 0, # 自定义宽度(像素)，0=自动
|
||
"image_custom_height": 0, # 自定义高度(像素)，0=自动
|
||
"image_window_title": "", # 投图窗口标题模板，可用 {name}；空=默认
|
||
"tts_voice": "", # 默认 TTS 音色，空=系统默认
|
||
"tts_rate": 0, # 语速 -10~10
|
||
"mp3_dir": "", # 本机音乐目录，空=自动找 音乐/Music
|
||
# ---- 下面三项是"边界可配置"：由电脑主人决定允许什么 ----
|
||
"file_roots": "", # 额外允许的目录，分号分隔
|
||
"extra_apps": "", # 自定义快捷程序：显示名=路径，分号分隔
|
||
"extra_commands": "", # 自定义命令别名：显示名=命令，分号分隔
|
||
"download_dir": "received",
|
||
"screen_max_width": 480, # 镜像帧最大宽度
|
||
# 镜像 JPEG 质量。
|
||
# 实测质量对编码耗时几乎没影响（都是 ~17ms），主要影响体积：
|
||
# 55 -> 30KB/帧 72 -> 45KB/帧 85 -> 70KB/帧
|
||
# 取 72：画质明显更清楚，局域网完全带得动。
|
||
"screen_quality": 72,
|
||
# 默认帧率。
|
||
# 实测抓屏 33ms + 编码 17ms = 50ms，也就是**硬件上限约 20 帧/秒**。
|
||
# 原来写 3 是白白浪费了 6 倍的能力。
|
||
# 取 12：够顺，又给抓屏留了余量，不至于把 CPU 占满。
|
||
# （画面静止时会自动降频，见 screen.py 的 _loop）
|
||
"screen_fps": 12,
|
||
}
|
||
|
||
|
||
class Store:
|
||
"""线程安全的 JSON 存储。"""
|
||
|
||
def __init__(self, directory: Path):
|
||
self.dir = Path(directory)
|
||
try:
|
||
self.dir.mkdir(parents=True, exist_ok=True)
|
||
probe = self.dir / ".write_test"
|
||
probe.write_text("ok", encoding="utf-8")
|
||
probe.unlink()
|
||
except OSError:
|
||
# 装到 Program Files 这类只读目录时，原来会直接崩在启动。
|
||
# 退到用户自己的 AppData —— 那里一定可写。
|
||
fallback = Path(os.environ.get("LOCALAPPDATA",
|
||
str(Path.home()))) / "LanRemote"
|
||
try:
|
||
fallback.mkdir(parents=True, exist_ok=True)
|
||
except OSError:
|
||
pass
|
||
self.dir = fallback
|
||
self._lock = threading.RLock()
|
||
self.settings_file = self.dir / "settings.json"
|
||
self.devices_file = self.dir / "devices.json"
|
||
self.settings = self._load(self.settings_file, dict(DEFAULT_SETTINGS))
|
||
self.devices = self._load(self.devices_file, {})
|
||
if not self.settings.get("server_name"):
|
||
self.settings["server_name"] = socket.gethostname()
|
||
|
||
# ---------- 基础读写 ----------
|
||
|
||
@staticmethod
|
||
def _load(path: Path, fallback):
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
if isinstance(data, dict):
|
||
if isinstance(fallback, dict) and fallback:
|
||
merged = dict(fallback)
|
||
merged.update(data)
|
||
return merged
|
||
return data
|
||
except (OSError, ValueError):
|
||
pass
|
||
return fallback
|
||
|
||
def save_settings(self) -> None:
|
||
with self._lock:
|
||
self._write(self.settings_file, self.settings)
|
||
|
||
def save_devices(self) -> None:
|
||
with self._lock:
|
||
self._write(self.devices_file, self.devices)
|
||
|
||
@staticmethod
|
||
def _write(path: Path, data) -> None:
|
||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||
with open(tmp, "w", encoding="utf-8") as fh:
|
||
json.dump(data, fh, ensure_ascii=False, indent=2)
|
||
os.replace(tmp, path)
|
||
|
||
def get(self, key: str, default=None):
|
||
with self._lock:
|
||
return self.settings.get(key, default)
|
||
|
||
def set(self, key: str, value) -> None:
|
||
with self._lock:
|
||
self.settings[key] = value
|
||
self.save_settings()
|
||
|
||
# ---------- 配对码 ----------
|
||
|
||
def new_pairing_code(self, digits: int = 6) -> str:
|
||
with self._lock:
|
||
code = "".join(secrets.choice("0123456789") for _ in range(digits))
|
||
self.settings["pairing_code"] = code
|
||
self.settings["pairing_created"] = time.time()
|
||
self.save_settings()
|
||
return code
|
||
|
||
MIN_CODE_LEN = 6
|
||
MAX_CODE_LEN = 16
|
||
|
||
def set_pairing_code(self, code: str) -> str:
|
||
"""设成用户自己的配对码。返回规范化后的值。
|
||
|
||
只存在本机 data/settings.json 里，不写进任何要分发的东西。
|
||
"""
|
||
cleaned = "".join(ch for ch in str(code or "").strip()
|
||
if ch.isdigit() or ch.isalpha())
|
||
if len(cleaned) < self.MIN_CODE_LEN:
|
||
raise ValueError(f"配对码至少 {self.MIN_CODE_LEN} 位")
|
||
if len(cleaned) > self.MAX_CODE_LEN:
|
||
raise ValueError(f"配对码最多 {self.MAX_CODE_LEN} 位")
|
||
with self._lock:
|
||
self.settings["pairing_code"] = cleaned
|
||
self.settings["pairing_created"] = time.time()
|
||
self.save_settings()
|
||
return cleaned
|
||
|
||
def pairing_code(self) -> str:
|
||
return self.get("pairing_code", "")
|
||
|
||
# ---------- 设备 ----------
|
||
|
||
def remember_device(self, device_id: str, name: str) -> str:
|
||
token = secrets.token_hex(16)
|
||
with self._lock:
|
||
self.devices[device_id] = {
|
||
"name": name or "未知设备",
|
||
"token": token,
|
||
"paired_at": time.time(),
|
||
"last_seen": time.time(),
|
||
}
|
||
self.save_devices()
|
||
return token
|
||
|
||
def verify_token(self, device_id: str, token: str) -> bool:
|
||
with self._lock:
|
||
rec = self.devices.get(device_id)
|
||
if not rec:
|
||
return False
|
||
ok = secrets.compare_digest(str(rec.get("token", "")), str(token or ""))
|
||
if ok:
|
||
with self._lock:
|
||
rec["last_seen"] = time.time()
|
||
self.save_devices()
|
||
return ok
|
||
|
||
def touch_device(self, device_id: str) -> None:
|
||
with self._lock:
|
||
rec = self.devices.get(device_id)
|
||
if rec:
|
||
rec["last_seen"] = time.time()
|
||
self.save_devices()
|
||
|
||
def forget_device(self, device_id: str) -> bool:
|
||
with self._lock:
|
||
existed = self.devices.pop(device_id, None) is not None
|
||
if existed:
|
||
self.save_devices()
|
||
return existed
|
||
|
||
def device_list(self) -> list[dict]:
|
||
with self._lock:
|
||
items = []
|
||
for dev_id, rec in self.devices.items():
|
||
items.append({
|
||
"id": dev_id,
|
||
"name": rec.get("name", ""),
|
||
"paired_at": rec.get("paired_at", 0),
|
||
"last_seen": rec.get("last_seen", 0),
|
||
})
|
||
return sorted(items, key=lambda d: d["last_seen"], reverse=True)
|
||
|
||
def received_dir(self) -> Path:
|
||
path = self.dir.parent / str(self.get("download_dir", "received"))
|
||
path.mkdir(parents=True, exist_ok=True)
|
||
return path
|
||
|
||
|
||
def app_dir() -> Path:
|
||
"""程序所在目录（兼容 PyInstaller 冻结环境）。"""
|
||
if getattr(sys, "frozen", False):
|
||
return Path(sys.executable).resolve().parent
|
||
return Path(__file__).resolve().parent.parent
|
||
|
||
|
||
def data_dir() -> Path:
|
||
return app_dir() / "data"
|