Echo-Tavern/app.py

from __future__ import annotations

import json
import re
import secrets
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode, urlparse
from typing import Any

import requests
from flask import Flask, jsonify, redirect, render_template, request, session, url_for
from flask_socketio import SocketIO, emit, join_room as sio_join_room
from werkzeug.security import check_password_hash, generate_password_hash

from config import Config, DEFAULT_SETTINGS, SETTINGS_FILE
from game.action_resolver import ActionResolver
from game.badges import (
    build_badge_progress,
    default_role_badges,
    normalize_badge_definitions,
    normalize_badge_id,
)
from game.card_modules import apply_modules_to_card
from game.room_manager import RoomManager
from game.state_manager import StateManager
from game.story_engine import StoryEngine
from services.deepseek_client import DeepSeekClient
from services.persistence import SQLitePersistence


ERROR_TEXTS = {
    "missing_room_or_player": "缺少房间信息，请重新进入房间。",
    "room_not_found": "房间不存在，请返回大厅重新加入。",
    "player_not_in_room": "你不在该房间中，请从大厅重新加入。",
    "only_host_can_start": "只有房主可以开始游戏。",
    "game_not_running": "游戏尚未开始，请先点击“房主开始”。",
    "player_not_found": "玩家信息不存在，请重新进入房间。",
    "empty_message": "聊天内容不能为空。",
    "empty_action": "行动内容不能为空。",
    "room_full": "房间已满，无法加入。",
    "no_pending_actions": "当前没有待处理行动。请先提交行动。",
    "round_resolving": "本轮正在结算，请稍候。",
    "action_already_submitted": "你已提交本轮行动，请等待 AI 结算后再提交下一次。",
    "game_in_progress": "本局已开始，不能中途加入。请等待下一局或使用原身份重连。",
    "game_already_started": "游戏已经开始，无需重复开始。",
    "players_not_ready": "仍有玩家未准备，无法开始。请等待所有玩家准备后再开始。",
    "game_ended": "本局已结束，请返回大厅创建新房间。",
    "username_taken": "登录账号已被占用。",
    "invalid_username": "登录账号格式无效，仅支持字母、数字、下划线，长度 3-20。",
    "weak_password": "密码至少需要 6 位字符。",
    "bad_credentials": "用户名或密码错误。",
    "auth_required": "请先登录账号。",
    "invalid_card_id": "游戏卡不存在。",
    "only_host_can_delete": "只有房主可以删除房间。",
    "card_delete_forbidden": "你只能删除自己创建的非官方游戏卡。",
    "card_in_use": "该游戏卡仍被房间使用中，不能删除。",
    "invalid_assist_prompt": "请先输入游戏卡创意描述。",
    "invalid_assist_modify_prompt": "请先输入修改指令。",
    "card_edit_forbidden": "你只能修改自己创建的游戏卡。",
    "invalid_card_payload": "游戏卡配置格式无效，请检查后重试。",
    "admin_required": "需要管理员权限。",
    "invalid_role": "权限组无效。",
    "cannot_remove_last_admin": "系统必须保留至少一名管理员。",
    "user_not_found": "用户不存在。",
    "only_host_can_transfer": "只有房主可以转交房主权限。",
    "cannot_transfer_to_self": "不能把房主权限转交给自己。",
    "target_player_not_found": "目标玩家不存在。",
    "only_host_can_kick": "只有房主可以踢出玩家。",
    "only_host_can_manage_public": "只有房主可以设置房间公开状态。",
    "only_lobby_can_kick": "游戏开始后不能踢人。",
    "cannot_kick_host": "不能踢出房主。",
    "spectator_readonly": "观战模式下不能执行该操作。",
    "player_is_dead": "你当前处于阵亡/失败状态，暂时不能提交行动。",
    "action_submission_timeout": "本轮已超时，无法再提交行动。",
    "round_waiting_for_players": "本轮仍在等待其他玩家或超时后再结算。",
    "ai_invalid_json": "AI 返回了无效 JSON，本轮未结算，请稍后重试。",
    "invite_invalid": "邀请参数无效。",
    "invite_self_forbidden": "不能邀请自己。",
    "invite_room_required": "请先选择要邀请进入的房间。",
    "invite_room_unavailable": "该房间当前不可邀请。",
    "invite_sender_not_in_room": "你不在该房间中，无法发起邀请。",
    "invite_not_found": "邀请不存在或已处理。",
    "invite_status_invalid": "邀请状态无效。",
    "oauth_identity_invalid": "OAuth 绑定参数无效。",
    "badge_locked": "该职业徽章尚未解锁。",
    "season_daily_card_not_completed": "需要先通关今日赛季日卡后才能解锁资料。",
    "badge_config_invalid": "职业徽章配置无效，请检查后重试。",
    "module_not_found": "扩展模组不存在。",
    "module_patch_invalid": "扩展模组补丁格式无效。",
    "points_locked": "积分账户已被锁定，请联系管理员。",
    "points_insufficient": "积分余额不足。",
    "invalid_points_amount": "积分数量无效。",
    "activation_code_invalid": "激活码无效。",
    "activation_code_used": "激活码已被使用。",
    "activation_code_expired": "激活码已过期。",
    "activation_code_unavailable": "激活码当前不可用。",
    "redeem_rate_limited": "操作过于频繁，请稍后再试。",
    "activation_code_not_found": "激活码不存在。",
    "invalid_activation_code_status": "激活码状态无效。",
    "shop_item_not_found": "商品不存在。",
    "invalid_shop_item": "商品参数无效。",
    "shop_item_unavailable": "商品未上架或不可购买。",
    "shop_item_out_of_stock": "商品库存不足。",
    "shop_item_limit_reached": "已达到该商品限购上限。",
    "shop_order_not_found": "订单不存在。",
    "shop_order_not_refundable": "该订单当前不可退款。",
    "tip_self_forbidden": "不能打赏自己。",
    "card_purchase_required": "该游戏卡为付费内容，请先购买后再创建房间。",
    "season_not_found": "当前没有可参与的赛季。",
    "season_inactive": "当前赛季未开启或已结束。",
    "season_task_not_found": "赛季任务不存在。",
    "season_task_not_completed": "赛季任务尚未完成。",
    "season_task_claimed": "该赛季任务已领取。",
    "season_card_not_found": "赛季活动卡片不存在。",
    "season_card_not_completed": "该活动卡片尚未完成。",
    "season_card_claimed": "该活动卡片已领取。",
    "season_reward_not_found": "赛季奖励不存在。",
    "season_reward_claimed": "该赛季奖励已领取。",
    "season_reward_locked": "赛季等级不足，暂不能领取该奖励。",
    "season_pass_required": "该奖励需要解锁高级通行证。",
}
ALLOWED_THEMES = {"minimal", "dopamine", "midnight", "mint", "sunset", "aurora", "graphite"}
ADMIN_SECTIONS = {
    "users",
    "rooms",
    "cards",
    "modules",
    "badges",
    "oauth",
    "activation_codes",
    "shop_items",
    "shop_orders",
    "points",
    "finance",
    "seasons",
}


def readable_error(error_code: str) -> str:
    if error_code.startswith("ai_call_failed:"):
        detail = error_code.split(":", 1)[1].strip()
        return f"AI 调用失败：{detail or 'unknown'}。本轮未结算，请重试。"
    if error_code.startswith("ai_invalid_schema:"):
        detail = error_code.split(":", 1)[1].strip()
        return f"AI 输出结构不符合约束（{detail or 'schema_error'}），本轮未结算，请重试。"
    return ERROR_TEXTS.get(error_code, f"操作失败：{error_code}")


def _utc_day_key(now_iso: str = "") -> str:
    if now_iso:
        dt = datetime.fromisoformat(now_iso.replace("Z", "+00:00"))
    else:
        dt = datetime.now(timezone.utc)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    dt = dt.astimezone(timezone.utc)
    return dt.strftime("%Y-%m-%d")


def normalize_theme(theme_value: Any, default: str = "dopamine") -> str:
    theme = str(theme_value or "").strip().lower()
    if theme in ALLOWED_THEMES:
        return theme
    return default


def normalize_admin_section(value: Any, default: str = "users") -> str:
    section = str(value or "").strip().lower()
    if section in ADMIN_SECTIONS:
        return section
    return default


def normalize_account_username(value: Any) -> str:
    text = str(value or "").strip().lower()
    if not text:
        return ""
    text = re.sub(r"[^a-z0-9_]+", "_", text).strip("_")
    if len(text) > 20:
        text = text[:20].rstrip("_")
    if not re.match(r"^[a-z0-9_]{3,20}$", text):
        return ""
    return text


def resolve_linuxdo_identity(userinfo: dict[str, Any]) -> tuple[str, str]:
    if not isinstance(userinfo, dict):
        raise ValueError("missing_user_id")

    nested_user = userinfo.get("user")
    user_payload = nested_user if isinstance(nested_user, dict) else {}

    raw_id = userinfo.get("id")
    if raw_id in {None, ""}:
        raw_id = user_payload.get("id")
    if raw_id in {None, ""}:
        raise ValueError("missing_user_id")

    username_candidates = [
        userinfo.get("username"),
        user_payload.get("username"),
        userinfo.get("login"),
        user_payload.get("login"),
        userinfo.get("preferred_username"),
        user_payload.get("preferred_username"),
    ]
    username = ""
    for candidate in username_candidates:
        normalized = normalize_account_username(candidate)
        if normalized:
            username = normalized
            break

    if not username:
        safe_id = re.sub(r"[^a-z0-9_]+", "_", str(raw_id).strip().lower()).strip("_")
        fallback = f"linuxdo_{safe_id}" if safe_id else "linuxdo_user"
        username = normalize_account_username(fallback) or "linuxdo_user"

    display_name = str(
        userinfo.get("name")
        or user_payload.get("name")
        or userinfo.get("username")
        or user_payload.get("username")
        or username
    ).strip()
    return username, (display_name[:24] or username)


def parse_json_object(raw_text: str) -> dict[str, Any] | None:
    text = str(raw_text or "").strip()
    if not text:
        return None
    candidates: list[str] = [text]
    # Try fenced JSON blocks first.
    fenced_blocks = re.findall(r"```(?:json)?\s*([\s\S]*?)```", text, flags=re.IGNORECASE)
    for block in fenced_blocks:
        block_text = block.strip()
        if block_text:
            candidates.append(block_text)
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        candidates.append(text[start : end + 1])

    decoder = json.JSONDecoder()
    for candidate in candidates:
        candidate = candidate.strip()
        if not candidate:
            continue
        try:
            payload = json.loads(candidate)
            if isinstance(payload, dict):
                return payload
        except json.JSONDecodeError:
            pass
        # Try scanning for the first decodable object to handle wrappers/noise.
        for idx, ch in enumerate(candidate):
            if ch != "{":
                continue
            try:
                payload, _ = decoder.raw_decode(candidate[idx:])
                if isinstance(payload, dict):
                    return payload
            except json.JSONDecodeError:
                continue
    return None


def parse_text_list(raw_value: Any, *, limit: int = 8) -> list[str]:
    if isinstance(raw_value, list):
        rows = [str(item).strip() for item in raw_value if str(item).strip()]
    else:
        rows = [row.strip() for row in re.split(r"[\n,，]", str(raw_value or "")) if row.strip()]
    return rows[:limit]


def parse_json_list(raw_value: Any) -> list[Any]:
    if isinstance(raw_value, list):
        return raw_value
    text = str(raw_value or "").strip()
    if not text:
        return []
    try:
        payload = json.loads(text)
        if isinstance(payload, list):
            return payload
    except json.JSONDecodeError:
        return []
    return []


ACHIEVEMENT_LEVEL_ALIASES = {
    "bronze": "bronze",
    "silver": "silver",
    "gold": "gold",
    "platinum": "platinum",
    "diamond": "diamond",
    "青铜": "bronze",
    "白银": "silver",
    "黄金": "gold",
    "铂金": "platinum",
    "钻石": "diamond",
}
ACHIEVEMENT_METRIC_ALIASES = {
    "plays": "plays",
    "play_count": "plays",
    "played_games": "plays",
    "completed_games": "plays",
    "wins": "wins",
    "win_count": "wins",
    "current_streak": "current_streak",
    "win_streak": "current_streak",
    "best_streak": "best_streak",
    "max_win_streak": "best_streak",
    "对局数": "plays",
    "完成局数": "plays",
    "胜场": "wins",
    "连胜": "current_streak",
    "最高连胜": "best_streak",
}


def parse_custom_achievements(raw_value: Any, *, limit: int = 20) -> list[dict[str, Any]]:
    rows = raw_value if isinstance(raw_value, list) else parse_json_list(raw_value)
    if not isinstance(rows, list):
        return []
    result: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    for idx, row in enumerate(rows[:limit]):
        if not isinstance(row, dict):
            continue
        name = str(row.get("name", "")).strip()[:48]
        if not name:
            continue

        raw_id = str(row.get("id", "")).strip().lower()
        if raw_id:
            ach_id = re.sub(r"[^a-z0-9_-]+", "_", raw_id).strip("_")[:64]
        else:
            seed = re.sub(r"[^a-z0-9_-]+", "_", name.lower()).strip("_")[:48] or f"custom_{idx+1}"
            ach_id = f"{seed}_{idx+1}"[:64]
        if not ach_id:
            continue
        if ach_id in seen_ids:
            continue
        seen_ids.add(ach_id)

        metric_raw = str(row.get("metric", "")).strip().lower()
        metric = ACHIEVEMENT_METRIC_ALIASES.get(metric_raw, "plays")

        tiers_raw = row.get("tiers", [])
        tiers_list = tiers_raw if isinstance(tiers_raw, list) else []
        tiers: list[dict[str, Any]] = []
        for tier_idx, tier in enumerate(tiers_list[:6]):
            if not isinstance(tier, dict):
                continue
            level_raw = str(tier.get("level", "")).strip().lower()
            level = ACHIEVEMENT_LEVEL_ALIASES.get(level_raw, "")
            if not level:
                continue
            threshold_raw = tier.get("threshold", 1)
            try:
                threshold = max(1, int(threshold_raw))
            except (TypeError, ValueError):
                threshold = 1
            title = str(tier.get("title", "")).strip()[:48] or name
            tiers.append(
                {
                    "level": level,
                    "threshold": threshold,
                    "title": title,
                    "sort": tier_idx,
                }
            )

        if not tiers:
            threshold_raw = row.get("threshold", 1)
            try:
                threshold = max(1, int(threshold_raw))
            except (TypeError, ValueError):
                threshold = 1
            level_raw = str(row.get("level", "bronze")).strip().lower()
            level = ACHIEVEMENT_LEVEL_ALIASES.get(level_raw, "bronze")
            tiers = [{"level": level, "threshold": threshold, "title": name, "sort": 0}]

        tiers.sort(key=lambda item: (item["threshold"], item["sort"]))
        for item in tiers:
            item.pop("sort", None)

        result.append(
            {
                "id": ach_id,
                "name": name,
                "description": str(row.get("description", "")).strip()[:180],
                "metric": metric,
                "tiers": tiers[:6],
            }
        )
    return result


def parse_condition_achievements(raw_value: Any, *, limit: int = 20) -> list[dict[str, Any]]:
    rows = raw_value if isinstance(raw_value, list) else parse_json_list(raw_value)
    if not isinstance(rows, list):
        return []

    allowed_types = {"ending_state", "flag", "metric", "round_no", "player_alive"}
    allowed_ops = {">=", "<=", ">", "<", "==", "!="}
    out: list[dict[str, Any]] = []
    seen_ids: set[str] = set()

    for idx, row in enumerate(rows[:limit]):
        if not isinstance(row, dict):
            continue
        name = str(row.get("name", "")).strip()[:48]
        if not name:
            continue
        raw_id = str(row.get("id", "")).strip().lower()
        if raw_id:
            ach_id = re.sub(r"[^a-z0-9_-]+", "_", raw_id).strip("_")[:64]
        else:
            seed = re.sub(r"[^a-z0-9_-]+", "_", name.lower()).strip("_")[:48] or f"condition_{idx+1}"
            ach_id = f"{seed}_{idx+1}"[:64]
        if not ach_id or ach_id in seen_ids:
            continue
        seen_ids.add(ach_id)

        level_raw = str(row.get("level", "bronze")).strip().lower()
        level = ACHIEVEMENT_LEVEL_ALIASES.get(level_raw, "bronze")
        title = str(row.get("title", "")).strip()[:48] or name

        raw_conditions = row.get("conditions", [])
        condition_rows = raw_conditions if isinstance(raw_conditions, list) else []
        conditions: list[dict[str, Any]] = []
        for cond in condition_rows[:12]:
            if not isinstance(cond, dict):
                continue
            cond_type = str(cond.get("type", "")).strip().lower()
            if cond_type not in allowed_types:
                continue
            op = str(cond.get("op", "==")).strip()
            if op not in allowed_ops:
                op = "=="
            normalized = {
                "type": cond_type,
                "op": op,
                "key": str(cond.get("key", "")).strip()[:72],
                "value": cond.get("value"),
            }
            if cond_type == "ending_state":
                normalized["value"] = str(cond.get("value", "")).strip()[:48]
            if cond_type in {"flag", "metric"} and not normalized["key"]:
                continue
            if cond_type == "player_alive" and not normalized["key"]:
                normalized["key"] = "self"
            conditions.append(normalized)

        if not conditions:
            continue

        out.append(
            {
                "id": ach_id,
                "name": name,
                "title": title,
                "description": str(row.get("description", "")).strip()[:180],
                "level": level,
                "conditions": conditions,
            }
        )
    return out


def parse_module_patch(raw_value: Any) -> dict[str, Any]:
    patch = raw_value if isinstance(raw_value, dict) else parse_json_object(raw_value)
    if not isinstance(patch, dict):
        return {}

    normalized: dict[str, Any] = {}

    list_fields = (
        "starter_goals_add",
        "player_roles_add",
        "interaction_modes_add",
        "system_rules_add",
        "npc_roster_add",
        "scene_blueprint_add",
        "ending_catalog_add",
        "hidden_truths_add",
        "feature_badges_add",
        "genre_tags_add",
        "special_choices_add",
    )
    for key in list_fields:
        if key in patch:
            normalized[key] = parse_text_list(patch.get(key, []), limit=24)

    if "end_conditions_add" in patch:
        end_rows = patch.get("end_conditions_add", [])
        if not isinstance(end_rows, list):
            end_rows = []
        normalized["end_conditions_add"] = [row for row in end_rows if isinstance(row, dict)][:24]

    if "custom_achievements_add" in patch:
        normalized["custom_achievements_add"] = parse_custom_achievements(
            patch.get("custom_achievements_add", []),
            limit=20,
        )
    if "condition_achievements_add" in patch:
        normalized["condition_achievements_add"] = parse_condition_achievements(
            patch.get("condition_achievements_add", []),
            limit=30,
        )

    runtime_raw = patch.get("runtime_options_patch", {})
    if isinstance(runtime_raw, dict):
        runtime_patch: dict[str, Any] = {}
        interaction_mode = str(runtime_raw.get("interaction_mode", "")).strip().lower()
        if interaction_mode in {"cooperative", "split", "hybrid"}:
            runtime_patch["interaction_mode"] = interaction_mode
        for key in (
            "sync_all_player_ai_replies",
            "allow_player_death",
            "allow_revive",
            "require_all_players_response",
        ):
            if key in runtime_raw:
                value = runtime_raw.get(key)
                if isinstance(value, bool):
                    runtime_patch[key] = value
                else:
                    text = str(value or "").strip().lower()
                    if text in {"1", "true", "yes", "on", "y"}:
                        runtime_patch[key] = True
                    elif text in {"0", "false", "no", "off", "n"}:
                        runtime_patch[key] = False
        if "response_timeout_seconds" in runtime_raw:
            try:
                runtime_patch["response_timeout_seconds"] = max(
                    5, min(300, int(runtime_raw.get("response_timeout_seconds", 45)))
                )
            except (TypeError, ValueError):
                runtime_patch["response_timeout_seconds"] = 45
        if runtime_patch:
            normalized["runtime_options_patch"] = runtime_patch

    module_flags = patch.get("module_flags", {})
    if isinstance(module_flags, dict):
        normalized["module_flags"] = module_flags

    return normalized


