智慧体育 AI 体测系统
技术全景分析

"""
本地 SQLite 数据库管理 - 设备端持久化层
使用 WAL 模式保证断电数据完整性（对应 Android Room ORM）
"""

class SyncStatus(IntEnum):
    PENDING   = 0  # 本地已写入，等待上传
    UPLOADING = 1  # 上传中
    SYNCED    = 2  # 服务端已确认
    FAILED    = 3  # 重试超限，进入死信
    DRAFT     = 4  # 违规/未完整，待审核列表

@dataclass
class TestRecord:
    record_id:     str   = field(default_factory=lambda: str(uuid.uuid4()))
    device_id:     str   = ""
    student_id:    str   = ""
    student_name:  str   = ""
    project:       str   = ""
    score:         float = 0.0
    grade:         str   = ""
    is_valid:      bool  = True
    violation:     str   = ""          # 违规描述，空=无违规
    motion_params: dict  = field(default_factory=dict)
    sync_status:   int   = SyncStatus.PENDING
    video_path:    str   = ""

    def to_upload_payload(self) -> dict:
        """序列化为上传给服务端的 JSON 载荷"""
        return { ... }

class LocalDatabase:
    def __init__(self, db_path: str = ":memory:"):
        self._conn = sqlite3.connect(db_path, check_same_thread=False)
        self._init_schema()

    def _init_schema(self):
        self._conn.executescript("""
            PRAGMA journal_mode=WAL;  -- 关键：WAL 模式防断电丢数据

            CREATE TABLE IF NOT EXISTS test_records (
                record_id    TEXT PRIMARY KEY,
                device_id    TEXT NOT NULL,
                student_id   TEXT NOT NULL,
                project      TEXT NOT NULL,
                score        REAL,
                grade        TEXT,
                sync_status  INTEGER DEFAULT 0,  -- SyncStatus 枚举
                retry_count  INTEGER DEFAULT 0,
                video_path   TEXT DEFAULT '',
                created_at   REAL
            );
            CREATE INDEX IF NOT EXISTS idx_records_sync
                ON test_records(sync_status, retry_count);
        """)

    def get_pending_records(self, limit: int = 20) -> List[TestRecord]:
        """获取待上传队列（PENDING + FAILED && retry < 5）"""
        rows = self._conn.execute("""
            SELECT * FROM test_records
            WHERE sync_status IN (0, 3)  -- PENDING or FAILED
              AND retry_count < 5
            ORDER BY created_at ASC
            LIMIT ?
        """, (limit,)).fetchall()
        return [self._row_to_record(r) for r in rows]

    def update_sync_status(self, record_id: str,
                           status: SyncStatus, retry_delta: int = 0):
        self._conn.execute("""
            UPDATE test_records
            SET sync_status = ?, retry_count = retry_count + ?
            WHERE record_id = ?
        """, (int(status), retry_delta, record_id))
        self._conn.commit()

"""
HTTP 上传引擎 - 含 HMAC 签名、指数退避重试、断点续传
对应 Android WorkManager OneTimeWorkRequest 调度
"""

class RequestSigner:
    """HMAC-SHA256 请求签名，防止数据篡改和重放攻击"""
    def sign(self, payload: dict, timestamp: int) -> str:
        body = f"{self.device_id}{timestamp}{json.dumps(payload, sort_keys=True)}"
        return hmac.new(self.secret_key, body.encode(), hashlib.sha256).hexdigest()

    def build_headers(self, payload: dict) -> dict:
        ts = int(time.time())
        return {
            "X-Device-Id":  self.device_id,
            "X-Timestamp":  str(ts),
            "X-Signature":  self.sign(payload, ts),  # 签名
        }

class ScoreUploader:
    MAX_RETRY    = 5
    BASE_BACKOFF = 2.0   # 指数退避基数（秒）

    def upload_single(self, record: TestRecord) -> bool:
        if not self.net.is_online:
            return False  # 离线，留在本地队列

        for attempt in range(1, self.MAX_RETRY + 1):
            try:
                resp = http_post(endpoint, payload, headers)
                if resp["code"] == 0:
                    self.db.update_sync_status(record_id, SYNCED)
                    return True  # ✅ 成功
            except NetworkError as e:
                # 指数退避：2¹=2s, 2²=4s, 2³=8s ...
                backoff = self.BASE_BACKOFF ** attempt + random.uniform(0, 1)
                time.sleep(backoff)

        # 超过最大重试 → 标记 FAILED，等待下次 flush
        self.db.update_sync_status(record_id, FAILED, retry_delta=1)
        return False

    def flush_pending_queue(self):
        """批量刷新离线积压的队列（网络恢复时自动触发）"""
        records = self.db.get_pending_records(limit=50)
        for r in records:
            if not self.net.is_online: break
            self.upload_single(r)

