feat: add standalone GM platform for stat and achievement overrides

gm_platform.py

@@ -0,0 +1,573 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import secrets
+import sqlite3
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from flask import Flask, flash, redirect, render_template_string, request, session, url_for
+
+
+WIN_ENDING_STATES = {"escaped", "sacrifice", "pact", "victory", "success", "win", "custom_end"}
+LEVELS = {"bronze", "silver", "gold", "platinum", "diamond"}
+
+
+@dataclass
+class UserStats:
+    plays: int
+    wins: int
+    total_rounds: int
+    best_streak: int
+
+
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def to_int(value: Any, default: int = 0, minimum: int = 0) -> int:
+    try:
+        out = int(value)
+    except (TypeError, ValueError):
+        out = default
+    return max(minimum, out)
+
+
+def normalize_level(value: Any) -> str:
+    text = str(value or "").strip().lower()
+    alias = {
+        "青铜": "bronze",
+        "白银": "silver",
+        "黄金": "gold",
+        "铂金": "platinum",
+        "钻石": "diamond",
+    }
+    text = alias.get(text, text)
+    if text in LEVELS:
+        return text
+    return "bronze"
+
+
+def ensure_schema(conn: sqlite3.Connection) -> None:
+    conn.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS gm_user_stat_overrides (
+            user_id TEXT PRIMARY KEY,
+            played_games INTEGER NOT NULL DEFAULT 0,
+            wins INTEGER NOT NULL DEFAULT 0,
+            total_rounds INTEGER NOT NULL DEFAULT 0,
+            best_streak INTEGER NOT NULL DEFAULT 0,
+            updated_at TEXT NOT NULL
+        );
+
+        CREATE TABLE IF NOT EXISTS gm_user_achievement_overrides (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id TEXT NOT NULL,
+            title TEXT NOT NULL,
+            name TEXT NOT NULL DEFAULT '',
+            description TEXT NOT NULL DEFAULT '',
+            level TEXT NOT NULL DEFAULT 'bronze',
+            source TEXT NOT NULL DEFAULT 'gm',
+            card_id TEXT NOT NULL DEFAULT '',
+            card_title TEXT NOT NULL DEFAULT '',
+            unlocked_at TEXT NOT NULL DEFAULT '',
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_gm_achievement_user
+        ON gm_user_achievement_overrides(user_id, unlocked_at DESC);
+        """
+    )
+    conn.commit()
+
+
+def create_app(db_path: str, gm_password: str) -> Flask:
+    app = Flask(__name__)
+    app.secret_key = os.environ.get("GM_SECRET_KEY", secrets.token_hex(16))
+    app.config["DB_PATH"] = db_path
+    app.config["GM_PASSWORD"] = gm_password
+
+    def connect() -> sqlite3.Connection:
+        conn = sqlite3.connect(app.config["DB_PATH"], check_same_thread=False)
+        conn.row_factory = sqlite3.Row
+        ensure_schema(conn)
+        return conn
+
+    def is_logged_in() -> bool:
+        return bool(session.get("gm_authed"))
+
+    def require_login() -> bool:
+        if is_logged_in():
+            return True
+        return False
+
+    def load_user(user_id: str) -> dict[str, Any] | None:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return None
+        with connect() as conn:
+            row = conn.execute(
+                """
+                SELECT user_id, username, display_name, role, updated_at
+                FROM users
+                WHERE user_id=?
+                """,
+                (safe_user_id,),
+            ).fetchone()
+            return dict(row) if row else None
+
+    def compute_base_stats(user_id: str) -> UserStats:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return UserStats(0, 0, 0, 0)
+        with connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT r.game_state, r.updated_at
+                FROM rooms r
+                JOIN players p ON p.room_id = r.room_id
+                WHERE p.user_id=? AND r.status='ended'
+                GROUP BY r.room_id
+                ORDER BY r.updated_at ASC
+                """,
+                (safe_user_id,),
+            ).fetchall()
+        plays = 0
+        wins = 0
+        total_rounds = 0
+        current_streak = 0
+        best_streak = 0
+        for row in rows:
+            item = dict(row)
+            state_raw = item.get("game_state")
+            try:
+                state = json.loads(state_raw) if isinstance(state_raw, str) else {}
+            except json.JSONDecodeError:
+                state = {}
+            if not isinstance(state, dict):
+                state = {}
+            plays += 1
+            ending_state = str(state.get("ending_state", "")).strip().lower()
+            is_win = ending_state in WIN_ENDING_STATES
+            if is_win:
+                wins += 1
+                current_streak += 1
+                best_streak = max(best_streak, current_streak)
+            else:
+                current_streak = 0
+            total_rounds += max(0, to_int(state.get("round_no", 0)))
+        return UserStats(plays, wins, total_rounds, best_streak)
+
+    def load_stat_override(user_id: str) -> UserStats:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return UserStats(0, 0, 0, 0)
+        with connect() as conn:
+            row = conn.execute(
+                """
+                SELECT played_games, wins, total_rounds, best_streak
+                FROM gm_user_stat_overrides
+                WHERE user_id=?
+                """,
+                (safe_user_id,),
+            ).fetchone()
+            if not row:
+                return UserStats(0, 0, 0, 0)
+            item = dict(row)
+            return UserStats(
+                plays=max(0, to_int(item.get("played_games", 0))),
+                wins=max(0, to_int(item.get("wins", 0))),
+                total_rounds=max(0, to_int(item.get("total_rounds", 0))),
+                best_streak=max(0, to_int(item.get("best_streak", 0))),
+            )
+
+    def load_achievement_overrides(user_id: str) -> list[dict[str, Any]]:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return []
+        with connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT id, title, name, description, level, source, card_title, unlocked_at, updated_at
+                FROM gm_user_achievement_overrides
+                WHERE user_id=?
+                ORDER BY unlocked_at DESC, id DESC
+                LIMIT 200
+                """,
+                (safe_user_id,),
+            ).fetchall()
+            return [dict(row) for row in rows]
+
+    @app.get("/login")
+    def login_page() -> str:
+        if is_logged_in():
+            return redirect(url_for("home"))
+        return render_template_string(
+            """
+            <!doctype html>
+            <html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
+            <title>GM 平台登录</title>
+            <style>
+              body { font-family: -apple-system, BlinkMacSystemFont, "PingFang SC", "Microsoft YaHei", sans-serif; margin: 0; background: #f5f7fb; color: #1f2a44; }
+              .card { width: min(460px, calc(100% - 2rem)); margin: 5rem auto; background: #fff; border: 1px solid #d8e0f1; border-radius: 14px; padding: 1.2rem; }
+              input, button { width: 100%; padding: .65rem .75rem; border-radius: 10px; border: 1px solid #c8d3ea; margin-top: .5rem; }
+              button { background: #2f6bff; color: #fff; border: none; font-weight: 700; cursor: pointer; }
+              .flash { color: #bf1e48; margin-top: .6rem; }
+            </style></head><body>
+              <main class="card">
+                <h1>GM 平台</h1>
+                <p>输入 GM 密码后可修改战绩和成就。</p>
+                <form method="post" action="{{ url_for('login_action') }}">
+                  <label for="password">密码</label>
+                  <input id="password" name="password" type="password" autofocus required>
+                  <button type="submit">登录</button>
+                </form>
+                {% with messages = get_flashed_messages() %}
+                  {% if messages %}<div class="flash">{{ messages[0] }}</div>{% endif %}
+                {% endwith %}
+              </main>
+            </body></html>
+            """
+        )
+
+    @app.post("/login")
+    def login_action() -> Any:
+        password = str(request.form.get("password", ""))
+        if password != app.config["GM_PASSWORD"]:
+            flash("密码错误")
+            return redirect(url_for("login_page"))
+        session["gm_authed"] = True
+        return redirect(url_for("home"))
+
+    @app.get("/logout")
+    def logout_action() -> Any:
+        session.pop("gm_authed", None)
+        return redirect(url_for("login_page"))
+
+    @app.get("/")
+    def home() -> Any:
+        if not require_login():
+            return redirect(url_for("login_page"))
+        q = str(request.args.get("q", "")).strip()
+        with connect() as conn:
+            if q:
+                like = f"%{q}%"
+                rows = conn.execute(
+                    """
+                    SELECT user_id, username, display_name, role, updated_at
+                    FROM users
+                    WHERE username LIKE ? OR display_name LIKE ?
+                    ORDER BY updated_at DESC
+                    LIMIT 120
+                    """,
+                    (like, like),
+                ).fetchall()
+            else:
+                rows = conn.execute(
+                    """
+                    SELECT user_id, username, display_name, role, updated_at
+                    FROM users
+                    ORDER BY updated_at DESC
+                    LIMIT 80
+                    """
+                ).fetchall()
+        users = [dict(row) for row in rows]
+        return render_template_string(
+            """
+            <!doctype html>
+            <html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
+            <title>GM 平台</title>
+            <style>
+              body { font-family: -apple-system, BlinkMacSystemFont, "PingFang SC", "Microsoft YaHei", sans-serif; margin: 0; background: #f4f7fd; color: #1f2a44; }
+              header { display: flex; justify-content: space-between; align-items: center; padding: 1rem; max-width: 1200px; margin: 0 auto; }
+              .wrap { max-width: 1200px; margin: 0 auto; padding: 0 1rem 1rem; }
+              .card { background: #fff; border: 1px solid #d8e0f1; border-radius: 14px; padding: 1rem; }
+              .search { display: flex; gap: .5rem; }
+              .search input { flex: 1; }
+              input, button { padding: .58rem .75rem; border-radius: 10px; border: 1px solid #c8d3ea; }
+              button { background: #2f6bff; color: #fff; border: none; font-weight: 700; cursor: pointer; }
+              ul { list-style: none; padding: 0; margin: .8rem 0 0; }
+              li { padding: .55rem; border-radius: 10px; border: 1px solid #e2e8f6; margin-bottom: .45rem; background: #f8faff; }
+              a { color: #2f6bff; text-decoration: none; font-weight: 700; }
+              .flash { margin-top: .6rem; color: #1e7e34; }
+            </style></head><body>
+              <header>
+                <h1>GM 平台</h1>
+                <a href="{{ url_for('logout_action') }}">退出</a>
+              </header>
+              <main class="wrap">
+                <section class="card">
+                  <form class="search" method="get" action="{{ url_for('home') }}">
+                    <input name="q" value="{{ q }}" placeholder="搜索用户名 / 显示名">
+                    <button type="submit">搜索</button>
+                  </form>
+                  {% with messages = get_flashed_messages() %}
+                    {% if messages %}<div class="flash">{{ messages[0] }}</div>{% endif %}
+                  {% endwith %}
+                  <ul>
+                    {% for u in users %}
+                    <li>
+                      <a href="{{ url_for('user_detail', user_id=u.user_id) }}">{{ u.username }}</a>
+                      · {{ u.display_name }} · {{ u.role }} · {{ u.updated_at }}
+                    </li>
+                    {% endfor %}
+                    {% if not users %}<li>没有找到用户</li>{% endif %}
+                  </ul>
+                </section>
+              </main>
+            </body></html>
+            """,
+            q=q,
+            users=users,
+        )
+
+    @app.get("/user/<user_id>")
+    def user_detail(user_id: str) -> Any:
+        if not require_login():
+            return redirect(url_for("login_page"))
+        user = load_user(user_id)
+        if not user:
+            flash("用户不存在")
+            return redirect(url_for("home"))
+
+        base = compute_base_stats(user_id)
+        override = load_stat_override(user_id)
+        combined = UserStats(
+            plays=base.plays + override.plays,
+            wins=base.wins + override.wins,
+            total_rounds=base.total_rounds + override.total_rounds,
+            best_streak=max(base.best_streak, override.best_streak),
+        )
+        achievements = load_achievement_overrides(user_id)
+
+        return render_template_string(
+            """
+            <!doctype html>
+            <html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
+            <title>GM 用户操作</title>
+            <style>
+              body { font-family: -apple-system, BlinkMacSystemFont, "PingFang SC", "Microsoft YaHei", sans-serif; margin: 0; background: #f4f7fd; color: #1f2a44; }
+              header, main { max-width: 1200px; margin: 0 auto; padding: 1rem; }
+              .grid { display: grid; grid-template-columns: 1.2fr 1fr; gap: .9rem; }
+              .card { background: #fff; border: 1px solid #d8e0f1; border-radius: 14px; padding: 1rem; }
+              input, select, textarea, button { width: 100%; padding: .58rem .75rem; border-radius: 10px; border: 1px solid #c8d3ea; margin-top: .45rem; }
+              textarea { min-height: 76px; resize: vertical; }
+              button { background: #2f6bff; color: #fff; border: none; font-weight: 700; cursor: pointer; }
+              .danger { background: #d2395c; }
+              ul { list-style: none; padding: 0; margin: .8rem 0 0; }
+              li { padding: .55rem; border-radius: 10px; border: 1px solid #e2e8f6; margin-bottom: .45rem; background: #f8faff; }
+              .row { display: grid; grid-template-columns: repeat(2, 1fr); gap: .6rem; }
+              .flash { margin-top: .6rem; color: #1e7e34; }
+              .muted { color: #61708f; font-size: .92rem; margin-top: .28rem; }
+              .inline { display: inline-flex; gap: .35rem; width: auto; align-items: center; }
+              .inline button { width: auto; margin-top: 0; padding: .4rem .55rem; }
+              .small { font-size: .88rem; color: #576988; }
+            </style></head><body>
+              <header>
+                <h1>GM 操作 · @{{ user.username }}</h1>
+                <div><a href="{{ url_for('home') }}">返回列表</a> · <a href="{{ url_for('logout_action') }}">退出</a></div>
+              </header>
+              <main class="grid">
+                <section class="card">
+                  <h2>战绩修改（目标值）</h2>
+                  <p class="muted">保存后会写入覆盖值：目标 - 原始。可直接改胜场、对局数、总回合、最高连胜。</p>
+                  <p class="small">当前原始：对局 {{ base.plays }} / 胜场 {{ base.wins }} / 回合 {{ base.total_rounds }} / 连胜 {{ base.best_streak }}</p>
+                  <p class="small">当前覆盖：对局 +{{ override.plays }} / 胜场 +{{ override.wins }} / 回合 +{{ override.total_rounds }} / 连胜下限 {{ override.best_streak }}</p>
+                  <p class="small">当前合计：对局 {{ combined.plays }} / 胜场 {{ combined.wins }} / 回合 {{ combined.total_rounds }} / 连胜 {{ combined.best_streak }}</p>
+
+                  <form method="post" action="{{ url_for('update_stats', user_id=user.user_id) }}">
+                    <div class="row">
+                      <div>
+                        <label for="target_plays">目标对局数</label>
+                        <input id="target_plays" name="target_plays" type="number" min="0" value="{{ combined.plays }}">
+                      </div>
+                      <div>
+                        <label for="target_wins">目标胜场</label>
+                        <input id="target_wins" name="target_wins" type="number" min="0" value="{{ combined.wins }}">
+                      </div>
+                      <div>
+                        <label for="target_rounds">目标总回合</label>
+                        <input id="target_rounds" name="target_rounds" type="number" min="0" value="{{ combined.total_rounds }}">
+                      </div>
+                      <div>
+                        <label for="target_best_streak">目标最高连胜</label>
+                        <input id="target_best_streak" name="target_best_streak" type="number" min="0" value="{{ combined.best_streak }}">
+                      </div>
+                    </div>
+                    <button type="submit">保存战绩目标</button>
+                  </form>
+                </section>
+
+                <section class="card">
+                  <h2>手动成就</h2>
+                  <p class="muted">可直接给该用户增加/删除成就，立即在个人主页显示。</p>
+                  <form method="post" action="{{ url_for('add_achievement', user_id=user.user_id) }}">
+                    <label for="title">成就标题</label>
+                    <input id="title" name="title" maxlength="48" required placeholder="例如：雾港总导演">
+                    <label for="name">成就名（可选）</label>
+                    <input id="name" name="name" maxlength="48" placeholder="例如：特别授勋">
+                    <label for="level">等级</label>
+                    <select id="level" name="level">
+                      <option value="bronze">bronze</option>
+                      <option value="silver">silver</option>
+                      <option value="gold">gold</option>
+                      <option value="platinum">platinum</option>
+                      <option value="diamond">diamond</option>
+                    </select>
+                    <label for="card_title">关联卡片名（可选）</label>
+                    <input id="card_title" name="card_title" maxlength="84">
+                    <label for="description">描述（可选）</label>
+                    <textarea id="description" name="description" maxlength="180"></textarea>
+                    <button type="submit">新增手动成就</button>
+                  </form>
+                </section>
+
+                <section class="card" style="grid-column: 1 / -1;">
+                  <h2>已发放手动成就</h2>
+                  {% with messages = get_flashed_messages() %}
+                    {% if messages %}<div class="flash">{{ messages[0] }}</div>{% endif %}
+                  {% endwith %}
+                  <ul>
+                    {% for a in achievements %}
+                    <li>
+                      <div><strong>{{ a.title }}</strong> · {{ a.level }} · {{ a.unlocked_at or '-' }}</div>
+                      <div class="small">{{ a.description }}</div>
+                      <div class="small">{{ a.card_title }}</div>
+                      <form class="inline" method="post" action="{{ url_for('delete_achievement', user_id=user.user_id, achievement_id=a.id) }}">
+                        <button type="submit" class="danger">删除</button>
+                      </form>
+                    </li>
+                    {% endfor %}
+                    {% if not achievements %}<li>暂无手动成就</li>{% endif %}
+                  </ul>
+                </section>
+              </main>
+            </body></html>
+            """,
+            user=user,
+            base=base,
+            override=override,
+            combined=combined,
+            achievements=achievements,
+        )
+
+    @app.post("/user/<user_id>/stats")
+    def update_stats(user_id: str) -> Any:
+        if not require_login():
+            return redirect(url_for("login_page"))
+        user = load_user(user_id)
+        if not user:
+            flash("用户不存在")
+            return redirect(url_for("home"))
+
+        base = compute_base_stats(user_id)
+        target_plays = to_int(request.form.get("target_plays", 0), minimum=0)
+        target_wins = to_int(request.form.get("target_wins", 0), minimum=0)
+        target_rounds = to_int(request.form.get("target_rounds", 0), minimum=0)
+        target_best_streak = to_int(request.form.get("target_best_streak", 0), minimum=0)
+
+        if target_wins > target_plays:
+            target_wins = target_plays
+
+        override_plays = max(0, target_plays - base.plays)
+        override_wins = max(0, target_wins - base.wins)
+        override_rounds = max(0, target_rounds - base.total_rounds)
+        override_best_streak = target_best_streak
+
+        with connect() as conn:
+            if override_plays == 0 and override_wins == 0 and override_rounds == 0 and override_best_streak == 0:
+                conn.execute("DELETE FROM gm_user_stat_overrides WHERE user_id=?", (user_id,))
+            else:
+                conn.execute(
+                    """
+                    INSERT INTO gm_user_stat_overrides(user_id, played_games, wins, total_rounds, best_streak, updated_at)
+                    VALUES (?, ?, ?, ?, ?, ?)
+                    ON CONFLICT(user_id) DO UPDATE SET
+                        played_games=excluded.played_games,
+                        wins=excluded.wins,
+                        total_rounds=excluded.total_rounds,
+                        best_streak=excluded.best_streak,
+                        updated_at=excluded.updated_at
+                    """,
+                    (user_id, override_plays, override_wins, override_rounds, override_best_streak, utc_now_iso()),
+                )
+            conn.commit()
+
+        flash("战绩覆盖已更新")
+        return redirect(url_for("user_detail", user_id=user_id))
+
+    @app.post("/user/<user_id>/achievement/add")
+    def add_achievement(user_id: str) -> Any:
+        if not require_login():
+            return redirect(url_for("login_page"))
+        user = load_user(user_id)
+        if not user:
+            flash("用户不存在")
+            return redirect(url_for("home"))
+
+        title = str(request.form.get("title", "")).strip()[:48]
+        if not title:
+            flash("成就标题不能为空")
+            return redirect(url_for("user_detail", user_id=user_id))
+        name = str(request.form.get("name", "")).strip()[:48] or title
+        description = str(request.form.get("description", "")).strip()[:180]
+        level = normalize_level(request.form.get("level", "bronze"))
+        card_title = str(request.form.get("card_title", "")).strip()[:84]
+
+        now = utc_now_iso()
+        with connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO gm_user_achievement_overrides(
+                    user_id, title, name, description, level, source, card_id, card_title, unlocked_at, created_at, updated_at
+                ) VALUES (?, ?, ?, ?, ?, 'gm', '', ?, ?, ?, ?)
+                """,
+                (user_id, title, name, description, level, card_title, now, now, now),
+            )
+            conn.commit()
+        flash("手动成就已添加")
+        return redirect(url_for("user_detail", user_id=user_id))
+
+    @app.post("/user/<user_id>/achievement/<int:achievement_id>/delete")
+    def delete_achievement(user_id: str, achievement_id: int) -> Any:
+        if not require_login():
+            return redirect(url_for("login_page"))
+        with connect() as conn:
+            conn.execute(
+                "DELETE FROM gm_user_achievement_overrides WHERE id=? AND user_id=?",
+                (int(achievement_id), user_id),
+            )
+            conn.commit()
+        flash("手动成就已删除")
+        return redirect(url_for("user_detail", user_id=user_id))
+
+    return app
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Echo Tavern GM 平台（独立可删除）")
+    parser.add_argument("--db", default="echo_tavern.db", help="SQLite 数据库路径")
+    parser.add_argument("--host", default="127.0.0.1", help="监听地址")
+    parser.add_argument("--port", type=int, default=8765, help="监听端口")
+    parser.add_argument(
+        "--password",
+        default=os.environ.get("GM_PASSWORD", "gm-admin-123456"),
+        help="GM 登录密码（也可用环境变量 GM_PASSWORD）",
+    )
+    args = parser.parse_args()
+
+    db_path = str(Path(args.db).expanduser().resolve())
+    app = create_app(db_path=db_path, gm_password=str(args.password))
+    print(f"[GM] DB: {db_path}")
+    print(f"[GM] URL: http://{args.host}:{args.port}")
+    print("[GM] 登录密码来自 --password 或环境变量 GM_PASSWORD")
+    app.run(host=args.host, port=int(args.port), debug=False)
+
+
+if __name__ == "__main__":
+    main()

services/persistence.py

@@ -183,6 +183,32 @@ class SQLitePersistence:
                     FOREIGN KEY(to_user_id) REFERENCES users(user_id),
                     FOREIGN KEY(room_id) REFERENCES rooms(room_id)
                 );
+
+                CREATE TABLE IF NOT EXISTS gm_user_stat_overrides (
+                    user_id TEXT PRIMARY KEY,
+                    played_games INTEGER NOT NULL DEFAULT 0,
+                    wins INTEGER NOT NULL DEFAULT 0,
+                    total_rounds INTEGER NOT NULL DEFAULT 0,
+                    best_streak INTEGER NOT NULL DEFAULT 0,
+                    updated_at TEXT NOT NULL,
+                    FOREIGN KEY(user_id) REFERENCES users(user_id)
+                );
+
+                CREATE TABLE IF NOT EXISTS gm_user_achievement_overrides (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id TEXT NOT NULL,
+                    title TEXT NOT NULL,
+                    name TEXT NOT NULL DEFAULT '',
+                    description TEXT NOT NULL DEFAULT '',
+                    level TEXT NOT NULL DEFAULT 'bronze',
+                    source TEXT NOT NULL DEFAULT 'gm',
+                    card_id TEXT NOT NULL DEFAULT '',
+                    card_title TEXT NOT NULL DEFAULT '',
+                    unlocked_at TEXT NOT NULL DEFAULT '',
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL,
+                    FOREIGN KEY(user_id) REFERENCES users(user_id)
+                );
                 """
             )
 
@@ -202,6 +228,9 @@ class SQLitePersistence:
             conn.execute(
                 "CREATE INDEX IF NOT EXISTS idx_players_user_id_room ON players(user_id, room_id)"
             )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_gm_achievement_user ON gm_user_achievement_overrides(user_id, unlocked_at DESC)"
+            )
 
             conn.commit()
             self._seed_default_cards(conn)
@@ -1266,6 +1295,37 @@ class SQLitePersistence:
             if updated_at and updated_at > entry["last_played_at"]:
                 entry["last_played_at"] = updated_at
 
+        for override in self.list_all_user_stat_overrides():
+            user_id = str(override.get("user_id", "")).strip()
+            if not user_id:
+                continue
+            entry = summary.get(user_id)
+            if entry is None:
+                user = self.get_user_by_id(user_id)
+                if user is None:
+                    continue
+                entry = {
+                    "user_id": user_id,
+                    "username": str(user.get("username", "")).strip(),
+                    "display_name": str(user.get("display_name", "")).strip(),
+                    "avatar_color": str(user.get("avatar_color", "#0070f2")).strip() or "#0070f2",
+                    "avatar_url": str(user.get("avatar_url", "")).strip(),
+                    "theme": str(user.get("theme", "dopamine")).strip() or "dopamine",
+                    "played_games": 0,
+                    "wins": 0,
+                    "total_rounds": 0,
+                    "last_played_at": "",
+                    "_seen_rooms": set(),
+                }
+                summary[user_id] = entry
+
+            entry["played_games"] += int(override.get("played_games", 0))
+            entry["wins"] += int(override.get("wins", 0))
+            entry["total_rounds"] += int(override.get("total_rounds", 0))
+            override_time = str(override.get("updated_at", "")).strip()
+            if override_time and override_time > entry["last_played_at"]:
+                entry["last_played_at"] = override_time
+
         results: list[dict[str, Any]] = []
         for entry in summary.values():
             entry.pop("_seen_rooms", None)
@@ -1326,6 +1386,225 @@ class SQLitePersistence:
             "last_played_at": str(row.get("last_played_at", "")).strip(),
         }
 
+    def get_user_stat_override(self, user_id: str) -> dict[str, Any]:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return {"played_games": 0, "wins": 0, "total_rounds": 0, "best_streak": 0, "updated_at": ""}
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                """
+                SELECT user_id, played_games, wins, total_rounds, best_streak, updated_at
+                FROM gm_user_stat_overrides
+                WHERE user_id=?
+                """,
+                (safe_user_id,),
+            ).fetchone()
+            if not row:
+                return {"played_games": 0, "wins": 0, "total_rounds": 0, "best_streak": 0, "updated_at": ""}
+            item = dict(row)
+            return {
+                "played_games": int(item.get("played_games", 0)),
+                "wins": int(item.get("wins", 0)),
+                "total_rounds": int(item.get("total_rounds", 0)),
+                "best_streak": int(item.get("best_streak", 0)),
+                "updated_at": str(item.get("updated_at", "")).strip(),
+            }
+
+    def list_all_user_stat_overrides(self) -> list[dict[str, Any]]:
+        with self._lock, self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT user_id, played_games, wins, total_rounds, best_streak, updated_at
+                FROM gm_user_stat_overrides
+                """
+            ).fetchall()
+            out: list[dict[str, Any]] = []
+            for row in rows:
+                item = dict(row)
+                out.append(
+                    {
+                        "user_id": str(item.get("user_id", "")).strip(),
+                        "played_games": int(item.get("played_games", 0)),
+                        "wins": int(item.get("wins", 0)),
+                        "total_rounds": int(item.get("total_rounds", 0)),
+                        "best_streak": int(item.get("best_streak", 0)),
+                        "updated_at": str(item.get("updated_at", "")).strip(),
+                    }
+                )
+            return out
+
+    def upsert_user_stat_override(
+        self,
+        user_id: str,
+        *,
+        played_games: int = 0,
+        wins: int = 0,
+        total_rounds: int = 0,
+        best_streak: int = 0,
+    ) -> dict[str, Any]:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            raise ValueError("user_not_found")
+        if self.get_user_by_id(safe_user_id) is None:
+            raise ValueError("user_not_found")
+
+        safe_played = max(0, int(played_games))
+        safe_wins = max(0, int(wins))
+        safe_rounds = max(0, int(total_rounds))
+        safe_best_streak = max(0, int(best_streak))
+        now = utc_now_iso()
+
+        with self._lock, self._connect() as conn:
+            if safe_played == 0 and safe_wins == 0 and safe_rounds == 0 and safe_best_streak == 0:
+                conn.execute("DELETE FROM gm_user_stat_overrides WHERE user_id=?", (safe_user_id,))
+                conn.commit()
+                return {"played_games": 0, "wins": 0, "total_rounds": 0, "best_streak": 0, "updated_at": ""}
+            conn.execute(
+                """
+                INSERT INTO gm_user_stat_overrides(user_id, played_games, wins, total_rounds, best_streak, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(user_id) DO UPDATE SET
+                    played_games=excluded.played_games,
+                    wins=excluded.wins,
+                    total_rounds=excluded.total_rounds,
+                    best_streak=excluded.best_streak,
+                    updated_at=excluded.updated_at
+                """,
+                (safe_user_id, safe_played, safe_wins, safe_rounds, safe_best_streak, now),
+            )
+            conn.commit()
+        return self.get_user_stat_override(safe_user_id)
+
+    def list_user_achievement_overrides(self, user_id: str, limit: int = 80) -> list[dict[str, Any]]:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return []
+        try:
+            safe_limit = max(1, min(300, int(limit)))
+        except (TypeError, ValueError):
+            safe_limit = 80
+        with self._lock, self._connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT id, user_id, title, name, description, level, source, card_id, card_title, unlocked_at, created_at, updated_at
+                FROM gm_user_achievement_overrides
+                WHERE user_id=?
+                ORDER BY unlocked_at DESC, id DESC
+                LIMIT ?
+                """,
+                (safe_user_id, safe_limit),
+            ).fetchall()
+            result: list[dict[str, Any]] = []
+            for row in rows:
+                item = dict(row)
+                level = self._normalize_achievement_level(item.get("level", "bronze"))
+                result.append(
+                    {
+                        "id": int(item.get("id", 0)),
+                        "user_id": safe_user_id,
+                        "title": str(item.get("title", "")).strip()[:48],
+                        "name": str(item.get("name", "")).strip()[:48],
+                        "description": str(item.get("description", "")).strip()[:180],
+                        "level": level,
+                        "level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
+                        "source": str(item.get("source", "gm")).strip()[:24] or "gm",
+                        "card_id": str(item.get("card_id", "")).strip()[:64],
+                        "card_title": str(item.get("card_title", "")).strip()[:84],
+                        "unlocked_at": str(item.get("unlocked_at", "")).strip(),
+                        "created_at": str(item.get("created_at", "")).strip(),
+                        "updated_at": str(item.get("updated_at", "")).strip(),
+                    }
+                )
+            return result
+
+    def create_user_achievement_override(
+        self,
+        user_id: str,
+        *,
+        title: str,
+        name: str = "",
+        description: str = "",
+        level: str = "bronze",
+        source: str = "gm",
+        card_id: str = "",
+        card_title: str = "",
+        unlocked_at: str = "",
+    ) -> dict[str, Any]:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id or self.get_user_by_id(safe_user_id) is None:
+            raise ValueError("user_not_found")
+        safe_title = str(title or "").strip()[:48]
+        if not safe_title:
+            raise ValueError("invalid_card_payload")
+        safe_name = str(name or "").strip()[:48] or safe_title
+        safe_desc = str(description or "").strip()[:180]
+        safe_level = self._normalize_achievement_level(level)
+        safe_source = str(source or "gm").strip()[:24] or "gm"
+        safe_card_id = str(card_id or "").strip()[:64]
+        safe_card_title = str(card_title or "").strip()[:84]
+        safe_unlocked_at = str(unlocked_at or "").strip()[:40] or utc_now_iso()
+        now = utc_now_iso()
+        with self._lock, self._connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO gm_user_achievement_overrides(
+                    user_id, title, name, description, level, source, card_id, card_title, unlocked_at, created_at, updated_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    safe_user_id,
+                    safe_title,
+                    safe_name,
+                    safe_desc,
+                    safe_level,
+                    safe_source,
+                    safe_card_id,
+                    safe_card_title,
+                    safe_unlocked_at,
+                    now,
+                    now,
+                ),
+            )
+            row_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
+            conn.commit()
+            row = conn.execute(
+                """
+                SELECT id, user_id, title, name, description, level, source, card_id, card_title, unlocked_at, created_at, updated_at
+                FROM gm_user_achievement_overrides
+                WHERE id=?
+                """,
+                (row_id,),
+            ).fetchone()
+            if not row:
+                raise ValueError("invalid_response")
+            item = dict(row)
+            safe_level = self._normalize_achievement_level(item.get("level", "bronze"))
+            item["level"] = safe_level
+            item["level_rank"] = int(ACHIEVEMENT_LEVEL_ORDER.get(safe_level, 1))
+            return item
+
+    def delete_user_achievement_override(self, user_id: str, achievement_id: int) -> bool:
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return False
+        try:
+            safe_id = int(achievement_id)
+        except (TypeError, ValueError):
+            return False
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                "SELECT id FROM gm_user_achievement_overrides WHERE id=? AND user_id=?",
+                (safe_id, safe_user_id),
+            ).fetchone()
+            if not row:
+                return False
+            conn.execute(
+                "DELETE FROM gm_user_achievement_overrides WHERE id=? AND user_id=?",
+                (safe_id, safe_user_id),
+            )
+            conn.commit()
+            return True
+
     def _normalize_achievement_level(self, raw_level: Any) -> str:
         level = str(raw_level or "").strip().lower()
         alias = {
@@ -1476,7 +1755,24 @@ class SQLitePersistence:
         return games
 
     def get_user_achievement_bundle(self, user_id: str) -> dict[str, Any]:
-        games = self.list_user_completed_games(user_id)
+        safe_user_id = str(user_id or "").strip()
+        if not safe_user_id:
+            return {
+                "summary": {
+                    "total_unlocked": 0,
+                    "system_unlocked": 0,
+                    "card_unlocked": 0,
+                    "gm_unlocked": 0,
+                    "highest_level": "",
+                    "completed_games": 0,
+                    "wins": 0,
+                    "best_streak": 0,
+                },
+                "items": [],
+            }
+
+        games = self.list_user_completed_games(safe_user_id)
+        stat_override = self.get_user_stat_override(safe_user_id)
         plays = 0
         wins = 0
         current_streak = 0
@@ -1530,6 +1826,45 @@ class SQLitePersistence:
                         }
                     )
 
+        final_metrics = {
+            "plays": max(0, plays + int(stat_override.get("played_games", 0))),
+            "wins": max(0, wins + int(stat_override.get("wins", 0))),
+            "current_streak": max(current_streak, int(stat_override.get("best_streak", 0))),
+            "best_streak": max(best_streak, int(stat_override.get("best_streak", 0))),
+        }
+        synthetic_unlock_time = str(stat_override.get("updated_at", "")).strip() or utc_now_iso()
+        for track in SYSTEM_ACHIEVEMENT_TRACKS:
+            track_id = str(track.get("id", "")).strip()
+            metric_key = self._normalize_achievement_metric(track.get("metric", "plays"))
+            metric_value = int(final_metrics.get(metric_key, 0))
+            for tier in track.get("tiers", []):
+                level = self._normalize_achievement_level(tier.get("level", "bronze"))
+                try:
+                    threshold = max(1, int(tier.get("threshold", 1)))
+                except (TypeError, ValueError):
+                    threshold = 1
+                item_key = f"{track_id}:{level}:{threshold}"
+                if item_key in system_seen:
+                    continue
+                if metric_value < threshold:
+                    continue
+                system_seen.add(item_key)
+                unlocked_system.append(
+                    {
+                        "source": "system",
+                        "track_id": track_id,
+                        "title": str(tier.get("title", "")).strip()[:48] or str(track.get("name", "")).strip(),
+                        "name": str(track.get("name", "")).strip()[:48] or "系统成就",
+                        "description": str(track.get("description", "")).strip()[:180],
+                        "level": level,
+                        "level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
+                        "threshold": threshold,
+                        "metric": metric_key,
+                        "metric_value": metric_value,
+                        "unlocked_at": synthetic_unlock_time,
+                    }
+                )
+
         card_games: dict[str, list[dict[str, Any]]] = {}
         for game in games:
             card_id = game["game_card_id"]
@@ -1602,7 +1937,29 @@ class SQLitePersistence:
                             }
                         )
 
-        all_items = unlocked_system + unlocked_card
+        unlocked_gm: list[dict[str, Any]] = []
+        for item in self.list_user_achievement_overrides(safe_user_id, limit=200):
+            level = self._normalize_achievement_level(item.get("level", "bronze"))
+            unlocked_gm.append(
+                {
+                    "source": str(item.get("source", "gm")).strip()[:24] or "gm",
+                    "track_id": str(item.get("id", "")),
+                    "title": str(item.get("title", "")).strip()[:48],
+                    "name": str(item.get("name", "")).strip()[:48] or str(item.get("title", "")).strip()[:48],
+                    "description": str(item.get("description", "")).strip()[:180],
+                    "level": level,
+                    "level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
+                    "threshold": 0,
+                    "metric": "manual",
+                    "metric_value": 0,
+                    "unlocked_at": str(item.get("unlocked_at", "")).strip(),
+                    "card_id": str(item.get("card_id", "")).strip(),
+                    "card_title": str(item.get("card_title", "")).strip(),
+                    "gm_id": int(item.get("id", 0)),
+                }
+            )
+
+        all_items = unlocked_system + unlocked_card + unlocked_gm
         all_items.sort(key=lambda row: (str(row.get("unlocked_at", "")), int(row.get("level_rank", 0))), reverse=True)
         highest = "bronze"
         highest_rank = 0
@@ -1617,10 +1974,11 @@ class SQLitePersistence:
                 "total_unlocked": len(all_items),
                 "system_unlocked": len(unlocked_system),
                 "card_unlocked": len(unlocked_card),
+                "gm_unlocked": len(unlocked_gm),
                 "highest_level": highest if highest_rank > 0 else "",
-                "completed_games": plays,
-                "wins": wins,
-                "best_streak": best_streak,
+                "completed_games": int(final_metrics["plays"]),
+                "wins": int(final_metrics["wins"]),
+                "best_streak": int(final_metrics["best_streak"]),
             },
             "items": all_items[:120],
         }