def normalize_assist_current(payload: Any) -> dict[str, Any]:
    raw = payload if isinstance(payload, dict) else {}

    data_scope = raw.get("data_scope", raw.get("data_scope_json", {}))
    if not isinstance(data_scope, dict):
        data_scope = parse_json_object(data_scope) or {}

    visibility = raw.get("visibility", raw.get("visibility_json", {}))
    if not isinstance(visibility, dict):
        visibility = parse_json_object(visibility) or {}

    systems = raw.get("systems", raw.get("systems_json", {}))
    if not isinstance(systems, dict):
        systems = parse_json_object(systems) or {}
    runtime_options = systems.get("runtime_options", {}) if isinstance(systems, dict) else {}
    if not isinstance(runtime_options, dict):
        runtime_options = {}

    interaction_mode = str(raw.get("interaction_mode", runtime_options.get("interaction_mode", "hybrid"))).strip().lower()
    if interaction_mode not in {"cooperative", "split", "hybrid"}:
        interaction_mode = "hybrid"

    def _to_bool(value: Any, default: bool) -> bool:
        if isinstance(value, bool):
            return value
        text = str(value or "").strip().lower()
        if text in {"1", "true", "yes", "on", "y"}:
            return True
        if text in {"0", "false", "no", "off", "n"}:
            return False
        return default

    sync_all_player_ai_replies = _to_bool(
        raw.get("sync_all_player_ai_replies", runtime_options.get("sync_all_player_ai_replies", False)),
        False,
    )
    allow_player_death = _to_bool(
        raw.get("allow_player_death", runtime_options.get("allow_player_death", False)),
        False,
    )
    allow_revive = _to_bool(
        raw.get("allow_revive", runtime_options.get("allow_revive", False)),
        False,
    )
    require_all_players_response = _to_bool(
        raw.get("require_all_players_response", runtime_options.get("require_all_players_response", False)),
        False,
    )
    timeout_raw = raw.get("response_timeout_seconds", runtime_options.get("response_timeout_seconds", 45))
    try:
        response_timeout_seconds = int(timeout_raw)
    except (TypeError, ValueError):
        response_timeout_seconds = 45
    response_timeout_seconds = max(5, min(300, response_timeout_seconds))

    metrics = raw.get("metrics", [])
    if not isinstance(metrics, list):
        metrics = parse_json_list(metrics)
    metrics = [row for row in metrics if isinstance(row, dict)][:20]

    status_tracks = raw.get("status_tracks", [])
    if not isinstance(status_tracks, list):
        status_tracks = parse_json_list(status_tracks)
    status_tracks = [row for row in status_tracks if isinstance(row, dict)][:20]

    end_conditions = raw.get("end_conditions", [])
    if not isinstance(end_conditions, list):
        end_conditions = parse_json_list(end_conditions)
    end_conditions = [row for row in end_conditions if isinstance(row, dict)][:30]

    return {
        "title": str(raw.get("title", "")).strip()[:84],
        "subtitle": str(raw.get("subtitle", "")).strip()[:84],
        "summary": str(raw.get("summary", "")).strip()[:420],
        "opening_text": str(raw.get("opening_text", "")).strip()[:660],
        "starter_goals": parse_text_list(raw.get("starter_goals", []), limit=8),
        "metrics": metrics,
        "status_tracks": status_tracks,
        "data_scope": data_scope,
        "visibility": visibility,
        "state_fields": parse_text_list(raw.get("state_fields", []), limit=8),
        "end_conditions": end_conditions,
        "systems": systems,
        "genre_tags": parse_text_list(raw.get("genre_tags", []), limit=10),
        "feature_badges": parse_text_list(raw.get("feature_badges", []), limit=12),
        "world_lore": str(raw.get("world_lore", "")).strip()[:3600],
        "core_conflict": str(raw.get("core_conflict", "")).strip()[:1800],
        "player_roles": parse_text_list(raw.get("player_roles", []), limit=12),
        "interaction_modes": parse_text_list(raw.get("interaction_modes", []), limit=12),
        "system_rules": parse_text_list(raw.get("system_rules", []), limit=14),
        "npc_roster": parse_text_list(raw.get("npc_roster", []), limit=16),
        "scene_blueprint": parse_text_list(raw.get("scene_blueprint", []), limit=16),
        "ending_catalog": parse_text_list(raw.get("ending_catalog", []), limit=12),
        "hidden_truths": parse_text_list(raw.get("hidden_truths", []), limit=12),
        "custom_achievements": parse_custom_achievements(raw.get("custom_achievements", []), limit=20),
        "inventory_label": str(raw.get("inventory_label", "")).strip()[:48],
        "clues_label": str(raw.get("clues_label", "")).strip()[:48],
        "choices_label": str(raw.get("choices_label", "")).strip()[:48],
        "scene_label": str(raw.get("scene_label", "")).strip()[:48],
        "hero_status_label": str(raw.get("hero_status_label", "")).strip()[:48],
        "hero_status_color": str(raw.get("hero_status_color", "")).strip()[:7],
        "cover_url": str(raw.get("cover_url", "")).strip()[:720],
        "room_background_url": str(raw.get("room_background_url", "")).strip()[:720],
        "interaction_mode": interaction_mode,
        "sync_all_player_ai_replies": sync_all_player_ai_replies,
        "allow_player_death": allow_player_death,
        "allow_revive": allow_revive,
        "require_all_players_response": require_all_players_response,
        "response_timeout_seconds": response_timeout_seconds,
        "design_notes": parse_text_list(raw.get("design_notes", []), limit=8),
        "condition_achievements": parse_condition_achievements(raw.get("condition_achievements", []), limit=30),
    }


def _merge_unique_texts(base_value: Any, patch_value: Any, *, limit: int) -> list[str]:
    base = parse_text_list(base_value, limit=limit * 3)
    patch = parse_text_list(patch_value, limit=limit * 3)
    merged = []
    seen: set[str] = set()
    for row in base + patch:
        if not row or row in seen:
            continue
        seen.add(row)
        merged.append(row)
        if len(merged) >= limit:
            break
    return merged


def _merge_object_list_by_key(base_value: Any, patch_value: Any, *, key_field: str, limit: int) -> list[dict[str, Any]]:
    base_rows = base_value if isinstance(base_value, list) else []
    patch_rows = patch_value if isinstance(patch_value, list) else []
    out: list[dict[str, Any]] = []
    index_by_key: dict[str, int] = {}

    def push_row(row: Any) -> None:
        if not isinstance(row, dict):
            return
        key = str(row.get(key_field, "")).strip()
        if key:
            if key in index_by_key:
                out[index_by_key[key]] = {**out[index_by_key[key]], **row}
                return
            index_by_key[key] = len(out)
        out.append(dict(row))

    for row in base_rows:
        push_row(row)
    for row in patch_rows:
        push_row(row)
    return out[:limit]


def _deep_merge(base: Any, patch: Any) -> Any:
    if not isinstance(base, dict) or not isinstance(patch, dict):
        return patch
    merged: dict[str, Any] = {**base}
    for key, value in patch.items():
        if key in merged and isinstance(merged[key], dict) and isinstance(value, dict):
            merged[key] = _deep_merge(merged[key], value)
        else:
            merged[key] = value
    return merged


def normalize_card_patch(raw_patch: Any) -> dict[str, Any]:
    patch = raw_patch if isinstance(raw_patch, dict) else {}
    normalized: dict[str, Any] = {}

    scalar_limits = {
        "title": 84,
        "subtitle": 84,
        "summary": 420,
        "opening_text": 660,
        "world_lore": 3600,
        "core_conflict": 1800,
        "inventory_label": 48,
        "clues_label": 48,
        "choices_label": 48,
        "scene_label": 48,
        "hero_status_label": 48,
        "hero_status_color": 7,
        "cover_url": 720,
        "room_background_url": 720,
        "interaction_mode": 16,
    }
    for key, limit in scalar_limits.items():
        if key in patch:
            normalized[key] = str(patch.get(key, "")).strip()[:limit]

    text_list_limits = {
        "starter_goals": 8,
        "state_fields": 8,
        "genre_tags": 10,
        "feature_badges": 12,
        "player_roles": 12,
        "interaction_modes": 12,
        "system_rules": 14,
        "npc_roster": 16,
        "scene_blueprint": 16,
        "ending_catalog": 12,
        "hidden_truths": 12,
        "design_notes": 8,
    }
    for key, limit in text_list_limits.items():
        if key in patch:
            normalized[key] = parse_text_list(patch.get(key, []), limit=limit)
    if "custom_achievements" in patch:
        normalized["custom_achievements"] = parse_custom_achievements(
            patch.get("custom_achievements", []), limit=20
        )
    if "condition_achievements" in patch:
        normalized["condition_achievements"] = parse_condition_achievements(
            patch.get("condition_achievements", []), limit=30
        )

    object_keys = {"data_scope", "visibility", "systems"}
    for key in object_keys:
        value = patch.get(key)
        if isinstance(value, dict):
            normalized[key] = value

    if isinstance(patch.get("metrics"), list):
        normalized["metrics"] = [row for row in patch.get("metrics", []) if isinstance(row, dict)][:20]
    if isinstance(patch.get("status_tracks"), list):
        normalized["status_tracks"] = [row for row in patch.get("status_tracks", []) if isinstance(row, dict)][:20]
    if isinstance(patch.get("end_conditions"), list):
        normalized["end_conditions"] = [row for row in patch.get("end_conditions", []) if isinstance(row, dict)][:30]

    for key in ("sync_all_player_ai_replies", "allow_player_death", "allow_revive", "require_all_players_response"):
        if key in patch:
            value = patch.get(key)
            if isinstance(value, bool):
                normalized[key] = value
            else:
                text = str(value or "").strip().lower()
                if text in {"1", "true", "yes", "on", "y"}:
                    normalized[key] = True
                elif text in {"0", "false", "no", "off", "n"}:
                    normalized[key] = False
    if "response_timeout_seconds" in patch:
        try:
            normalized["response_timeout_seconds"] = max(5, min(300, int(patch.get("response_timeout_seconds", 45))))
        except (TypeError, ValueError):
            normalized["response_timeout_seconds"] = 45

    return normalized


def apply_card_patch(current: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]:
    merged = dict(current)

    replace_keys = {
        "title",
        "subtitle",
        "summary",
        "opening_text",
        "world_lore",
        "core_conflict",
        "inventory_label",
        "clues_label",
        "choices_label",
        "scene_label",
        "hero_status_label",
        "hero_status_color",
        "cover_url",
        "room_background_url",
        "interaction_mode",
    }
    for key in replace_keys:
        if key in patch and str(patch.get(key, "")).strip():
            merged[key] = patch[key]

    text_list_limits = {
        "starter_goals": 8,
        "state_fields": 8,
        "genre_tags": 10,
        "feature_badges": 12,
        "player_roles": 12,
        "interaction_modes": 12,
        "system_rules": 14,
        "npc_roster": 16,
        "scene_blueprint": 16,
        "ending_catalog": 12,
        "hidden_truths": 12,
        "design_notes": 8,
    }
    for key, limit in text_list_limits.items():
        if key in patch:
            merged[key] = _merge_unique_texts(merged.get(key, []), patch.get(key, []), limit=limit)

    if "metrics" in patch:
        merged["metrics"] = _merge_object_list_by_key(
            merged.get("metrics", []), patch.get("metrics", []), key_field="key", limit=20
        )
    if "status_tracks" in patch:
        merged["status_tracks"] = _merge_object_list_by_key(
            merged.get("status_tracks", []), patch.get("status_tracks", []), key_field="key", limit=20
        )
    if "end_conditions" in patch:
        merged["end_conditions"] = _merge_object_list_by_key(
            merged.get("end_conditions", []), patch.get("end_conditions", []), key_field="id", limit=30
        )

    if "data_scope" in patch:
        merged["data_scope"] = _deep_merge(merged.get("data_scope", {}), patch.get("data_scope", {}))
    if "visibility" in patch:
        merged["visibility"] = _deep_merge(merged.get("visibility", {}), patch.get("visibility", {}))
    if "systems" in patch:
        merged["systems"] = _deep_merge(merged.get("systems", {}), patch.get("systems", {}))
    if "custom_achievements" in patch:
        merged["custom_achievements"] = parse_custom_achievements(
            patch.get("custom_achievements", []), limit=20
        )
    if "condition_achievements" in patch:
        merged["condition_achievements"] = parse_condition_achievements(
            patch.get("condition_achievements", []), limit=30
        )
    for key in ("sync_all_player_ai_replies", "allow_player_death", "allow_revive", "require_all_players_response"):
        if key in patch and isinstance(patch.get(key), bool):
            merged[key] = patch[key]
    if "response_timeout_seconds" in patch:
        try:
            merged["response_timeout_seconds"] = max(5, min(300, int(patch.get("response_timeout_seconds", 45))))
        except (TypeError, ValueError):
            merged["response_timeout_seconds"] = 45

    return merged


