diff --git a/gm_platform.py b/gm_platform.py new file mode 100644 index 0000000..1a9ff0b --- /dev/null +++ b/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( + """ + + + GM 平台登录 + +
+

GM 平台

+

输入 GM 密码后可修改战绩和成就。

+
+ + + +
+ {% with messages = get_flashed_messages() %} + {% if messages %}
{{ messages[0] }}
{% endif %} + {% endwith %} +
+ + """ + ) + + @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( + """ + + + GM 平台 + +
+

GM 平台

+ 退出 +
+
+
+
+ + +
+ {% with messages = get_flashed_messages() %} + {% if messages %}
{{ messages[0] }}
{% endif %} + {% endwith %} + +
+
+ + """, + q=q, + users=users, + ) + + @app.get("/user/") + 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( + """ + + + GM 用户操作 + +
+

GM 操作 · @{{ user.username }}

+
返回列表 · 退出
+
+
+
+

战绩修改（目标值）

+

保存后会写入覆盖值：目标 - 原始。可直接改胜场、对局数、总回合、最高连胜。

+

当前原始：对局 {{ base.plays }} / 胜场 {{ base.wins }} / 回合 {{ base.total_rounds }} / 连胜 {{ base.best_streak }}

+

当前覆盖：对局 +{{ override.plays }} / 胜场 +{{ override.wins }} / 回合 +{{ override.total_rounds }} / 连胜下限 {{ override.best_streak }}

+

当前合计：对局 {{ combined.plays }} / 胜场 {{ combined.wins }} / 回合 {{ combined.total_rounds }} / 连胜 {{ combined.best_streak }}

+ +
+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+ +
+
+ +
+

手动成就

+

可直接给该用户增加/删除成就，立即在个人主页显示。

+
+ + + + + + + + + + + +
+
+ +
+

已发放手动成就

+ {% with messages = get_flashed_messages() %} + {% if messages %}
{{ messages[0] }}
{% endif %} + {% endwith %} +
    + {% for a in achievements %} +
  • +
    {{ a.title }} · {{ a.level }} · {{ a.unlocked_at or '-' }}
    +
    {{ a.description }}
    +
    {{ a.card_title }}
    +
    + +
    +
    • + {% endfor %} + {% if not achievements %}
  • 暂无手动成就
    • {% endif %} +
+
+
+ + """, + user=user, + base=base, + override=override, + combined=combined, + achievements=achievements, + ) + + @app.post("/user//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//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//achievement//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() diff --git a/services/persistence.py b/services/persistence.py index 6562cbb..2c71fa6 100644 --- a/services/persistence.py +++ b/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], }