ykxmz/windows/lanremote/protocol.py

# -*- coding: utf-8 -*-
"""通信协议常量与编解码。

设计目标：双端零外部依赖。
  * Windows 端：Python 标准库 socket + json
  * 安卓端：java.net.Socket + 系统自带 org.json

传输格式：TCP 长连接，每条消息一行 UTF-8 JSON，以 \\n 结尾。
二进制负载（图片/音频/文件）用 base64 分块传输。
"""
from __future__ import annotations

import base64
import json

DEFAULT_PORT = 8765
PROTOCOL_VERSION = 1
MAX_LINE_BYTES = 8 * 1024 * 1024        # 单行上限 8MB（约 6MB 原始数据）
CHUNK_SIZE = 256 * 1024
# 单次传输（文件 / 图片 / 音频）的总体积上限
MAX_TOTAL_BYTES = 256 * 1024 * 1024                 # 分块大小 256KB

# ---- 客户端 -> 服务端 命令 ----
CMD_PAIR = "pair"
CMD_AUTH = "auth"
CMD_PING = "ping"
CMD_IMAGE = "image"
CMD_AUDIO = "audio"
CMD_MEDIA = "media"
CMD_CLIP_SET = "clip_set"
CMD_CLIP_GET = "clip_get"
CMD_POWER = "power"
CMD_RUN = "run"
CMD_SAY = "say"
CMD_VOICE_LIST = "voice_list"
CMD_SET_VOICE = "set_voice"
CMD_AUDIO_LIST = "audio_list"
CMD_SFX_LIST = "sfx_list"               # 列出可用音效（内置 + 用户自己放的）
CMD_SFX_PLAY = "sfx_play"               # 播放某个音效
CMD_PLAY_LOCAL = "play_local"
CMD_SHUTDOWN_TIMER = "shutdown_timer"   # 定时关机 / 取消
CMD_MONITOR = "monitor"                 # 关闭 / 打开显示器
CMD_OPEN_URL = "open_url"               # 在电脑浏览器打开网址
CMD_QUICK = "quick"                     # 快捷启动程序
CMD_OPEN_FOLDER = "open_folder"         # 打开常用文件夹
CMD_PROC_LIST = "proc_list"             # 进程列表
CMD_PROC_KILL = "proc_kill"             # 结束进程
CMD_BATTERY = "battery"                 # 电量
CMD_DISK = "disk"                       # 磁盘占用
CMD_PERF = "perf"                       # CPU / 内存占用
CMD_CLIP_HISTORY = "clip_history"       # 剪贴板历史
CMD_SET_WALLPAPER = "set_wallpaper"     # 把收到的图设为壁纸
CMD_SAY_CLIP = "say_clip"               # 朗读剪贴板
CMD_FEATURE_LIST = "feature_list"       # 电脑端能做哪些功能
CMD_FEATURE_RUN = "feature_run"         # 执行某个功能
CMD_AUDIO_PAUSE = "audio_pause"         # 暂停电脑正在播的音频
CMD_AUDIO_RESUME = "audio_resume"       # 继续播放
CMD_AUDIO_STATUS = "audio_status"       # 当前播放状态
CMD_NOTIFY = "notify"
CMD_FILE_PUSH = "file_push"
CMD_FILE_LIST = "file_list"
CMD_FILE_PULL = "file_pull"
CMD_INFO = "info"
# 静默自动更新：从**用户自己的中转站**取新版本包，静默装好并重启。
# action=check 只看不动；action=start 真的更新。
CMD_UPDATE = "update"
# 远程鼠标控制：{"action":"move|click|down|up|scroll", ...}
CMD_MOUSE = "mouse"
# 一键显示桌面（Win+D）。用户要求的快捷栏固定按钮。
CMD_SHOW_DESKTOP = "show_desktop"
CMD_SCREEN_START = "screen_start"
CMD_SCREEN_STOP = "screen_stop"
CMD_SCREENSHOT = "screenshot"

# ---- 服务端 -> 客户端 事件 ----
EVT_FILE = "file"
EVT_CLIP = "clip"
EVT_STATUS = "status"
EVT_LOG = "log"
EVT_FRAME = "frame"
EVT_SCREEN_STOPPED = "screen_stopped"

MEDIA_ACTIONS = {
    "playpause": 0xB3,
    "next": 0xB0,
    "prev": 0xB1,
    "stop": 0xB2,
    "volup": 0xAF,
    "voldown": 0xAE,
    "mute": 0xAD,
}

POWER_ACTIONS = {"shutdown", "restart", "lock", "sleep", "abort"}

# 手表端会用到的中文动作名（便于 UI 直接显示）
MEDIA_LABELS = {
    "playpause": "播放/暂停",
    "next": "下一曲",
    "prev": "上一曲",
    "stop": "停止",
    "volup": "音量+",
    "voldown": "音量-",
    "mute": "静音",
}

POWER_LABELS = {
    "shutdown": "关机",
    "restart": "重启",
    "lock": "锁屏",
    "sleep": "睡眠",
    "abort": "取消关机",
}


def dumps(obj: dict) -> bytes:
    """编码成一行 JSON（UTF-8 + 换行）。"""
    return (json.dumps(obj, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8")


def loads(line: bytes) -> dict:
    """解析一行 JSON。"""
    return json.loads(line.decode("utf-8", errors="replace").strip())


def b64encode(data: bytes) -> str:
    return base64.b64encode(data).decode("ascii")


def b64decode(text: str) -> bytes:
    return base64.b64decode(text.encode("ascii"))


def chunk_bytes(data: bytes, size: int = CHUNK_SIZE):
    """把字节流切成若干块，产出 (seq, is_last, chunk)。"""
    if not data:
        yield 0, True, b""
        return
    total = (len(data) + size - 1) // size
    for index in range(total):
        piece = data[index * size:(index + 1) * size]
        yield index, index == total - 1, piece


class ChunkTooLarge(Exception):
    """分块累积超过上限，或顺序不对。"""


class ChunkAssembler:
    """接收端：按 seq 收集分块，拼回完整数据。

    MAX_LINE_BYTES 只管单行，累积体积必须另外设上限 ——
    否则已配对设备可以一直发 chunk 把内存吃干（最后 bytes() 还要再翻一倍）。
    """

    def __init__(self, limit: int = MAX_TOTAL_BYTES):
        self._buf = bytearray()
        self._expected = 0
        self._limit = int(limit)

    def feed(self, seq: int, is_last: bool, data: bytes) -> bytes | None:
        """返回完整数据（当收到最后一块时），否则 None。

        顺序不对或超限时抛 ChunkTooLarge，不静默拼出坏文件。
        """
        if seq == 0:
            self._buf = bytearray()
            self._expected = 0
        if seq != self._expected:
            self._buf = bytearray()
            self._expected = 0
            raise ChunkTooLarge(
                f"分块顺序不对（收到 {seq}，应为 {self._expected}）")
        if len(self._buf) + len(data) > self._limit:
            self._buf = bytearray()
            self._expected = 0
            raise ChunkTooLarge(f"文件超过上限（{self._limit // 1048576} MB）")
        self._buf.extend(data)
        self._expected = seq + 1
        if is_last:
            result = bytes(self._buf)
            self._buf = bytearray()
            self._expected = 0
            return result
        return None

    def reset(self) -> None:
        self._buf = bytearray()
        self._expected = 0