class VideoUploader:
    """视频分片断点续传（类 TUS 协议）"""
    CHUNK_SIZE = 512 * 1024  # 512 KB / 片

    def _upload_video(self, task: VideoUploadTask):
        # Step1: 向服务端申请上传会话，获取已接收字节偏移
        offset = self._init_upload_session(task)  # 断点位置

        # Step2: 从断点处分片循环上传
        while offset < task.file_size:
            if not self.net.is_online:
                # 保存断点，下次从此处继续
                self.db.update_video_progress(task_id, offset, PENDING)
                return

            chunk = read_file(task.local_path, offset, CHUNK_SIZE)
            headers["Content-Range"] = f"bytes {offset}-{offset+len(chunk)-1}/{task.file_size}"
            if upload_chunk(chunk, headers):
                offset += len(chunk)

        # Step3: 通知服务端合并所有分片
        self._finalize_upload(task)

"""
MQTT 实时推送 - 上行事件 + 下行指令 + 排行榜广播
生产使用 Eclipse Paho MQTT 客户端库（Android）
"""

# Topic 规范
TOPICS = {
    "上行事件": "device/{device_id}/events",
    "下行指令": "device/{device_id}/commands",
    "心跳":    "device/{device_id}/heartbeat",
    "排行榜":  "school/{school_id}/ranking/{project}",
}

class DeviceMQTTClient:
    def publish_score_event(self, record_payload: dict):
        """
        测试完成后，HTTP 上传的同时也通过 MQTT 发布事件，
        触发服务端实时重算排行榜并广播
        """
        self.broker.publish(
            topic   = f"device/{self.device_id}/events",
            payload = {
                "event_type": "score_completed",
                "data":       record_payload,
            }
        )

    @mqtt_device.on_command("update_test_plan")
    def on_test_plan(payload):
        """接收服务端下发的新测试计划"""
        save_to_local_db(payload)

    @mqtt_device.on_command("sync_students")
    def on_sync_students(payload):
        """接收学生底库更新指令，拉取增量数据"""
        pull_student_delta(last_sync_ts)

class ServerMQTTPusher:
    """服务端：收到成绩事件后，重算排行榜并广播到所有设备"""

    def _update_ranking(self, record: dict):
        project = record["project"]
        # Redis ZSET 维护排行榜（生产）/ 内存列表（模拟）
        ranking = self.rankings.setdefault(project, [])
        ranking.append((record["score"], record["student_name"]))
        ranking.sort(reverse=True)  # 分值高优先

        # 广播给所有订阅该项目排行榜的设备屏幕
        self.broker.publish(
            topic   = f"school/{self.school_id}/ranking/{project}",
            payload = {
                "top1_name":  ranking[0][1],
                "top1_score": ranking[0][0],
                "total":      len(ranking),
            }
        )

"""
模拟校园服务端 - Python 内置 http.server
对应生产环境：Spring Boot / FastAPI
"""

# API 路由表
ROUTES = {
    "POST  /api/v1/records":         "接收成绩上传（幂等）",
    "POST  /api/v1/videos/init":     "初始化视频上传会话",
    "PATCH /api/v1/videos/chunk":    "接收视频分片",
    "POST  /api/v1/videos/complete": "完成视频上传，触发合并",
    "HEAD  /health":                 "健康探测",
    "GET   /api/v1/students/sync":   "下发学生底库（设备拉取）",
}

def _handle_score_upload(self, body: dict):
    record_id = body.get("record_id")

    # 幂等设计：record_id 已存在则直接返回成功
    if record_id in _records_store:
        self._respond(200, {"code": 0, "msg": "ok (idempotent)"})
        return

    # 持久化（模拟 MySQL 写入）
    _records_store[record_id] = {
        **body,
        "server_received_at": time.time(),
    }
    self._respond(200, {"code": 0, "record_id": record_id})

def _handle_video_init(self, body: dict):
    """支持断点续传：返回已接收字节偏移"""
    task_id = body["task_id"]
    existing = _video_sessions.get(task_id)

    if existing:
        offset = existing["offset"]  # 从断点继续
    else:
        offset = 0
        _video_sessions[task_id] = {"offset": 0, "size": body["file_size"]}

    self._respond(200, {"code": 0, "offset": offset})

def _handle_student_sync(self):
    """下发学生底库增量（含人脸特征向量）"""
    students = [
        {
            "student_id":    "S001",
            "name":          "张伟",
            "face_feature":  "BASE64_FEATURE_VECTOR",  # 非原始图像
        },
        # ...
    ]
    self._respond(200, {"students": students, "sync_ts": int(time.time())})