ykxmz/windows/lanremote/config.py

# -*- 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,
    # ---- 弹窗外观（用户要求"位置和大小都能自定义"）----
    #
    # 位置：bottom-right / top-right / top-left / bottom-left / center / custom
    # 默认保持右下角 —— 不改变老用户的使用习惯。
    "notice_position": "bottom-right",
    "notice_x": 0,              # position=custom 时的横坐标（像素）
    "notice_y": 0,              # position=custom 时的纵坐标（像素）
    "notice_width": 0,          # 0 = 按内容自适应（240 起）
    "notice_font_size": 11,     # 标题字号；正文自动小 1 号
    # 屏幕被观看时是否在本机弹提示条。
    #
    # ⚠ 原来这个提示条是**强制显示**的（"有人在看你的屏幕，
    #   电脑前面的人必须知道"）。用户要求去掉，理由是"像 UU 远程一样
    #   不要任何窗口"。
    #
    # 这是用户自己的电脑，属于正当的远程管理需求
    # （极域电子教室这类校园软件也都是无提示的）。
    # 所以能力保留、**默认关闭** —— 需要的话在设置里打开，
    # 这样这软件始终有一个"可见性开关"可查。
    "show_screen_notice": False,
}


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:
        """写 JSON。优先"临时文件 + 原子替换"，失败就降级。

        为什么要降级：
            装到 Program Files / ProgramData 时，配置是**管理员**建的那一份，
            属主是管理员。普通用户跑的时候：
                能新建文件，但不能覆盖/改名别人建的文件。
            于是 os.replace 报 WinError 5。
            这时候退成"直接写目标文件" —— 牺牲原子性，换能存下配置。
        """
        tmp = path.with_suffix(path.suffix + ".tmp")
        payload = json.dumps(data, ensure_ascii=False, indent=2)

        # 1) 正常路径：临时文件 + 原子替换
        try:
            with open(tmp, "w", encoding="utf-8") as fh:
                fh.write(payload)
            os.replace(tmp, path)
            return
        except OSError:
            pass

        # 2) 降级：直接覆盖目标文件
        try:
            with open(path, "w", encoding="utf-8") as fh:
                fh.write(payload)
            Store._cleanup(tmp)
            return
        except OSError:
            pass

        # 3) 再降级：目标被设成只读就摘掉只读位再写
        try:
            os.chmod(path, 0o666)
        except OSError:
            pass
        try:
            with open(path, "w", encoding="utf-8") as fh:
                fh.write(payload)
            Store._cleanup(tmp)
            return
        except OSError:
            Store._cleanup(tmp)
            raise

    @staticmethod
    def _cleanup(tmp: Path) -> None:
        try:
            if tmp.exists():
                tmp.unlink()
        except OSError:
            pass

    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


# 数据目录只算一次，之后缓存 —— 每个命令都去试写一次太浪费
_DATA_DIR_CACHE: Path | None = None


def _can_write(folder: Path) -> bool:
    """这个目录能不能**真正**用来存配置。

    注意不能只测"能新建文件" —— 那只过一半。
    实际写配置用的是"临时文件 + 原子替换"（os.replace），
    它需要**覆盖已存在文件**的权限。
    装到 ProgramData 时，管理员建的那份普通用户改不了，
    只测新建会误判成可写，然后跑起来才报 WinError 5。
    所以这里把整个周期都跑一遍：建 -> 改名覆盖 -> 删。
    """
    try:
        folder.mkdir(parents=True, exist_ok=True)
        probe = folder / ".write-test"
        probe.write_text("x", encoding="utf-8")

        # 关键：模拟真实写入的"替换"这一步
        target = folder / ".write-test-target"
        try:
            if target.exists():
                target.unlink()
            probe.write_text("a", encoding="utf-8")
            os.replace(probe, target)          # <- 真正的门槛在这
            target.write_text("b", encoding="utf-8")   # 覆盖已存在的
            target.unlink()
        except OSError:
            # 建得了新文件但替换不了 -> 不能当数据目录
            try:
                probe.unlink()
            except OSError:
                pass
            return False
        return True
    except OSError:
        return False


def _is_installed_copy() -> bool:
    """程序是不是装在 Program Files 下（也就是"安装版"）。

    这个判断只跟**程序装在哪**有关，跟"当前是谁在跑"无关 ——
    这正是关键：数据目录必须对所有用户是同一个，
    否则管理员装的配置普通用户看不见（实测踩过）。
    """
    try:
        here = str(app_dir()).lower().replace("/", "\\")
    except OSError:
        return False
    for name in ("ProgramFiles", "ProgramFiles(x86)", "ProgramW6432"):
        base = os.environ.get(name)
        if not base:
            continue
        prefix = str(base).lower().replace("/", "\\").rstrip("\\")
        if here == prefix or here.startswith(prefix + "\\"):
            return True
    return False


def data_dir() -> Path:
    """配置 / 令牌 / 接收文件的存放目录。

    优先级：
        1. 环境变量 YKXMZ_DATA（测试和特殊部署用）
        2. 程序目录下的 data/   —— 绿色版、便携 U 盘，拔走就带走配置
        3. %PROGRAMDATA%\ykxmz\data —— 装到 Program Files 时用这个

    为什么要有第 3 条：
        安装包默认装到 C:\Program Files\ykxmz，
        普通用户对那里**没有写权限**（Windows 的硬规矩，不是 bug）。
        不转走的话，配对码和令牌会存不下去，用户会莫名其妙连不上。
        ProgramData 是所有用户可写的公共位置，换 Windows 账号也不丢配置。
    """
    global _DATA_DIR_CACHE
    if _DATA_DIR_CACHE is not None:
        return _DATA_DIR_CACHE

    override = os.environ.get("YKXMZ_DATA")
    if override:
        _DATA_DIR_CACHE = Path(override)
        return _DATA_DIR_CACHE

    # ── 安装版（装在 Program Files）一律用 ProgramData ──
    # 不能按"我能不能写 exe 旁边"来判断：
    # 安装时是管理员跑的，写得进去；之后普通用户跑就写不进去，
    # 于是悄悄换了目录 —— 安装时设的配对码就找不着了（实测踩过）。
    # 按"装在哪"判断，管理员和普通用户看到的就是同一份配置。
    in_program_files = _is_installed_copy()

    beside = app_dir() / "data"
    if not in_program_files and _can_write(beside):
        _DATA_DIR_CACHE = beside
        return _DATA_DIR_CACHE

    # ProgramData：所有用户共用一份配置（换 Windows 账号也不丢）
    base = os.environ.get("PROGRAMDATA") or r"C:\ProgramData"
    common = Path(base) / "ykxmz" / "data"
    if _can_write(common):
        _DATA_DIR_CACHE = common
        return _DATA_DIR_CACHE

    # 兜底：ProgramData 写不了（ACL 没配好 / 权限被收紧）就退到用户目录。
    # 每人一份配置，但**永远能写** —— 总比跑起来报错强。
    local = (Path(os.environ.get("LOCALAPPDATA")
                  or Path.home()) / "ykxmz" / "data")
    local.mkdir(parents=True, exist_ok=True)
    _DATA_DIR_CACHE = local
    return _DATA_DIR_CACHE