def create_app(config_object: type[Config] = Config) -> tuple[Flask, SocketIO]:
    app = Flask(__name__)
    app.config.from_object(config_object)

    socketio = SocketIO(app, cors_allowed_origins="*", async_mode="threading")

    persistence = SQLitePersistence(app.config["DB_PATH"])
    story_engine = StoryEngine()
    state_manager = StateManager(persistence, story_engine)
    action_resolver = ActionResolver()
    deepseek_client = DeepSeekClient(
        api_key=app.config.get("LLM_API_KEY", app.config.get("DEEPSEEK_API_KEY", "")),
        base_url=app.config.get("LLM_BASE_URL", app.config.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com")),
        model=app.config.get("LLM_MODEL", app.config.get("DEEPSEEK_MODEL", "deepseek-chat")),
        timeout=float(app.config.get("LLM_TIMEOUT", app.config.get("DEEPSEEK_TIMEOUT", 25))),
        connect_timeout=float(
            app.config.get("LLM_CONNECT_TIMEOUT", app.config.get("DEEPSEEK_CONNECT_TIMEOUT", 8))
        ),
        read_timeout=float(
            app.config.get("LLM_READ_TIMEOUT", app.config.get("DEEPSEEK_READ_TIMEOUT", app.config.get("DEEPSEEK_TIMEOUT", 25)))
        ),
        retries=int(app.config.get("LLM_RETRIES", app.config.get("DEEPSEEK_RETRIES", 2))),
        extra_headers=app.config.get("LLM_EXTRA_HEADERS", {}),
        force_json_mode=bool(app.config.get("LLM_FORCE_JSON_MODE", True)),
    )
    room_manager = RoomManager(
        persistence,
        state_manager,
        story_engine,
        action_resolver,
        deepseek_client,
        max_players=int(app.config.get("MAX_ROOM_PLAYERS", 8)),
        max_action_length=int(app.config.get("MAX_ACTION_LENGTH", 280)),
        badge_definitions_provider=lambda: get_role_badges_config(),
    )

    app.extensions["persistence"] = persistence
    app.extensions["story_engine"] = story_engine
    app.extensions["room_manager"] = room_manager

    def load_runtime_settings() -> dict[str, Any]:
        defaults = dict(DEFAULT_SETTINGS)
        defaults["linuxdo_oauth"] = dict(DEFAULT_SETTINGS.get("linuxdo_oauth", {}))
        defaults["role_badges"] = default_role_badges()
        defaults["points_system"] = dict(DEFAULT_SETTINGS.get("points_system", {}))
        defaults["season_system"] = dict(DEFAULT_SETTINGS.get("season_system", {}))
        try:
            payload = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return defaults
        if not isinstance(payload, dict):
            return defaults

        merged = dict(defaults)
        merged.update(payload)
        raw_oauth = payload.get("linuxdo_oauth", {})
        if isinstance(raw_oauth, dict):
            merged["linuxdo_oauth"] = {**defaults["linuxdo_oauth"], **raw_oauth}
        else:
            merged["linuxdo_oauth"] = dict(defaults["linuxdo_oauth"])
        raw_points = payload.get("points_system", {})
        if isinstance(raw_points, dict):
            merged["points_system"] = {**defaults["points_system"], **raw_points}
        else:
            merged["points_system"] = dict(defaults["points_system"])
        raw_season = payload.get("season_system", {})
        if isinstance(raw_season, dict):
            merged["season_system"] = {**defaults["season_system"], **raw_season}
        else:
            merged["season_system"] = dict(defaults["season_system"])
        if "role_badges" in payload:
            merged["role_badges"] = normalize_badge_definitions(
                payload.get("role_badges", []),
                default_to_builtin=False,
            )
        else:
            merged["role_badges"] = default_role_badges()
        return merged

    def save_runtime_settings(settings_payload: dict[str, Any]) -> None:
        SETTINGS_FILE.write_text(
            json.dumps(settings_payload, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )

    def load_settings_file_payload() -> dict[str, Any]:
        try:
            raw = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return {}
        return raw if isinstance(raw, dict) else {}

    def _is_valid_http_url(url_text: Any) -> bool:
        url = str(url_text or "").strip()
        if not url:
            return False
        parsed = urlparse(url)
        return parsed.scheme in {"http", "https"} and bool(parsed.netloc)

    def get_linuxdo_oauth_config(*, mask_secret: bool = False) -> dict[str, Any]:
        defaults = DEFAULT_SETTINGS.get("linuxdo_oauth", {})
        runtime = load_runtime_settings().get("linuxdo_oauth", {})
        raw = runtime if isinstance(runtime, dict) else {}
        cfg = {
            "enabled": bool(raw.get("enabled", defaults.get("enabled", False))),
            "client_id": str(raw.get("client_id", defaults.get("client_id", ""))).strip(),
            "client_secret": str(raw.get("client_secret", defaults.get("client_secret", ""))).strip(),
            "authorize_url": str(raw.get("authorize_url", defaults.get("authorize_url", ""))).strip(),
            "token_url": str(raw.get("token_url", defaults.get("token_url", ""))).strip(),
            "userinfo_url": str(raw.get("userinfo_url", defaults.get("userinfo_url", ""))).strip(),
            "scope": str(raw.get("scope", defaults.get("scope", ""))).strip(),
        }
        if mask_secret:
            secret = cfg["client_secret"]
            cfg["client_secret"] = "" if not secret else ("*" * 8)
            cfg["client_secret_configured"] = bool(secret)
        return cfg

    def get_role_badges_config(*, fallback_to_default: bool = True) -> list[dict[str, Any]]:
        runtime = load_runtime_settings()
        raw = runtime.get("role_badges", [])
        badges = normalize_badge_definitions(raw, default_to_builtin=False)
        if not badges and fallback_to_default:
            return default_role_badges()
        return badges

    def get_points_system_config() -> dict[str, Any]:
        raw = load_runtime_settings().get("points_system", {})
        cfg = raw if isinstance(raw, dict) else {}
        def _to_int(value: Any, default: int, minimum: int, maximum: int) -> int:
            try:
                out = int(value)
            except (TypeError, ValueError):
                out = int(default)
            return max(minimum, min(maximum, out))
        return {
            "enabled": bool(cfg.get("enabled", True)),
            "room_create_fee_enabled": bool(cfg.get("room_create_fee_enabled", False)),
            "room_create_fee": _to_int(cfg.get("room_create_fee", 0), 0, 0, 1000000),
            "redeem_ip_limit": _to_int(cfg.get("redeem_ip_limit", 12), 12, 1, 100),
            "redeem_user_limit": _to_int(cfg.get("redeem_user_limit", 10), 10, 1, 100),
        }

    def get_season_system_config() -> dict[str, Any]:
        raw = load_runtime_settings().get("season_system", {})
        cfg = raw if isinstance(raw, dict) else {}

        def _to_int(value: Any, default: int, minimum: int, maximum: int) -> int:
            try:
                out = int(value)
            except (TypeError, ValueError):
                out = int(default)
            return max(minimum, min(maximum, out))

        refresh_hours_raw = cfg.get("refresh_hours", [0, 8, 18])
        refresh_hours: list[int] = []
        if isinstance(refresh_hours_raw, list):
            for row in refresh_hours_raw[:8]:
                try:
                    val = int(row)
                except (TypeError, ValueError):
                    continue
                val = max(0, min(23, val))
                if val not in refresh_hours:
                    refresh_hours.append(val)
        refresh_hours.sort()
        if not refresh_hours:
            refresh_hours = [0, 8, 18]

        return {
            "enabled": bool(cfg.get("enabled", True)),
            "auto_create_enabled": bool(cfg.get("auto_create_enabled", True)),
            "auto_generate_theme": bool(cfg.get("auto_generate_theme", True)),
            "season_duration_days": _to_int(cfg.get("season_duration_days", 30), 30, 7, 120),
            "rest_days": _to_int(cfg.get("rest_days", 3), 3, 0, 14),
            "daily_card_count": _to_int(cfg.get("daily_card_count", 3), 3, 1, 8),
            "refresh_hours": refresh_hours,
            "manual_refresh_cost": _to_int(cfg.get("manual_refresh_cost", 60), 60, 0, 99999),
            "premium_pass_price": _to_int(cfg.get("premium_pass_price", 680), 680, 0, 999999),
        }

    def oauth_error_redirect(message: str) -> Any:
        return redirect(url_for("login_page", oauth_error=message))

    def current_user() -> dict[str, Any] | None:
        user_id = session.get("user_id")
        if not user_id:
            return None
        return persistence.get_user_by_id(str(user_id))

    @app.before_request
    def mark_session_activity() -> None:
        if request.endpoint == "static":
            return
        user_id = str(session.get("user_id", "")).strip()
        if not user_id:
            return
        persistence.touch_user_activity(user_id, is_online=True)

    def serialize_user(user: dict[str, Any] | None) -> dict[str, Any] | None:
        if user is None:
            return None
        return {
            "user_id": user["user_id"],
            "username": user["username"],
            "display_name": user.get("display_name", user["username"]),
            "public_name": user["username"],
            "role": user.get("role", "player"),
            "avatar_color": user.get("avatar_color", "#0070f2"),
            "avatar_url": user.get("avatar_url", ""),
            "theme": normalize_theme(user.get("theme", "minimal")),
            "bio": user.get("bio", ""),
            "active_badge": normalize_badge_id(user.get("active_badge", ""), get_role_badges_config()),
        }

    def serialize_public_user(user: dict[str, Any] | None) -> dict[str, Any] | None:
        if user is None:
            return None
        return {
            "user_id": user["user_id"],
            "username": user["username"],
            "display_name": user.get("display_name", user["username"]),
            "avatar_color": user.get("avatar_color", "#0070f2"),
            "avatar_url": user.get("avatar_url", ""),
            "theme": normalize_theme(user.get("theme", "dopamine")),
            "bio": user.get("bio", ""),
            "created_at": user.get("created_at", ""),
            "updated_at": user.get("updated_at", ""),
            "active_badge": normalize_badge_id(user.get("active_badge", ""), get_role_badges_config()),
        }

    def build_user_badges_payload(user: dict[str, Any]) -> dict[str, Any]:
        stats = persistence.get_user_game_stats(user["user_id"])
        progress = build_badge_progress(stats, user.get("active_badge", ""), get_role_badges_config())
        return {
            "active_badge": progress.get("active_badge", ""),
            "badges": progress.get("badges", []),
            "unlocked_ids": progress.get("unlocked_ids", []),
            "stats_snapshot": progress.get("stats_snapshot", {}),
        }

    def build_user_points_payload(user_id: str) -> dict[str, Any]:
        account = persistence.get_points_account(user_id)
        rewards = persistence.evaluate_achievement_point_rewards(user_id)
        if rewards:
            account = persistence.get_points_account(user_id)
        return {
            "account": account,
            "new_rewards": rewards,
        }

    def build_user_profile_payload(user: dict[str, Any], *, viewer_user_id: str = "") -> dict[str, Any]:
        presence = persistence.get_user_presence(user["user_id"])
        stats = persistence.get_user_game_stats(user["user_id"])
        achievements = persistence.get_user_achievement_bundle(user["user_id"])
        recent_rooms = persistence.list_user_rooms(user["user_id"], limit=8)
        for row in recent_rooms:
            card = persistence.get_game_card(row.get("game_card_id", ""))
            row["card_title"] = card["title"] if card else "未知剧本"
            row["card_cover_url"] = card.get("cover_url", "") if card else ""
        can_view_points = bool(viewer_user_id and str(viewer_user_id) == str(user["user_id"]))
        active_season = persistence.get_active_season()
        season_profile = None
        if active_season is not None:
            try:
                season_profile = persistence.get_or_create_user_season_profile(
                    user["user_id"],
                    str(active_season.get("season_id", "")).strip(),
                )
            except ValueError:
                season_profile = None
        equipped_decorations = persistence.get_user_equipped_decorations(user["user_id"])
        return {
            "user": serialize_public_user(user),
            "presence": presence,
            "stats": stats,
            "badges": build_user_badges_payload(user),
            "points": build_user_points_payload(user["user_id"]) if can_view_points else {"account": None, "new_rewards": []},
            "achievements": achievements,
            "recent_rooms": recent_rooms,
            "season": {
                "active_season": active_season,
                "profile": season_profile,
                "equipped_decorations": equipped_decorations,
            },
        }

    def build_room_summary_payload(room_id: str) -> dict[str, Any]:
        def _to_int(value: Any, default: int = 0) -> int:
            try:
                return int(value)
            except (TypeError, ValueError):
                return int(default)

        safe_room_id = str(room_id or "").strip().upper()
        room = persistence.get_room(safe_room_id)
        if room is None:
            raise ValueError("room_not_found")

        state = room.get("game_state", {})
        if not isinstance(state, dict):
            state = {}
        global_flags = state.get("global_flags", {})
        if not isinstance(global_flags, dict):
            global_flags = {}
        runtime_options = global_flags.get("runtime_options", {})
        if not isinstance(runtime_options, dict):
            runtime_options = {}

        players = persistence.list_players(safe_room_id)
        actions = persistence.list_actions(safe_room_id, limit=1200)
        messages = persistence.list_messages(safe_room_id, limit=600)
        card = persistence.get_game_card(str(room.get("game_card_id", "")))
        user_cache: dict[str, dict[str, Any]] = {}
        for player in players:
            user_id = str(player.get("user_id") or "").strip()
            if not user_id or user_id in user_cache:
                continue
            user_cache[user_id] = persistence.get_user_by_id(user_id) or {}

        contributions: dict[str, dict[str, Any]] = {}
        for player in players:
            player_id = str(player.get("player_id", "")).strip()
            if not player_id:
                continue
            contributions[player_id] = {
                "actions": 0,
                "primary": 0,
                "support": 0,
                "blocked": 0,
                "chats": 0,
                "score": 0.0,
            }

        for row in actions:
            player_id = str(row.get("player_id", "")).strip()
            if player_id in contributions:
                contributions[player_id]["actions"] += 1

        for row in messages:
            if str(row.get("kind", "")).strip() != "chat":
                continue
            player_id = str(row.get("player_id", "")).strip()
            if player_id in contributions:
                contributions[player_id]["chats"] += 1

        action_history = state.get("action_history", [])
        if isinstance(action_history, list):
            for history_row in action_history:
                if not isinstance(history_row, dict):
                    continue
                merged = history_row.get("merged_actions", {})
                if not isinstance(merged, dict):
                    continue
                ordered = merged.get("ordered_actions", [])
                if not isinstance(ordered, list):
                    continue
                for action_row in ordered:
                    if not isinstance(action_row, dict):
                        continue
                    player_id = str(action_row.get("player_id", "")).strip()
                    if player_id not in contributions:
                        continue
                    resolution = str(action_row.get("resolution", "primary")).strip().lower()
                    if resolution == "primary":
                        contributions[player_id]["primary"] += 1
                    elif resolution == "support":
                        contributions[player_id]["support"] += 1
                    elif resolution == "blocked":
                        contributions[player_id]["blocked"] += 1

        for score_row in contributions.values():
            score_row["score"] = round(
                score_row["actions"] * 1.2
                + score_row["primary"] * 3.0
                + score_row["support"] * 1.8
                + score_row["blocked"] * 0.6
                + score_row["chats"] * 0.45,
                2,
            )

        mvp_player_id = ""
        mvp_score = -1.0
        for player_id, score_row in contributions.items():
            key = (
                float(score_row.get("score", 0.0)),
                int(score_row.get("primary", 0)),
                int(score_row.get("actions", 0)),
            )
            best = (
                float(mvp_score),
                int(contributions.get(mvp_player_id, {}).get("primary", 0)),
                int(contributions.get(mvp_player_id, {}).get("actions", 0)),
            )
            if key > best:
                mvp_player_id = player_id
                mvp_score = float(score_row.get("score", 0.0))

        player_states = global_flags.get("player_states", {})
        if not isinstance(player_states, dict):
            player_states = {}

        status_labels = {
            "alive": "存活",
            "down": "倒地",
            "dead": "阵亡",
            "failed": "失败",
            "revived": "复活",
        }

        player_rows: list[dict[str, Any]] = []
        alive_count = 0
        for player in players:
            player_id = str(player.get("player_id", "")).strip()
            state_row = player_states.get(player_id, {})
            if not isinstance(state_row, dict):
                state_row = {}
            raw_status = str(state_row.get("status", "")).strip().lower()
            is_alive = bool(state_row.get("is_alive", True))
            if raw_status in {"dead", "failed"}:
                is_alive = False
            if raw_status in {"alive", "revived"}:
                is_alive = True
            if not raw_status:
                raw_status = "alive" if is_alive else "dead"
            if raw_status == "down":
                is_alive = False
            if is_alive:
                alive_count += 1

            user_id = str(player.get("user_id") or "").strip()
            user_row = user_cache.get(user_id, {})
            score_row = contributions.get(player_id, {})
            player_rows.append(
                {
                    "player_id": player_id,
                    "name": player.get("name", player_id),
                    "username": user_row.get("username", ""),
                    "display_name": user_row.get("display_name", ""),
                    "avatar_color": user_row.get("avatar_color", "#0070f2"),
                    "avatar_url": user_row.get("avatar_url", ""),
                    "active_badge": user_row.get("active_badge", ""),
                    "status": raw_status,
                    "status_label": status_labels.get(raw_status, raw_status or "未知"),
                    "is_alive": is_alive,
                    "deaths": max(0, _to_int(state_row.get("deaths", 0), 0)),
                    "revives": max(0, _to_int(state_row.get("revives", 0), 0)),
                    "contribution": score_row,
                    "is_mvp": player_id == mvp_player_id,
                }
            )

        mvp = next((row for row in player_rows if row.get("player_id") == mvp_player_id), None)
        ending_summary = str(global_flags.get("ending_summary", "")).strip()
        if not ending_summary:
            history = state.get("action_history", [])
            if isinstance(history, list) and history:
                tail = history[-1] if isinstance(history[-1], dict) else {}
                ending_summary = str(tail.get("narration", "")).strip()
        if not ending_summary:
            ending_summary = str(state.get("scene_description", "")).strip()

        interaction_mode = str(runtime_options.get("interaction_mode", "hybrid")).strip().lower()
        if interaction_mode not in {"cooperative", "split", "hybrid"}:
            interaction_mode = "hybrid"
        card_content_pack = global_flags.get("card_content_pack", {})
        if not isinstance(card_content_pack, dict):
            card_content_pack = {}
        enabled_modules = card_content_pack.get("enabled_modules", [])
        if not isinstance(enabled_modules, list):
            enabled_modules = []

        return {
            "room_id": safe_room_id,
            "status": room.get("status", ""),
            "card_id": str(room.get("game_card_id", "")).strip(),
            "card_title": card.get("title", "未知剧本") if card else "未知剧本",
            "ending_state": str(state.get("ending_state", "")).strip(),
            "ending_summary": ending_summary,
            "rounds": max(0, _to_int(state.get("round_no", 0), 0)),
            "chapter": max(0, _to_int(state.get("chapter", 0), 0)),
            "total_actions": len(actions),
            "total_players": len(player_rows),
            "alive_count": alive_count,
            "mvp": mvp,
            "players": player_rows,
            "interaction_mode": interaction_mode,
            "sync_all_player_ai_replies": bool(runtime_options.get("sync_all_player_ai_replies", False)),
            "allow_player_death": bool(runtime_options.get("allow_player_death", False)),
            "allow_revive": bool(runtime_options.get("allow_revive", False)),
            "require_all_players_response": bool(runtime_options.get("require_all_players_response", False)),
            "response_timeout_seconds": max(5, min(300, _to_int(runtime_options.get("response_timeout_seconds", 45), 45))),
            "enabled_modules": enabled_modules[:12],
            "module_count": len(enabled_modules),
            "updated_at": str(room.get("updated_at", "")).strip(),
        }

    def build_auto_season_blueprint(
        *,
        theme_hint: str = "",
        duration_days: int = 30,
        pass_price: int = 680,
    ) -> dict[str, Any]:
        safe_theme = str(theme_hint or "").strip()[:40] or "迷雾电波"
        now = datetime.now(timezone.utc)
        start_at = now.isoformat()
        end_at = (now + timedelta(days=max(7, min(120, int(duration_days))))).isoformat()
        season_name = f"{safe_theme}赛季"
        description = f"围绕“{safe_theme}”展开的连续故事赛季，包含每日刷新卡片、周常任务与通行证奖励。"
        story_seed = f"城市边缘的信号塔持续发出异常广播，所有线索都指向{safe_theme}背后的真相。"
        story_outline = (
            "第一阶段收集碎片情报，第二阶段完成多线挑战，第三阶段进入赛季终章并解锁纪念奖励。"
        )
        announcements = [
            {
                "title": f"{season_name} 开启公告",
                "content": (
                    f"{season_name} 现已开启。每日 3 次刷新活动卡片，完成任务可提升赛季等级并领取限定装扮。"
                ),
            }
        ]
        task_templates = [
            {
                "task_type": "daily",
                "title": "每日热身",
                "description": "完成 1 局任意剧本。",
                "event_key": "complete_game",
                "target_value": 1,
                "reward_points": 16,
                "reward_xp": 36,
                "reward_season_coin": 4,
                "weight": 100,
                "difficulty": "easy",
            },
            {
                "task_type": "daily",
                "title": "行动日志",
                "description": "累计提交 10 次行动。",
                "event_key": "submit_action",
                "target_value": 10,
                "reward_points": 14,
                "reward_xp": 34,
                "reward_season_coin": 4,
                "weight": 90,
                "difficulty": "easy",
            },
            {
                "task_type": "weekly",
                "title": "周常远征",
                "description": "本周完成 5 局游戏。",
                "event_key": "complete_game",
                "target_value": 5,
                "reward_points": 52,
                "reward_xp": 120,
                "reward_season_coin": 14,
                "weight": 100,
                "difficulty": "normal",
            },
            {
                "task_type": "season",
                "title": "赛季主线",
                "description": "整个赛季完成 35 局游戏。",
                "event_key": "complete_game",
                "target_value": 35,
                "reward_points": 220,
                "reward_xp": 480,
                "reward_season_coin": 80,
                "weight": 100,
                "difficulty": "hard",
            },
        ]
        card_pools = [
            {
                "card_type": "daily",
                "title_hint": f"{safe_theme}·晨雾",
                "story_hook": "无线电频道出现重复坐标，你需要确认来源。",
                "objective_template": "完成 1 局并记录关键事件。",
                "hint_template": "优先完成基础目标，稳拿收益。",
                "event_key": "complete_game",
                "target_value": 1,
                "reward_points": 12,
                "reward_xp": 28,
                "rarity": "normal",
                "weight": 120,
            },
            {
                "card_type": "rotating",
                "title_hint": f"{safe_theme}·回廊",
                "story_hook": "平行片段开始互相干扰，你需要同步多方线索。",
                "objective_template": "累计提交 8 次行动。",
                "hint_template": "组队可更快完成目标。",
                "event_key": "submit_action",
                "target_value": 8,
                "reward_points": 20,
                "reward_xp": 52,
                "rarity": "rare",
                "weight": 88,
            },
            {
                "card_type": "surprise",
                "title_hint": f"{safe_theme}·暗潮",
                "story_hook": "隐藏档案突然解密，风险和收益同时提升。",
                "objective_template": "赢下 1 局并保持存活。",
                "hint_template": "建议协作推进，提高成功率。",
                "event_key": "win_game",
                "target_value": 1,
                "reward_points": 50,
                "reward_xp": 120,
                "rarity": "epic",
                "weight": 20,
            },
        ]
        pass_rewards: list[dict[str, Any]] = []
        for level in range(1, 31):
            pass_rewards.append(
                {
                    "reward_level": level,
                    "reward_track": "free",
                    "title": f"免费档 Lv.{level}",
                    "reward_points": 10 if level % 5 else 26,
                    "reward_xp": 0,
                    "reward_season_coin": 2 if level % 3 == 0 else 0,
                    "reward_payload": {"titles": []},
                }
            )
            payload = {}
            if level == 15:
                payload = {"avatar_frames": ["season_auto_frame_epic"]}
            if level == 30:
                payload = {
                    "titles": ["season_auto_title_legend"],
                    "avatar_frames": ["season_auto_frame_legend"],
                }
            pass_rewards.append(
                {
                    "reward_level": level,
                    "reward_track": "premium",
                    "title": f"高级档 Lv.{level}",
                    "reward_points": 18 if level % 5 else 58,
                    "reward_xp": 0,
                    "reward_season_coin": 6 if level % 2 == 0 else 3,
                    "reward_payload": payload,
                }
            )
        return {
            "season": {
                "name": season_name,
                "theme": safe_theme,
                "description": description,
                "cover_url": "/static/img/cards/fogharbor.svg",
                "accent_color": "#4f6dff",
                "story_seed": story_seed,
                "story_outline": story_outline,
                "start_at": start_at,
                "end_at": end_at,
                "status": "active",
                "source": "ai_auto",
                "pass_price": max(0, int(pass_price)),
                "config": {
                    "duration_days": max(7, min(120, int(duration_days))),
                    "daily_card_count": 3,
                    "weekly_task_count": 3,
                    "daily_task_count": 3,
                    "refresh_hours": [0, 8, 18],
                    "manual_refresh_cost": 60,
                    "xp_per_level": 100,
                },
            },
            "announcements": announcements,
            "task_templates": task_templates,
            "card_pools": card_pools,
            "pass_rewards": pass_rewards,
            "titles": [
                {"title_id": "season_auto_title_legend", "name": f"{safe_theme}传奇", "color_code": "#ffd166", "rarity": "legendary"},
            ],
            "avatar_frames": [
                {"frame_id": "season_auto_frame_epic", "name": f"{safe_theme}回响框", "rarity": "epic"},
                {"frame_id": "season_auto_frame_legend", "name": f"{safe_theme}王冠框", "rarity": "legendary"},
            ],
        }

    def build_ai_daily_card_payload(season: dict[str, Any], *, day_key: str) -> dict[str, Any] | None:
        cfg = season.get("config", {})
        if not isinstance(cfg, dict) or not cfg.get("ai_daily_card_enabled"):
            return None
        theme = str(season.get("theme", "")).strip()[:80] or "赛季主题"
        season_name = str(season.get("name", "")).strip()[:80] or theme
        tone = str(cfg.get("ai_daily_card_tone", "")).strip() or "悬疑冒险"
        difficulty = str(cfg.get("ai_daily_card_difficulty", "")).strip() or "中等"
        prompt_extra = str(cfg.get("ai_daily_card_prompt_extra", "")).strip()[:400]
        cover_url = str(season.get("cover_url", "")).strip()[:720] or "/static/img/cards/fogharbor.svg"
        schema_hint = {
            "title": "string",
            "subtitle": "string",
            "summary": "string",
            "opening_text": "string",
            "starter_goals": ["string", "string", "string"],
            "metrics": [{"key": "string", "label": "string", "min": 0, "max": 100, "initial": 0, "color": "#RRGGBB"}],
            "status_tracks": [{"key": "string", "label": "string", "min": 0, "max": 100, "color": "#RRGGBB"}],
            "hero_status_label": "string",
            "hero_status_color": "#RRGGBB",
            "genre_tags": ["string"],
            "feature_badges": ["string"],
            "world_lore": "string",
            "core_conflict": "string",
            "npc_roster": ["string"],
            "scene_blueprint": ["string"],
            "ending_catalog": ["string"],
            "cover_url": "string",
            "room_background_url": "string",
        }
        messages = [
            {
                "role": "system",
                "content": (
                    "你是多人联机文字冒险的高级策划。"
                    "只输出一个合法 JSON 对象，不要输出 markdown 或解释。"
                    "内容使用中文，要求是完整可游玩的剧本卡。"
                ),
            },
            {
                "role": "user",
                "content": (
                    f"赛季名称：{season_name}\n"
                    f"赛季主题：{theme}\n"
                    f"日期：{day_key}\n"
                    f"赛季背景：{str(season.get('story_seed', '')).strip()}\n"
                    f"赛季大纲：{str(season.get('story_outline', '')).strip()}\n"
                    f"风格：{tone}\n难度：{difficulty}\n"
                    f"{'额外要求：' + prompt_extra if prompt_extra else ''}\n"
                    "输出必须包含以下字段：\n"
                    f"{json.dumps(schema_hint, ensure_ascii=False)}\n"
                    "约束：starter_goals 3 条；metrics 2-4 条；status_tracks 1-3 条；"
                    "genre_tags 2-6 条；feature_badges 2-6 条；"
                    "world_lore 与 core_conflict 必须是完整中文段落。"
                ),
            },
        ]
        ai_result = deepseek_client.chat_json(messages, temperature=0.7, max_tokens=1800)
        if not ai_result.get("ok"):
            return None
        draft = parse_json_object(ai_result.get("content", ""))
        if not isinstance(draft, dict):
            return None
        if not str(draft.get("cover_url", "")).strip():
            draft["cover_url"] = cover_url
        if not str(draft.get("room_background_url", "")).strip():
            draft["room_background_url"] = cover_url
        try:
            return build_card_config(draft)
        except ValueError:
            return None

    def get_or_create_season_daily_card(season_id: str) -> dict[str, Any]:
        season = persistence.get_season(season_id)
        if season is None:
            raise ValueError("season_not_found")
        day_key = _utc_day_key()
        existing = persistence.get_season_daily_game_card(season_id, day_key=day_key)
        if existing is not None:
            cfg = season.get("config", {})
            ai_enabled = bool(isinstance(cfg, dict) and cfg.get("ai_daily_card_enabled"))
            if ai_enabled:
                card_id = str(existing.get("card_id", "")).strip()
                card = persistence.get_game_card(card_id) if card_id else None
                legacy_mark = False
                if card:
                    legacy_mark = (
                        "赛季日卡" in str(card.get("title", ""))
                        or "页脚" in str(card.get("opening_text", ""))
                        or any("解锁一条资料" in str(g) for g in (card.get("starter_goals") or []))
                    )
                if legacy_mark:
                    ai_payload = build_ai_daily_card_payload(season, day_key=day_key)
                    if ai_payload:
                        try:
                            persistence.update_game_card(
                                card_id,
                                title=ai_payload["title"],
                                subtitle=ai_payload["subtitle"],
                                summary=ai_payload["summary"],
                                cover_url=ai_payload["cover_url"],
                                opening_text=ai_payload["opening_text"],
                                starter_goals=ai_payload["starter_goals"],
                                state_schema=ai_payload["state_schema"],
                                content_pack=ai_payload["content_pack"],
                            )
                        except ValueError:
                            pass
            return existing
        ai_payload = build_ai_daily_card_payload(season, day_key=day_key)
        if ai_payload:
            return persistence.ensure_season_daily_game_card(season_id, card_payload=ai_payload, source="ai")
        return persistence.ensure_season_daily_game_card(season_id)

    def ensure_active_season() -> dict[str, Any] | None:
        season_cfg = get_season_system_config()
        if not season_cfg.get("enabled", True):
            return None
        active = persistence.get_active_season()
        if active is not None:
            persistence.ensure_season_start_announcement(str(active.get("season_id", "")).strip())
            return active
        if not season_cfg.get("auto_create_enabled", True):
            return None
        theme_hint = "迷雾电波" if season_cfg.get("auto_generate_theme", True) else "标准赛季"
        blueprint = build_auto_season_blueprint(
            theme_hint=theme_hint,
            duration_days=int(season_cfg.get("season_duration_days", 30)),
            pass_price=int(season_cfg.get("premium_pass_price", 680)),
        )
        season_payload = blueprint.get("season", {}) if isinstance(blueprint, dict) else {}
        if not isinstance(season_payload, dict):
            return None
        try:
            created = persistence.create_season(
                name=str(season_payload.get("name", "自动赛季")),
                theme=str(season_payload.get("theme", "")),
                description=str(season_payload.get("description", "")),
                cover_url=str(season_payload.get("cover_url", "")),
                accent_color=str(season_payload.get("accent_color", "#4f6dff")),
                story_seed=str(season_payload.get("story_seed", "")),
                story_outline=str(season_payload.get("story_outline", "")),
                start_at=str(season_payload.get("start_at", "")),
                end_at=str(season_payload.get("end_at", "")),
                status=str(season_payload.get("status", "active")),
                source=str(season_payload.get("source", "ai_auto")),
                pass_price=int(season_payload.get("pass_price", 680) or 680),
                config=season_payload.get("config", {}) if isinstance(season_payload.get("config"), dict) else {},
                created_by="system",
            )
        except ValueError:
            return persistence.get_active_season()
        season_id = str(created.get("season_id", "")).strip()
        for item in blueprint.get("announcements", []):
            if not isinstance(item, dict):
                continue
            try:
                persistence.create_season_announcement(
                    season_id,
                    title=str(item.get("title", "赛季公告")),
                    content=str(item.get("content", "")),
                    created_by="system",
                )
            except ValueError:
                continue
        for item in blueprint.get("task_templates", []):
            if not isinstance(item, dict):
                continue
            try:
                persistence.upsert_season_task_template(season_id, item)
            except ValueError:
                continue
        for item in blueprint.get("card_pools", []):
            if not isinstance(item, dict):
                continue
            try:
                persistence.upsert_season_card_pool(season_id, item)
            except ValueError:
                continue
        for item in blueprint.get("pass_rewards", []):
            if not isinstance(item, dict):
                continue
            try:
                persistence.upsert_season_pass_reward(season_id, item)
            except ValueError:
                continue
        for idx, item in enumerate(blueprint.get("titles", [])):
            if not isinstance(item, dict):
                continue
            try:
                persistence.upsert_season_task_template(
                    season_id,
                    {
                        "task_type": "achievement",
                        "title": str(item.get("name", "赛季称号")),
                        "description": "隐藏成就占位模板，可在后台继续扩展。",
                        "event_key": "complete_game",
                        "target_value": 999999,
                        "reward_payload": {"titles": [str(item.get("title_id", ""))]},
                        "is_active": False,
                        "sort_order": 900 + idx,
                    },
                )
            except ValueError:
                continue
        return persistence.get_active_season()

    def build_active_season_payload(user_id: str) -> dict[str, Any]:
        active = ensure_active_season()
        rest = persistence.get_season_by_time_status("rest")
        season = active or rest
        if season is None:
            raise ValueError("season_not_found")
        season_id = str(season.get("season_id", "")).strip()
        profile = persistence.get_or_create_user_season_profile(user_id, season_id)
        persistence.ensure_user_season_tasks(user_id, season_id)
        cards_bundle = persistence.ensure_user_season_cards(user_id, season_id) if str(season.get("time_status")) == "running" else {"cards": [], "refresh_key": "", "next_refresh_at": "", "generated": False}
        announcements = persistence.list_season_announcements(season_id, limit=10)
        tasks = persistence.list_user_season_tasks(user_id, season_id, limit=1200)
        rewards = persistence.list_season_pass_rewards(season_id, limit=1200)
        claimed_reward_ids = persistence.list_user_season_reward_claims(user_id, season_id)
        leaderboard = persistence.get_season_leaderboard(season_id, limit=20)
        decorations = persistence.list_user_decorations(user_id, limit=400)
        equipped = persistence.get_user_equipped_decorations(user_id)
        daily_game_card = get_or_create_season_daily_card(season_id) if str(season.get("time_status")) in {"running", "rest"} else {}
        intel = persistence.list_user_season_intel(user_id, season_id, limit=400)
        next_refresh_at = str(cards_bundle.get("next_refresh_at", "")).strip()
        if not next_refresh_at and cards_bundle.get("cards"):
            first_card = cards_bundle.get("cards", [])[0] if isinstance(cards_bundle.get("cards"), list) else {}
            if isinstance(first_card, dict):
                next_refresh_at = str(first_card.get("expires_at", "")).strip()
        return {
            "season": season,
            "active_season": active,
            "rest_season": rest,
            "profile": profile,
            "announcements": announcements,
            "tasks": tasks,
            "cards": cards_bundle.get("cards", []),
            "refresh_key": cards_bundle.get("refresh_key", ""),
            "next_refresh_at": next_refresh_at,
            "daily_game_card": daily_game_card,
            "intel": intel,
            "rewards": rewards,
            "claimed_reward_ids": sorted(claimed_reward_ids),
            "leaderboard": leaderboard,
            "decorations": decorations,
            "equipped_decorations": equipped,
            "system_config": get_season_system_config(),
        }

    def pick_unique_username(base_username: str, provider_user_id: str) -> str:
        base = normalize_account_username(base_username) or "linuxdo_user"
        if persistence.get_user_by_username(base) is None:
            return base
        suffix_seed = re.sub(r"[^a-z0-9]+", "", str(provider_user_id or "").lower())[-6:] or "id"
        for seq in range(1, 80):
            suffix = f"_{suffix_seed}{seq}" if seq > 1 else f"_{suffix_seed}"
            candidate = (base[: max(3, 20 - len(suffix))] + suffix)[:20]
            candidate = normalize_account_username(candidate)
            if not candidate:
                continue
            if persistence.get_user_by_username(candidate) is None:
                return candidate
        return f"linuxdo_{secrets.token_hex(3)}"[:20]

    def require_admin_json() -> tuple[dict[str, Any] | None, Any | None]:
        user = current_user()
        if user is None:
            return None, (jsonify({"ok": False, "error": readable_error("auth_required")}), 401)
        if str(user.get("role", "player")) != "admin":
            return None, (jsonify({"ok": False, "error": readable_error("admin_required")}), 403)
        return user, None

    @app.get("/")
    def index() -> str:
        return render_template("index.html", user=serialize_user(current_user()))

    @app.get("/rooms")
    def rooms_page() -> str:
        return render_template("rooms.html", user=serialize_user(current_user()))

    @app.get("/leaderboard")
    def leaderboard_page() -> str:
        return render_template("leaderboard.html", user=serialize_user(current_user()))

    @app.get("/wallet")
    def wallet_page() -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        return render_template("wallet.html", user=serialize_user(user))

    @app.get("/store")
    def store_page() -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        return render_template("store.html", user=serialize_user(user))

    @app.get("/season")
    def season_page() -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        return render_template("season.html", user=serialize_user(user))

    @app.get("/season/chronicle")
    def season_chronicle_page() -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        return render_template("season_chronicle.html", user=serialize_user(user))

    @app.get("/season/<season_id>/chronicle")
    def season_chronicle_view_page(season_id: str) -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        safe_season_id = str(season_id or "").strip()[:80]
        if not safe_season_id:
            return redirect(url_for("season_page"))
        return render_template(
            "season_chronicle.html",
            user=serialize_user(user),
        )

    @app.get("/season/<season_id>/cards")
    def season_cards_page(season_id: str) -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        safe_season_id = str(season_id or "").strip()[:80]
        if not safe_season_id:
            return redirect(url_for("season_page"))
        return render_template(
            "season_cards.html",
            user=serialize_user(user),
            season_id=safe_season_id,
        )

    @app.get("/admin/seasons")
    def season_admin_page() -> Any:
        user = current_user()
        if user is None or str(user.get("role", "player")) != "admin":
            return redirect(url_for("index"))
        return render_template("season_admin.html", user=serialize_user(user))

    @app.get("/users/<username>")
    def user_profile_page(username: str) -> str:
        safe_username = normalize_account_username(username)
        return render_template(
            "user_profile.html",
            user=serialize_user(current_user()),
            profile_username=safe_username or "",
        )

    @app.get("/users/<username>/achievements")
    def user_achievements_page(username: str) -> str:
        safe_username = normalize_account_username(username)
        return render_template(
            "achievements.html",
            user=serialize_user(current_user()),
            profile_username=safe_username or "",
        )

    @app.get("/plaza")
    def plaza_page() -> str:
        return render_template("plaza.html", user=serialize_user(current_user()))

    @app.get("/cards/new")
    def card_new_page() -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        return render_template("card_editor.html", user=serialize_user(user), card_id="")

    @app.get("/cards/<card_id>/edit")
    def card_edit_page(card_id: str) -> Any:
        user = current_user()
        if user is None:
            return redirect(url_for("login_page"))
        card = persistence.get_game_card(card_id.strip())
        if card is None:
            return redirect(url_for("plaza_page"))
        is_admin = str(user.get("role", "player")) == "admin"
        is_creator = str(card.get("creator_user_id") or "") == str(user["user_id"])
        if not is_admin and not is_creator:
            return redirect(url_for("plaza_page"))
        return render_template("card_editor.html", user=serialize_user(user), card_id=card["card_id"])

    @app.get("/login")
    def login_page() -> str:
        oauth_error = str(request.args.get("oauth_error", "")).strip()[:120]
        return render_template("login.html", user=serialize_user(current_user()), oauth_error=oauth_error)

    @app.get("/register")
    def register_page() -> str:
        return render_template("register.html", user=serialize_user(current_user()))

    @app.get("/admin")
    def admin_page() -> Any:
        user = current_user()
        if user is None or str(user.get("role", "player")) != "admin":
            return redirect(url_for("index"))
        return render_template(
            "admin.html",
            user=serialize_user(user),
            admin_section=normalize_admin_section(request.args.get("section", "users")),
        )

    @app.get("/admin/<section>")
    def admin_section_page(section: str) -> Any:
        user = current_user()
        if user is None or str(user.get("role", "player")) != "admin":
            return redirect(url_for("index"))
        return render_template(
            "admin.html",
            user=serialize_user(user),
            admin_section=normalize_admin_section(section),
        )

    @app.post("/api/auth/register")
    def auth_register_api() -> Any:
        payload = request.get_json(silent=True) or {}
        username = str(payload.get("username", "")).strip().lower()
        password = str(payload.get("password", ""))
        display_name = username

        if not re.match(r"^[a-zA-Z0-9_]{3,20}$", username):
            return jsonify({"ok": False, "error": readable_error("invalid_username")}), 400
        if len(password) < 6:
            return jsonify({"ok": False, "error": readable_error("weak_password")}), 400

        try:
            user = persistence.create_user(
                username=username,
                password_hash=generate_password_hash(password, method="pbkdf2:sha256"),
                display_name=display_name[:24] or username,
            )
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

        session["user_id"] = user["user_id"]
        persistence.touch_user_activity(user["user_id"], is_online=True)
        return jsonify({"ok": True, "user": serialize_user(user)})

    @app.post("/api/auth/login")
    def auth_login_api() -> Any:
        payload = request.get_json(silent=True) or {}
        username = str(payload.get("username", "")).strip().lower()
        password = str(payload.get("password", ""))

        user = persistence.get_user_by_username(username)
        if user is None or not check_password_hash(user["password_hash"], password):
            return jsonify({"ok": False, "error": readable_error("bad_credentials")}), 400

        session["user_id"] = user["user_id"]
        persistence.touch_user_activity(user["user_id"], is_online=True)
        return jsonify({"ok": True, "user": serialize_user(user)})

    @app.post("/api/auth/logout")
    def auth_logout_api() -> Any:
        user_id = str(session.get("user_id", "")).strip()
        if user_id:
            persistence.touch_user_activity(user_id, is_online=False)
        session.pop("user_id", None)
        return jsonify({"ok": True})

    @app.get("/api/auth/oauth/linuxdo/start")
    def auth_oauth_linuxdo_start_api() -> Any:
        cfg = get_linuxdo_oauth_config()
        if not cfg.get("enabled"):
            return jsonify({"ok": False, "error": "Linux DO OAuth 未启用。"}), 400
        required_fields = ["client_id", "client_secret", "authorize_url", "token_url", "userinfo_url"]
        missing = [field for field in required_fields if not str(cfg.get(field, "")).strip()]
        if missing:
            return jsonify({"ok": False, "error": f"Linux DO OAuth 配置缺失：{', '.join(missing)}"}), 400
        for url_field in ("authorize_url", "token_url", "userinfo_url"):
            if not _is_valid_http_url(cfg.get(url_field, "")):
                return jsonify({"ok": False, "error": f"Linux DO OAuth 配置无效：{url_field}"}), 400

        state = secrets.token_urlsafe(24)
        session["linuxdo_oauth_state"] = state
        callback_url = url_for("auth_oauth_linuxdo_callback_api", _external=True)
        query = {
            "client_id": cfg["client_id"],
            "response_type": "code",
            "redirect_uri": callback_url,
            "state": state,
        }
        scope = str(cfg.get("scope", "")).strip()
        if scope:
            query["scope"] = scope
        authorize_url = f"{cfg['authorize_url']}?{urlencode(query)}"
        return redirect(authorize_url)

    @app.get("/api/auth/oauth/linuxdo/callback")
    def auth_oauth_linuxdo_callback_api() -> Any:
        cfg = get_linuxdo_oauth_config()
        if not cfg.get("enabled"):
            return oauth_error_redirect("Linux DO OAuth 未启用")

        state = str(request.args.get("state", "")).strip()
        code = str(request.args.get("code", "")).strip()
        expected_state = str(session.pop("linuxdo_oauth_state", "")).strip()
        if not code:
            return oauth_error_redirect("缺少授权 code")
        if not state or state != expected_state:
            return oauth_error_redirect("state 校验失败")

        callback_url = url_for("auth_oauth_linuxdo_callback_api", _external=True)
        try:
            token_resp = requests.post(
                cfg["token_url"],
                data={
                    "grant_type": "authorization_code",
                    "code": code,
                    "client_id": cfg["client_id"],
                    "client_secret": cfg["client_secret"],
                    "redirect_uri": callback_url,
                },
                timeout=12,
            )
            token_resp.raise_for_status()
            token_payload = token_resp.json()
        except Exception as exc:
            return oauth_error_redirect(f"获取 access_token 失败: {exc}")

        access_token = str(token_payload.get("access_token", "")).strip()
        if not access_token:
            return oauth_error_redirect("access_token 为空")

        try:
            userinfo_resp = requests.get(
                cfg["userinfo_url"],
                headers={"Authorization": f"Bearer {access_token}"},
                timeout=12,
            )
            userinfo_resp.raise_for_status()
            userinfo = userinfo_resp.json()
            if not isinstance(userinfo, dict):
                return oauth_error_redirect("userinfo 响应格式无效")
        except Exception as exc:
            return oauth_error_redirect(f"获取用户信息失败: {exc}")

        try:
            username, display_name = resolve_linuxdo_identity(userinfo)
        except ValueError:
            return oauth_error_redirect("用户信息缺少 id")

        nested_user = userinfo.get("user")
        user_payload = nested_user if isinstance(nested_user, dict) else {}
        provider_user_id = str(userinfo.get("id") or user_payload.get("id") or "").strip()
        if not provider_user_id:
            return oauth_error_redirect("用户信息缺少 id")

        user = None
        identity = persistence.get_oauth_identity("linuxdo", provider_user_id)
        if identity:
            user = persistence.get_user_by_id(identity.get("user_id"))

        if user is None:
            final_username = pick_unique_username(username, provider_user_id)
            random_password = secrets.token_urlsafe(32)
            try:
                user = persistence.create_user(
                    username=final_username,
                    password_hash=generate_password_hash(random_password, method="pbkdf2:sha256"),
                    display_name=display_name,
                )
            except ValueError as exc:
                return oauth_error_redirect(f"创建本地账号失败: {readable_error(str(exc))}")
            try:
                persistence.bind_oauth_identity(
                    "linuxdo",
                    provider_user_id,
                    user["user_id"],
                    username_snapshot=username,
                )
            except ValueError as exc:
                return oauth_error_redirect(f"绑定 OAuth 身份失败: {readable_error(str(exc))}")

        session["user_id"] = user["user_id"]
        persistence.touch_user_activity(user["user_id"], is_online=True)
        return redirect(url_for("index"))

    @app.get("/api/me")
    def me_api() -> Any:
        user = current_user()
        badges = build_user_badges_payload(user) if user else {"active_badge": "", "badges": [], "unlocked_ids": []}
        points = build_user_points_payload(user["user_id"]) if user else {"account": None, "new_rewards": []}
        return jsonify(
            {"ok": True, "logged_in": bool(user), "user": serialize_user(user), "badges": badges, "points": points}
        )

    @app.get("/api/me/badges")
    def me_badges_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        return jsonify({"ok": True, "badges": build_user_badges_payload(user)})

    @app.patch("/api/me/profile")
    def me_profile_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        payload = request.get_json(silent=True) or {}
        avatar_color = payload.get("avatar_color")
        avatar_url = payload.get("avatar_url")
        theme = payload.get("theme")
        bio = payload.get("bio")
        active_badge = payload.get("active_badge")

        if avatar_color is not None:
            avatar_color = str(avatar_color).strip()
            if not re.match(r"^#[0-9a-fA-F]{6}$", avatar_color):
                avatar_color = user.get("avatar_color", "#0070f2")
        if avatar_url is not None:
            avatar_url = str(avatar_url).strip()[:280]
            if avatar_url and not (
                avatar_url.startswith("/")
                or avatar_url.startswith("http://")
                or avatar_url.startswith("https://")
            ):
                avatar_url = ""
        if bio is not None:
            bio = str(bio).strip()[:160]
        if theme is not None:
            theme = normalize_theme(theme, normalize_theme(user.get("theme", "minimal")))
        if active_badge is not None:
            badge_definitions = get_role_badges_config()
            badge_id = normalize_badge_id(active_badge, badge_definitions)
            if badge_id:
                progress = build_badge_progress(
                    persistence.get_user_game_stats(user["user_id"]),
                    badge_id,
                    badge_definitions,
                )
                if badge_id not in set(progress.get("unlocked_ids", [])):
                    return jsonify({"ok": False, "error": readable_error("badge_locked")}), 400
                active_badge = badge_id
            else:
                active_badge = ""

        try:
            updated = persistence.update_user_profile(
                user["user_id"],
                avatar_color=avatar_color,
                avatar_url=avatar_url,
                theme=theme,
                bio=bio,
                active_badge=active_badge,
            )
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

        return jsonify({"ok": True, "user": serialize_user(updated), "badges": build_user_badges_payload(updated)})

    @app.get("/api/me/points")
    def me_points_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        payload = build_user_points_payload(user["user_id"])
        return jsonify({"ok": True, "points": payload})

    @app.get("/api/points/transactions")
    def points_transactions_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        limit_raw = request.args.get("limit", 80)
        tx_type = str(request.args.get("tx_type", "")).strip()
        try:
            limit = max(1, min(400, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 80
        rows = persistence.list_user_point_transactions(user["user_id"], limit=limit, tx_type=tx_type)
        return jsonify({"ok": True, "transactions": rows})

    @app.post("/api/points/redeem")
    def points_redeem_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        payload = request.get_json(silent=True) or {}
        code = str(payload.get("code", "")).strip().upper()
        if not code:
            return jsonify({"ok": False, "error": readable_error("activation_code_invalid")}), 400
        points_cfg = get_points_system_config()
        ip_address = str(request.headers.get("X-Forwarded-For", "") or request.remote_addr or "").split(",")[0].strip()
        try:
            result = persistence.redeem_activation_code(
                user["user_id"],
                code,
                ip_address=ip_address,
                ip_limit=int(points_cfg.get("redeem_ip_limit", 12)),
                user_limit=int(points_cfg.get("redeem_user_limit", 10)),
                window_seconds=60,
            )
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/me/entitlements")
    def me_entitlements_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        entitlement_type = str(request.args.get("type", "")).strip().lower()
        rows = persistence.list_user_entitlements(user["user_id"], entitlement_type=entitlement_type, status="active", limit=400)
        return jsonify({"ok": True, "entitlements": rows})

    @app.get("/api/seasons/active")
    def active_season_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        try:
            payload = build_active_season_payload(user["user_id"])
            return jsonify({"ok": True, **payload})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 404

    @app.get("/api/seasons/history")
    def season_history_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        rows = persistence.list_seasons(limit=40)
        return jsonify({"ok": True, "seasons": rows})

    @app.get("/api/seasons/<season_id>/cards")
    def season_cards_api(season_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        safe_season_id = str(season_id or "").strip()[:80]
        if not safe_season_id:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 400
        season = persistence.get_season(safe_season_id)
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        cards = persistence.list_game_cards(world_id=f"season:{safe_season_id}")
        return jsonify({"ok": True, "season": season, "season_id": safe_season_id, "cards": cards})

    @app.get("/api/seasons/active/daily-card")
    def active_season_daily_game_card_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season() or persistence.get_season_by_time_status("rest")
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        try:
            daily = get_or_create_season_daily_card(season_id)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "season_id": season_id, "daily_game_card": daily})

    @app.get("/api/seasons/active/chronicle")
    def active_season_chronicle_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season() or persistence.get_season_by_time_status("rest")
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        try:
            items = persistence.build_season_chronicle_items(user["user_id"], season_id, limit_days=90)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "season": season, "season_id": season_id, "items": items})

    @app.get("/api/seasons/<season_id>/chronicle")
    def season_chronicle_api(season_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        safe_season_id = str(season_id or "").strip()[:80]
        if not safe_season_id:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 400
        season = persistence.get_season(safe_season_id)
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        try:
            items = persistence.build_season_chronicle_items(user["user_id"], safe_season_id, limit_days=180)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "season": season, "season_id": safe_season_id, "items": items})

    @app.post("/api/seasons/active/intel/unlock")
    def active_season_intel_unlock_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season() or persistence.get_season_by_time_status("rest")
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        payload = request.get_json(silent=True) or {}
        day_key = str(payload.get("day_key", "")).strip()
        intel_key = str(payload.get("intel_key", "")).strip()
        if not day_key:
            return jsonify({"ok": False, "error": readable_error("invalid_card_payload")}), 400
        try:
            unlocked = persistence.unlock_season_intel(user["user_id"], season_id, day_key=day_key, intel_key=intel_key)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "season_id": season_id, "unlocked": unlocked})

    @app.get("/api/seasons/active/intel")
    def active_season_intel_list_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season() or persistence.get_season_by_time_status("rest")
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        rows = persistence.list_user_season_intel(user["user_id"], season_id, limit=800)
        return jsonify({"ok": True, "season_id": season_id, "intel": rows})

    @app.post("/api/seasons/active/pass/unlock")
    def unlock_active_season_pass_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        try:
            profile = persistence.unlock_user_season_pass(user["user_id"], season_id)
            return jsonify({"ok": True, "profile": profile, "season_id": season_id})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/seasons/active/tasks")
    def active_season_tasks_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        task_type = str(request.args.get("type", "")).strip().lower()
        persistence.ensure_user_season_tasks(user["user_id"], season_id)
        tasks = persistence.list_user_season_tasks(
            user["user_id"],
            season_id,
            task_type=task_type,
            include_claimed=True,
            limit=1200,
        )
        return jsonify({"ok": True, "season_id": season_id, "tasks": tasks})

    @app.post("/api/seasons/active/tasks/progress")
    def active_season_tasks_progress_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        payload = request.get_json(silent=True) or {}
        event_key = str(payload.get("event_key", "")).strip().lower()
        amount_raw = payload.get("amount", 1)
        context = payload.get("context", {})
        try:
            amount = max(1, int(amount_raw))
        except (TypeError, ValueError):
            amount = 1
        if not event_key:
            return jsonify({"ok": False, "error": readable_error("invalid_card_payload")}), 400
        task_result = persistence.progress_user_season_tasks(
            user["user_id"],
            season_id,
            event_key=event_key,
            amount=amount,
            context=context if isinstance(context, dict) else {},
        )
        card_result = persistence.progress_user_season_cards(
            user["user_id"],
            season_id,
            event_key=event_key,
            amount=amount,
        )
        return jsonify({"ok": True, "season_id": season_id, "task_result": task_result, "card_result": card_result})

    @app.post("/api/seasons/active/tasks/<int:task_id>/claim")
    def active_season_task_claim_api(task_id: int) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        try:
            result = persistence.claim_user_season_task(user["user_id"], season_id, task_id)
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.post("/api/seasons/active/tasks/claim-all")
    def active_season_tasks_claim_all_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        result = persistence.claim_all_user_season_tasks(user["user_id"], season_id)
        return jsonify({"ok": True, "result": result})

    @app.get("/api/seasons/active/cards")
    def active_season_cards_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        bundle = persistence.ensure_user_season_cards(user["user_id"], season_id)
        cards = persistence.list_user_season_cards(
            user["user_id"],
            season_id,
            refresh_key=str(bundle.get("refresh_key", "")),
            include_expired=False,
            limit=200,
        )
        return jsonify(
            {
                "ok": True,
                "season_id": season_id,
                "refresh_key": bundle.get("refresh_key", ""),
                "next_refresh_at": bundle.get("next_refresh_at", ""),
                "generated": bool(bundle.get("generated", False)),
                "cards": cards,
            }
        )

    @app.post("/api/seasons/active/cards/<int:card_id>/claim")
    def active_season_card_claim_api(card_id: int) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        try:
            result = persistence.claim_user_season_card(user["user_id"], season_id, card_id)
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.post("/api/seasons/active/cards/<int:card_id>/refresh")
    def active_season_card_refresh_api(card_id: int) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        payload = request.get_json(silent=True) or {}
        req_cost_raw = payload.get("cost")
        season_cfg = get_season_system_config()
        refresh_cost = int(season_cfg.get("manual_refresh_cost", 60))
        if req_cost_raw is not None:
            try:
                refresh_cost = max(0, int(req_cost_raw))
            except (TypeError, ValueError):
                refresh_cost = int(season_cfg.get("manual_refresh_cost", 60))
        try:
            result = persistence.refresh_user_season_card(
                user["user_id"],
                season_id,
                card_id,
                refresh_cost=refresh_cost,
            )
            return jsonify({"ok": True, "result": result, "refresh_cost": refresh_cost})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/seasons/active/rewards")
    def active_season_rewards_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        profile = persistence.get_or_create_user_season_profile(user["user_id"], season_id)
        rewards = persistence.list_season_pass_rewards(season_id, limit=1600)
        claimed = sorted(persistence.list_user_season_reward_claims(user["user_id"], season_id))
        return jsonify({"ok": True, "season_id": season_id, "profile": profile, "rewards": rewards, "claimed_reward_ids": claimed})

    @app.post("/api/seasons/active/rewards/<reward_id>/claim")
    def active_season_reward_claim_api(reward_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        season = ensure_active_season()
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        season_id = str(season.get("season_id", "")).strip()
        try:
            result = persistence.claim_season_pass_reward(user["user_id"], season_id, reward_id.strip())
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/seasons/decorations")
    def season_decorations_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        rows = persistence.list_user_decorations(user["user_id"], limit=600)
        equipped = persistence.get_user_equipped_decorations(user["user_id"])
        return jsonify({"ok": True, "decorations": rows, "equipped": equipped})

    @app.post("/api/seasons/decorations/equip")
    def season_decorations_equip_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        payload = request.get_json(silent=True) or {}
        decoration_type = str(payload.get("decoration_type", "")).strip().lower()
        decoration_id = str(payload.get("decoration_id", "")).strip()
        if decoration_type not in {"title", "avatar_frame"} or not decoration_id:
            return jsonify({"ok": False, "error": readable_error("invalid_card_payload")}), 400
        try:
            equipped = persistence.equip_user_decoration(user["user_id"], decoration_type, decoration_id)
            return jsonify({"ok": True, "equipped": equipped})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/store/items")
    def store_items_api() -> Any:
        user = current_user()
        status = "active"
        item_type = str(request.args.get("type", "")).strip().lower()
        keyword = str(request.args.get("q", "")).strip()
        limit_raw = request.args.get("limit", 120)
        try:
            limit = max(1, min(800, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 120
        items = persistence.list_shop_items(status=status, item_type=item_type, keyword=keyword, limit=limit)
        entitlements = persistence.list_user_entitlements(user["user_id"], status="active", limit=1200) if user else []
        owned_ref_set = {str(row.get("ref_id", "")).strip() for row in entitlements if str(row.get("ref_id", "")).strip()}
        owned_type_set = {
            (str(row.get("entitlement_type", "")).strip(), str(row.get("ref_id", "")).strip())
            for row in entitlements
        }
        result = []
        for row in items:
            item = dict(row)
            meta = item.get("metadata", {}) if isinstance(item.get("metadata"), dict) else {}
            ref_id = str(meta.get("ref_id") or meta.get("card_id") or meta.get("module_id") or item.get("item_id", "")).strip()
            ent_type = str(item.get("item_type", "")).strip().lower()
            owner_key = {
                "game_card": "game_card",
                "card": "game_card",
                "module": "module",
                "decoration": "decoration",
                "privilege": "privilege",
                "bundle": "bundle",
            }.get(ent_type, ent_type)
            item["owned"] = bool((owner_key, ref_id) in owned_type_set or ref_id in owned_ref_set)
            result.append(item)
        return jsonify({"ok": True, "items": result})

    @app.get("/api/store/items/<item_id>")
    def store_item_detail_api(item_id: str) -> Any:
        user = current_user()
        item = persistence.get_shop_item(item_id)
        if item is None or str(item.get("status", "")).strip().lower() != "active":
            return jsonify({"ok": False, "error": readable_error("shop_item_not_found")}), 404
        entitlements = persistence.list_user_entitlements(user["user_id"], status="active", limit=1200) if user else []
        meta = item.get("metadata", {}) if isinstance(item.get("metadata"), dict) else {}
        ref_id = str(meta.get("ref_id") or meta.get("card_id") or meta.get("module_id") or item.get("item_id", "")).strip()
        ent_type = str(item.get("item_type", "")).strip().lower()
        owner_key = {
            "game_card": "game_card",
            "card": "game_card",
            "module": "module",
            "decoration": "decoration",
            "privilege": "privilege",
            "bundle": "bundle",
        }.get(ent_type, ent_type)
        item["owned"] = bool(
            any(
                str(row.get("entitlement_type", "")).strip() == owner_key
                and str(row.get("ref_id", "")).strip() == ref_id
                and str(row.get("status", "")).strip() == "active"
                for row in entitlements
            )
        )
        return jsonify({"ok": True, "item": item})

    @app.post("/api/store/purchase")
    def store_purchase_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        payload = request.get_json(silent=True) or {}
        item_id = str(payload.get("item_id", "")).strip()
        qty_raw = payload.get("quantity", 1)
        try:
            quantity = max(1, min(99, int(qty_raw)))
        except (TypeError, ValueError):
            quantity = 1
        note = str(payload.get("note", "")).strip()[:180]
        try:
            result = persistence.purchase_shop_item(user["user_id"], item_id, quantity=quantity, note=note)
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/store/orders")
    def store_orders_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        limit_raw = request.args.get("limit", 80)
        try:
            limit = max(1, min(500, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 80
        rows = persistence.list_user_shop_orders(user["user_id"], limit=limit)
        return jsonify({"ok": True, "orders": rows})

    @app.post("/api/store/tip")
    def store_tip_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        payload = request.get_json(silent=True) or {}
        to_user_id = str(payload.get("to_user_id", "")).strip()
        room_id = str(payload.get("room_id", "")).strip().upper()
        message = str(payload.get("message", "")).strip()[:180]
        amount_raw = payload.get("amount", 0)
        try:
            amount = max(1, min(100000, int(amount_raw)))
        except (TypeError, ValueError):
            amount = 0
        if amount <= 0:
            return jsonify({"ok": False, "error": readable_error("invalid_points_amount")}), 400
        try:
            result = persistence.tip_user_points(
                user["user_id"],
                to_user_id,
                amount,
                room_id=room_id,
                message=message,
            )
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    def build_card_config(payload: dict[str, Any]) -> dict[str, Any]:
        def parse_bool(value: Any, default: bool) -> bool:
            if isinstance(value, bool):
                return value
            text = str(value or "").strip().lower()
            if text in {"1", "true", "yes", "y", "on"}:
                return True
            if text in {"0", "false", "no", "n", "off"}:
                return False
            return default

        title = str(payload.get("title", "")).strip()[:84]
        subtitle = str(payload.get("subtitle", "")).strip()[:84]
        summary = str(payload.get("summary", "")).strip()[:420]
        cover_url = str(payload.get("cover_url", "")).strip()[:720] or "/static/img/cards/fogharbor.svg"
        opening_text = str(payload.get("opening_text", "")).strip()[:660]
        if not title or not summary or not opening_text:
            raise ValueError("invalid_card_payload")

        starter_goals = parse_text_list(payload.get("starter_goals", []), limit=8)
        if not starter_goals:
            starter_goals = ["先确认队伍分工", "与 NPC 交谈获取线索", "推进第一轮行动"]

        raw_metrics = payload.get("metrics", [])
        metric_rows = raw_metrics if isinstance(raw_metrics, list) else parse_json_list(raw_metrics)
        metrics: list[dict[str, Any]] = []
        if isinstance(metric_rows, list):
            for idx, row in enumerate(metric_rows[:8]):
                if not isinstance(row, dict):
                    continue
                key = str(row.get("key", "")).strip()[:96]
                label = str(row.get("label", "")).strip()[:72]
                if not key or not label:
                    continue
                min_val = row.get("min", 0)
                max_val = row.get("max", 10)
                init_val = row.get("initial", 0)
                try:
                    min_num = int(min_val)
                    max_num = int(max_val)
                    init_num = int(init_val)
                except (TypeError, ValueError):
                    min_num, max_num, init_num = 0, 10, 0
                if max_num < min_num:
                    max_num = min_num
                init_num = max(min_num, min(max_num, init_num))
                color = str(row.get("color", "")).strip()
                if not re.match(r"^#[0-9a-fA-F]{6}$", color):
                    color = ""
                metrics.append(
                    {
                        "key": key,
                        "label": label,
                        "min": min_num,
                        "max": max_num,
                        "initial": init_num,
                        "color": color,
                    }
                )

        if not metrics:
            state_fields = parse_text_list(payload.get("state_fields", []), limit=4)
            for idx, label in enumerate(state_fields):
                safe_key = re.sub(r"[^a-z0-9_]+", "_", label.lower())[:20] or f"metric_{idx+1}"
                metrics.append({"key": safe_key, "label": label, "min": 0, "max": 100, "initial": 0, "color": ""})
        if not metrics:
            metrics = [
                {"key": "tension", "label": "局势紧张度", "min": 0, "max": 100, "initial": 15, "color": "#ff9f43"},
                {"key": "progress", "label": "推进度", "min": 0, "max": 100, "initial": 0, "color": "#3b82f6"},
            ]

        metric_by_key = {row["key"]: row for row in metrics}
        hero_status_label = str(payload.get("hero_status_label", "")).strip()
        hero_status_key = str(payload.get("hero_status_key", "")).strip()
        hero_status_color = str(payload.get("hero_status_color", "")).strip()
        if not re.match(r"^#[0-9a-fA-F]{6}$", hero_status_color):
            hero_status_color = ""

        selected_metric = metric_by_key.get(hero_status_key) if hero_status_key else None
        if selected_metric is None and hero_status_label:
            selected_metric = next((row for row in metrics if row["label"] == hero_status_label), None)
        if selected_metric is None:
            selected_metric = metrics[0]

        raw_tracks = payload.get("status_tracks", [])
        track_rows = raw_tracks if isinstance(raw_tracks, list) else parse_json_list(raw_tracks)
        status_tracks: list[dict[str, Any]] = []
        for row in track_rows[:6]:
            if isinstance(row, str):
                metric = metric_by_key.get(row.strip())
                if metric:
                    status_tracks.append(
                        {
                            "key": metric["key"],
                            "label": metric["label"],
                            "min": metric["min"],
                            "max": metric["max"],
                            "color": metric.get("color") or "",
                        }
                    )
            elif isinstance(row, dict):
                key = str(row.get("key", "")).strip()
                if not key or key not in metric_by_key:
                    continue
                metric = metric_by_key[key]
                color = str(row.get("color", "")).strip()
                if not re.match(r"^#[0-9a-fA-F]{6}$", color):
                    color = metric.get("color", "")
                raw_min = row.get("min", metric["min"])
                raw_max = row.get("max", metric["max"])
                try:
                    track_min = int(raw_min)
                    track_max = int(raw_max)
                except (TypeError, ValueError):
                    track_min = int(metric["min"])
                    track_max = int(metric["max"])
                if track_max < track_min:
                    track_max = track_min
                status_tracks.append(
                    {
                        "key": key,
                    "label": str(row.get("label", metric["label"])).strip()[:72] or metric["label"],
                        "min": track_min,
                        "max": track_max,
                        "color": color,
                    }
                )
        if not status_tracks:
            status_tracks = [
                {
                    "key": selected_metric["key"],
                    "label": hero_status_label or selected_metric["label"],
                    "min": selected_metric["min"],
                    "max": selected_metric["max"],
                    "color": hero_status_color or selected_metric.get("color", ""),
                }
            ]

        labels = {
            "inventory": str(payload.get("inventory_label", "")).strip() or "我的道具",
            "clues": str(payload.get("clues_label", "")).strip() or "关键信息",
            "choices": str(payload.get("choices_label", "")).strip() or "可执行动作",
            "scene": str(payload.get("scene_label", "")).strip() or "当前场景",
        }

        scope_raw = payload.get("data_scope", {})
        if not isinstance(scope_raw, dict) or not scope_raw:
            scope_json_raw = payload.get("data_scope_json", {})
            scope_raw = scope_json_raw if isinstance(scope_json_raw, dict) else (parse_json_object(scope_json_raw) or {})
        allowed_scope = {"shared", "per_player"}
        data_scope = {
            "metrics": str(scope_raw.get("metrics", "shared")).strip().lower(),
            "inventory": str(scope_raw.get("inventory", "per_player")).strip().lower(),
            "clues": str(scope_raw.get("clues", "shared")).strip().lower(),
        }
        for key in list(data_scope.keys()):
            if data_scope[key] not in allowed_scope:
                data_scope[key] = "shared" if key != "inventory" else "per_player"

        visibility_raw = payload.get("visibility", {})
        if not isinstance(visibility_raw, dict) or not visibility_raw:
            visibility_json_raw = payload.get("visibility_json", {})
            visibility_raw = visibility_json_raw if isinstance(visibility_json_raw, dict) else (parse_json_object(visibility_json_raw) or {})
        visibility = {
            "show_all_player_metrics": parse_bool(visibility_raw.get("show_all_player_metrics", True), True),
            "show_all_player_inventory": parse_bool(visibility_raw.get("show_all_player_inventory", True), True),
        }

        state_schema = {
            "metrics": metrics,
            "labels": labels,
            "data_scope": data_scope,
            "visibility": visibility,
            "ui": {
                "room_background_url": str(payload.get("room_background_url", "")).strip()[:720] or cover_url,
                "hero_status": {
                    "key": selected_metric["key"],
                    "label": hero_status_label or selected_metric["label"],
                    "min": selected_metric["min"],
                    "max": selected_metric["max"],
                    "color": hero_status_color or selected_metric.get("color", "") or "#ffb500",
                },
                "status_tracks": status_tracks,
            },
        }

        systems_json_raw = payload.get("systems_json", {})
        systems_payload = systems_json_raw if isinstance(systems_json_raw, dict) else parse_json_object(systems_json_raw)
        if not isinstance(systems_payload, dict):
            systems_payload = {}

        runtime_options_raw = systems_payload.get("runtime_options", {}) if isinstance(systems_payload, dict) else {}
        if not isinstance(runtime_options_raw, dict):
            runtime_options_raw = {}
        interaction_mode = str(
            payload.get("interaction_mode", runtime_options_raw.get("interaction_mode", "hybrid"))
        ).strip().lower()
        if interaction_mode not in {"cooperative", "split", "hybrid"}:
            interaction_mode = "hybrid"
        runtime_options = {
            "interaction_mode": interaction_mode,
            "sync_all_player_ai_replies": parse_bool(
                payload.get(
                    "sync_all_player_ai_replies",
                    runtime_options_raw.get("sync_all_player_ai_replies", False),
                ),
                False,
            ),
            "allow_player_death": parse_bool(
                payload.get("allow_player_death", runtime_options_raw.get("allow_player_death", False)),
                False,
            ),
            "allow_revive": parse_bool(
                payload.get("allow_revive", runtime_options_raw.get("allow_revive", False)),
                False,
            ),
            "require_all_players_response": parse_bool(
                payload.get(
                    "require_all_players_response",
                    runtime_options_raw.get("require_all_players_response", False),
                ),
                False,
            ),
        }
        timeout_raw = payload.get("response_timeout_seconds", runtime_options_raw.get("response_timeout_seconds", 45))
        try:
            response_timeout_seconds = int(timeout_raw)
        except (TypeError, ValueError):
            response_timeout_seconds = 45
        runtime_options["response_timeout_seconds"] = max(5, min(300, response_timeout_seconds))
        systems_payload["runtime_options"] = runtime_options

        content_pack = {
            "genre_tags": parse_text_list(payload.get("genre_tags", []), limit=10),
            "feature_badges": parse_text_list(payload.get("feature_badges", []), limit=12),
            "world_lore": str(payload.get("world_lore", "")).strip()[:3600],
            "core_conflict": str(payload.get("core_conflict", "")).strip()[:1800],
            "player_roles": parse_text_list(payload.get("player_roles", []), limit=12),
            "interaction_modes": parse_text_list(payload.get("interaction_modes", []), limit=12),
            "system_rules": parse_text_list(payload.get("system_rules", []), limit=14),
            "npc_roster": parse_text_list(payload.get("npc_roster", []), limit=16),
            "scene_blueprint": parse_text_list(payload.get("scene_blueprint", []), limit=16),
            "ending_catalog": parse_text_list(payload.get("ending_catalog", []), limit=12),
            "hidden_truths": parse_text_list(payload.get("hidden_truths", []), limit=12),
            "end_conditions": parse_json_list(payload.get("end_conditions", []))[:20],
            "systems": systems_payload,
            "custom_achievements": parse_custom_achievements(payload.get("custom_achievements", []), limit=20),
            "condition_achievements": parse_condition_achievements(payload.get("condition_achievements", []), limit=30),
        }
        if not content_pack["genre_tags"]:
            content_pack["genre_tags"] = ["自定义冒险"]
        if not content_pack["feature_badges"]:
            content_pack["feature_badges"] = ["多人协作", "AI主持"]
        if not content_pack["world_lore"]:
            content_pack["world_lore"] = f"{title} 的世界被异常事件覆盖，玩家需要在信息不完整中推进调查。"
        if not content_pack["core_conflict"]:
            content_pack["core_conflict"] = "在有限回合中达成目标，同时处理队伍内分歧与外部风险。"

        return {
            "title": title,
            "subtitle": subtitle,
            "summary": summary,
            "cover_url": cover_url,
            "opening_text": opening_text,
            "starter_goals": starter_goals,
            "state_schema": state_schema,
            "content_pack": content_pack,
        }

    def parse_module_ids(raw_value: Any) -> list[str]:
        rows = raw_value if isinstance(raw_value, list) else parse_json_list(raw_value)
        if not isinstance(rows, list):
            return []
        out: list[str] = []
        seen: set[str] = set()
        for row in rows[:20]:
            module_id = str(row or "").strip()
            if not module_id or module_id in seen:
                continue
            seen.add(module_id)
            out.append(module_id)
        return out

    def normalize_module_payload(raw_payload: Any) -> dict[str, Any]:
        payload = raw_payload if isinstance(raw_payload, dict) else {}
        return {
            "title": str(payload.get("title", "")).strip()[:84],
            "summary": str(payload.get("summary", "")).strip()[:240],
            "patch_json": parse_module_patch(payload.get("patch_json", {})),
        }

    def load_modules_for_card(base_card_id: str, module_ids: list[str]) -> list[dict[str, Any]]:
        available = persistence.list_card_modules(base_card_id, status="published")
        if not module_ids:
            return []
        by_id = {str(row.get("module_id", "")).strip(): row for row in available}
        selected: list[dict[str, Any]] = []
        for module_id in module_ids:
            row = by_id.get(module_id)
            if row:
                selected.append(row)
        return selected

    @app.get("/api/cards")
    def list_cards_api() -> Any:
        cards = persistence.list_game_cards(world_id="global")
        user = current_user()
        price_map = persistence.get_game_card_price_map_for_user(user.get("user_id") if user else None)
        rows: list[dict[str, Any]] = []
        for card in cards:
            item = dict(card)
            card_id = str(item.get("card_id", "")).strip()
            pricing = price_map.get(card_id)
            is_creator = bool(user and str(item.get("creator_user_id") or "") == str(user.get("user_id", "")))
            owned = bool(pricing.get("owned", False)) if isinstance(pricing, dict) else False
            if is_creator:
                owned = True
            item["shop_price"] = int(pricing.get("price", 0)) if isinstance(pricing, dict) else 0
            item["shop_item_id"] = str(pricing.get("item_id", "")).strip() if isinstance(pricing, dict) else ""
            item["shop_item_name"] = str(pricing.get("item_name", "")).strip() if isinstance(pricing, dict) else ""
            item["owned"] = bool(owned or (item["shop_price"] <= 0))
            item["requires_purchase"] = bool(item["shop_price"] > 0 and not item["owned"])
            rows.append(item)
        return jsonify({"ok": True, "cards": rows})

    @app.get("/api/cards/<card_id>")
    def get_card_api(card_id: str) -> Any:
        card = persistence.get_game_card(card_id.strip())
        if card is None:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 404
        return jsonify({"ok": True, "card": card})

    @app.get("/api/cards/<card_id>/modules")
    def list_card_modules_api(card_id: str) -> Any:
        safe_card_id = card_id.strip()
        if persistence.get_game_card(safe_card_id) is None:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 404
        modules = persistence.list_card_modules(safe_card_id, status="published")
        return jsonify({"ok": True, "modules": modules})

    @app.post("/api/cards/<card_id>/modules")
    def create_card_module_api(card_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        safe_card_id = card_id.strip()
        if persistence.get_game_card(safe_card_id) is None:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 404

        payload = normalize_module_payload(request.get_json(silent=True) or {})
        if not payload["title"]:
            return jsonify({"ok": False, "error": readable_error("invalid_card_payload")}), 400
        if not payload["patch_json"]:
            return jsonify({"ok": False, "error": readable_error("module_patch_invalid")}), 400

        try:
            module = persistence.create_card_module(
                base_card_id=safe_card_id,
                title=payload["title"],
                summary=payload["summary"],
                patch_json=payload["patch_json"],
                creator_user_id=user["user_id"],
            )
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "module": module})

    @app.patch("/api/modules/<module_id>")
    def update_card_module_api(module_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        module = persistence.get_card_module(module_id.strip())
        if module is None:
            return jsonify({"ok": False, "error": readable_error("module_not_found")}), 404

        payload_raw = request.get_json(silent=True) or {}
        payload = normalize_module_payload(payload_raw)
        title = payload["title"] if "title" in payload_raw else None
        summary = payload["summary"] if "summary" in payload_raw else None
        patch_json = payload["patch_json"] if "patch_json" in payload_raw else None
        if "patch_json" in payload_raw and not patch_json:
            return jsonify({"ok": False, "error": readable_error("module_patch_invalid")}), 400

        is_admin = str(user.get("role", "player")) == "admin"
        try:
            updated = persistence.update_card_module(
                module_id.strip(),
                title=title,
                summary=summary,
                patch_json=patch_json,
                actor_user_id=user["user_id"],
                as_admin=is_admin,
            )
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "module": updated})

    @app.delete("/api/modules/<module_id>")
    def delete_card_module_api(module_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        module = persistence.get_card_module(module_id.strip())
        if module is None:
            return jsonify({"ok": False, "error": readable_error("module_not_found")}), 404
        is_admin = str(user.get("role", "player")) == "admin"
        try:
            persistence.delete_card_module(module_id.strip(), user["user_id"], as_admin=is_admin)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "module_id": module_id.strip()})

    @app.get("/api/admin/users")
    def admin_users_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        users = persistence.list_users(limit=600)
        return jsonify({"ok": True, "users": users})

    @app.patch("/api/admin/users/<user_id>/role")
    def admin_update_user_role_api(user_id: str) -> Any:
        me, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        role = str(payload.get("role", "")).strip()
        target_id = user_id.strip()
        try:
            updated = persistence.update_user_role(target_id, role)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

        if me and target_id == me["user_id"] and role != "admin":
            session.pop("user_id", None)
        return jsonify({"ok": True, "user": serialize_user(updated)})

    @app.get("/api/admin/rooms")
    def admin_rooms_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        rows = persistence.list_rooms_admin(limit=400)
        card_map = {row["card_id"]: row for row in persistence.list_game_cards()}
        for row in rows:
            row["card_title"] = card_map.get(row.get("game_card_id", ""), {}).get("title", "未知剧本")
        return jsonify({"ok": True, "rooms": rows})

    @app.delete("/api/admin/rooms/<room_id>")
    def admin_delete_room_api(room_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        safe_room_id = room_id.strip().upper()
        try:
            room_manager.force_delete_room(safe_room_id)
            return jsonify({"ok": True, "room_id": safe_room_id})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/cards")
    def admin_cards_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        cards = persistence.list_game_cards()
        for card in cards:
            creator = persistence.get_user_by_id(card.get("creator_user_id"))
            card["creator_display_name"] = creator["username"] if creator else ("官方" if card.get("is_official") else "玩家")
        return jsonify({"ok": True, "cards": cards})

    @app.get("/api/admin/modules")
    def admin_modules_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        modules = persistence.list_card_modules_admin(limit=800)
        return jsonify({"ok": True, "modules": modules})

    @app.delete("/api/admin/modules/<module_id>")
    def admin_delete_module_api(module_id: str) -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        safe_module_id = str(module_id or "").strip()
        if not safe_module_id:
            return jsonify({"ok": False, "error": readable_error("module_not_found")}), 404
        module = persistence.get_card_module(safe_module_id)
        if module is None:
            return jsonify({"ok": False, "error": readable_error("module_not_found")}), 404
        try:
            persistence.delete_card_module(safe_module_id, user["user_id"], as_admin=True)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        return jsonify({"ok": True, "module_id": safe_module_id})

    @app.get("/api/admin/activation-codes")
    def admin_activation_codes_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        status = str(request.args.get("status", "")).strip().lower()
        batch_no = str(request.args.get("batch_no", "")).strip()
        keyword = str(request.args.get("q", "")).strip()
        limit_raw = request.args.get("limit", 300)
        try:
            limit = max(1, min(3000, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 300
        rows = persistence.list_activation_codes(status=status, batch_no=batch_no, keyword=keyword, limit=limit)
        return jsonify({"ok": True, "codes": rows})

    @app.post("/api/admin/activation-codes/generate")
    def admin_generate_activation_codes_api() -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        try:
            points = max(1, int(payload.get("points", 0)))
            quantity = max(1, min(5000, int(payload.get("quantity", 1))))
        except (TypeError, ValueError):
            return jsonify({"ok": False, "error": readable_error("invalid_points_amount")}), 400
        expires_at = str(payload.get("expires_at", "")).strip()[:40]
        batch_no = str(payload.get("batch_no", "")).strip()[:64]
        remark = str(payload.get("remark", "")).strip()[:180]
        code_prefix = str(payload.get("code_prefix", "ET")).strip().upper()[:8]
        rows = persistence.generate_activation_codes(
            points=points,
            quantity=quantity,
            expires_at=expires_at,
            batch_no=batch_no,
            remark=remark,
            created_by=user["user_id"],
            code_prefix=code_prefix or "ET",
        )
        return jsonify({"ok": True, "codes": rows, "count": len(rows)})

    @app.patch("/api/admin/activation-codes/<code>")
    def admin_update_activation_code_api(code: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        status = payload.get("status")
        expires_at = payload.get("expires_at")
        remark = payload.get("remark")
        try:
            row = persistence.update_activation_code(
                code,
                status=status if status is not None else None,
                expires_at=expires_at if expires_at is not None else None,
                remark=remark if remark is not None else None,
            )
            return jsonify({"ok": True, "code": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/shop/items")
    def admin_shop_items_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        status = str(request.args.get("status", "")).strip().lower()
        item_type = str(request.args.get("type", "")).strip().lower()
        keyword = str(request.args.get("q", "")).strip()
        limit_raw = request.args.get("limit", 500)
        try:
            limit = max(1, min(5000, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 500
        rows = persistence.list_shop_items(status=status, item_type=item_type, keyword=keyword, limit=limit)
        return jsonify({"ok": True, "items": rows})

    @app.post("/api/admin/shop/items")
    def admin_create_shop_item_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        try:
            row = persistence.create_shop_item(
                name=str(payload.get("name", "")).strip(),
                description=str(payload.get("description", "")).strip(),
                image_url=str(payload.get("image_url", "")).strip(),
                item_type=str(payload.get("item_type", "decoration")).strip().lower(),
                subtype=str(payload.get("subtype", "")).strip(),
                price=int(payload.get("price", 0) or 0),
                stock=int(payload.get("stock", -1) if payload.get("stock", -1) is not None else -1),
                purchase_limit=int(payload.get("purchase_limit", 0) or 0),
                status=str(payload.get("status", "active")).strip().lower(),
                sort_order=int(payload.get("sort_order", 0) or 0),
                is_recommended=bool(payload.get("is_recommended", False)),
                metadata=payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {},
                item_id=str(payload.get("item_id", "")).strip().lower(),
            )
            return jsonify({"ok": True, "item": row})
        except (TypeError, ValueError):
            return jsonify({"ok": False, "error": readable_error("invalid_shop_item")}), 400

    @app.patch("/api/admin/shop/items/<item_id>")
    def admin_update_shop_item_api(item_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        try:
            row = persistence.update_shop_item(item_id, payload if isinstance(payload, dict) else {})
            return jsonify({"ok": True, "item": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.delete("/api/admin/shop/items/<item_id>")
    def admin_delete_shop_item_api(item_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        try:
            persistence.delete_shop_item(item_id)
            return jsonify({"ok": True, "item_id": str(item_id or "").strip().lower()})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/shop/orders")
    def admin_shop_orders_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        status = str(request.args.get("status", "")).strip().lower()
        user_id = str(request.args.get("user_id", "")).strip()
        limit_raw = request.args.get("limit", 400)
        try:
            limit = max(1, min(5000, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 400
        rows = persistence.list_shop_orders_admin(status=status, user_id=user_id, limit=limit)
        return jsonify({"ok": True, "orders": rows})

    @app.post("/api/admin/shop/orders/<order_id>/refund")
    def admin_refund_shop_order_api(order_id: str) -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        note = str(payload.get("note", "")).strip()[:180]
        try:
            row = persistence.refund_shop_order(order_id, operator_user_id=user["user_id"], note=note)
            return jsonify({"ok": True, "order": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/points/transactions")
    def admin_points_transactions_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        user_id = str(request.args.get("user_id", "")).strip()
        limit_raw = request.args.get("limit", 300)
        try:
            limit = max(1, min(5000, int(limit_raw)))
        except (TypeError, ValueError):
            limit = 300
        rows = persistence.list_point_transactions_admin(user_id=user_id, limit=limit)
        return jsonify({"ok": True, "transactions": rows})

    @app.patch("/api/admin/users/<user_id>/points/adjust")
    def admin_adjust_user_points_api(user_id: str) -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        delta_raw = payload.get("delta", 0)
        note = str(payload.get("note", "")).strip()[:180]
        count_recharge = bool(payload.get("count_recharge", False))
        try:
            delta = int(delta_raw)
        except (TypeError, ValueError):
            return jsonify({"ok": False, "error": readable_error("invalid_points_amount")}), 400
        try:
            result = persistence.admin_adjust_user_points(
                user_id.strip(),
                delta,
                note=note,
                operator_user_id=user["user_id"],
                count_recharge=count_recharge,
            )
            return jsonify({"ok": True, "result": result})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.patch("/api/admin/users/<user_id>/points/lock")
    def admin_update_user_points_lock_api(user_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        is_locked = payload.get("is_locked")
        frozen_balance = payload.get("frozen_balance")
        frozen_value = None
        if frozen_balance is not None:
            try:
                frozen_value = max(0, int(frozen_balance))
            except (TypeError, ValueError):
                return jsonify({"ok": False, "error": readable_error("invalid_points_amount")}), 400
        try:
            account = persistence.update_points_lock(
                user_id.strip(),
                is_locked=bool(is_locked) if is_locked is not None else None,
                frozen_balance=frozen_value,
            )
            return jsonify({"ok": True, "account": account})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/finance/summary")
    def admin_finance_summary_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        days_raw = request.args.get("days", 30)
        try:
            days = max(1, min(365, int(days_raw)))
        except (TypeError, ValueError):
            days = 30
        summary = persistence.get_finance_summary(days=days)
        return jsonify({"ok": True, "summary": summary})

    @app.get("/api/admin/settings/season-system")
    def admin_season_system_get_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        return jsonify({"ok": True, "config": get_season_system_config()})

    @app.patch("/api/admin/settings/season-system")
    def admin_season_system_patch_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        if not isinstance(payload, dict):
            payload = {}
        current_raw = load_settings_file_payload()
        defaults = get_season_system_config()
        next_cfg = dict(defaults)
        for key in (
            "enabled",
            "auto_create_enabled",
            "auto_generate_theme",
            "season_duration_days",
            "daily_card_count",
            "refresh_hours",
            "manual_refresh_cost",
            "premium_pass_price",
        ):
            if key in payload:
                next_cfg[key] = payload.get(key)
        current_raw["season_system"] = next_cfg
        save_runtime_settings(current_raw)
        return jsonify({"ok": True, "config": get_season_system_config()})

    @app.get("/api/admin/seasons")
    def admin_seasons_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        rows = persistence.list_seasons(limit=80)
        return jsonify({"ok": True, "seasons": rows, "active": persistence.get_active_season()})

    @app.post("/api/admin/seasons/auto-generate")
    def admin_auto_generate_season_api() -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        season_cfg = get_season_system_config()
        theme_hint = str(payload.get("theme_hint", "")).strip()[:40] or "迷雾电波"
        duration_days = int(payload.get("duration_days", season_cfg.get("season_duration_days", 30)) or 30)
        pass_price = int(payload.get("pass_price", season_cfg.get("premium_pass_price", 680)) or 680)
        blueprint = build_auto_season_blueprint(
            theme_hint=theme_hint,
            duration_days=duration_days,
            pass_price=pass_price,
        )
        season_payload = blueprint.get("season", {}) if isinstance(blueprint, dict) else {}
        try:
            season = persistence.create_season(
                name=str(season_payload.get("name", "自动赛季")),
                theme=str(season_payload.get("theme", "")),
                description=str(season_payload.get("description", "")),
                cover_url=str(season_payload.get("cover_url", "")),
                accent_color=str(season_payload.get("accent_color", "#4f6dff")),
                story_seed=str(season_payload.get("story_seed", "")),
                story_outline=str(season_payload.get("story_outline", "")),
                start_at=str(season_payload.get("start_at", "")),
                end_at=str(season_payload.get("end_at", "")),
                status=str(season_payload.get("status", "active")),
                source=str(season_payload.get("source", "ai_auto")),
                pass_price=int(season_payload.get("pass_price", 680) or 680),
                config=season_payload.get("config", {}) if isinstance(season_payload.get("config"), dict) else {},
                created_by=str(user.get("user_id", "")),
            )
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        season_id = str(season.get("season_id", "")).strip()
        for item in blueprint.get("announcements", []):
            if isinstance(item, dict):
                try:
                    persistence.create_season_announcement(
                        season_id,
                        title=str(item.get("title", "赛季公告")),
                        content=str(item.get("content", "")),
                        created_by=str(user.get("user_id", "")),
                    )
                except ValueError:
                    pass
        for item in blueprint.get("task_templates", []):
            if isinstance(item, dict):
                try:
                    persistence.upsert_season_task_template(season_id, item)
                except ValueError:
                    pass
        for item in blueprint.get("card_pools", []):
            if isinstance(item, dict):
                try:
                    persistence.upsert_season_card_pool(season_id, item)
                except ValueError:
                    pass
        for item in blueprint.get("pass_rewards", []):
            if isinstance(item, dict):
                try:
                    persistence.upsert_season_pass_reward(season_id, item)
                except ValueError:
                    pass
        for item in blueprint.get("titles", []):
            if isinstance(item, dict):
                try:
                    persistence.upsert_season_title(season_id, item, title_id=str(item.get("title_id", "")))
                except ValueError:
                    pass
        for item in blueprint.get("avatar_frames", []):
            if isinstance(item, dict):
                try:
                    persistence.upsert_season_avatar_frame(season_id, item, frame_id=str(item.get("frame_id", "")))
                except ValueError:
                    pass
        return jsonify({"ok": True, "season": persistence.get_season(season_id), "blueprint": blueprint})

    @app.post("/api/admin/seasons")
    def admin_create_season_api() -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        if not isinstance(payload, dict):
            payload = {}
        try:
            season = persistence.create_season(
                name=str(payload.get("name", "")),
                theme=str(payload.get("theme", "")),
                description=str(payload.get("description", "")),
                cover_url=str(payload.get("cover_url", "")),
                accent_color=str(payload.get("accent_color", "#4f6dff")),
                story_seed=str(payload.get("story_seed", "")),
                story_outline=str(payload.get("story_outline", "")),
                start_at=str(payload.get("start_at", "")),
                end_at=str(payload.get("end_at", "")),
                status=str(payload.get("status", "active")),
                source=str(payload.get("source", "manual")),
                pass_price=int(payload.get("pass_price", 680) or 680),
                config=payload.get("config", {}) if isinstance(payload.get("config"), dict) else {},
                created_by=str(user.get("user_id", "")),
                season_id=str(payload.get("season_id", "")),
            )
            return jsonify({"ok": True, "season": season})
        except (TypeError, ValueError) as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.patch("/api/admin/seasons/<season_id>")
    def admin_update_season_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        try:
            season = persistence.update_season(season_id.strip(), payload if isinstance(payload, dict) else {})
            return jsonify({"ok": True, "season": season})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/seasons/<season_id>/dashboard")
    def admin_season_dashboard_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        season = persistence.get_season(season_id.strip())
        if season is None:
            return jsonify({"ok": False, "error": readable_error("season_not_found")}), 404
        dashboard = persistence.get_season_dashboard(season_id.strip())
        return jsonify({"ok": True, "season": season, "dashboard": dashboard})

    @app.post("/api/admin/seasons/<season_id>/announce")
    def admin_season_announce_api(season_id: str) -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        title = str(payload.get("title", "")).strip()
        content = str(payload.get("content", "")).strip()
        if not title:
            return jsonify({"ok": False, "error": readable_error("invalid_card_payload")}), 400
        try:
            row = persistence.create_season_announcement(
                season_id.strip(),
                title=title,
                content=content,
                created_by=str(user.get("user_id", "")),
            )
            return jsonify({"ok": True, "announcement": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/seasons/<season_id>/tasks")
    def admin_season_tasks_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        task_type = str(request.args.get("type", "")).strip().lower()
        rows = persistence.list_season_task_templates(season_id.strip(), task_type=task_type, active_only=False, limit=1200)
        return jsonify({"ok": True, "task_templates": rows})

    @app.post("/api/admin/seasons/<season_id>/tasks")
    def admin_season_task_upsert_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        template_id = str(payload.get("template_id", "")).strip()
        try:
            row = persistence.upsert_season_task_template(
                season_id.strip(),
                payload if isinstance(payload, dict) else {},
                template_id=template_id,
            )
            return jsonify({"ok": True, "task_template": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/seasons/<season_id>/card-pools")
    def admin_season_card_pools_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        rows = persistence.list_season_card_pools(season_id.strip(), active_only=False, limit=1200)
        return jsonify({"ok": True, "card_pools": rows})

    @app.post("/api/admin/seasons/<season_id>/card-pools")
    def admin_season_card_pool_upsert_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        pool_id = str(payload.get("pool_id", "")).strip()
        try:
            row = persistence.upsert_season_card_pool(
                season_id.strip(),
                payload if isinstance(payload, dict) else {},
                pool_id=pool_id,
            )
            return jsonify({"ok": True, "card_pool": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/seasons/<season_id>/pass-rewards")
    def admin_season_pass_rewards_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        rows = persistence.list_season_pass_rewards(season_id.strip(), limit=2400)
        return jsonify({"ok": True, "pass_rewards": rows})

    @app.post("/api/admin/seasons/<season_id>/pass-rewards")
    def admin_season_pass_reward_upsert_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        reward_id = str(payload.get("reward_id", "")).strip()
        try:
            row = persistence.upsert_season_pass_reward(
                season_id.strip(),
                payload if isinstance(payload, dict) else {},
                reward_id=reward_id,
            )
            return jsonify({"ok": True, "pass_reward": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/seasons/<season_id>/titles")
    def admin_season_titles_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        rows = persistence.list_season_titles(season_id.strip(), active_only=False, limit=1200)
        return jsonify({"ok": True, "titles": rows})

    @app.post("/api/admin/seasons/<season_id>/titles")
    def admin_season_title_upsert_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        title_id = str(payload.get("title_id", "")).strip()
        try:
            row = persistence.upsert_season_title(
                season_id.strip(),
                payload if isinstance(payload, dict) else {},
                title_id=title_id,
            )
            return jsonify({"ok": True, "title": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/seasons/<season_id>/avatar-frames")
    def admin_season_avatar_frames_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        rows = persistence.list_season_avatar_frames(season_id.strip(), active_only=False, limit=1200)
        return jsonify({"ok": True, "avatar_frames": rows})

    @app.post("/api/admin/seasons/<season_id>/avatar-frames")
    def admin_season_avatar_frame_upsert_api(season_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        frame_id = str(payload.get("frame_id", "")).strip()
        try:
            row = persistence.upsert_season_avatar_frame(
                season_id.strip(),
                payload if isinstance(payload, dict) else {},
                frame_id=frame_id,
            )
            return jsonify({"ok": True, "avatar_frame": row})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/admin/settings/oauth/linuxdo")
    def admin_linuxdo_oauth_get_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        return jsonify({"ok": True, "config": get_linuxdo_oauth_config(mask_secret=True)})

    @app.get("/api/admin/settings/badges")
    def admin_badges_get_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        return jsonify(
            {
                "ok": True,
                "badges": get_role_badges_config(),
                "defaults": default_role_badges(),
            }
        )

    @app.patch("/api/admin/settings/badges")
    def admin_badges_patch_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        raw_badges = payload.get("badges")
        if not isinstance(raw_badges, list):
            return jsonify({"ok": False, "error": readable_error("badge_config_invalid")}), 400

        badges = normalize_badge_definitions(raw_badges, default_to_builtin=False)
        if not badges:
            return jsonify({"ok": False, "error": readable_error("badge_config_invalid")}), 400

        raw_settings = load_settings_file_payload()
        raw_settings["role_badges"] = badges
        save_runtime_settings(raw_settings)
        return jsonify({"ok": True, "badges": get_role_badges_config()})

    @app.patch("/api/admin/settings/oauth/linuxdo")
    def admin_linuxdo_oauth_patch_api() -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp

        payload = request.get_json(silent=True) or {}
        if not isinstance(payload, dict):
            payload = {}

        raw_settings = load_settings_file_payload()

        defaults = DEFAULT_SETTINGS.get("linuxdo_oauth", {})
        existing_raw = raw_settings.get("linuxdo_oauth", {})
        existing_cfg = {**defaults, **(existing_raw if isinstance(existing_raw, dict) else {})}

        enabled = bool(payload.get("enabled", existing_cfg.get("enabled", False)))
        client_id = str(payload.get("client_id", existing_cfg.get("client_id", ""))).strip()
        incoming_secret = payload.get("client_secret")
        if incoming_secret is None:
            client_secret = str(existing_cfg.get("client_secret", "")).strip()
        else:
            normalized_secret = str(incoming_secret).strip()
            client_secret = normalized_secret or str(existing_cfg.get("client_secret", "")).strip()
        authorize_url = str(payload.get("authorize_url", existing_cfg.get("authorize_url", ""))).strip()
        token_url = str(payload.get("token_url", existing_cfg.get("token_url", ""))).strip()
        userinfo_url = str(payload.get("userinfo_url", existing_cfg.get("userinfo_url", ""))).strip()
        scope = str(payload.get("scope", existing_cfg.get("scope", ""))).strip()

        for field, url_value in (
            ("authorize_url", authorize_url),
            ("token_url", token_url),
            ("userinfo_url", userinfo_url),
        ):
            if not _is_valid_http_url(url_value):
                return jsonify({"ok": False, "error": f"Linux DO OAuth 配置无效：{field}"}), 400

        if enabled and not client_id:
            return jsonify({"ok": False, "error": "Linux DO OAuth 配置缺失：client_id"}), 400
        if enabled and not client_secret:
            return jsonify({"ok": False, "error": "Linux DO OAuth 配置缺失：client_secret"}), 400

        raw_settings["linuxdo_oauth"] = {
            "enabled": enabled,
            "client_id": client_id,
            "client_secret": client_secret,
            "authorize_url": authorize_url,
            "token_url": token_url,
            "userinfo_url": userinfo_url,
            "scope": scope,
        }
        save_runtime_settings(raw_settings)

        return jsonify({"ok": True, "config": get_linuxdo_oauth_config(mask_secret=True)})

    @app.patch("/api/admin/cards/<card_id>")
    def admin_patch_card_api(card_id: str) -> Any:
        _, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        payload = request.get_json(silent=True) or {}
        is_official = bool(payload.get("is_official", False))
        try:
            card = persistence.set_game_card_official(card_id.strip(), is_official)
            return jsonify({"ok": True, "card": card})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.delete("/api/admin/cards/<card_id>")
    def admin_delete_card_api(card_id: str) -> Any:
        user, error_resp = require_admin_json()
        if error_resp:
            return error_resp
        try:
            persistence.delete_game_card(card_id.strip(), user["user_id"], as_admin=True)
            return jsonify({"ok": True, "card_id": card_id.strip()})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.delete("/api/cards/<card_id>")
    def delete_card_api(card_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        try:
            persistence.delete_game_card(card_id.strip(), user["user_id"])
            return jsonify({"ok": True, "card_id": card_id.strip()})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.post("/api/cards/assist")
    def assist_card_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        payload = request.get_json(silent=True) or {}
        concept = str(payload.get("concept", "")).strip()
        tone = str(payload.get("tone", "")).strip() or "悬疑冒险"
        difficulty = str(payload.get("difficulty", "")).strip() or "中等"
        if not concept:
            return jsonify({"ok": False, "error": readable_error("invalid_assist_prompt")}), 400

        # Keep schema concise to avoid model truncation and JSON parse failures.
        schema_hint = {
            "title": "string",
            "subtitle": "string",
            "summary": "string",
            "opening_text": "string",
            "starter_goals": ["string", "string", "string"],
            "metrics": [{"key": "string", "label": "string", "min": 0, "max": 100, "initial": 0, "color": "#RRGGBB"}],
            "status_tracks": [{"key": "string", "label": "string", "min": 0, "max": 100, "color": "#RRGGBB"}],
            "data_scope": {"metrics": "shared|per_player", "inventory": "shared|per_player", "clues": "shared|per_player"},
            "visibility": {"show_all_player_metrics": "boolean", "show_all_player_inventory": "boolean"},
            "state_fields": ["string", "string", "string"],
            "end_conditions": [{"id": "string", "metric": "string", "op": ">=|<=|>|<|==|!=", "value": 0, "ending_state": "string", "ending_summary": "string"}],
            "systems": {"combat": {"enabled": True}, "dice": {"enabled": True}},
            "interaction_mode": "cooperative|split|hybrid",
            "sync_all_player_ai_replies": "boolean",
            "allow_player_death": "boolean",
            "allow_revive": "boolean",
            "require_all_players_response": "boolean",
            "response_timeout_seconds": "number",
            "genre_tags": ["string"],
            "feature_badges": ["string"],
            "world_lore": "string",
            "core_conflict": "string",
            "player_roles": ["string"],
            "interaction_modes": ["string"],
            "system_rules": ["string"],
            "npc_roster": ["string"],
            "scene_blueprint": ["string"],
            "ending_catalog": ["string"],
            "hidden_truths": ["string"],
            "custom_achievements": [{"id": "string", "name": "string", "metric": "plays|wins|current_streak|best_streak", "tiers": [{"level": "bronze|silver|gold|platinum|diamond", "threshold": 1, "title": "string"}]}],
            "condition_achievements": [{"id": "string", "name": "string", "level": "bronze|silver|gold|platinum|diamond", "conditions": [{"type": "ending_state|flag|metric|round_no|player_alive", "key": "string(optional)", "op": ">=|<=|>|<|==|!=", "value": "any"}]}],
            "labels": {"inventory": "string", "clues": "string", "choices": "string", "scene": "string"},
            "hero_status_label": "string",
            "hero_status_color": "#RRGGBB",
            "cover_url": "string",
            "room_background_url": "string",
            "design_notes": ["string", "string", "string"],
        }
        messages_primary = [
            {
                "role": "system",
                "content": (
                    "你是多人联机文字冒险的高级策划。"
                    "必须只输出一个合法 JSON 对象，不能输出 markdown 或解释。"
                    "所有文案使用中文，内容具体可落地。"
                ),
            },
            {
                "role": "user",
                "content": (
                    "根据以下创意生成游戏卡草稿："
                    f"\n创意：{concept}\n风格：{tone}\n难度：{difficulty}"
                    "\n输出必须包含这些字段（可留空数组但字段必须存在）："
                    f"\n{json.dumps(schema_hint, ensure_ascii=False)}"
                    "\n约束：starter_goals 3 条；metrics 2-4 条；status_tracks 2-4 条；"
                    "genre_tags 3-6 条；feature_badges 3-6 条；design_notes 3 条；"
                    "world_lore 与 core_conflict 需为非空中文段落。"
                ),
            },
        ]
        ai_result = deepseek_client.chat_json(messages_primary, temperature=0.7, max_tokens=1800)
        draft: dict[str, Any] | None = None
        ai_error = ""
        ai_error_detail = ""

        if ai_result.get("ok"):
            draft = parse_json_object(ai_result.get("content", ""))
            if not isinstance(draft, dict):
                ai_error = "invalid_json_from_model"
        else:
            ai_error = str(ai_result.get("error", "assist_ai_failed"))
            ai_error_detail = str(ai_result.get("detail", ""))
            return jsonify(
                {
                    "ok": False,
                    "error": f"AI 生成失败：{ai_error}",
                    "ai_error": ai_error,
                    "ai_error_detail": ai_error_detail,
                }
            ), 502

        # Second pass: ask the model to repair to strict JSON if first output is unparsable.
        if not isinstance(draft, dict) and ai_result.get("ok"):
            repair_messages = [
                {
                    "role": "system",
                    "content": "你是 JSON 修复器。你只能输出一个合法 JSON 对象，不能输出其他文本。",
                },
                {
                    "role": "user",
                    "content": (
                        "把下面文本修复成合法 JSON 对象，且字段至少包含："
                        f"{json.dumps(schema_hint, ensure_ascii=False)}"
                        "\n文本如下：\n"
                        + str(ai_result.get("content", ""))[:5000]
                    ),
                },
            ]
            repair_result = deepseek_client.chat_json(repair_messages, temperature=0.2, max_tokens=1800)
            if repair_result.get("ok"):
                draft = parse_json_object(repair_result.get("content", ""))
            if not isinstance(draft, dict):
                ai_error = f"{ai_error or 'invalid_json_from_model'};repair_failed"
                if not ai_error_detail:
                    ai_error_detail = str(repair_result.get("detail", ""))

        if not isinstance(draft, dict):
            return jsonify(
                {
                    "ok": False,
                    "error": f"AI 返回内容无法解析为合法 JSON：{ai_error or 'invalid_json_from_model'}",
                    "ai_error": ai_error or "invalid_json_from_model",
                    "ai_error_detail": ai_error_detail,
                }
            ), 502

        def _to_str_list(value: Any, limit: int = 5) -> list[str]:
            if not isinstance(value, list):
                return []
            return [str(item).strip() for item in value if str(item).strip()][:limit]

        labels = draft.get("labels", {}) if isinstance(draft.get("labels"), dict) else {}
        timeout_draft_raw = draft.get("response_timeout_seconds", 45)
        try:
            timeout_draft = int(timeout_draft_raw)
        except (TypeError, ValueError):
            timeout_draft = 45
        result = {
            "title": str(draft.get("title", ""))[:84],
            "subtitle": str(draft.get("subtitle", ""))[:84],
            "summary": str(draft.get("summary", ""))[:420],
            "opening_text": str(draft.get("opening_text", ""))[:660],
            "starter_goals": _to_str_list(draft.get("starter_goals"), 5),
            "metrics": draft.get("metrics", []) if isinstance(draft.get("metrics"), list) else [],
            "status_tracks": draft.get("status_tracks", []) if isinstance(draft.get("status_tracks"), list) else [],
            "data_scope": draft.get("data_scope", {}) if isinstance(draft.get("data_scope"), dict) else {},
            "visibility": draft.get("visibility", {}) if isinstance(draft.get("visibility"), dict) else {},
            "state_fields": _to_str_list(draft.get("state_fields"), 4),
            "end_conditions": draft.get("end_conditions", []) if isinstance(draft.get("end_conditions"), list) else [],
            "systems": draft.get("systems", {}) if isinstance(draft.get("systems"), dict) else {},
            "interaction_mode": str(draft.get("interaction_mode", "hybrid"))[:16],
            "sync_all_player_ai_replies": bool(draft.get("sync_all_player_ai_replies", False)),
            "allow_player_death": bool(draft.get("allow_player_death", False)),
            "allow_revive": bool(draft.get("allow_revive", False)),
            "require_all_players_response": bool(draft.get("require_all_players_response", False)),
            "response_timeout_seconds": max(5, min(300, timeout_draft)),
            "genre_tags": _to_str_list(draft.get("genre_tags"), 8),
            "feature_badges": _to_str_list(draft.get("feature_badges"), 8),
            "world_lore": str(draft.get("world_lore", ""))[:3600],
            "core_conflict": str(draft.get("core_conflict", ""))[:1800],
            "player_roles": _to_str_list(draft.get("player_roles"), 8),
            "interaction_modes": _to_str_list(draft.get("interaction_modes"), 8),
            "system_rules": _to_str_list(draft.get("system_rules"), 10),
            "npc_roster": _to_str_list(draft.get("npc_roster"), 12),
            "scene_blueprint": _to_str_list(draft.get("scene_blueprint"), 10),
            "ending_catalog": _to_str_list(draft.get("ending_catalog"), 8),
            "hidden_truths": _to_str_list(draft.get("hidden_truths"), 8),
            "custom_achievements": parse_custom_achievements(draft.get("custom_achievements", []), limit=20),
            "condition_achievements": parse_condition_achievements(draft.get("condition_achievements", []), limit=30),
            "inventory_label": str(labels.get("inventory", ""))[:48],
            "clues_label": str(labels.get("clues", ""))[:48],
            "choices_label": str(labels.get("choices", ""))[:48],
            "scene_label": str(labels.get("scene", ""))[:48],
            "hero_status_label": str(draft.get("hero_status_label", ""))[:48],
            "hero_status_color": str(draft.get("hero_status_color", ""))[:7],
            "cover_url": str(draft.get("cover_url", ""))[:720],
            "room_background_url": str(draft.get("room_background_url", ""))[:720],
            "design_notes": _to_str_list(draft.get("design_notes"), 5),
        }
        if not result["title"]:
            result["title"] = f"{''.join(concept.split())[:24] or '新世界'}行动录"
        if not result["subtitle"]:
            result["subtitle"] = f"{tone} / 多人合作"
        if not result["summary"]:
            result["summary"] = f"围绕「{concept}」展开的多人冒险，玩家需要共享信息并做出关键抉择。"
        if not result["opening_text"]:
            result["opening_text"] = f"你们抵达「{concept}」现场，异常规则已经启动。"
        if not re.match(r"^#[0-9a-fA-F]{6}$", result["hero_status_color"]):
            result["hero_status_color"] = "#ff8c2f"
        if not result["cover_url"]:
            result["cover_url"] = "/static/img/cards/fogharbor.svg"
        if not result["room_background_url"]:
            result["room_background_url"] = result["cover_url"]
        if not result["starter_goals"]:
            result["starter_goals"] = ["先统一目标分工", "与关键NPC交谈", "推进第一轮调查"]
        if not result["state_fields"]:
            result["state_fields"] = ["紧张度", "推进度", "资源储备"]
        if not result["metrics"]:
            result["metrics"] = [
                {"key": "tension", "label": "紧张度", "min": 0, "max": 100, "initial": 25, "color": "#ff8c2f"},
                {"key": "progress", "label": "推进度", "min": 0, "max": 100, "initial": 0, "color": "#3b82f6"},
            ]
        if not result["status_tracks"]:
            result["status_tracks"] = [
                {"key": "tension", "label": "紧张度", "min": 0, "max": 100, "color": "#ff8c2f"},
                {"key": "progress", "label": "推进度", "min": 0, "max": 100, "color": "#3b82f6"},
            ]
        if not result["data_scope"]:
            result["data_scope"] = {"metrics": "per_player", "inventory": "per_player", "clues": "shared"}
        if not result["visibility"]:
            result["visibility"] = {"show_all_player_metrics": True, "show_all_player_inventory": True}
        if not result["genre_tags"]:
            result["genre_tags"] = ["悬疑", "合作", "多结局"]
        if not result["feature_badges"]:
            result["feature_badges"] = ["剧情分支", "团队决策"]
        if not result["world_lore"]:
            result["world_lore"] = (
                f"在「{concept}」中，现实会被集体记忆改写。每一次公开讨论都会改变世界规则，"
                "玩家必须持续校验彼此信息，才能接近真相。"
            )
        if not result["core_conflict"]:
            result["core_conflict"] = "队伍需要在有限时间内兼顾生存、调查与立场分歧，并决定是否牺牲局部利益换取全局突破。"
        if not result["player_roles"]:
            result["player_roles"] = ["调查者", "交涉者", "执行者", "记录者"]
        if not result["interaction_modes"]:
            result["interaction_modes"] = ["公开讨论", "分组行动", "协商投票", "紧急决断"]
        if not result["system_rules"]:
            result["system_rules"] = ["每轮至少一人提交推进行动", "冲突行动会增加系统压力", "关键结论需至少两人交叉验证"]
        if not result["npc_roster"]:
            result["npc_roster"] = ["引导型 NPC：提供线索但有所保留", "对抗型 NPC：制造误导与压力", "中立观察者：记录并反馈玩家选择"]
        if not result["scene_blueprint"]:
            result["scene_blueprint"] = ["开场：规则显现", "中段：线索分叉", "后段：矛盾爆发", "终局：代价抉择"]
        if not result["ending_catalog"]:
            result["ending_catalog"] = ["全员生还并破解真相", "达成目标但付出牺牲", "内部分裂导致失败", "触发隐藏路线结局"]
        if not result["hidden_truths"]:
            result["hidden_truths"] = ["关键真相不会一次性出现", "NPC 阵营会被玩家行为动态影响"]
        if not result["design_notes"]:
            result["design_notes"] = ["先给玩家明确短目标", "每幕至少提供一次分歧选择", "关键线索建议可被不同路径验证"]

        return jsonify(
            {
                "ok": True,
                "draft": result,
                "source": "ai",
                "ai_error": ai_error,
                "ai_error_detail": ai_error_detail,
            }
        )

    @app.post("/api/cards/assist_patch")
    def assist_card_patch_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        payload = request.get_json(silent=True) or {}
        instruction = str(payload.get("instruction", "")).strip()
        if not instruction:
            return jsonify({"ok": False, "error": readable_error("invalid_assist_modify_prompt")}), 400

        current_raw = payload.get("current", {})
        current = normalize_assist_current(current_raw)

        patch_hint = {
            "summary": "string(optional)",
            "opening_text": "string(optional)",
            "starter_goals": ["string(optional)"],
            "metrics": [{"key": "string", "label": "string"}],
            "status_tracks": [{"key": "string", "label": "string"}],
            "end_conditions": [{"id": "string", "metric": "string"}],
            "systems": {"any_system": {"enabled": True}},
            "interaction_mode": "cooperative|split|hybrid(optional)",
            "sync_all_player_ai_replies": "boolean(optional)",
            "allow_player_death": "boolean(optional)",
            "allow_revive": "boolean(optional)",
            "require_all_players_response": "boolean(optional)",
            "response_timeout_seconds": "number(optional)",
            "feature_badges": ["string(optional)"],
            "system_rules": ["string(optional)"],
            "player_roles": ["string(optional)"],
            "interaction_modes": ["string(optional)"],
            "npc_roster": ["string(optional)"],
            "scene_blueprint": ["string(optional)"],
            "design_notes": ["string(optional)"],
            "custom_achievements": [{"id": "string", "name": "string", "metric": "plays|wins|current_streak|best_streak", "tiers": [{"level": "bronze", "threshold": 1, "title": "string"}]}],
            "condition_achievements": [{"id": "string", "name": "string", "level": "bronze", "conditions": [{"type": "ending_state|flag|metric|round_no|player_alive", "key": "string(optional)", "op": ">=|<=|>|<|==|!=", "value": "any"}]}],
        }
        messages = [
            {
                "role": "system",
                "content": (
                    "你是游戏卡增量修改助手。"
                    "你只能输出一个合法 JSON 对象，内容是 patch（仅包含需要修改的字段）。"
                    "禁止输出 markdown、解释、前后缀。"
                ),
            },
            {
                "role": "user",
                "content": (
                    f"修改指令：{instruction}\n"
                    "当前游戏卡内容（JSON）：\n"
                    f"{json.dumps(current, ensure_ascii=False)}\n"
                    "返回 patch JSON，允许字段示例：\n"
                    f"{json.dumps(patch_hint, ensure_ascii=False)}\n"
                    "要求：尽量保留现有内容，按需新增或局部修改，不要把全部字段重写。"
                ),
            },
        ]

        ai_result = deepseek_client.chat_json(messages, temperature=0.5, max_tokens=1400)
        ai_error = ""
        ai_error_detail = ""
        patch_payload: dict[str, Any] | None = None

        if ai_result.get("ok"):
            patch_payload = parse_json_object(ai_result.get("content", ""))
            if not isinstance(patch_payload, dict):
                ai_error = "invalid_json_from_model"
        else:
            ai_error = str(ai_result.get("error", "assist_ai_failed"))
            ai_error_detail = str(ai_result.get("detail", ""))
            return jsonify(
                {
                    "ok": False,
                    "error": f"AI 修改失败：{ai_error}",
                    "ai_error": ai_error,
                    "ai_error_detail": ai_error_detail,
                }
            ), 502

        if not isinstance(patch_payload, dict) and ai_result.get("ok"):
            repair_messages = [
                {
                    "role": "system",
                    "content": "你是 JSON 修复器。只输出一个合法 JSON 对象（patch）。",
                },
                {
                    "role": "user",
                    "content": (
                        f"把以下文本修复成 JSON patch：\n{str(ai_result.get('content', ''))[:5000]}"
                    ),
                },
            ]
            repair_result = deepseek_client.chat_json(repair_messages, temperature=0.2, max_tokens=1000)
            if repair_result.get("ok"):
                patch_payload = parse_json_object(repair_result.get("content", ""))
            if not isinstance(patch_payload, dict):
                ai_error = f"{ai_error or 'invalid_json_from_model'};repair_failed"
                if not ai_error_detail:
                    ai_error_detail = str(repair_result.get("detail", ""))

        if not isinstance(patch_payload, dict):
            return jsonify(
                {
                    "ok": False,
                    "error": f"AI 返回 patch 不是合法 JSON：{ai_error or 'invalid_json_from_model'}",
                    "ai_error": ai_error or "invalid_json_from_model",
                    "ai_error_detail": ai_error_detail,
                }
            ), 502

        patch = normalize_card_patch(patch_payload)
        if not patch:
            return jsonify(
                {
                    "ok": False,
                    "error": "AI 返回 patch 为空，未执行修改。",
                    "ai_error": ai_error or "empty_patch_from_model",
                    "ai_error_detail": ai_error_detail,
                }
            ), 502

        updated = apply_card_patch(current, patch)

        return jsonify(
            {
                "ok": True,
                "patch": patch,
                "updated": updated,
                "source": "ai",
                "ai_error": ai_error,
                "ai_error_detail": ai_error_detail,
            }
        )

    @app.post("/api/cards")
    def create_card_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        payload = request.get_json(silent=True) or {}
        try:
            card_config = build_card_config(payload)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400
        card = persistence.create_game_card(
            title=card_config["title"],
            subtitle=card_config["subtitle"],
            summary=card_config["summary"],
            cover_url=card_config["cover_url"],
            opening_text=card_config["opening_text"],
            starter_goals=card_config["starter_goals"],
            state_schema=card_config["state_schema"],
            content_pack=card_config["content_pack"],
            creator_user_id=user["user_id"],
        )
        return jsonify({"ok": True, "card": card})

    @app.patch("/api/cards/<card_id>")
    def update_card_api(card_id: str) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        card = persistence.get_game_card(card_id.strip())
        if card is None:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 404

        is_admin = str(user.get("role", "player")) == "admin"
        is_creator = str(card.get("creator_user_id") or "") == str(user["user_id"])
        if not is_admin and not is_creator:
            return jsonify({"ok": False, "error": readable_error("card_edit_forbidden")}), 403

        if bool(card.get("is_official")) and not is_admin:
            return jsonify({"ok": False, "error": readable_error("card_edit_forbidden")}), 403

        payload = request.get_json(silent=True) or {}
        try:
            card_config = build_card_config(payload)
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

        try:
            updated = persistence.update_game_card(
                card["card_id"],
                title=card_config["title"],
                subtitle=card_config["subtitle"],
                summary=card_config["summary"],
                cover_url=card_config["cover_url"],
                opening_text=card_config["opening_text"],
                starter_goals=card_config["starter_goals"],
                state_schema=card_config["state_schema"],
                content_pack=card_config["content_pack"],
            )
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

        return jsonify({"ok": True, "card": updated})

    @app.delete("/api/rooms/<room_id>")
    def delete_room_api(room_id: str) -> Any:
        room_id = room_id.strip().upper()
        payload = request.get_json(silent=True) or {}
        player_id = str(payload.get("player_id", "")).strip() or str(
            request.args.get("player_id", "")
        ).strip()
        if not player_id:
            return jsonify({"ok": False, "error": readable_error("missing_room_or_player")}), 400
        user = current_user()
        try:
            room_manager.delete_room(
                room_id,
                player_id,
                requester_user_id=(user or {}).get("user_id"),
            )
            return jsonify({"ok": True, "room_id": room_id})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.get("/api/me/rooms")
    def my_rooms_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": True, "rooms": []})

        rows = persistence.list_user_rooms(user["user_id"], limit=12)
        for row in rows:
            card = persistence.get_game_card(row.get("game_card_id", ""))
            row["card_title"] = card["title"] if card else "未知剧本"
        return jsonify({"ok": True, "rooms": rows})

    @app.get("/api/users/search")
    def users_search_api() -> Any:
        q = str(request.args.get("q", "")).strip()
        limit_raw = request.args.get("limit", "20")
        try:
            limit = max(1, min(100, int(str(limit_raw))))
        except ValueError:
            limit = 20
        rows = persistence.search_users(q, limit=limit)
        users = []
        for row in rows:
            profile = serialize_public_user(row)
            if profile is None:
                continue
            profile["presence"] = persistence.get_user_presence(row["user_id"])
            profile["stats"] = persistence.get_user_game_stats(row["user_id"])
            users.append(profile)
        return jsonify({"ok": True, "q": q, "users": users})

    @app.get("/api/users/<username>/profile")
    def user_profile_api(username: str) -> Any:
        safe_username = normalize_account_username(username)
        if not safe_username:
            return jsonify({"ok": False, "error": readable_error("user_not_found")}), 404
        target = persistence.get_user_by_username(safe_username)
        if target is None:
            return jsonify({"ok": False, "error": readable_error("user_not_found")}), 404
        viewer = current_user()
        return jsonify(
            {
                "ok": True,
                "profile": build_user_profile_payload(target, viewer_user_id=(viewer or {}).get("user_id", "")),
            }
        )

    @app.get("/api/users/<username>/achievements")
    def user_achievements_api(username: str) -> Any:
        safe_username = normalize_account_username(username)
        if not safe_username:
            return jsonify({"ok": False, "error": readable_error("user_not_found")}), 404
        target = persistence.get_user_by_username(safe_username)
        if target is None:
            return jsonify({"ok": False, "error": readable_error("user_not_found")}), 404
        bundle = persistence.get_user_achievement_bundle(target["user_id"])
        return jsonify(
            {
                "ok": True,
                "username": target.get("username", ""),
                "display_name": target.get("display_name", target.get("username", "")),
                "achievements": bundle,
            }
        )

    @app.get("/api/invites")
    def list_invites_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        status = str(request.args.get("status", "pending")).strip().lower() or "pending"
        limit_raw = request.args.get("limit", "20")
        try:
            limit = max(1, min(100, int(str(limit_raw))))
        except ValueError:
            limit = 20
        rows = persistence.list_user_invitations(user["user_id"], status=status, limit=limit)
        for row in rows:
            card = persistence.get_game_card(row.get("room_game_card_id", ""))
            row["card_title"] = card["title"] if card else "未知剧本"
        return jsonify({"ok": True, "invitations": rows})

    @app.get("/api/leaderboard")
    def leaderboard_api() -> Any:
        limit_raw = request.args.get("limit", "20")
        try:
            limit = max(1, min(100, int(str(limit_raw))))
        except ValueError:
            limit = 20
        rows = persistence.list_player_leaderboard(limit=limit)
        return jsonify({"ok": True, "leaderboard": rows})

    @app.get("/api/rooms/public")
    def public_rooms_api() -> Any:
        limit_raw = request.args.get("limit", "30")
        try:
            limit = max(1, min(100, int(str(limit_raw))))
        except ValueError:
            limit = 30
        rows = persistence.list_public_rooms(limit=limit)
        for row in rows:
            card = persistence.get_game_card(row.get("game_card_id", ""))
            row["card_title"] = card["title"] if card else "未知剧本"
            row["card_cover_url"] = card.get("cover_url", "") if card else ""
            row["card_subtitle"] = card.get("subtitle", "") if card else ""
        return jsonify({"ok": True, "rooms": rows})

    def join_room_for_user(
        room_id: str,
        user: dict[str, Any] | None,
        *,
        player_id: str | None = None,
    ) -> tuple[dict[str, Any], int]:
        if user and user.get("user_id"):
            persistence.touch_user_activity(str(user["user_id"]), is_online=True)
        safe_room_id = str(room_id or "").strip().upper()
        safe_player_id = str(player_id or "").strip() or None
        if not safe_room_id:
            return {"ok": False, "error": "请输入房间码"}, 400

        room = persistence.get_room(safe_room_id)
        if room is None:
            return {"ok": False, "error": readable_error("room_not_found")}, 400

        if room["status"] == "lobby":
            if user is None:
                return {"ok": False, "error": readable_error("auth_required")}, 401
            name = str(user["username"]).strip()
            user_id = user["user_id"]
            if not name:
                return {"ok": False, "error": "用户名无效"}, 400
            try:
                join_info, _ = room_manager.join_room(safe_room_id, name, safe_player_id, user_id=user_id)
                return (
                    {
                        "ok": True,
                        "room_id": safe_room_id,
                        "player_id": join_info["player_id"],
                        "player_name": join_info["player_name"],
                        "is_spectator": False,
                        "redirect": url_for(
                            "room_page",
                            room_id=safe_room_id,
                            player_id=join_info["player_id"],
                        ),
                    },
                    200,
                )
            except ValueError as exc:
                return {"ok": False, "error": readable_error(str(exc))}, 400

        if user:
            existing = persistence.get_player_by_user(safe_room_id, user["user_id"])
            if existing:
                try:
                    join_info, _ = room_manager.join_room(
                        safe_room_id,
                        str(user["username"]).strip(),
                        existing["player_id"],
                        user_id=user["user_id"],
                    )
                    return (
                        {
                            "ok": True,
                            "room_id": safe_room_id,
                            "player_id": join_info["player_id"],
                            "player_name": join_info["player_name"],
                            "is_spectator": False,
                            "redirect": url_for(
                                "room_page",
                                room_id=safe_room_id,
                                player_id=join_info["player_id"],
                            ),
                        },
                        200,
                    )
                except ValueError:
                    pass

        return (
            {
                "ok": True,
                "room_id": safe_room_id,
                "is_spectator": True,
                "redirect": url_for("room_page", room_id=safe_room_id, spectator=1),
            },
            200,
        )

    @app.post("/api/invites")
    def create_invite_api() -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401

        payload = request.get_json(silent=True) or {}
        to_user_id = str(payload.get("to_user_id", "")).strip()
        to_username = normalize_account_username(payload.get("to_username"))
        room_id = str(payload.get("room_id", "")).strip().upper()
        message = str(payload.get("message", "")).strip()[:200]

        if not room_id:
            return jsonify({"ok": False, "error": readable_error("invite_room_required")}), 400
        if not to_user_id and not to_username:
            return jsonify({"ok": False, "error": readable_error("invite_invalid")}), 400

        target = None
        if to_user_id:
            target = persistence.get_user_by_id(to_user_id)
        if target is None and to_username:
            target = persistence.get_user_by_username(to_username)
        if target is None:
            return jsonify({"ok": False, "error": readable_error("user_not_found")}), 404
        if str(target["user_id"]) == str(user["user_id"]):
            return jsonify({"ok": False, "error": readable_error("invite_self_forbidden")}), 400

        room = persistence.get_room(room_id)
        if room is None:
            return jsonify({"ok": False, "error": readable_error("room_not_found")}), 400
        if room["status"] not in {"lobby", "running"}:
            return jsonify({"ok": False, "error": readable_error("invite_room_unavailable")}), 400

        sender_player = persistence.get_player_by_user(room_id, user["user_id"])
        if sender_player is None:
            return jsonify({"ok": False, "error": readable_error("invite_sender_not_in_room")}), 400

        invite = persistence.create_invitation(
            from_user_id=user["user_id"],
            to_user_id=target["user_id"],
            room_id=room_id,
            message=message,
        )
        invite["from_username"] = user["username"]
        invite["to_username"] = target["username"]
        return jsonify({"ok": True, "invitation": invite})

    @app.post("/api/invites/<int:invite_id>/accept")
    def accept_invite_api(invite_id: int) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        invite = persistence.get_pending_invitation(invite_id, user["user_id"])
        if invite is None:
            return jsonify({"ok": False, "error": readable_error("invite_not_found")}), 404
        persistence.update_invitation_status(invite_id, user["user_id"], "accepted")
        payload, status = join_room_for_user(invite.get("room_id", ""), user)
        payload["invite_id"] = invite_id
        return jsonify(payload), status

    @app.post("/api/invites/<int:invite_id>/dismiss")
    def dismiss_invite_api(invite_id: int) -> Any:
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        try:
            invite = persistence.update_invitation_status(invite_id, user["user_id"], "dismissed")
            return jsonify({"ok": True, "invitation": invite})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.post("/api/rooms/create")
    def create_room_api() -> Any:
        payload = request.get_json(silent=True) or {}
        user = current_user()
        if user is None:
            return jsonify({"ok": False, "error": readable_error("auth_required")}), 401
        card_id = str(payload.get("card_id", "")).strip() or "fogharbor"
        base_card = persistence.get_game_card(card_id)
        if base_card is None:
            return jsonify({"ok": False, "error": readable_error("invalid_card_id")}), 400
        card_price_map = persistence.get_game_card_price_map_for_user(user["user_id"])
        card_pricing = card_price_map.get(card_id, {})
        is_creator = str(base_card.get("creator_user_id") or "") == str(user.get("user_id", ""))
        is_admin = str(user.get("role", "player")) == "admin"
        if card_pricing and not bool(card_pricing.get("owned", False)) and not is_creator and not is_admin:
            return jsonify({"ok": False, "error": readable_error("card_purchase_required")}), 400
        module_ids = parse_module_ids(payload.get("module_ids", []))
        selected_modules = load_modules_for_card(card_id, module_ids)
        card, enabled_modules = apply_modules_to_card(base_card, selected_modules)

        name = str(user["username"]).strip()
        user_id = user["user_id"]

        if not name:
            return jsonify({"ok": False, "error": "用户名无效"}), 400

        room_fee_tx = None
        points_cfg = get_points_system_config()
        if points_cfg.get("enabled") and points_cfg.get("room_create_fee_enabled"):
            room_fee = max(0, int(points_cfg.get("room_create_fee", 0)))
            if room_fee > 0:
                try:
                    fee_payload = persistence.debit_user_points(
                        user_id,
                        room_fee,
                        tx_type="room_create_fee",
                        source="room_create",
                        note=f"创建房间消耗（剧本卡 {card_id}）",
                        count_spent=True,
                    )
                    room_fee_tx = fee_payload.get("transaction")
                except ValueError as exc:
                    return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

        room = room_manager.create_room(name, user_id=user_id, game_card=card)
        return jsonify(
            {
                "ok": True,
                "room_id": room["room_id"],
                "player_id": room["player_id"],
                "player_name": room["player_name"],
                "redirect": url_for("room_page", room_id=room["room_id"], player_id=room["player_id"]),
                "enabled_modules": enabled_modules,
                "room_fee_transaction": room_fee_tx,
            }
        )

    @app.patch("/api/rooms/<room_id>/visibility")
    def set_room_visibility_api(room_id: str) -> Any:
        safe_room_id = room_id.strip().upper()
        payload = request.get_json(silent=True) or {}
        player_id = str(payload.get("player_id", "")).strip()
        is_public = bool(payload.get("is_public", False))
        if not player_id:
            return jsonify({"ok": False, "error": readable_error("missing_room_or_player")}), 400
        try:
            result = room_manager.set_room_public(safe_room_id, player_id, is_public)
            snapshot = room_manager.get_snapshot(safe_room_id)
            return jsonify({"ok": True, "room_id": result["room_id"], "is_public": result["is_public"], "snapshot": snapshot})
        except ValueError as exc:
            return jsonify({"ok": False, "error": readable_error(str(exc))}), 400

    @app.post("/api/rooms/join")
    def join_room_api() -> Any:
        payload = request.get_json(silent=True) or {}
        room_id = str(payload.get("room_id", "")).strip().upper()
        player_id = str(payload.get("player_id", "")).strip() or None
        user = current_user()
        payload_out, status = join_room_for_user(room_id, user, player_id=player_id)
        return jsonify(payload_out), status

    @app.get("/invite/<room_id>")
    def invite_link_page(room_id: str) -> Any:
        safe_room_id = room_id.strip().upper()
        room = persistence.get_room(safe_room_id)
        if room is None:
            return redirect(url_for("index"))

        mode = str(request.args.get("mode", "join")).strip().lower()
        user = current_user()
        if mode == "spectate" or room["status"] != "lobby":
            return redirect(url_for("room_page", room_id=safe_room_id, spectator=1))

        if user is None:
            return redirect(url_for("login_page"))

        existing = persistence.get_player_by_user(safe_room_id, user["user_id"])
        if existing:
            return redirect(
                url_for("room_page", room_id=safe_room_id, player_id=existing["player_id"])
            )
        return redirect(url_for("index", invite=safe_room_id))

    @app.get("/api/rooms/<room_id>/invite")
    def room_invite_api(room_id: str) -> Any:
        safe_room_id = room_id.strip().upper()
        room = persistence.get_room(safe_room_id)
        if room is None:
            return jsonify({"ok": False, "error": readable_error("room_not_found")}), 404
        base = request.host_url.rstrip("/")
        return jsonify(
            {
                "ok": True,
                "room_id": safe_room_id,
                "join_link": f"{base}/invite/{safe_room_id}?mode=join",
                "spectate_link": f"{base}/invite/{safe_room_id}?mode=spectate",
            }
        )

    @app.get("/room/<room_id>/summary")
    def room_summary_page(room_id: str) -> Any:
        safe_room_id = room_id.strip().upper()
        try:
            summary = build_room_summary_payload(safe_room_id)
        except ValueError:
            return redirect(url_for("index"))
        if summary.get("status") != "ended":
            return redirect(url_for("room_page", room_id=safe_room_id))
        return render_template(
            "room_summary.html",
            room_id=safe_room_id,
            user=serialize_user(current_user()),
        )

    @app.get("/room/<room_id>")
    def room_page(room_id: str) -> Any:
        room_id = room_id.strip().upper()
        player_id = request.args.get("player_id", "").strip()
        spectator = str(request.args.get("spectator", "0")).strip() in {"1", "true", "yes"}
        user = current_user()

        try:
            snapshot = room_manager.get_snapshot(room_id)
        except ValueError:
            return redirect(url_for("index"))

        if spectator:
            if snapshot["status"] == "lobby":
                return redirect(url_for("index"))
            return render_template(
                "room.html",
                room_id=room_id,
                player_id="",
                player_name="观战者",
                spectator=True,
                user=serialize_user(user),
            )

        if not player_id and user:
            row = persistence.get_player_by_user(room_id, user["user_id"])
            if row:
                return redirect(url_for("room_page", room_id=room_id, player_id=row["player_id"]))

        if not player_id:
            return redirect(url_for("index"))

        player = next((row for row in snapshot["players"] if row["player_id"] == player_id), None)
        if player is None:
            return redirect(url_for("index"))

        if player.get("user_id") and ((not user) or player.get("user_id") != user["user_id"]):
            return redirect(url_for("index"))

        return render_template(
            "room.html",
            room_id=room_id,
            player_id=player_id,
            player_name=player["name"],
            spectator=False,
            user=serialize_user(user),
        )

    @app.get("/api/rooms/<room_id>/snapshot")
    def room_snapshot_api(room_id: str) -> Any:
        try:
            return jsonify({"ok": True, "snapshot": room_manager.get_snapshot(room_id.upper())})
        except ValueError:
            return jsonify({"ok": False, "error": "room_not_found"}), 404

    @app.get("/api/rooms/<room_id>/summary")
    def room_summary_api(room_id: str) -> Any:
        safe_room_id = room_id.strip().upper()
        try:
            summary = build_room_summary_payload(safe_room_id)
        except ValueError:
            return jsonify({"ok": False, "error": readable_error("room_not_found")}), 404
        user = current_user()
        if user and str(summary.get("status", "")).strip().lower() == "ended":
            season_progress = persistence.apply_room_summary_season_progress(user["user_id"], summary)
            summary["season_progress"] = season_progress
        return jsonify({"ok": True, "summary": summary})

    @socketio.on("join_room_socket")
    def on_join_room_socket(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        spectator = bool(data.get("spectator", False))
        if not room_id:
            emit("error_message", {"error": readable_error("missing_room_or_player")})
            return

        try:
            snapshot = room_manager.get_snapshot(room_id)
        except ValueError:
            emit("error_message", {"error": readable_error("room_not_found")})
            return

        if spectator:
            if snapshot["status"] == "lobby":
                emit("error_message", {"error": "当前房间尚未开局，观战暂不可用。"})
                return
            sio_join_room(room_id)
            emit("room_snapshot", snapshot)
            emit("system_message", {"content": "观战者已连接。"}, to=room_id)
            return

        if not player_id:
            emit("error_message", {"error": readable_error("missing_room_or_player")})
            return

        player = next((row for row in snapshot["players"] if row["player_id"] == player_id), None)
        if player is None:
            emit("error_message", {"error": readable_error("player_not_in_room")})
            return

        room_manager.attach_socket(room_id, player_id, request.sid)
        sio_join_room(room_id)
        emit("room_snapshot", snapshot)
        emit("system_message", {"content": f"{player['name']} 已连接。"}, to=room_id)
        emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)

    @socketio.on("request_snapshot")
    def on_request_snapshot(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        if not room_id:
            return
        try:
            emit("room_snapshot", room_manager.get_snapshot(room_id))
        except ValueError:
            emit("error_message", {"error": readable_error("room_not_found")})

    @socketio.on("toggle_ready")
    def on_toggle_ready(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        ready = bool(data.get("ready", False))
        if not player_id:
            emit("error_message", {"error": readable_error("spectator_readonly")})
            return
        try:
            room_manager.set_ready(room_id, player_id, ready)
            player = persistence.get_player(room_id, player_id)
            name = player["name"] if player else player_id
            emit(
                "system_message",
                {"content": f"{name} {'已准备' if ready else '取消准备'}"},
                to=room_id,
            )
            emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})

    @socketio.on("start_game")
    def on_start_game(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        if not player_id:
            emit("error_message", {"error": readable_error("spectator_readonly")})
            return
        try:
            room_manager.start_game(room_id, player_id)
            emit("system_message", {"content": "游戏开始：第一幕开启。"}, to=room_id)
            emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})

    @socketio.on("transfer_host")
    def on_transfer_host(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        target_player_id = str(data.get("target_player_id", "")).strip()
        if not player_id:
            emit("error_message", {"error": readable_error("spectator_readonly")})
            return
        try:
            room_manager.transfer_host(room_id, player_id, target_player_id)
            emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})

    @socketio.on("kick_player")
    def on_kick_player(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        target_player_id = str(data.get("target_player_id", "")).strip()
        if not player_id:
            emit("error_message", {"error": readable_error("spectator_readonly")})
            return
        try:
            room_manager.kick_player(room_id, player_id, target_player_id)
            emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})

    @socketio.on("send_chat")
    def on_send_chat(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        content = str(data.get("content", ""))
        if not player_id:
            emit("error_message", {"error": readable_error("spectator_readonly")})
            return
        try:
            room_manager.submit_chat(room_id, player_id, content)
            emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})

    @socketio.on("submit_action")
    def on_submit_action(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()
        content = str(data.get("content", ""))
        if not player_id:
            emit("error_message", {"error": readable_error("spectator_readonly")})
            return

        try:
            result = room_manager.submit_action(room_id, player_id, content)
            emit(
                "system_message",
                {"content": f"{result.get('player_name', player_id)} 已提交行动，等待处理。"},
                to=room_id,
            )
            emit("pending_action_update", result, to=room_id)
            if result.get("should_auto_resolve"):
                resolved = room_manager.resolve_round(room_id, player_id)
                emit("round_resolved", resolved["round_result"], to=room_id)
                emit("room_snapshot", resolved["snapshot"], to=room_id)
                snapshot = resolved["snapshot"] or {}
                game_state = snapshot.get("game_state", {}) if isinstance(snapshot, dict) else {}
                if snapshot.get("status") == "ended" or game_state.get("game_phase") == "ended":
                    emit(
                        "game_ended",
                        {
                            "room_id": room_id,
                            "ending_state": game_state.get("ending_state"),
                            "ending_summary": game_state.get("global_flags", {}).get("ending_summary"),
                            "summary_url": url_for("room_summary_page", room_id=room_id),
                        },
                        to=room_id,
                    )
                for private_player_id, messages in resolved["private_updates"].items():
                    sid = room_manager.sid_for_player(room_id, private_player_id)
                    if sid:
                        emit(
                            "private_updates",
                            {"messages": messages, "player_id": private_player_id},
                            to=sid,
                        )
            else:
                emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})
            try:
                emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
            except ValueError:
                pass
        except Exception as exc:  # pragma: no cover - runtime safety net
            emit("error_message", {"error": f"处理失败：{exc}"})

    @socketio.on("resolve_round")
    def on_resolve_round(data: dict[str, Any]) -> None:
        room_id = str(data.get("room_id", "")).strip().upper()
        player_id = str(data.get("player_id", "")).strip()

        try:
            result = room_manager.resolve_round(room_id, player_id)
            emit("round_resolved", result["round_result"], to=room_id)
            emit("room_snapshot", result["snapshot"], to=room_id)
            snapshot = result["snapshot"] or {}
            game_state = snapshot.get("game_state", {}) if isinstance(snapshot, dict) else {}
            if snapshot.get("status") == "ended" or game_state.get("game_phase") == "ended":
                emit(
                    "game_ended",
                    {
                        "room_id": room_id,
                        "ending_state": game_state.get("ending_state"),
                        "ending_summary": game_state.get("global_flags", {}).get("ending_summary"),
                        "summary_url": url_for("room_summary_page", room_id=room_id),
                    },
                    to=room_id,
                )

            for private_player_id, messages in result["private_updates"].items():
                sid = room_manager.sid_for_player(room_id, private_player_id)
                if sid:
                    emit(
                        "private_updates",
                        {"messages": messages, "player_id": private_player_id},
                        to=sid,
                    )
        except ValueError as exc:
            emit("error_message", {"error": readable_error(str(exc))})
            try:
                emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
            except ValueError:
                pass
        except Exception as exc:  # pragma: no cover - runtime safety net
            emit("error_message", {"error": f"处理失败：{exc}"})

    @socketio.on("disconnect")
    def on_disconnect() -> None:
        mapping = room_manager.detach_socket(request.sid)
        if not mapping:
            return
        room_id, player_id = mapping
        player = persistence.get_player(room_id, player_id)
        name = player["name"] if player else player_id
        try:
            emit("system_message", {"content": f"{name} 已离线。"}, to=room_id)
            emit("room_snapshot", room_manager.get_snapshot(room_id), to=room_id)
        except ValueError:
            pass

    return app, socketio


app, socketio = create_app()


if __name__ == "__main__":
    socketio.run(
        app,
        host=str(app.config.get("APP_HOST", "0.0.0.0")),
        port=int(app.config.get("APP_PORT", 8095)),
        debug=bool(app.config.get("DEBUG", True)),
        allow_unsafe_werkzeug=True,
    )