1247 lines
53 KiB
Python
1247 lines
53 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import threading
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from game.models import utc_now_iso
|
|
|
|
|
|
DEFAULT_CARD_ID = "fogharbor"
|
|
|
|
|
|
class SQLitePersistence:
|
|
def __init__(self, db_path: str):
|
|
self.db_path = str(Path(db_path))
|
|
self._lock = threading.RLock()
|
|
self.init_schema()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(self.db_path, check_same_thread=False)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def _ensure_column(self, conn: sqlite3.Connection, table: str, column: str, ddl: str) -> None:
|
|
rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
|
|
columns = {row["name"] for row in rows}
|
|
if column not in columns:
|
|
conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")
|
|
|
|
def init_schema(self) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
user_id TEXT PRIMARY KEY,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
display_name TEXT NOT NULL,
|
|
role TEXT NOT NULL DEFAULT 'player',
|
|
avatar_color TEXT NOT NULL DEFAULT '#0070f2',
|
|
avatar_url TEXT NOT NULL DEFAULT '',
|
|
theme TEXT NOT NULL DEFAULT 'dopamine',
|
|
bio TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS game_cards (
|
|
card_id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
subtitle TEXT NOT NULL DEFAULT '',
|
|
summary TEXT NOT NULL,
|
|
cover_url TEXT NOT NULL,
|
|
opening_text TEXT NOT NULL,
|
|
starter_goals TEXT NOT NULL,
|
|
state_schema TEXT NOT NULL DEFAULT '{}',
|
|
content_pack TEXT NOT NULL DEFAULT '{}',
|
|
creator_user_id TEXT,
|
|
is_official INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS rooms (
|
|
room_id TEXT PRIMARY KEY,
|
|
host_player_id TEXT NOT NULL,
|
|
is_public INTEGER NOT NULL DEFAULT 0,
|
|
status TEXT NOT NULL,
|
|
chapter INTEGER NOT NULL,
|
|
scene_id TEXT NOT NULL,
|
|
game_state TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS players (
|
|
player_id TEXT PRIMARY KEY,
|
|
room_id TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
is_host INTEGER NOT NULL DEFAULT 0,
|
|
is_ready INTEGER NOT NULL DEFAULT 0,
|
|
is_online INTEGER NOT NULL DEFAULT 1,
|
|
joined_at TEXT NOT NULL,
|
|
last_seen TEXT NOT NULL,
|
|
FOREIGN KEY(room_id) REFERENCES rooms(room_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
room_id TEXT NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
player_id TEXT,
|
|
visibility TEXT NOT NULL DEFAULT 'public',
|
|
content TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY(room_id) REFERENCES rooms(room_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS actions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
room_id TEXT NOT NULL,
|
|
round_no INTEGER NOT NULL,
|
|
player_id TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY(room_id) REFERENCES rooms(room_id)
|
|
);
|
|
"""
|
|
)
|
|
|
|
self._ensure_column(conn, "players", "user_id", "TEXT")
|
|
self._ensure_column(conn, "rooms", "game_card_id", f"TEXT NOT NULL DEFAULT '{DEFAULT_CARD_ID}'")
|
|
self._ensure_column(conn, "rooms", "is_public", "INTEGER NOT NULL DEFAULT 0")
|
|
self._ensure_column(conn, "game_cards", "state_schema", "TEXT NOT NULL DEFAULT '{}'")
|
|
self._ensure_column(conn, "game_cards", "content_pack", "TEXT NOT NULL DEFAULT '{}'")
|
|
self._ensure_column(conn, "users", "role", "TEXT NOT NULL DEFAULT 'player'")
|
|
self._ensure_column(conn, "users", "avatar_url", "TEXT NOT NULL DEFAULT ''")
|
|
self._ensure_column(conn, "users", "theme", "TEXT NOT NULL DEFAULT 'dopamine'")
|
|
|
|
conn.commit()
|
|
self._seed_default_cards(conn)
|
|
conn.commit()
|
|
|
|
def _seed_default_cards(self, conn: sqlite3.Connection) -> None:
|
|
now = utc_now_iso()
|
|
cards = [
|
|
{
|
|
"card_id": "fogharbor",
|
|
"title": "雾港旅店",
|
|
"subtitle": "合作推理 / 诡异悬疑",
|
|
"summary": "暴风雨夜，一间旅店困住了所有人。你们要在规则与谎言中找出真相。",
|
|
"cover_url": "/static/img/cards/fogharbor.svg",
|
|
"opening_text": "暴风雨夜，你们推开雾港旅店大门，空气里有潮湿铁锈味。",
|
|
"starter_goals": ["先查看大厅可疑区域", "与旅店老板交谈", "确定队伍分工"],
|
|
"state_schema": {
|
|
"metrics": [
|
|
{"key": "team_stability", "label": "队伍稳定度", "min": 0, "max": 100, "initial": 68, "color": "#ff8c42"},
|
|
{"key": "truth_progress", "label": "真相推进", "min": 0, "max": 100, "initial": 5, "color": "#4b7cff"},
|
|
{"key": "night_rule_pressure", "label": "夜规压迫", "min": 0, "max": 100, "initial": 20, "color": "#8b5cf6"},
|
|
],
|
|
"labels": {
|
|
"inventory": "个人携行物",
|
|
"clues": "线索墙",
|
|
"choices": "可执行指令",
|
|
"scene": "当前区域",
|
|
},
|
|
"scene_labels": {
|
|
"lobby": "大厅",
|
|
"dining_room": "餐厅",
|
|
"corridor": "走廊",
|
|
"guest_room": "客房",
|
|
"basement": "地下室",
|
|
},
|
|
"data_scope": {
|
|
"metrics": "per_player",
|
|
"inventory": "per_player",
|
|
"clues": "shared",
|
|
},
|
|
"visibility": {
|
|
"show_all_player_metrics": True,
|
|
"show_all_player_inventory": True,
|
|
},
|
|
"ui": {
|
|
"room_background_url": "/static/img/cards/fogharbor.svg",
|
|
"hero_status": {
|
|
"key": "night_rule_pressure",
|
|
"label": "夜规压迫",
|
|
"min": 0,
|
|
"max": 100,
|
|
"color": "#8b5cf6",
|
|
},
|
|
"status_tracks": [
|
|
{"key": "night_rule_pressure", "label": "夜规压迫", "min": 0, "max": 100, "color": "#8b5cf6"},
|
|
{"key": "truth_progress", "label": "真相推进", "min": 0, "max": 100, "color": "#4b7cff"},
|
|
{"key": "team_stability", "label": "队伍稳定度", "min": 0, "max": 100, "color": "#ff8c42"},
|
|
],
|
|
},
|
|
},
|
|
"content_pack": {
|
|
"genre_tags": ["悬疑", "诡异", "合作推理"],
|
|
"feature_badges": ["多结局", "NPC交锋", "规则怪谈"],
|
|
"world_lore": "雾港旅店坐落在暴雨断航的旧码头，每逢风暴夜就会重复一次失踪事件。",
|
|
"core_conflict": "玩家要在旅店规则与NPC谎言中找出失踪者去向，并决定是否牺牲某人换取逃离。",
|
|
"player_roles": ["调查者", "谈判者", "记录者", "潜行者"],
|
|
"interaction_modes": ["合作调查", "分头行动", "公开投票", "私下交易"],
|
|
"system_rules": [
|
|
"每轮至少提交一条调查或交涉行动",
|
|
"危险值越高，NPC越不可信",
|
|
"公开讨论可降低误判，但会抬升紧张度",
|
|
],
|
|
"npc_roster": [
|
|
"旅店老板：掌握旧账本，言辞含糊",
|
|
"女仆：知道地下室入口，害怕被追责",
|
|
"沉默旅客：与失踪者有关，但拒绝明说",
|
|
],
|
|
"scene_blueprint": [
|
|
"大厅：建立怀疑对象与分工",
|
|
"餐厅：线索交叉验证",
|
|
"走廊/客房：揭示隐藏关系",
|
|
"地下室：结局抉择",
|
|
],
|
|
"ending_catalog": ["成功逃离", "真相伴随牺牲", "被旅店困住", "与隐藏势力交易"],
|
|
"hidden_truths": ["失踪者并非单独行动", "旅店规则有人为操控", "地下室仪式可被篡改"],
|
|
"systems": {
|
|
"investigation": {"enabled": True, "suggestion": "多人交叉验证提高真相推进"},
|
|
"social": {"enabled": True, "trust_field": "team_stability"},
|
|
},
|
|
"end_conditions": [
|
|
{
|
|
"id": "escape_success",
|
|
"metric": "truth_progress",
|
|
"op": ">=",
|
|
"value": 88,
|
|
"flag": "escape_route_open",
|
|
"flag_equals": True,
|
|
"ending_state": "escaped",
|
|
"ending_summary": "你们破解了旅店夜规并打开逃生路线，成功离开雾港。",
|
|
},
|
|
{
|
|
"id": "night_consumed",
|
|
"metric": "night_rule_pressure",
|
|
"op": ">=",
|
|
"value": 100,
|
|
"ending_state": "trapped",
|
|
"ending_summary": "夜规彻底接管旅店，队伍被困在永夜轮回里。",
|
|
},
|
|
],
|
|
},
|
|
},
|
|
{
|
|
"card_id": "sunken_arcade",
|
|
"title": "沉没游乐城",
|
|
"subtitle": "探索冒险 / 机关解谜",
|
|
"summary": "被海水吞没的游乐城仍在夜里通电，机械与记忆一起复苏。",
|
|
"cover_url": "/static/img/cards/sunken_arcade.svg",
|
|
"opening_text": "霓虹在海雾中一闪一灭，沉没游乐城的闸门缓缓开启。",
|
|
"starter_goals": ["寻找主控电源", "确认出口是否畅通", "排查游乐设施异常"],
|
|
"state_schema": {
|
|
"metrics": [
|
|
{"key": "pressure", "label": "水压警戒", "min": 0, "max": 120, "initial": 24, "color": "#26b7ff"},
|
|
{"key": "power_grid", "label": "电网稳定", "min": 0, "max": 100, "initial": 40, "color": "#7affb3"},
|
|
{"key": "team_oxygen", "label": "队伍氧量", "min": 0, "max": 100, "initial": 85, "color": "#ffd166"},
|
|
],
|
|
"labels": {
|
|
"inventory": "潜行装备",
|
|
"clues": "异常记录",
|
|
"choices": "可执行指令",
|
|
"scene": "当前区段",
|
|
},
|
|
"data_scope": {
|
|
"metrics": "per_player",
|
|
"inventory": "shared",
|
|
"clues": "shared",
|
|
},
|
|
"visibility": {
|
|
"show_all_player_metrics": True,
|
|
"show_all_player_inventory": True,
|
|
},
|
|
"ui": {
|
|
"room_background_url": "/static/img/cards/sunken_arcade.svg",
|
|
"hero_status": {
|
|
"key": "pressure",
|
|
"label": "水压警戒",
|
|
"min": 0,
|
|
"max": 120,
|
|
"color": "#26b7ff",
|
|
},
|
|
"status_tracks": [
|
|
{"key": "pressure", "label": "水压警戒", "min": 0, "max": 120, "color": "#26b7ff"},
|
|
{"key": "power_grid", "label": "电网稳定", "min": 0, "max": 100, "color": "#7affb3"},
|
|
{"key": "team_oxygen", "label": "队伍氧量", "min": 0, "max": 100, "color": "#ffd166"},
|
|
],
|
|
},
|
|
},
|
|
"content_pack": {
|
|
"genre_tags": ["科幻", "探索", "解谜"],
|
|
"feature_badges": ["资源管理", "机关联动", "团队协作"],
|
|
"world_lore": "沉没游乐城位于海沟边缘，夜晚会短暂恢复供电并广播失落游客名单。",
|
|
"core_conflict": "玩家必须在水压持续上升前修复主控系统，决定牺牲供电区还是冒险突破。",
|
|
"player_roles": ["工程师", "侦查员", "安全员"],
|
|
"interaction_modes": ["区域分队", "同步解谜", "风险分摊"],
|
|
"system_rules": [
|
|
"每轮需声明要优先处理的设施",
|
|
"电网稳定度过低会关闭部分选项",
|
|
"冒险行动收益高但会显著增加水压警戒",
|
|
],
|
|
"npc_roster": [
|
|
"售票机器人：可提供旧地图但需要通电",
|
|
"失忆保安：记得撤离路线碎片",
|
|
"海沟回声：会伪装成队友语音",
|
|
],
|
|
"scene_blueprint": ["入口检修区", "霓虹主街", "维护中枢", "深层观景廊", "总控塔"],
|
|
"ending_catalog": ["全员撤离", "部分区段沉没", "系统失控永久封锁"],
|
|
"hidden_truths": ["事故由内部测试引发", "主控塔有备用逃生通道"],
|
|
"systems": {
|
|
"hazard": {"enabled": True, "primary_metric": "pressure"},
|
|
"resource": {"enabled": True, "shared_inventory": True},
|
|
},
|
|
"end_conditions": [
|
|
{
|
|
"id": "arcade_escape",
|
|
"metric": "power_grid",
|
|
"op": ">=",
|
|
"value": 85,
|
|
"flag": "escape_route_open",
|
|
"flag_equals": True,
|
|
"ending_state": "escaped",
|
|
"ending_summary": "你们重启了主控网络并打开泄压门，全员脱离沉没游乐城。",
|
|
},
|
|
{
|
|
"id": "oxygen_exhausted",
|
|
"metric": "team_oxygen",
|
|
"op": "<=",
|
|
"value": 0,
|
|
"ending_state": "trapped",
|
|
"ending_summary": "氧量耗尽，队伍被困在黑暗海沟中。",
|
|
},
|
|
],
|
|
},
|
|
},
|
|
{
|
|
"card_id": "skyrail_omega",
|
|
"title": "天轨终点站",
|
|
"subtitle": "科幻悬疑 / 团队抉择",
|
|
"summary": "无人列车驶向失落终点，车厢里的每个人都藏着不同目的。",
|
|
"cover_url": "/static/img/cards/skyrail_omega.svg",
|
|
"opening_text": "列车广播响起：下一站，终点站。窗外却只有无尽云海。",
|
|
"starter_goals": ["检查列车广播来源", "盘问可疑乘客", "确认车头控制权"],
|
|
"state_schema": {
|
|
"metrics": [
|
|
{"key": "suspicion", "label": "全列车怀疑度", "min": 0, "max": 100, "initial": 30, "color": "#ff6a6a"},
|
|
{"key": "signal_integrity", "label": "信号完整度", "min": 0, "max": 100, "initial": 55, "color": "#4d84ff"},
|
|
{"key": "authority", "label": "队伍控制权", "min": 0, "max": 100, "initial": 35, "color": "#7affb3"},
|
|
],
|
|
"labels": {
|
|
"inventory": "随身模块",
|
|
"clues": "乘客情报",
|
|
"choices": "可选路线",
|
|
"scene": "当前车厢",
|
|
},
|
|
"data_scope": {
|
|
"metrics": "shared",
|
|
"inventory": "per_player",
|
|
"clues": "shared",
|
|
},
|
|
"visibility": {
|
|
"show_all_player_metrics": True,
|
|
"show_all_player_inventory": True,
|
|
},
|
|
"ui": {
|
|
"room_background_url": "/static/img/cards/skyrail_omega.svg",
|
|
"hero_status": {
|
|
"key": "signal_integrity",
|
|
"label": "信号完整度",
|
|
"min": 0,
|
|
"max": 100,
|
|
"color": "#4d84ff",
|
|
},
|
|
"status_tracks": [
|
|
{"key": "signal_integrity", "label": "信号完整度", "min": 0, "max": 100, "color": "#4d84ff"},
|
|
{"key": "authority", "label": "队伍控制权", "min": 0, "max": 100, "color": "#7affb3"},
|
|
{"key": "suspicion", "label": "全列车怀疑度", "min": 0, "max": 100, "color": "#ff6a6a"},
|
|
],
|
|
},
|
|
},
|
|
"content_pack": {
|
|
"genre_tags": ["悬疑", "科幻", "心理博弈"],
|
|
"feature_badges": ["信任系统", "隐私事件", "阵营抉择"],
|
|
"world_lore": "天轨列车在云海上循环行驶，乘客档案会在每次广播后被篡改。",
|
|
"core_conflict": "玩家要查明列车终点的真实含义，并在互不信任的乘客间维持秩序。",
|
|
"player_roles": ["安保官", "情报员", "调度员", "医生"],
|
|
"interaction_modes": ["信息交换", "临时联盟", "公开审讯"],
|
|
"system_rules": [
|
|
"怀疑度过高会触发乘客暴动",
|
|
"信号完整度决定可访问车厢范围",
|
|
"隐私情报可影响公开投票结果",
|
|
],
|
|
"npc_roster": [
|
|
"乘务长：强行维持秩序",
|
|
"失忆科学家：掌握终点数据",
|
|
"匿名旅客：不断制造误导",
|
|
],
|
|
"scene_blueprint": ["前段车厢", "中段控制桥", "封闭实验舱", "终点前置区"],
|
|
"ending_catalog": ["重置列车并返航", "牺牲车厢换取生存", "接受终点协议"],
|
|
"hidden_truths": ["终点站并不存在实体站台", "广播系统受外部势力操纵"],
|
|
"systems": {
|
|
"trust": {"enabled": True, "metric": "suspicion"},
|
|
"command": {"enabled": True, "metric": "authority"},
|
|
},
|
|
"end_conditions": [
|
|
{
|
|
"id": "retake_control",
|
|
"metric": "authority",
|
|
"op": ">=",
|
|
"value": 88,
|
|
"flag": "escape_route_open",
|
|
"flag_equals": True,
|
|
"ending_state": "escaped",
|
|
"ending_summary": "你们重掌列车控制桥，终点协议被解除，列车返航。",
|
|
},
|
|
{
|
|
"id": "riot_failure",
|
|
"metric": "suspicion",
|
|
"op": ">=",
|
|
"value": 95,
|
|
"ending_state": "trapped",
|
|
"ending_summary": "乘客暴动失控，列车陷入不可逆循环。",
|
|
},
|
|
],
|
|
},
|
|
},
|
|
]
|
|
|
|
count_row = conn.execute("SELECT COUNT(*) AS c FROM game_cards").fetchone()
|
|
is_empty = bool(count_row and int(count_row["c"]) == 0)
|
|
for card in cards:
|
|
if is_empty:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO game_cards(
|
|
card_id, title, subtitle, summary, cover_url, opening_text,
|
|
starter_goals, state_schema, content_pack, creator_user_id, is_official, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
card["card_id"],
|
|
card["title"],
|
|
card["subtitle"],
|
|
card["summary"],
|
|
card["cover_url"],
|
|
card["opening_text"],
|
|
json.dumps(card["starter_goals"], ensure_ascii=False),
|
|
json.dumps(card["state_schema"], ensure_ascii=False),
|
|
json.dumps(card.get("content_pack", {}), ensure_ascii=False),
|
|
None,
|
|
1,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
else:
|
|
row = conn.execute(
|
|
"SELECT card_id, state_schema, content_pack FROM game_cards WHERE card_id=?",
|
|
(card["card_id"],),
|
|
).fetchone()
|
|
if row is None:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO game_cards(
|
|
card_id, title, subtitle, summary, cover_url, opening_text,
|
|
starter_goals, state_schema, content_pack, creator_user_id, is_official, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
card["card_id"],
|
|
card["title"],
|
|
card["subtitle"],
|
|
card["summary"],
|
|
card["cover_url"],
|
|
card["opening_text"],
|
|
json.dumps(card["starter_goals"], ensure_ascii=False),
|
|
json.dumps(card["state_schema"], ensure_ascii=False),
|
|
json.dumps(card.get("content_pack", {}), ensure_ascii=False),
|
|
None,
|
|
1,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
UPDATE game_cards
|
|
SET
|
|
title=?,
|
|
subtitle=?,
|
|
summary=?,
|
|
cover_url=?,
|
|
opening_text=?,
|
|
starter_goals=?,
|
|
state_schema=?,
|
|
content_pack=?,
|
|
is_official=1,
|
|
updated_at=?
|
|
WHERE card_id=?
|
|
""",
|
|
(
|
|
card["title"],
|
|
card["subtitle"],
|
|
card["summary"],
|
|
card["cover_url"],
|
|
card["opening_text"],
|
|
json.dumps(card["starter_goals"], ensure_ascii=False),
|
|
json.dumps(card["state_schema"], ensure_ascii=False),
|
|
json.dumps(card.get("content_pack", {}), ensure_ascii=False),
|
|
now,
|
|
card["card_id"],
|
|
),
|
|
)
|
|
|
|
def create_room(
|
|
self,
|
|
room_id: str,
|
|
host_player_id: str,
|
|
game_state: dict[str, Any],
|
|
*,
|
|
game_card_id: str = DEFAULT_CARD_ID,
|
|
is_public: bool = False,
|
|
) -> None:
|
|
now = utc_now_iso()
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO rooms(
|
|
room_id, host_player_id, is_public, status, chapter, scene_id,
|
|
game_state, game_card_id, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
room_id,
|
|
host_player_id,
|
|
1 if is_public else 0,
|
|
game_state.get("game_phase", "lobby"),
|
|
int(game_state.get("chapter", 0)),
|
|
game_state.get("scene_id", "lobby"),
|
|
json.dumps(game_state, ensure_ascii=False),
|
|
game_card_id,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
def add_or_update_player(
|
|
self,
|
|
room_id: str,
|
|
player_id: str,
|
|
name: str,
|
|
*,
|
|
is_host: bool = False,
|
|
is_ready: bool = False,
|
|
is_online: bool = True,
|
|
user_id: str | None = None,
|
|
) -> None:
|
|
now = utc_now_iso()
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO players(
|
|
player_id, room_id, name, is_host, is_ready, is_online, user_id, joined_at, last_seen
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(player_id) DO UPDATE SET
|
|
name=excluded.name,
|
|
is_host=excluded.is_host,
|
|
is_ready=excluded.is_ready,
|
|
is_online=excluded.is_online,
|
|
user_id=COALESCE(excluded.user_id, players.user_id),
|
|
last_seen=excluded.last_seen
|
|
""",
|
|
(
|
|
player_id,
|
|
room_id,
|
|
name,
|
|
1 if is_host else 0,
|
|
1 if is_ready else 0,
|
|
1 if is_online else 0,
|
|
user_id,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
def set_room_host(self, room_id: str, host_player_id: str) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"UPDATE rooms SET host_player_id=?, updated_at=? WHERE room_id=?",
|
|
(host_player_id, utc_now_iso(), room_id),
|
|
)
|
|
conn.execute(
|
|
"UPDATE players SET is_host=CASE WHEN player_id=? THEN 1 ELSE 0 END WHERE room_id=?",
|
|
(host_player_id, room_id),
|
|
)
|
|
conn.commit()
|
|
|
|
def get_room(self, room_id: str) -> dict[str, Any] | None:
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute("SELECT * FROM rooms WHERE room_id=?", (room_id,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
result = dict(row)
|
|
result["game_state"] = json.loads(result["game_state"])
|
|
result["is_public"] = bool(result.get("is_public", 0))
|
|
return result
|
|
|
|
def update_room_public(self, room_id: str, is_public: bool) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"UPDATE rooms SET is_public=?, updated_at=? WHERE room_id=?",
|
|
(1 if is_public else 0, utc_now_iso(), room_id),
|
|
)
|
|
conn.commit()
|
|
|
|
def list_players(self, room_id: str) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM players WHERE room_id=? ORDER BY joined_at ASC", (room_id,)
|
|
).fetchall()
|
|
players = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item["is_host"] = bool(item["is_host"])
|
|
item["is_ready"] = bool(item["is_ready"])
|
|
item["is_online"] = bool(item["is_online"])
|
|
players.append(item)
|
|
return players
|
|
|
|
def get_player(self, room_id: str, player_id: str) -> dict[str, Any] | None:
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM players WHERE room_id=? AND player_id=?", (room_id, player_id)
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
item = dict(row)
|
|
item["is_host"] = bool(item["is_host"])
|
|
item["is_ready"] = bool(item["is_ready"])
|
|
item["is_online"] = bool(item["is_online"])
|
|
return item
|
|
|
|
def get_player_by_user(self, room_id: str, user_id: str) -> dict[str, Any] | None:
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM players WHERE room_id=? AND user_id=? ORDER BY joined_at ASC LIMIT 1",
|
|
(room_id, user_id),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
item = dict(row)
|
|
item["is_host"] = bool(item["is_host"])
|
|
item["is_ready"] = bool(item["is_ready"])
|
|
item["is_online"] = bool(item["is_online"])
|
|
return item
|
|
|
|
def update_player_ready(self, room_id: str, player_id: str, ready: bool) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"UPDATE players SET is_ready=?, last_seen=? WHERE room_id=? AND player_id=?",
|
|
(1 if ready else 0, utc_now_iso(), room_id, player_id),
|
|
)
|
|
conn.commit()
|
|
|
|
def update_player_online(self, room_id: str, player_id: str, is_online: bool) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"UPDATE players SET is_online=?, last_seen=? WHERE room_id=? AND player_id=?",
|
|
(1 if is_online else 0, utc_now_iso(), room_id, player_id),
|
|
)
|
|
conn.commit()
|
|
|
|
def remove_player(self, room_id: str, player_id: str) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute("DELETE FROM players WHERE room_id=? AND player_id=?", (room_id, player_id))
|
|
conn.commit()
|
|
|
|
def update_room_state(self, room_id: str, status: str, game_state: dict[str, Any]) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
UPDATE rooms
|
|
SET status=?, chapter=?, scene_id=?, game_state=?, updated_at=?
|
|
WHERE room_id=?
|
|
""",
|
|
(
|
|
status,
|
|
int(game_state.get("chapter", 0)),
|
|
game_state.get("scene_id", "lobby"),
|
|
json.dumps(game_state, ensure_ascii=False),
|
|
utc_now_iso(),
|
|
room_id,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
def append_message(
|
|
self,
|
|
room_id: str,
|
|
kind: str,
|
|
content: str,
|
|
*,
|
|
player_id: str | None = None,
|
|
visibility: str = "public",
|
|
) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO messages(room_id, kind, player_id, visibility, content, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(room_id, kind, player_id, visibility, content, utc_now_iso()),
|
|
)
|
|
conn.commit()
|
|
|
|
def list_messages(self, room_id: str, limit: int = 200) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, room_id, kind, player_id, visibility, content, created_at
|
|
FROM messages
|
|
WHERE room_id=?
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
""",
|
|
(room_id, limit),
|
|
).fetchall()
|
|
return [dict(row) for row in reversed(rows)]
|
|
|
|
def append_action(self, room_id: str, round_no: int, player_id: str, content: str) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO actions(room_id, round_no, player_id, content, created_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(room_id, round_no, player_id, content, utc_now_iso()),
|
|
)
|
|
conn.commit()
|
|
|
|
def delete_room(self, room_id: str) -> None:
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute("DELETE FROM actions WHERE room_id=?", (room_id,))
|
|
conn.execute("DELETE FROM messages WHERE room_id=?", (room_id,))
|
|
conn.execute("DELETE FROM players WHERE room_id=?", (room_id,))
|
|
conn.execute("DELETE FROM rooms WHERE room_id=?", (room_id,))
|
|
conn.commit()
|
|
|
|
def load_room_snapshot(self, room_id: str) -> dict[str, Any] | None:
|
|
room = self.get_room(room_id)
|
|
if room is None:
|
|
return None
|
|
players = self.list_players(room_id)
|
|
messages = self.list_messages(room_id)
|
|
return {
|
|
"room": room,
|
|
"players": players,
|
|
"messages": messages,
|
|
}
|
|
|
|
def create_user(self, username: str, password_hash: str, display_name: str) -> dict[str, Any]:
|
|
user_id = uuid.uuid4().hex
|
|
now = utc_now_iso()
|
|
with self._lock, self._connect() as conn:
|
|
admin_row = conn.execute(
|
|
"SELECT COUNT(*) AS c FROM users WHERE role='admin'"
|
|
).fetchone()
|
|
role = "admin" if int(admin_row["c"] if admin_row else 0) == 0 else "player"
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO users(
|
|
user_id, username, password_hash, display_name, role, avatar_color, avatar_url, theme, bio, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(user_id, username, password_hash, display_name, role, "#0070f2", "", "dopamine", "", now, now),
|
|
)
|
|
conn.commit()
|
|
except sqlite3.IntegrityError as exc:
|
|
raise ValueError("username_taken") from exc
|
|
user = self.get_user_by_id(user_id)
|
|
if user is None:
|
|
raise ValueError("user_create_failed")
|
|
return user
|
|
|
|
def get_user_by_username(self, username: str) -> dict[str, Any] | None:
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_user_by_id(self, user_id: str | None) -> dict[str, Any] | None:
|
|
if not user_id:
|
|
return None
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute("SELECT * FROM users WHERE user_id=?", (user_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def update_user_profile(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
display_name: str | None = None,
|
|
avatar_color: str | None = None,
|
|
avatar_url: str | None = None,
|
|
theme: str | None = None,
|
|
bio: str | None = None,
|
|
) -> dict[str, Any]:
|
|
existing = self.get_user_by_id(user_id)
|
|
if existing is None:
|
|
raise ValueError("user_not_found")
|
|
|
|
merged_name = display_name if display_name is not None else existing["display_name"]
|
|
merged_color = avatar_color if avatar_color is not None else existing["avatar_color"]
|
|
merged_avatar_url = avatar_url if avatar_url is not None else existing.get("avatar_url", "")
|
|
merged_theme = theme if theme is not None else existing.get("theme", "minimal")
|
|
merged_bio = bio if bio is not None else existing["bio"]
|
|
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
UPDATE users
|
|
SET display_name=?, avatar_color=?, avatar_url=?, theme=?, bio=?, updated_at=?
|
|
WHERE user_id=?
|
|
""",
|
|
(merged_name, merged_color, merged_avatar_url, merged_theme, merged_bio, utc_now_iso(), user_id),
|
|
)
|
|
conn.commit()
|
|
|
|
updated = self.get_user_by_id(user_id)
|
|
if updated is None:
|
|
raise ValueError("user_not_found")
|
|
return updated
|
|
|
|
def list_user_rooms(self, user_id: str, limit: int = 10) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT r.room_id, r.status, r.scene_id, r.chapter, r.game_card_id,
|
|
r.is_public, p.player_id, p.name AS player_name, p.is_host, p.is_online,
|
|
r.updated_at
|
|
FROM players p
|
|
JOIN rooms r ON r.room_id = p.room_id
|
|
WHERE p.user_id=?
|
|
ORDER BY r.updated_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(user_id, limit),
|
|
).fetchall()
|
|
results = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item["is_public"] = bool(item.get("is_public", 0))
|
|
item["is_host"] = bool(item["is_host"])
|
|
item["is_online"] = bool(item["is_online"])
|
|
results.append(item)
|
|
return results
|
|
|
|
def list_public_rooms(self, limit: int = 40) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT
|
|
r.room_id,
|
|
r.status,
|
|
r.chapter,
|
|
r.scene_id,
|
|
r.game_card_id,
|
|
r.updated_at,
|
|
r.created_at,
|
|
hp.name AS host_name,
|
|
(
|
|
SELECT COUNT(*) FROM players p WHERE p.room_id = r.room_id
|
|
) AS player_count,
|
|
(
|
|
SELECT COUNT(*) FROM players p WHERE p.room_id = r.room_id AND p.is_online = 1
|
|
) AS online_count
|
|
FROM rooms r
|
|
LEFT JOIN players hp ON hp.player_id = r.host_player_id
|
|
WHERE r.is_public = 1 AND r.status IN ('lobby', 'running')
|
|
ORDER BY CASE WHEN r.status='running' THEN 0 ELSE 1 END ASC, r.updated_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def _is_winning_ending(self, ending_state: Any) -> bool:
|
|
ending = str(ending_state or "").strip().lower()
|
|
if not ending:
|
|
return False
|
|
return ending in {
|
|
"escaped",
|
|
"sacrifice",
|
|
"pact",
|
|
"victory",
|
|
"success",
|
|
"win",
|
|
"custom_end",
|
|
}
|
|
|
|
def list_player_leaderboard(self, limit: int = 20) -> list[dict[str, Any]]:
|
|
try:
|
|
safe_limit = max(1, min(100, int(limit)))
|
|
except (TypeError, ValueError):
|
|
safe_limit = 20
|
|
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT
|
|
u.user_id,
|
|
u.username,
|
|
u.display_name,
|
|
u.avatar_color,
|
|
u.avatar_url,
|
|
u.theme,
|
|
p.room_id,
|
|
r.game_state,
|
|
r.updated_at
|
|
FROM users u
|
|
JOIN players p ON p.user_id = u.user_id
|
|
JOIN rooms r ON r.room_id = p.room_id
|
|
WHERE r.status = 'ended'
|
|
ORDER BY r.updated_at DESC
|
|
"""
|
|
).fetchall()
|
|
|
|
summary: dict[str, dict[str, Any]] = {}
|
|
for row in rows:
|
|
item = dict(row)
|
|
user_id = str(item.get("user_id", "")).strip()
|
|
room_id = str(item.get("room_id", "")).strip()
|
|
if not user_id or not room_id:
|
|
continue
|
|
|
|
entry = summary.get(user_id)
|
|
if entry is None:
|
|
entry = {
|
|
"user_id": user_id,
|
|
"username": str(item.get("username", "")).strip(),
|
|
"display_name": str(item.get("display_name", "")).strip(),
|
|
"avatar_color": str(item.get("avatar_color", "#0070f2")).strip() or "#0070f2",
|
|
"avatar_url": str(item.get("avatar_url", "")).strip(),
|
|
"theme": str(item.get("theme", "dopamine")).strip() or "dopamine",
|
|
"played_games": 0,
|
|
"wins": 0,
|
|
"total_rounds": 0,
|
|
"last_played_at": "",
|
|
"_seen_rooms": set(),
|
|
}
|
|
summary[user_id] = entry
|
|
|
|
seen_rooms = entry["_seen_rooms"]
|
|
if room_id in seen_rooms:
|
|
continue
|
|
seen_rooms.add(room_id)
|
|
|
|
game_state_raw = item.get("game_state")
|
|
try:
|
|
game_state = json.loads(game_state_raw) if isinstance(game_state_raw, str) else {}
|
|
except json.JSONDecodeError:
|
|
game_state = {}
|
|
if not isinstance(game_state, dict):
|
|
game_state = {}
|
|
|
|
entry["played_games"] += 1
|
|
if self._is_winning_ending(game_state.get("ending_state")):
|
|
entry["wins"] += 1
|
|
|
|
round_no_raw = game_state.get("round_no", 0)
|
|
try:
|
|
round_no = max(0, int(round_no_raw))
|
|
except (TypeError, ValueError):
|
|
round_no = 0
|
|
entry["total_rounds"] += round_no
|
|
|
|
updated_at = str(item.get("updated_at", "")).strip()
|
|
if updated_at and updated_at > entry["last_played_at"]:
|
|
entry["last_played_at"] = updated_at
|
|
|
|
results: list[dict[str, Any]] = []
|
|
for entry in summary.values():
|
|
entry.pop("_seen_rooms", None)
|
|
played = int(entry["played_games"])
|
|
wins = int(entry["wins"])
|
|
win_rate = round((wins / played) * 100, 1) if played else 0.0
|
|
entry["win_rate"] = win_rate
|
|
entry["score"] = wins * 100 + played * 12 + int(entry["total_rounds"])
|
|
results.append(entry)
|
|
|
|
results.sort(
|
|
key=lambda x: (
|
|
int(x.get("wins", 0)),
|
|
int(x.get("played_games", 0)),
|
|
int(x.get("score", 0)),
|
|
str(x.get("last_played_at", "")),
|
|
str(x.get("username", "")),
|
|
),
|
|
reverse=True,
|
|
)
|
|
|
|
ranked = results[:safe_limit]
|
|
for idx, row in enumerate(ranked, start=1):
|
|
row["rank"] = idx
|
|
return ranked
|
|
|
|
def list_users(self, limit: int = 500) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT user_id, username, display_name, role, avatar_color, avatar_url, bio, created_at, updated_at
|
|
, theme
|
|
FROM users
|
|
ORDER BY created_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def update_user_role(self, user_id: str, role: str) -> dict[str, Any]:
|
|
role = str(role).strip()
|
|
if role not in {"player", "moderator", "admin"}:
|
|
raise ValueError("invalid_role")
|
|
existing = self.get_user_by_id(user_id)
|
|
if existing is None:
|
|
raise ValueError("user_not_found")
|
|
|
|
with self._lock, self._connect() as conn:
|
|
if existing.get("role") == "admin" and role != "admin":
|
|
row = conn.execute("SELECT COUNT(*) AS c FROM users WHERE role='admin'").fetchone()
|
|
admin_count = int(row["c"]) if row else 0
|
|
if admin_count <= 1:
|
|
raise ValueError("cannot_remove_last_admin")
|
|
|
|
conn.execute(
|
|
"UPDATE users SET role=?, updated_at=? WHERE user_id=?",
|
|
(role, utc_now_iso(), user_id),
|
|
)
|
|
conn.commit()
|
|
|
|
updated = self.get_user_by_id(user_id)
|
|
if updated is None:
|
|
raise ValueError("user_not_found")
|
|
return updated
|
|
|
|
def list_rooms_admin(self, limit: int = 300) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT
|
|
r.room_id,
|
|
r.status,
|
|
r.chapter,
|
|
r.scene_id,
|
|
r.game_card_id,
|
|
r.host_player_id,
|
|
hp.name AS host_name,
|
|
r.created_at,
|
|
r.updated_at,
|
|
(
|
|
SELECT COUNT(*) FROM players p WHERE p.room_id = r.room_id
|
|
) AS player_count
|
|
FROM rooms r
|
|
LEFT JOIN players hp ON hp.player_id = r.host_player_id
|
|
ORDER BY r.updated_at DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def _build_card_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
|
|
item = dict(row)
|
|
try:
|
|
item["starter_goals"] = json.loads(item.get("starter_goals", "[]"))
|
|
except json.JSONDecodeError:
|
|
item["starter_goals"] = []
|
|
try:
|
|
item["state_schema"] = json.loads(item.get("state_schema", "{}"))
|
|
except json.JSONDecodeError:
|
|
item["state_schema"] = {}
|
|
try:
|
|
item["content_pack"] = json.loads(item.get("content_pack", "{}"))
|
|
except json.JSONDecodeError:
|
|
item["content_pack"] = {}
|
|
item["is_official"] = bool(item.get("is_official", 0))
|
|
return item
|
|
|
|
def list_game_cards(self) -> list[dict[str, Any]]:
|
|
with self._lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT * FROM game_cards
|
|
ORDER BY is_official DESC, created_at DESC
|
|
"""
|
|
).fetchall()
|
|
return [self._build_card_payload(row) for row in rows]
|
|
|
|
def get_game_card(self, card_id: str) -> dict[str, Any] | None:
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute("SELECT * FROM game_cards WHERE card_id=?", (card_id,)).fetchone()
|
|
if row is None:
|
|
return None
|
|
return self._build_card_payload(row)
|
|
|
|
def get_default_card(self) -> dict[str, Any] | None:
|
|
return self.get_game_card(DEFAULT_CARD_ID)
|
|
|
|
def create_game_card(
|
|
self,
|
|
*,
|
|
title: str,
|
|
subtitle: str,
|
|
summary: str,
|
|
cover_url: str,
|
|
opening_text: str,
|
|
starter_goals: list[str],
|
|
state_schema: dict[str, Any] | None,
|
|
content_pack: dict[str, Any] | None,
|
|
creator_user_id: str,
|
|
) -> dict[str, Any]:
|
|
base_slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "card"
|
|
card_id = base_slug[:32]
|
|
suffix = 2
|
|
while self.get_game_card(card_id) is not None:
|
|
card_id = f"{base_slug[:28]}-{suffix}"
|
|
suffix += 1
|
|
|
|
now = utc_now_iso()
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO game_cards(
|
|
card_id, title, subtitle, summary, cover_url, opening_text,
|
|
starter_goals, state_schema, content_pack, creator_user_id, is_official, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
card_id,
|
|
title,
|
|
subtitle,
|
|
summary,
|
|
cover_url,
|
|
opening_text,
|
|
json.dumps(starter_goals, ensure_ascii=False),
|
|
json.dumps(state_schema or {}, ensure_ascii=False),
|
|
json.dumps(content_pack or {}, ensure_ascii=False),
|
|
creator_user_id,
|
|
0,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
card = self.get_game_card(card_id)
|
|
if card is None:
|
|
raise ValueError("card_create_failed")
|
|
return card
|
|
|
|
def update_game_card(
|
|
self,
|
|
card_id: str,
|
|
*,
|
|
title: str,
|
|
subtitle: str,
|
|
summary: str,
|
|
cover_url: str,
|
|
opening_text: str,
|
|
starter_goals: list[str],
|
|
state_schema: dict[str, Any] | None,
|
|
content_pack: dict[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
existing = self.get_game_card(card_id)
|
|
if existing is None:
|
|
raise ValueError("invalid_card_id")
|
|
|
|
with self._lock, self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
UPDATE game_cards
|
|
SET
|
|
title=?,
|
|
subtitle=?,
|
|
summary=?,
|
|
cover_url=?,
|
|
opening_text=?,
|
|
starter_goals=?,
|
|
state_schema=?,
|
|
content_pack=?,
|
|
updated_at=?
|
|
WHERE card_id=?
|
|
""",
|
|
(
|
|
title,
|
|
subtitle,
|
|
summary,
|
|
cover_url,
|
|
opening_text,
|
|
json.dumps(starter_goals, ensure_ascii=False),
|
|
json.dumps(state_schema or {}, ensure_ascii=False),
|
|
json.dumps(content_pack or {}, ensure_ascii=False),
|
|
utc_now_iso(),
|
|
card_id,
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
card = self.get_game_card(card_id)
|
|
if card is None:
|
|
raise ValueError("invalid_card_id")
|
|
return card
|
|
|
|
def delete_game_card(self, card_id: str, user_id: str, *, as_admin: bool = False) -> None:
|
|
card = self.get_game_card(card_id)
|
|
if card is None:
|
|
raise ValueError("invalid_card_id")
|
|
if not as_admin:
|
|
if card.get("is_official"):
|
|
raise ValueError("card_delete_forbidden")
|
|
if str(card.get("creator_user_id") or "") != str(user_id):
|
|
raise ValueError("card_delete_forbidden")
|
|
|
|
with self._lock, self._connect() as conn:
|
|
active_row = conn.execute(
|
|
"""
|
|
SELECT COUNT(*) AS c
|
|
FROM rooms
|
|
WHERE game_card_id=? AND status IN ('lobby', 'running')
|
|
""",
|
|
(card_id,),
|
|
).fetchone()
|
|
active_count = int(active_row["c"]) if active_row else 0
|
|
if active_count > 0:
|
|
raise ValueError("card_in_use")
|
|
|
|
conn.execute("DELETE FROM game_cards WHERE card_id=?", (card_id,))
|
|
conn.commit()
|
|
|
|
def set_game_card_official(self, card_id: str, is_official: bool) -> dict[str, Any]:
|
|
with self._lock, self._connect() as conn:
|
|
row = conn.execute("SELECT card_id FROM game_cards WHERE card_id=?", (card_id,)).fetchone()
|
|
if row is None:
|
|
raise ValueError("invalid_card_id")
|
|
conn.execute(
|
|
"UPDATE game_cards SET is_official=?, updated_at=? WHERE card_id=?",
|
|
(1 if is_official else 0, utc_now_iso(), card_id),
|
|
)
|
|
conn.commit()
|
|
card = self.get_game_card(card_id)
|
|
if card is None:
|
|
raise ValueError("invalid_card_id")
|
|
return card
|