# -*- 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"