Files

8805 lines
374 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import json
import re
import secrets
import sqlite3
import string
import threading
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from game.models import utc_now_iso
DEFAULT_CARD_ID = "fogharbor"
PRESENCE_ACTIVE_WINDOW_SECONDS = 300
ACHIEVEMENT_LEVEL_ORDER = {
"bronze": 1,
"silver": 2,
"gold": 3,
"platinum": 4,
"diamond": 5,
}
SYSTEM_ACHIEVEMENT_TRACKS: list[dict[str, Any]] = [
{
"id": "story_journey",
"name": "迷雾旅程",
"description": "完成对局次数达到里程碑。",
"metric": "plays",
"tiers": [
{"level": "bronze", "threshold": 1, "title": "初入迷雾"},
{"level": "silver", "threshold": 10, "title": "港湾熟客"},
{"level": "gold", "threshold": 30, "title": "终幕见证者"},
{"level": "platinum", "threshold": 60, "title": "编年史守望"},
],
},
{
"id": "victory_hunt",
"name": "破局凯歌",
"description": "累计胜场达到里程碑。",
"metric": "wins",
"tiers": [
{"level": "bronze", "threshold": 1, "title": "首胜燃灯"},
{"level": "silver", "threshold": 6, "title": "凯旋策士"},
{"level": "gold", "threshold": 20, "title": "风暴统筹者"},
{"level": "platinum", "threshold": 45, "title": "终局指挥官"},
],
},
{
"id": "streak_legend",
"name": "连捷神话",
"description": "达成连胜里程碑。",
"metric": "best_streak",
"tiers": [
{"level": "bronze", "threshold": 2, "title": "双连锋芒"},
{"level": "silver", "threshold": 4, "title": "四连长虹"},
{"level": "gold", "threshold": 6, "title": "不败领航"},
{"level": "diamond", "threshold": 9, "title": "神话连冠"},
],
},
]
ACHIEVEMENT_POINT_REWARD_TRACKS: list[dict[str, Any]] = [
{
"id": "recharge_journey",
"metric": "recharged",
"tiers": [
{"threshold": 500, "points": 20},
{"threshold": 2000, "points": 80},
{"threshold": 5000, "points": 220},
],
},
{
"id": "spend_journey",
"metric": "spent",
"tiers": [
{"threshold": 300, "points": 10},
{"threshold": 1200, "points": 45},
{"threshold": 3000, "points": 130},
],
},
]
SEASON_XP_PER_LEVEL = 100
SEASON_MAX_LEVEL = 120
SEASON_TASK_TYPES = {"daily", "weekly", "season", "achievement"}
SEASON_CARD_RARITY_MULTIPLIER = {
"normal": 1.0,
"rare": 1.35,
"epic": 1.85,
"legendary": 2.6,
}
SEASON_STORY_OPENERS = [
"潮雾在天台翻涌,远处的霓虹像被潮汐擦亮。",
"失真的广播再次响起,仿佛有人在幕布后低声提醒。",
"旧档案的封条被悄悄揭开,纸页边缘还留着余温。",
"钟摆越过整点时,街区的光影突然反向流动。",
]
SEASON_STORY_TWISTS = [
"你上次留下的选择在今天产生了反作用。",
"一位沉默角色开始主动与你交换线索。",
"原本安全的路径被新的规则临时封锁。",
"看似无关的碎片在本轮首次拼接成完整图案。",
]
SEASON_HINT_POOL = [
"优先完成一次稳定推进,先拿到基础奖励。",
"如果队伍在线,尝试组队类目标,效率更高。",
"本卡目标可跨局累积,不必在一局内硬冲。",
"保留一次刷新机会,可能刷出高价值惊喜卡。",
]
def _utc_day_key(now_iso: str = "") -> str:
dt = datetime.fromisoformat((now_iso or utc_now_iso()).replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
dt = dt.astimezone(timezone.utc)
return dt.strftime("%Y-%m-%d")
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,
world_id TEXT NOT NULL DEFAULT 'global',
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 card_modules (
module_id TEXT PRIMARY KEY,
base_card_id TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT NOT NULL DEFAULT '',
patch_json TEXT NOT NULL DEFAULT '{}',
creator_user_id TEXT NOT NULL,
is_official INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'published',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(base_card_id) REFERENCES game_cards(card_id),
FOREIGN KEY(creator_user_id) REFERENCES users(user_id)
);
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)
);
CREATE TABLE IF NOT EXISTS oauth_identities (
provider TEXT NOT NULL,
provider_user_id TEXT NOT NULL,
user_id TEXT NOT NULL,
username_snapshot TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(provider, provider_user_id),
UNIQUE(provider, user_id),
FOREIGN KEY(user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS invitations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
from_user_id TEXT NOT NULL,
to_user_id TEXT NOT NULL,
room_id TEXT NOT NULL,
message TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'pending',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
responded_at TEXT,
FOREIGN KEY(from_user_id) REFERENCES users(user_id),
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)
);
CREATE TABLE IF NOT EXISTS user_points_accounts (
user_id TEXT PRIMARY KEY,
balance INTEGER NOT NULL DEFAULT 0,
total_recharged INTEGER NOT NULL DEFAULT 0,
total_spent INTEGER NOT NULL DEFAULT 0,
frozen_balance INTEGER NOT NULL DEFAULT 0,
is_locked INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS user_point_transactions (
tx_id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
change_amount INTEGER NOT NULL,
balance_after INTEGER NOT NULL,
tx_type TEXT NOT NULL,
source TEXT NOT NULL DEFAULT '',
related_id TEXT NOT NULL DEFAULT '',
note TEXT NOT NULL DEFAULT '',
expires_at TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'applied',
operator_user_id TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS user_point_lots (
lot_id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
source_tx_id INTEGER,
original_amount INTEGER NOT NULL,
remaining_amount INTEGER NOT NULL,
expires_at TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(source_tx_id) REFERENCES user_point_transactions(tx_id)
);
CREATE TABLE IF NOT EXISTS activation_codes (
code TEXT PRIMARY KEY,
points INTEGER NOT NULL,
status TEXT NOT NULL DEFAULT 'unused',
batch_no TEXT NOT NULL DEFAULT '',
remark TEXT NOT NULL DEFAULT '',
expires_at TEXT NOT NULL DEFAULT '',
created_by TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
used_by_user_id TEXT NOT NULL DEFAULT '',
used_at TEXT NOT NULL DEFAULT '',
used_ip TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS activation_code_attempts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ip_address TEXT NOT NULL DEFAULT '',
user_id TEXT NOT NULL DEFAULT '',
code TEXT NOT NULL DEFAULT '',
success INTEGER NOT NULL DEFAULT 0,
error_code TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS shop_items (
item_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
image_url TEXT NOT NULL DEFAULT '',
item_type TEXT NOT NULL DEFAULT 'decoration',
subtype TEXT NOT NULL DEFAULT '',
price INTEGER NOT NULL DEFAULT 0,
stock INTEGER NOT NULL DEFAULT -1,
purchase_limit INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active',
sort_order INTEGER NOT NULL DEFAULT 0,
is_recommended INTEGER NOT NULL DEFAULT 0,
metadata_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS shop_orders (
order_id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
item_id TEXT NOT NULL,
item_snapshot TEXT NOT NULL DEFAULT '{}',
price INTEGER NOT NULL DEFAULT 0,
quantity INTEGER NOT NULL DEFAULT 1,
total_cost INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'paid',
refund_note TEXT NOT NULL DEFAULT '',
operator_user_id TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
refunded_at TEXT NOT NULL DEFAULT '',
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(item_id) REFERENCES shop_items(item_id)
);
CREATE TABLE IF NOT EXISTS user_entitlements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
entitlement_type TEXT NOT NULL,
ref_id TEXT NOT NULL DEFAULT '',
source_order_id TEXT NOT NULL DEFAULT '',
payload_json TEXT NOT NULL DEFAULT '{}',
status TEXT NOT NULL DEFAULT 'active',
expires_at TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS user_tips (
id INTEGER PRIMARY KEY AUTOINCREMENT,
from_user_id TEXT NOT NULL,
to_user_id TEXT NOT NULL,
amount INTEGER NOT NULL,
room_id TEXT NOT NULL DEFAULT '',
message TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
FOREIGN KEY(from_user_id) REFERENCES users(user_id),
FOREIGN KEY(to_user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS achievement_point_rewards (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
reward_key TEXT NOT NULL,
points INTEGER NOT NULL,
created_at TEXT NOT NULL,
UNIQUE(user_id, reward_key),
FOREIGN KEY(user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS seasons (
season_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
theme TEXT NOT NULL DEFAULT '',
description TEXT NOT NULL DEFAULT '',
cover_url TEXT NOT NULL DEFAULT '',
accent_color TEXT NOT NULL DEFAULT '#4f6dff',
story_seed TEXT NOT NULL DEFAULT '',
story_outline TEXT NOT NULL DEFAULT '',
start_at TEXT NOT NULL,
end_at TEXT NOT NULL,
rest_end_at TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'active',
source TEXT NOT NULL DEFAULT 'manual',
pass_price INTEGER NOT NULL DEFAULT 680,
created_by TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS season_announcements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
season_id TEXT NOT NULL,
title TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
created_by TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS user_season_profiles (
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
pass_type TEXT NOT NULL DEFAULT 'free',
level INTEGER NOT NULL DEFAULT 1,
xp INTEGER NOT NULL DEFAULT 0,
season_points INTEGER NOT NULL DEFAULT 0,
total_claimed_cards INTEGER NOT NULL DEFAULT 0,
completed_tasks INTEGER NOT NULL DEFAULT 0,
last_refresh_key TEXT NOT NULL DEFAULT '',
last_generated_at TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(user_id, season_id),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_story_threads (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
thread_summary TEXT NOT NULL DEFAULT '',
thread_state_json TEXT NOT NULL DEFAULT '{}',
last_generated_at TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL,
UNIQUE(user_id, season_id),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_activity_cards (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
refresh_key TEXT NOT NULL,
title TEXT NOT NULL,
story_text TEXT NOT NULL DEFAULT '',
objective TEXT NOT NULL DEFAULT '',
hint_text TEXT NOT NULL DEFAULT '',
reward_points INTEGER NOT NULL DEFAULT 0,
reward_xp INTEGER NOT NULL DEFAULT 0,
rarity TEXT NOT NULL DEFAULT 'normal',
status TEXT NOT NULL DEFAULT 'active',
claimed_at TEXT NOT NULL DEFAULT '',
expires_at TEXT NOT NULL DEFAULT '',
payload_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
claimed_tx_id TEXT NOT NULL DEFAULT '',
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_task_templates (
template_id TEXT PRIMARY KEY,
season_id TEXT NOT NULL,
task_type TEXT NOT NULL DEFAULT 'daily',
title TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
event_key TEXT NOT NULL DEFAULT '',
target_value INTEGER NOT NULL DEFAULT 1,
difficulty TEXT NOT NULL DEFAULT 'normal',
weight INTEGER NOT NULL DEFAULT 100,
reward_points INTEGER NOT NULL DEFAULT 0,
reward_xp INTEGER NOT NULL DEFAULT 0,
reward_season_coin INTEGER NOT NULL DEFAULT 0,
reward_payload_json TEXT NOT NULL DEFAULT '{}',
constraints_json TEXT NOT NULL DEFAULT '{}',
refresh_policy TEXT NOT NULL DEFAULT 'auto',
is_active INTEGER NOT NULL DEFAULT 1,
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS user_season_tasks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
template_id TEXT NOT NULL,
task_type TEXT NOT NULL DEFAULT 'daily',
refresh_key TEXT NOT NULL DEFAULT '',
title TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
event_key TEXT NOT NULL DEFAULT '',
target_value INTEGER NOT NULL DEFAULT 1,
progress_value INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active',
reward_points INTEGER NOT NULL DEFAULT 0,
reward_xp INTEGER NOT NULL DEFAULT 0,
reward_season_coin INTEGER NOT NULL DEFAULT 0,
reward_payload_json TEXT NOT NULL DEFAULT '{}',
constraints_json TEXT NOT NULL DEFAULT '{}',
expires_at TEXT NOT NULL DEFAULT '',
claimed_at TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(user_id, season_id, template_id, refresh_key),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id),
FOREIGN KEY(template_id) REFERENCES season_task_templates(template_id)
);
CREATE TABLE IF NOT EXISTS season_pass_rewards (
reward_id TEXT PRIMARY KEY,
season_id TEXT NOT NULL,
reward_level INTEGER NOT NULL DEFAULT 1,
reward_track TEXT NOT NULL DEFAULT 'free',
title TEXT NOT NULL DEFAULT '',
reward_points INTEGER NOT NULL DEFAULT 0,
reward_xp INTEGER NOT NULL DEFAULT 0,
reward_season_coin INTEGER NOT NULL DEFAULT 0,
reward_payload_json TEXT NOT NULL DEFAULT '{}',
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(season_id, reward_level, reward_track, sort_order),
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS user_season_reward_claims (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
reward_id TEXT NOT NULL,
claimed_at TEXT NOT NULL,
UNIQUE(user_id, season_id, reward_id),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id),
FOREIGN KEY(reward_id) REFERENCES season_pass_rewards(reward_id)
);
CREATE TABLE IF NOT EXISTS season_card_pools (
pool_id TEXT PRIMARY KEY,
season_id TEXT NOT NULL,
card_type TEXT NOT NULL DEFAULT 'rotating',
title_hint TEXT NOT NULL DEFAULT '',
story_hook TEXT NOT NULL DEFAULT '',
objective_template TEXT NOT NULL DEFAULT '',
hint_template TEXT NOT NULL DEFAULT '',
event_key TEXT NOT NULL DEFAULT '',
target_value INTEGER NOT NULL DEFAULT 1,
reward_points INTEGER NOT NULL DEFAULT 0,
reward_xp INTEGER NOT NULL DEFAULT 0,
rarity TEXT NOT NULL DEFAULT 'normal',
weight INTEGER NOT NULL DEFAULT 100,
min_level INTEGER NOT NULL DEFAULT 1,
payload_json TEXT NOT NULL DEFAULT '{}',
is_active INTEGER NOT NULL DEFAULT 1,
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_titles (
title_id TEXT PRIMARY KEY,
season_id TEXT NOT NULL,
name TEXT NOT NULL,
color_code TEXT NOT NULL DEFAULT '#ffffff',
rarity TEXT NOT NULL DEFAULT 'normal',
acquire_type TEXT NOT NULL DEFAULT 'season_reward',
image_url TEXT NOT NULL DEFAULT '',
effect_json TEXT NOT NULL DEFAULT '{}',
is_active INTEGER NOT NULL DEFAULT 1,
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_avatar_frames (
frame_id TEXT PRIMARY KEY,
season_id TEXT NOT NULL,
name TEXT NOT NULL,
style_json TEXT NOT NULL DEFAULT '{}',
rarity TEXT NOT NULL DEFAULT 'normal',
acquire_type TEXT NOT NULL DEFAULT 'season_reward',
preview_url TEXT NOT NULL DEFAULT '',
is_active INTEGER NOT NULL DEFAULT 1,
sort_order INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS user_decorations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL DEFAULT '',
decoration_type TEXT NOT NULL,
decoration_id TEXT NOT NULL,
source TEXT NOT NULL DEFAULT 'season',
expires_at TEXT NOT NULL DEFAULT '',
is_equipped INTEGER NOT NULL DEFAULT 0,
payload_json TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(user_id, decoration_type, decoration_id),
FOREIGN KEY(user_id) REFERENCES users(user_id)
);
CREATE TABLE IF NOT EXISTS season_room_progress_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
room_id TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE(user_id, season_id, room_id),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_daily_game_cards (
day_key TEXT NOT NULL,
season_id TEXT NOT NULL,
card_id TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT NOT NULL DEFAULT '',
guide_text TEXT NOT NULL DEFAULT '',
intro_text TEXT NOT NULL DEFAULT '',
locked_intel_json TEXT NOT NULL DEFAULT '[]',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(day_key, season_id),
UNIQUE(season_id, card_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id),
FOREIGN KEY(card_id) REFERENCES game_cards(card_id)
);
CREATE TABLE IF NOT EXISTS user_season_intel_unlocks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
day_key TEXT NOT NULL,
intel_key TEXT NOT NULL,
intel_title TEXT NOT NULL DEFAULT '',
intel_text TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
UNIQUE(user_id, season_id, day_key, intel_key),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id)
);
CREATE TABLE IF NOT EXISTS season_daily_card_completions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
season_id TEXT NOT NULL,
day_key TEXT NOT NULL,
room_id TEXT NOT NULL,
card_id TEXT NOT NULL,
ending_state TEXT NOT NULL DEFAULT '',
ending_summary TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
UNIQUE(user_id, season_id, day_key),
UNIQUE(user_id, season_id, room_id),
FOREIGN KEY(user_id) REFERENCES users(user_id),
FOREIGN KEY(season_id) REFERENCES seasons(season_id),
FOREIGN KEY(card_id) REFERENCES game_cards(card_id)
);
CREATE TABLE IF NOT EXISTS season_story_snapshots (
season_id TEXT PRIMARY KEY,
guide_text TEXT NOT NULL DEFAULT '',
story_text TEXT NOT NULL DEFAULT '',
cards_count INTEGER NOT NULL DEFAULT 0,
intel_count INTEGER NOT NULL DEFAULT 0,
built_at TEXT NOT NULL,
FOREIGN KEY(season_id) REFERENCES seasons(season_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, "game_cards", "world_id", "TEXT NOT NULL DEFAULT 'global'")
conn.execute("CREATE INDEX IF NOT EXISTS idx_game_cards_world_created ON game_cards(world_id, created_at DESC)")
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'")
self._ensure_column(conn, "users", "is_online", "INTEGER NOT NULL DEFAULT 0")
self._ensure_column(conn, "users", "last_active_at", "TEXT NOT NULL DEFAULT ''")
self._ensure_column(conn, "users", "active_badge", "TEXT NOT NULL DEFAULT ''")
self._ensure_column(conn, "seasons", "config_json", "TEXT NOT NULL DEFAULT '{}'")
self._ensure_column(conn, "seasons", "rest_end_at", "TEXT NOT NULL DEFAULT ''")
self._ensure_column(conn, "season_activity_cards", "event_key", "TEXT NOT NULL DEFAULT ''")
self._ensure_column(conn, "season_activity_cards", "target_value", "INTEGER NOT NULL DEFAULT 1")
self._ensure_column(conn, "season_activity_cards", "progress_value", "INTEGER NOT NULL DEFAULT 0")
self._ensure_column(conn, "season_activity_cards", "card_type", "TEXT NOT NULL DEFAULT 'rotating'")
self._ensure_column(conn, "season_activity_cards", "source_pool_id", "TEXT NOT NULL DEFAULT ''")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_invitations_to_status ON invitations(to_user_id, status, created_at DESC)"
)
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.execute(
"CREATE INDEX IF NOT EXISTS idx_card_modules_base_status ON card_modules(base_card_id, status, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_points_tx_user_created ON user_point_transactions(user_id, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_point_lots_user_status_exp ON user_point_lots(user_id, status, expires_at, created_at)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_activation_codes_status_created ON activation_codes(status, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_activation_attempts_ip_created ON activation_code_attempts(ip_address, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_activation_attempts_user_created ON activation_code_attempts(user_id, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_shop_items_status_sort ON shop_items(status, sort_order DESC, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_shop_orders_user_created ON shop_orders(user_id, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_shop_orders_status_created ON shop_orders(status, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_entitlements_user_type_status ON user_entitlements(user_id, entitlement_type, status, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_seasons_status_time ON seasons(status, start_at, end_at)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_announce_season_created ON season_announcements(season_id, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_season_profile_updated ON user_season_profiles(season_id, updated_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_thread_user_season ON season_story_threads(user_id, season_id)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_cards_user_slot ON season_activity_cards(user_id, season_id, refresh_key, status, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_cards_status ON season_activity_cards(season_id, status, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_task_tpl_lookup ON season_task_templates(season_id, task_type, is_active, sort_order, updated_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_season_tasks_lookup ON user_season_tasks(user_id, season_id, task_type, status, refresh_key, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_season_tasks_event ON user_season_tasks(user_id, season_id, event_key, status)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_pass_reward_lookup ON season_pass_rewards(season_id, reward_level, reward_track, sort_order)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_reward_claims_user ON user_season_reward_claims(user_id, season_id, claimed_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_card_pool_lookup ON season_card_pools(season_id, card_type, is_active, weight, sort_order)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_decorations_lookup ON user_decorations(user_id, decoration_type, is_equipped, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_room_progress_unique ON season_room_progress_logs(user_id, season_id, room_id)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_season_daily_cards_season_created ON season_daily_game_cards(season_id, created_at DESC)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_user_season_intel_user_season ON user_season_intel_unlocks(user_id, season_id, created_at DESC)"
)
conn.commit()
self._seed_default_cards(conn)
self._seed_default_season_assets(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 _seed_default_season_assets(self, conn: sqlite3.Connection) -> None:
now_dt = datetime.now(timezone.utc)
now = now_dt.isoformat()
count_row = conn.execute("SELECT COUNT(*) AS c FROM seasons").fetchone()
season_count = int(dict(count_row or {}).get("c", 0))
season_id = ""
if season_count <= 0:
start_at = (now_dt - timedelta(days=1)).isoformat()
end_at = (now_dt + timedelta(days=29)).isoformat()
season_id = f"s{now_dt.strftime('%Y%m')}"
conn.execute(
"""
INSERT INTO seasons(
season_id, name, theme, description, cover_url, accent_color, story_seed, story_outline,
start_at, end_at, rest_end_at, status, source, pass_price, config_json, created_by, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
season_id,
"启航赛季",
"迷雾电波",
"首个赛季:在异常电波与雾夜都市中追踪真相碎片。",
"/static/img/cards/fogharbor.svg",
"#4f6dff",
"电波中反复提到一段被抹去的档案编号。",
"从零散异象入手,拼接一条完整的城市夜行叙事。",
start_at,
end_at,
(now_dt + timedelta(days=32)).isoformat(),
"active",
"seed",
680,
json.dumps(
{
"duration_days": 30,
"rest_days": 3,
"daily_card_count": 3,
"refresh_hours": [0, 8, 18],
"manual_refresh_cost": 60,
"xp_per_level": SEASON_XP_PER_LEVEL,
},
ensure_ascii=False,
),
"system",
now,
now,
),
)
else:
row = conn.execute(
"SELECT season_id FROM seasons ORDER BY start_at DESC, created_at DESC LIMIT 1"
).fetchone()
season_id = str(dict(row or {}).get("season_id", "")).strip()
if not season_id:
return
announce_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_announcements WHERE season_id=?",
(season_id,),
).fetchone()
if int(dict(announce_row or {}).get("c", 0)) <= 0:
conn.execute(
"""
INSERT INTO season_announcements(season_id, title, content, created_by, created_at)
VALUES (?, ?, ?, ?, ?)
""",
(
season_id,
"赛季开启公告",
"启航赛季现已开放:完成每日卡片、推进每周任务、冲击赛季等级,解锁限定称号与头像框。",
"system",
now,
),
)
template_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_task_templates WHERE season_id=?",
(season_id,),
).fetchone()
if int(dict(template_row or {}).get("c", 0)) <= 0:
templates = [
(
f"{season_id}_daily_play",
"daily",
"热身对局",
"完成 1 局任意剧本。",
"complete_game",
1,
"easy",
100,
16,
36,
4,
),
(
f"{season_id}_daily_action",
"daily",
"行动记录",
"累计提交 10 次行动。",
"submit_action",
10,
"easy",
95,
14,
34,
4,
),
(
f"{season_id}_daily_win",
"daily",
"小队首胜",
"赢下 1 局游戏。",
"win_game",
1,
"normal",
88,
18,
44,
5,
),
(
f"{season_id}_weekly_play",
"weekly",
"周常远征",
"本周完成 5 局游戏。",
"complete_game",
5,
"normal",
100,
52,
120,
14,
),
(
f"{season_id}_weekly_win",
"weekly",
"周常捷报",
"本周获得 3 场胜利。",
"win_game",
3,
"normal",
92,
68,
156,
18,
),
(
f"{season_id}_season_play",
"season",
"赛季长线",
"整个赛季累计完成 35 局游戏。",
"complete_game",
35,
"hard",
100,
220,
480,
80,
),
(
f"{season_id}_season_mvp",
"season",
"核心贡献者",
"累计获得 8 次 MVP。",
"mvp_count",
8,
"hard",
90,
250,
520,
90,
),
]
for idx, row in enumerate(templates):
conn.execute(
"""
INSERT INTO season_task_templates(
template_id, season_id, task_type, title, description, event_key, target_value,
difficulty, weight, reward_points, reward_xp, reward_season_coin,
reward_payload_json, constraints_json, refresh_policy, is_active, sort_order,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'auto', 1, ?, ?, ?)
""",
(
row[0],
season_id,
row[1],
row[2],
row[3],
row[4],
row[5],
row[6],
row[7],
row[8],
row[9],
row[10],
"{}",
"{}",
idx,
now,
now,
),
)
pool_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_card_pools WHERE season_id=?",
(season_id,),
).fetchone()
if int(dict(pool_row or {}).get("c", 0)) <= 0:
pools = [
(
f"{season_id}_daily_1",
"daily",
"雾幕巡查",
"追踪失真的无线电片段。",
"完成 1 局对局并记录关键线索。",
"优先完成基础目标,保留刷新次数。",
"complete_game",
1,
12,
28,
"normal",
120,
),
(
f"{season_id}_rotating_1",
"rotating",
"并行剧场",
"平行世界中的同一条街出现两套规则。",
"累计提交 8 次行动并触发一次团队协作。",
"如果队伍在线,优先冲刺协作玩法。",
"submit_action",
8,
20,
52,
"rare",
88,
),
(
f"{season_id}_surprise_1",
"surprise",
"暗潮来信",
"你收到一封未署名的旧案回执。",
"赢下 1 局并保持存活。",
"难度较高,建议组队尝试。",
"win_game",
1,
50,
120,
"epic",
20,
),
]
for idx, row in enumerate(pools):
conn.execute(
"""
INSERT INTO season_card_pools(
pool_id, season_id, card_type, title_hint, story_hook, objective_template, hint_template,
event_key, target_value, reward_points, reward_xp, rarity, weight, min_level,
payload_json, is_active, sort_order, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, '{}', 1, ?, ?, ?)
""",
(
row[0],
season_id,
row[1],
row[2],
row[3],
row[4],
row[5],
row[6],
row[7],
row[8],
row[9],
row[10],
row[11],
idx,
now,
now,
),
)
reward_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_pass_rewards WHERE season_id=?",
(season_id,),
).fetchone()
if int(dict(reward_row or {}).get("c", 0)) <= 0:
for level in range(1, 31):
free_reward_id = f"{season_id}_free_{level}"
premium_reward_id = f"{season_id}_premium_{level}"
free_points = 10 if level % 5 else 26
premium_points = 18 if level % 5 else 58
free_payload = {}
premium_payload = {}
if level == 8:
free_payload = {"titles": [f"{season_id}_title_pathfinder"]}
if level == 15:
premium_payload = {"avatar_frames": [f"{season_id}_frame_neon"]}
if level == 30:
premium_payload = {
"titles": [f"{season_id}_title_legend"],
"avatar_frames": [f"{season_id}_frame_legend"],
}
conn.execute(
"""
INSERT INTO season_pass_rewards(
reward_id, season_id, reward_level, reward_track, title, reward_points, reward_xp,
reward_season_coin, reward_payload_json, sort_order, created_at, updated_at
) VALUES (?, ?, ?, 'free', ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
free_reward_id,
season_id,
level,
f"免费档 Lv.{level}",
free_points,
0,
2 if level % 3 == 0 else 0,
json.dumps(free_payload, ensure_ascii=False),
level,
now,
now,
),
)
conn.execute(
"""
INSERT INTO season_pass_rewards(
reward_id, season_id, reward_level, reward_track, title, reward_points, reward_xp,
reward_season_coin, reward_payload_json, sort_order, created_at, updated_at
) VALUES (?, ?, ?, 'premium', ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
premium_reward_id,
season_id,
level,
f"高级档 Lv.{level}",
premium_points,
0,
6 if level % 2 == 0 else 3,
json.dumps(premium_payload, ensure_ascii=False),
level,
now,
now,
),
)
title_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_titles WHERE season_id=?",
(season_id,),
).fetchone()
if int(dict(title_row or {}).get("c", 0)) <= 0:
title_defs = [
(f"{season_id}_title_pathfinder", "雾幕先行者", "#9ec7ff", "rare"),
(f"{season_id}_title_legend", "电波传奇", "#ffd166", "legendary"),
]
for idx, row in enumerate(title_defs):
conn.execute(
"""
INSERT INTO season_titles(
title_id, season_id, name, color_code, rarity, acquire_type, image_url,
effect_json, is_active, sort_order, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, 'season_reward', '', '{}', 1, ?, ?, ?)
""",
(row[0], season_id, row[1], row[2], row[3], idx, now, now),
)
frame_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_avatar_frames WHERE season_id=?",
(season_id,),
).fetchone()
if int(dict(frame_row or {}).get("c", 0)) <= 0:
frame_defs = [
(f"{season_id}_frame_neon", "霓虹折线框", "rare"),
(f"{season_id}_frame_legend", "王冠回声框", "legendary"),
]
for idx, row in enumerate(frame_defs):
conn.execute(
"""
INSERT INTO season_avatar_frames(
frame_id, season_id, name, style_json, rarity, acquire_type, preview_url,
is_active, sort_order, created_at, updated_at
) VALUES (?, ?, ?, '{}', ?, 'season_reward', '', 1, ?, ?, ?)
""",
(row[0], season_id, row[1], row[2], idx, now, now),
)
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 list_actions(self, room_id: str, limit: int = 600) -> list[dict[str, Any]]:
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT id, room_id, round_no, player_id, content, created_at
FROM actions
WHERE room_id=?
ORDER BY id ASC
LIMIT ?
""",
(room_id, max(1, int(limit))),
).fetchall()
return [dict(row) for row in 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,
is_online, last_active_at, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id,
username,
password_hash,
display_name,
role,
"#0070f2",
"",
"dopamine",
"",
0,
"",
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 touch_user_activity(self, user_id: str, *, is_online: bool | None = True) -> None:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return
now = utc_now_iso()
with self._lock, self._connect() as conn:
if is_online is None:
conn.execute(
"""
UPDATE users
SET last_active_at=?, updated_at=?
WHERE user_id=?
""",
(now, now, safe_user_id),
)
else:
conn.execute(
"""
UPDATE users
SET is_online=?, last_active_at=?, updated_at=?
WHERE user_id=?
""",
(1 if is_online else 0, now, now, safe_user_id),
)
conn.commit()
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,
active_badge: 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"]
merged_active_badge = active_badge if active_badge is not None else existing.get("active_badge", "")
with self._lock, self._connect() as conn:
conn.execute(
"""
UPDATE users
SET display_name=?, avatar_color=?, avatar_url=?, theme=?, bio=?, active_badge=?, updated_at=?
WHERE user_id=?
""",
(
merged_name,
merged_color,
merged_avatar_url,
merged_theme,
merged_bio,
str(merged_active_badge or "").strip()[:36],
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 search_users(self, query: str, limit: int = 20) -> list[dict[str, Any]]:
text = str(query or "").strip()
try:
safe_limit = max(1, min(100, int(limit)))
except (TypeError, ValueError):
safe_limit = 20
with self._lock, self._connect() as conn:
if text:
like = f"%{text}%"
rows = conn.execute(
"""
SELECT
user_id,
username,
display_name,
role,
avatar_color,
avatar_url,
theme,
bio,
created_at,
updated_at
FROM users
WHERE username LIKE ? OR display_name LIKE ?
ORDER BY CASE WHEN username = ? THEN 0 ELSE 1 END ASC, updated_at DESC
LIMIT ?
""",
(like, like, text.lower(), safe_limit),
).fetchall()
else:
rows = conn.execute(
"""
SELECT
user_id,
username,
display_name,
role,
avatar_color,
avatar_url,
theme,
bio,
created_at,
updated_at
FROM users
ORDER BY updated_at DESC
LIMIT ?
""",
(safe_limit,),
).fetchall()
return [dict(row) for row in rows]
def _parse_iso_datetime(self, raw_value: Any) -> datetime | None:
text = str(raw_value or "").strip()
if not text:
return None
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed
def _is_recent_iso(self, raw_value: Any, seconds: int = PRESENCE_ACTIVE_WINDOW_SECONDS) -> bool:
parsed = self._parse_iso_datetime(raw_value)
if parsed is None:
return False
return (datetime.now(timezone.utc) - parsed) <= timedelta(seconds=max(1, int(seconds)))
def _max_iso(self, *values: Any) -> str:
latest_text = ""
latest_time: datetime | None = None
for value in values:
text = str(value or "").strip()
if not text:
continue
parsed = self._parse_iso_datetime(text)
if parsed is None:
continue
if latest_time is None or parsed > latest_time:
latest_time = parsed
latest_text = text
return latest_text
def get_user_presence(self, user_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return {
"is_online": False,
"session_online": False,
"active_room_id": "",
"active_room_status": "",
"active_player_id": "",
"active_player_name": "",
"last_seen": "",
}
with self._lock, self._connect() as conn:
user_row = conn.execute(
"SELECT is_online, last_active_at FROM users WHERE user_id=?",
(safe_user_id,),
).fetchone()
online_row = conn.execute(
"""
SELECT
p.player_id,
p.name,
p.room_id,
p.last_seen,
r.status
FROM players p
JOIN rooms r ON r.room_id = p.room_id
WHERE p.user_id=? AND p.is_online=1 AND r.status IN ('lobby', 'running')
ORDER BY p.last_seen DESC
LIMIT 1
""",
(safe_user_id,),
).fetchone()
active_row = conn.execute(
"""
SELECT
p.player_id,
p.name,
p.room_id,
p.last_seen,
p.is_online,
r.status
FROM players p
JOIN rooms r ON r.room_id = p.room_id
WHERE p.user_id=? AND r.status IN ('lobby', 'running')
ORDER BY p.is_online DESC, p.last_seen DESC
LIMIT 1
""",
(safe_user_id,),
).fetchone()
latest_row = conn.execute(
"""
SELECT last_seen
FROM players
WHERE user_id=?
ORDER BY last_seen DESC
LIMIT 1
""",
(safe_user_id,),
).fetchone()
user_payload = dict(user_row) if user_row else {}
session_online = bool(user_payload.get("is_online")) and self._is_recent_iso(
user_payload.get("last_active_at"), PRESENCE_ACTIVE_WINDOW_SECONDS
)
player_online = online_row is not None
is_online = bool(session_online or player_online)
active_payload = dict(online_row or active_row or {})
last_seen = self._max_iso(
user_payload.get("last_active_at"),
(dict(latest_row) if latest_row else {}).get("last_seen"),
active_payload.get("last_seen"),
)
return {
"is_online": is_online,
"session_online": session_online,
"active_room_id": str(active_payload.get("room_id", "")).strip() if is_online else "",
"active_room_status": str(active_payload.get("status", "")).strip() if is_online else "",
"active_player_id": str(active_payload.get("player_id", "")).strip() if is_online else "",
"active_player_name": str(active_payload.get("name", "")).strip() if is_online else "",
"last_seen": last_seen,
}
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 | None = 20) -> list[dict[str, Any]]:
safe_limit: int | None
if limit is None:
safe_limit = None
else:
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
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)
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] if safe_limit is not None else results
for idx, row in enumerate(ranked, start=1):
row["rank"] = idx
return ranked
def get_user_game_stats(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,
"win_rate": 0.0,
"total_rounds": 0,
"score": 0,
"rank": None,
"last_played_at": "",
}
leaderboard = self.list_player_leaderboard(limit=None)
row = next((item for item in leaderboard if str(item.get("user_id", "")) == safe_user_id), None)
if row is None:
return {
"played_games": 0,
"wins": 0,
"win_rate": 0.0,
"total_rounds": 0,
"score": 0,
"rank": None,
"last_played_at": "",
}
return {
"played_games": int(row.get("played_games", 0)),
"wins": int(row.get("wins", 0)),
"win_rate": float(row.get("win_rate", 0.0)),
"total_rounds": int(row.get("total_rounds", 0)),
"score": int(row.get("score", 0)),
"rank": int(row.get("rank", 0)) if row.get("rank") is not None else None,
"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 = {
"青铜": "bronze",
"白银": "silver",
"黄金": "gold",
"铂金": "platinum",
"钻石": "diamond",
}
if level in alias:
return alias[level]
if level in ACHIEVEMENT_LEVEL_ORDER:
return level
return "bronze"
def _normalize_achievement_metric(self, raw_metric: Any) -> str:
metric = str(raw_metric or "").strip().lower()
mapping = {
"plays": "plays",
"play_count": "plays",
"played_games": "plays",
"completed_games": "plays",
"wins": "wins",
"win_count": "wins",
"current_streak": "current_streak",
"win_streak": "current_streak",
"best_streak": "best_streak",
"max_win_streak": "best_streak",
"对局数": "plays",
"完成局数": "plays",
"胜场": "wins",
"连胜": "current_streak",
"最高连胜": "best_streak",
}
return mapping.get(metric, "plays")
def _normalize_custom_achievements(self, raw_rows: Any, *, limit: int = 20) -> list[dict[str, Any]]:
rows = raw_rows if isinstance(raw_rows, list) else []
out: list[dict[str, Any]] = []
seen: set[str] = set()
for idx, row in enumerate(rows[:limit]):
if not isinstance(row, dict):
continue
name = str(row.get("name", "")).strip()[:48]
if not name:
continue
raw_id = str(row.get("id", "")).strip().lower()
if raw_id:
ach_id = re.sub(r"[^a-z0-9_-]+", "_", raw_id).strip("_")[:64]
else:
seed = re.sub(r"[^a-z0-9_-]+", "_", name.lower()).strip("_")[:48] or f"custom_{idx+1}"
ach_id = f"{seed}_{idx+1}"[:64]
if not ach_id or ach_id in seen:
continue
seen.add(ach_id)
metric = self._normalize_achievement_metric(row.get("metric", "plays"))
tiers_raw = row.get("tiers", [])
tiers: list[dict[str, Any]] = []
if isinstance(tiers_raw, list):
for tier_idx, tier in enumerate(tiers_raw[:6]):
if not isinstance(tier, dict):
continue
level = self._normalize_achievement_level(tier.get("level", "bronze"))
try:
threshold = max(1, int(tier.get("threshold", 1)))
except (TypeError, ValueError):
threshold = 1
title = str(tier.get("title", "")).strip()[:48] or name
tiers.append(
{
"level": level,
"threshold": threshold,
"title": title,
"_idx": tier_idx,
}
)
if not tiers:
level = self._normalize_achievement_level(row.get("level", "bronze"))
try:
threshold = max(1, int(row.get("threshold", 1)))
except (TypeError, ValueError):
threshold = 1
tiers = [{"level": level, "threshold": threshold, "title": name, "_idx": 0}]
tiers.sort(
key=lambda item: (int(item.get("threshold", 1)), int(ACHIEVEMENT_LEVEL_ORDER.get(item["level"], 1)), item.get("_idx", 0))
)
for tier in tiers:
tier.pop("_idx", None)
out.append(
{
"id": ach_id,
"name": name,
"description": str(row.get("description", "")).strip()[:180],
"metric": metric,
"tiers": tiers[:6],
}
)
return out
def _compare_condition(self, actual: Any, op: str, expected: Any) -> bool:
numeric_ops = {">", ">=", "<", "<="}
if op in numeric_ops:
if not isinstance(actual, (int, float)) or not isinstance(expected, (int, float)):
return False
if op == ">":
return actual > expected
if op == ">=":
return actual >= expected
if op == "<":
return actual < expected
if op == "<=":
return actual <= expected
if op == "==":
return actual == expected
if op == "!=":
return actual != expected
return False
def _normalize_condition_achievements(self, raw_rows: Any, *, limit: int = 30) -> list[dict[str, Any]]:
rows = raw_rows if isinstance(raw_rows, list) else []
out: list[dict[str, Any]] = []
seen: set[str] = set()
allowed_types = {"ending_state", "flag", "metric", "round_no", "player_alive"}
allowed_ops = {">", ">=", "<", "<=", "==", "!="}
for idx, row in enumerate(rows[:limit]):
if not isinstance(row, dict):
continue
name = str(row.get("name", "")).strip()[:48]
if not name:
continue
raw_id = str(row.get("id", "")).strip().lower()
if raw_id:
ach_id = re.sub(r"[^a-z0-9_-]+", "_", raw_id).strip("_")[:64]
else:
seed = re.sub(r"[^a-z0-9_-]+", "_", name.lower()).strip("_")[:48] or f"condition_{idx+1}"
ach_id = f"{seed}_{idx+1}"[:64]
if not ach_id or ach_id in seen:
continue
seen.add(ach_id)
conditions_raw = row.get("conditions", [])
conditions: list[dict[str, Any]] = []
if isinstance(conditions_raw, list):
for cond in conditions_raw[:12]:
if not isinstance(cond, dict):
continue
cond_type = str(cond.get("type", "")).strip().lower()
if cond_type not in allowed_types:
continue
op = str(cond.get("op", "==")).strip()
if op not in allowed_ops:
op = "=="
item = {
"type": cond_type,
"key": str(cond.get("key", "")).strip()[:72],
"op": op,
"value": cond.get("value"),
}
if cond_type == "ending_state":
item["value"] = str(cond.get("value", "")).strip()[:48]
if cond_type in {"flag", "metric"} and not item["key"]:
continue
if cond_type == "player_alive" and not item["key"]:
item["key"] = "self"
conditions.append(item)
if not conditions:
continue
level = self._normalize_achievement_level(row.get("level", "bronze"))
out.append(
{
"id": ach_id,
"name": name,
"title": str(row.get("title", "")).strip()[:48] or name,
"description": str(row.get("description", "")).strip()[:180],
"level": level,
"conditions": conditions,
}
)
return out
def _condition_metric_value(self, game: dict[str, Any], key: str) -> Any:
state = game.get("game_state", {})
if not isinstance(state, dict):
return None
global_flags = state.get("global_flags", {})
if not isinstance(global_flags, dict):
global_flags = {}
metric_values = global_flags.get("metric_values", {})
if isinstance(metric_values, dict) and key in metric_values:
return metric_values.get(key)
metric_by_player = global_flags.get("metric_values_by_player", {})
player_id = str(game.get("player_id", "")).strip()
if isinstance(metric_by_player, dict) and player_id:
player_values = metric_by_player.get(player_id, {})
if isinstance(player_values, dict) and key in player_values:
return player_values.get(key)
return global_flags.get(key)
def _condition_player_alive(self, game: dict[str, Any], key: str) -> Any:
state = game.get("game_state", {})
if not isinstance(state, dict):
return None
global_flags = state.get("global_flags", {})
if not isinstance(global_flags, dict):
return None
player_states = global_flags.get("player_states", {})
if not isinstance(player_states, dict):
return None
player_id = str(game.get("player_id", "")).strip() if key in {"", "self"} else key
if not player_id:
return None
row = player_states.get(player_id, {})
if not isinstance(row, dict):
return None
return bool(row.get("is_alive", True))
def _count_matched_conditions(self, game: dict[str, Any], conditions: list[dict[str, Any]]) -> int:
state = game.get("game_state", {})
if not isinstance(state, dict):
state = {}
global_flags = state.get("global_flags", {})
if not isinstance(global_flags, dict):
global_flags = {}
matched = 0
for cond in conditions:
cond_type = str(cond.get("type", "")).strip().lower()
op = str(cond.get("op", "==")).strip() or "=="
key = str(cond.get("key", "")).strip()
expected = cond.get("value")
actual: Any = None
if cond_type == "ending_state":
actual = str(state.get("ending_state", "")).strip()
elif cond_type == "round_no":
actual = state.get("round_no", 0)
elif cond_type == "flag":
actual = global_flags.get(key)
elif cond_type == "metric":
actual = self._condition_metric_value(game, key)
elif cond_type == "player_alive":
actual = self._condition_player_alive(game, key)
if self._compare_condition(actual, op, expected):
matched += 1
return matched
def list_user_completed_games(self, user_id: str) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return []
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT
r.room_id,
r.game_card_id,
r.updated_at,
r.game_state,
p.player_id
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()
games: list[dict[str, Any]] = []
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 = {}
ending_state = str(state.get("ending_state", "")).strip()
try:
round_no = max(0, int(state.get("round_no", 0)))
except (TypeError, ValueError):
round_no = 0
games.append(
{
"room_id": str(item.get("room_id", "")).strip(),
"game_card_id": str(item.get("game_card_id", DEFAULT_CARD_ID)).strip() or DEFAULT_CARD_ID,
"player_id": str(item.get("player_id", "")).strip(),
"ended_at": str(item.get("updated_at", "")).strip(),
"ending_state": ending_state,
"is_win": self._is_winning_ending(ending_state),
"round_no": round_no,
"game_state": state,
}
)
return games
def get_user_achievement_bundle(self, user_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
empty_section = {
"unlocked": [],
"locked": [],
"all": [],
"unlocked_count": 0,
"total_count": 0,
}
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": [],
"official": dict(empty_section),
"card": dict(empty_section),
}
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
best_streak = 0
unlocked_system: list[dict[str, Any]] = []
system_seen: set[str] = set()
for game in games:
plays += 1
if game["is_win"]:
wins += 1
current_streak += 1
best_streak = max(best_streak, current_streak)
else:
current_streak = 0
metric_map = {
"plays": plays,
"wins": wins,
"current_streak": current_streak,
"best_streak": best_streak,
}
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(metric_map.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": game["ended_at"],
}
)
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,
}
)
system_unlock_index = {
f"{row.get('track_id')}:{row.get('level')}:{int(row.get('threshold', 0))}": row
for row in unlocked_system
}
card_games: dict[str, list[dict[str, Any]]] = {}
for game in games:
card_id = game["game_card_id"]
card_games.setdefault(card_id, []).append(game)
unlocked_card: list[dict[str, Any]] = []
for card_id, rows in card_games.items():
card = self.get_game_card(card_id) or {}
card_title = str(card.get("title", "未知剧本")).strip() or "未知剧本"
content_pack = card.get("content_pack", {})
custom_defs = self._normalize_custom_achievements(
content_pack.get("custom_achievements", []) if isinstance(content_pack, dict) else [],
limit=20,
)
condition_defs = self._normalize_condition_achievements(
content_pack.get("condition_achievements", []) if isinstance(content_pack, dict) else [],
limit=30,
)
if not custom_defs and not condition_defs:
continue
plays_count = 0
wins_count = 0
streak_count = 0
best_streak_count = 0
seen: set[str] = set()
for game in rows:
plays_count += 1
if game["is_win"]:
wins_count += 1
streak_count += 1
best_streak_count = max(best_streak_count, streak_count)
else:
streak_count = 0
metric_map = {
"plays": plays_count,
"wins": wins_count,
"current_streak": streak_count,
"best_streak": best_streak_count,
}
for ach in custom_defs:
ach_id = str(ach.get("id", "")).strip() or "card_achievement"
ach_name = str(ach.get("name", "")).strip()[:48] or "卡片成就"
ach_desc = str(ach.get("description", "")).strip()[:180]
metric_key = self._normalize_achievement_metric(ach.get("metric", "plays"))
metric_value = int(metric_map.get(metric_key, 0))
for tier in ach.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
key = f"{card_id}:{ach_id}:{level}:{threshold}"
if key in seen:
continue
if metric_value < threshold:
continue
seen.add(key)
unlocked_card.append(
{
"source": "card",
"track_id": ach_id,
"title": str(tier.get("title", "")).strip()[:48] or ach_name,
"name": ach_name,
"description": ach_desc,
"level": level,
"level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
"threshold": threshold,
"metric": metric_key,
"metric_value": metric_value,
"unlocked_at": game["ended_at"],
"card_id": card_id,
"card_title": card_title,
}
)
for ach in condition_defs:
ach_id = str(ach.get("id", "")).strip() or "card_condition"
ach_name = str(ach.get("name", "")).strip()[:48] or "条件成就"
ach_desc = str(ach.get("description", "")).strip()[:180]
level = self._normalize_achievement_level(ach.get("level", "bronze"))
key = f"{card_id}:{ach_id}:{level}:1"
if key in seen:
continue
conditions = ach.get("conditions", [])
if not isinstance(conditions, list):
conditions = []
if not conditions:
continue
if self._count_matched_conditions(game, conditions) < len(conditions):
continue
seen.add(key)
unlocked_card.append(
{
"source": "card",
"track_id": ach_id,
"title": str(ach.get("title", "")).strip()[:48] or ach_name,
"name": ach_name,
"description": ach_desc,
"level": level,
"level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
"threshold": 1,
"metric": "conditions",
"metric_value": 1,
"unlocked_at": game["ended_at"],
"card_id": card_id,
"card_title": card_title,
}
)
card_unlock_index = {
f"{row.get('card_id')}:{row.get('track_id')}:{row.get('level')}:{int(row.get('threshold', 0))}": row
for row in 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,
"progress_value": 1,
"progress_percent": 100,
"unlocked": True,
"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)),
}
)
official_rows: list[dict[str, Any]] = []
for track in SYSTEM_ACHIEVEMENT_TRACKS:
track_id = str(track.get("id", "")).strip()
track_name = str(track.get("name", "")).strip()[:48] or "系统成就"
track_desc = str(track.get("description", "")).strip()[:180]
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
idx_key = f"{track_id}:{level}:{threshold}"
unlocked = system_unlock_index.get(idx_key)
progress_value = min(metric_value, threshold)
progress_percent = min(100, int(round((progress_value / threshold) * 100))) if threshold > 0 else 0
official_rows.append(
{
"source": "system",
"track_id": track_id,
"track_name": track_name,
"title": str(tier.get("title", "")).strip()[:48] or track_name,
"name": track_name,
"description": track_desc,
"level": level,
"level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
"threshold": threshold,
"metric": metric_key,
"metric_value": metric_value,
"progress_value": progress_value,
"progress_percent": progress_percent,
"unlocked": bool(unlocked),
"unlocked_at": str((unlocked or {}).get("unlocked_at", "")).strip(),
}
)
official_rows.sort(
key=lambda row: (
not bool(row.get("unlocked")),
int(row.get("level_rank", 0)),
int(row.get("threshold", 0)),
str(row.get("track_id", "")),
)
)
all_cards = self.list_game_cards()
card_rows: list[dict[str, Any]] = []
for card in all_cards:
card_id = str(card.get("card_id", "")).strip()
if not card_id:
continue
card_title = str(card.get("title", "未知剧本")).strip() or "未知剧本"
rows = card_games.get(card_id, [])
content_pack = card.get("content_pack", {})
custom_defs = self._normalize_custom_achievements(
content_pack.get("custom_achievements", []) if isinstance(content_pack, dict) else [],
limit=20,
)
condition_defs = self._normalize_condition_achievements(
content_pack.get("condition_achievements", []) if isinstance(content_pack, dict) else [],
limit=30,
)
if not custom_defs and not condition_defs:
continue
plays_count = 0
wins_count = 0
streak_count = 0
best_streak_count = 0
for row in rows:
plays_count += 1
if row.get("is_win"):
wins_count += 1
streak_count += 1
best_streak_count = max(best_streak_count, streak_count)
else:
streak_count = 0
metric_map = {
"plays": plays_count,
"wins": wins_count,
"current_streak": streak_count,
"best_streak": best_streak_count,
}
for ach in custom_defs:
ach_id = str(ach.get("id", "")).strip() or "card_achievement"
ach_name = str(ach.get("name", "")).strip()[:48] or "卡片成就"
ach_desc = str(ach.get("description", "")).strip()[:180]
metric_key = self._normalize_achievement_metric(ach.get("metric", "plays"))
metric_value = int(metric_map.get(metric_key, 0))
for tier in ach.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
idx_key = f"{card_id}:{ach_id}:{level}:{threshold}"
unlocked = card_unlock_index.get(idx_key)
progress_value = min(metric_value, threshold)
progress_percent = min(100, int(round((progress_value / threshold) * 100))) if threshold > 0 else 0
card_rows.append(
{
"source": "card",
"track_id": ach_id,
"track_name": ach_name,
"title": str(tier.get("title", "")).strip()[:48] or ach_name,
"name": ach_name,
"description": ach_desc,
"level": level,
"level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
"threshold": threshold,
"metric": metric_key,
"metric_value": metric_value,
"progress_value": progress_value,
"progress_percent": progress_percent,
"unlocked": bool(unlocked),
"unlocked_at": str((unlocked or {}).get("unlocked_at", "")).strip(),
"card_id": card_id,
"card_title": card_title,
}
)
for ach in condition_defs:
ach_id = str(ach.get("id", "")).strip() or "card_condition"
ach_name = str(ach.get("name", "")).strip()[:48] or "条件成就"
ach_desc = str(ach.get("description", "")).strip()[:180]
level = self._normalize_achievement_level(ach.get("level", "bronze"))
idx_key = f"{card_id}:{ach_id}:{level}:1"
unlocked = card_unlock_index.get(idx_key)
conditions = ach.get("conditions", [])
if not isinstance(conditions, list):
conditions = []
threshold = max(1, len(conditions))
best_match = 0
for row in rows:
matched = self._count_matched_conditions(row, conditions)
if matched > best_match:
best_match = matched
progress_value = min(best_match, threshold)
progress_percent = min(100, int(round((progress_value / threshold) * 100))) if threshold > 0 else 0
card_rows.append(
{
"source": "card",
"track_id": ach_id,
"track_name": ach_name,
"title": str(ach.get("title", "")).strip()[:48] or ach_name,
"name": ach_name,
"description": ach_desc,
"level": level,
"level_rank": int(ACHIEVEMENT_LEVEL_ORDER.get(level, 1)),
"threshold": threshold,
"metric": "conditions",
"metric_value": best_match,
"progress_value": progress_value,
"progress_percent": progress_percent,
"unlocked": bool(unlocked),
"unlocked_at": str((unlocked or {}).get("unlocked_at", "")).strip(),
"card_id": card_id,
"card_title": card_title,
}
)
card_rows.sort(
key=lambda row: (
not bool(row.get("unlocked")),
str(row.get("card_title", "")),
int(row.get("level_rank", 0)),
int(row.get("threshold", 0)),
str(row.get("track_id", "")),
)
)
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
for item in all_items:
rank = int(item.get("level_rank", 0))
if rank > highest_rank:
highest_rank = rank
highest = str(item.get("level", "bronze")).strip()
return {
"summary": {
"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": int(final_metrics["plays"]),
"wins": int(final_metrics["wins"]),
"best_streak": int(final_metrics["best_streak"]),
},
"items": all_items[:120],
"official": {
"unlocked": [row for row in official_rows if row.get("unlocked")] + unlocked_gm[:120],
"locked": [row for row in official_rows if not row.get("unlocked")],
"all": official_rows + unlocked_gm[:120],
"unlocked_count": len([row for row in official_rows if row.get("unlocked")]) + len(unlocked_gm),
"total_count": len(official_rows) + len(unlocked_gm),
},
"card": {
"unlocked": [row for row in card_rows if row.get("unlocked")],
"locked": [row for row in card_rows if not row.get("unlocked")],
"all": card_rows,
"unlocked_count": len([row for row in card_rows if row.get("unlocked")]),
"total_count": len(card_rows),
},
}
def _json_object(self, raw_value: Any) -> dict[str, Any]:
if isinstance(raw_value, dict):
return raw_value
if not isinstance(raw_value, str):
return {}
try:
payload = json.loads(raw_value)
except json.JSONDecodeError:
return {}
return payload if isinstance(payload, dict) else {}
def _json_list(self, raw_value: Any) -> list[Any]:
if isinstance(raw_value, list):
return raw_value
if not isinstance(raw_value, str):
return []
try:
payload = json.loads(raw_value)
except json.JSONDecodeError:
return []
return payload if isinstance(payload, list) else []
def _season_status_by_time(self, row: dict[str, Any], *, now_iso: str = "") -> str:
now_text = str(now_iso or "").strip() or utc_now_iso()
start_at = str(row.get("start_at", "")).strip()
end_at = str(row.get("end_at", "")).strip()
rest_end_at = str(row.get("rest_end_at", "")).strip()
manual_status = str(row.get("status", "")).strip().lower()
if manual_status in {"settling", "结算中"}:
return "settling"
if manual_status in {"ended", "已结束", "archived"}:
return "ended"
if start_at and now_text < start_at:
return "upcoming"
if end_at and now_text >= end_at:
if rest_end_at and now_text < rest_end_at:
return "rest"
return "ended"
if manual_status in {"draft", "未开始"}:
return "upcoming"
return "running"
def _build_season_payload(self, row: sqlite3.Row | dict[str, Any], *, now_iso: str = "") -> dict[str, Any]:
item = dict(row)
item["pass_price"] = self._to_int(item.get("pass_price", 680), 680, 0)
item["config"] = self._json_object(item.get("config_json", "{}"))
item["time_status"] = self._season_status_by_time(item, now_iso=now_iso)
return item
def list_seasons(self, *, limit: int = 20) -> list[dict[str, Any]]:
safe_limit = max(1, min(200, int(limit)))
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT *
FROM seasons
ORDER BY start_at DESC, created_at DESC
LIMIT ?
""",
(safe_limit,),
).fetchall()
now_iso = utc_now_iso()
return [self._build_season_payload(row, now_iso=now_iso) for row in rows]
def get_season(self, season_id: str) -> dict[str, Any] | None:
safe_id = str(season_id or "").strip()
if not safe_id:
return None
with self._lock, self._connect() as conn:
row = conn.execute("SELECT * FROM seasons WHERE season_id=?", (safe_id,)).fetchone()
if row is None:
return None
return self._build_season_payload(row, now_iso=utc_now_iso())
def get_active_season(self) -> dict[str, Any] | None:
seasons = self.list_seasons(limit=24)
now_iso = utc_now_iso()
for row in seasons:
if self._season_status_by_time(row, now_iso=now_iso) == "running":
payload = dict(row)
payload["time_status"] = "running"
return payload
return None
def get_season_by_time_status(self, time_status: str) -> dict[str, Any] | None:
safe_status = str(time_status or "").strip().lower()
if safe_status not in {"upcoming", "running", "rest", "ended", "settling"}:
return None
seasons = self.list_seasons(limit=24)
now_iso = utc_now_iso()
for row in seasons:
if self._season_status_by_time(row, now_iso=now_iso) == safe_status:
payload = dict(row)
payload["time_status"] = safe_status
return payload
return None
def create_season(
self,
*,
name: str,
theme: str = "",
description: str = "",
cover_url: str = "",
accent_color: str = "#4f6dff",
story_seed: str = "",
story_outline: str = "",
start_at: str = "",
end_at: str = "",
status: str = "active",
source: str = "manual",
pass_price: int = 680,
config: dict[str, Any] | None = None,
created_by: str = "",
season_id: str = "",
) -> dict[str, Any]:
safe_name = str(name or "").strip()[:80]
if not safe_name:
raise ValueError("invalid_card_payload")
safe_theme = str(theme or "").strip()[:80]
safe_description = str(description or "").strip()[:360]
safe_cover = str(cover_url or "").strip()[:720]
safe_accent = str(accent_color or "").strip()[:16] or "#4f6dff"
safe_story_seed = str(story_seed or "").strip()[:240]
safe_story_outline = str(story_outline or "").strip()[:1200]
safe_start = str(start_at or "").strip()[:40] or utc_now_iso()
safe_end = str(end_at or "").strip()[:40]
if not safe_end:
parsed = self._parse_iso_datetime(safe_start)
if parsed is None:
parsed = datetime.now(timezone.utc)
safe_end = (parsed + timedelta(days=30)).isoformat()
if safe_end <= safe_start:
raise ValueError("invalid_card_payload")
safe_config = config if isinstance(config, dict) else {}
rest_days = self._to_int(safe_config.get("rest_days", 0), 0, 0)
rest_end = ""
end_dt = self._parse_iso_datetime(safe_end)
if end_dt is not None and rest_days > 0:
rest_end = (end_dt + timedelta(days=rest_days)).isoformat()
safe_status = str(status or "active").strip().lower()[:24] or "active"
safe_source = str(source or "manual").strip().lower()[:24] or "manual"
safe_creator = str(created_by or "").strip()[:64]
safe_pass_price = self._to_int(pass_price, 680, 0)
safe_id = str(season_id or "").strip().lower()[:64]
if not safe_id:
seed = re.sub(r"[^a-z0-9]+", "-", safe_name.lower()).strip("-")[:36] or "season"
safe_id = f"{seed}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"[:64]
now = utc_now_iso()
with self._lock, self._connect() as conn:
exists = conn.execute("SELECT season_id FROM seasons WHERE season_id=?", (safe_id,)).fetchone()
if exists:
raise ValueError("invalid_card_payload")
conn.execute(
"""
INSERT INTO seasons(
season_id, name, theme, description, cover_url, accent_color, story_seed, story_outline,
start_at, end_at, rest_end_at, status, source, pass_price, config_json, created_by, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_id,
safe_name,
safe_theme,
safe_description,
safe_cover,
safe_accent,
safe_story_seed,
safe_story_outline,
safe_start,
safe_end,
rest_end,
safe_status,
safe_source,
safe_pass_price,
json.dumps(safe_config, ensure_ascii=False),
safe_creator,
now,
now,
),
)
conn.commit()
season = self.get_season(safe_id)
if season is None:
raise ValueError("invalid_response")
return season
def update_season(self, season_id: str, patch: dict[str, Any]) -> dict[str, Any]:
current = self.get_season(season_id)
if current is None:
raise ValueError("invalid_card_id")
payload = patch if isinstance(patch, dict) else {}
merged = dict(current)
for key, limit in (
("name", 80),
("theme", 80),
("description", 360),
("cover_url", 720),
("accent_color", 16),
("story_seed", 240),
("story_outline", 1200),
("status", 24),
("source", 24),
("start_at", 40),
("end_at", 40),
):
if key in payload:
merged[key] = str(payload.get(key, "")).strip()[:limit]
if "pass_price" in payload:
merged["pass_price"] = self._to_int(payload.get("pass_price", merged.get("pass_price", 680)), 680, 0)
rest_end_from_payload = None
if "rest_end_at" in payload:
rest_end_from_payload = str(payload.get("rest_end_at", "")).strip()[:40]
merged["rest_end_at"] = rest_end_from_payload
config_patch = payload.get("config")
if isinstance(config_patch, dict):
current_cfg = current.get("config", {})
merged["config"] = {**(current_cfg if isinstance(current_cfg, dict) else {}), **config_patch}
else:
merged["config"] = current.get("config", {})
start_at = str(merged.get("start_at", "")).strip()
end_at = str(merged.get("end_at", "")).strip()
if start_at and end_at and end_at <= start_at:
raise ValueError("invalid_card_payload")
rest_days = self._to_int(merged.get("config", {}).get("rest_days", 0), 0, 0)
if rest_end_from_payload is None and rest_days > 0 and end_at:
end_dt = self._parse_iso_datetime(end_at)
if end_dt is not None:
merged["rest_end_at"] = (end_dt + timedelta(days=rest_days)).isoformat()
with self._lock, self._connect() as conn:
conn.execute(
"""
UPDATE seasons
SET name=?, theme=?, description=?, cover_url=?, accent_color=?, story_seed=?, story_outline=?,
start_at=?, end_at=?, rest_end_at=?, status=?, source=?, pass_price=?, config_json=?, updated_at=?
WHERE season_id=?
""",
(
str(merged.get("name", "")).strip()[:80],
str(merged.get("theme", "")).strip()[:80],
str(merged.get("description", "")).strip()[:360],
str(merged.get("cover_url", "")).strip()[:720],
str(merged.get("accent_color", "#4f6dff")).strip()[:16] or "#4f6dff",
str(merged.get("story_seed", "")).strip()[:240],
str(merged.get("story_outline", "")).strip()[:1200],
start_at,
end_at,
str(merged.get("rest_end_at", "")).strip()[:40],
str(merged.get("status", "active")).strip()[:24] or "active",
str(merged.get("source", "manual")).strip()[:24] or "manual",
self._to_int(merged.get("pass_price", 680), 680, 0),
json.dumps(merged.get("config", {}), ensure_ascii=False),
utc_now_iso(),
str(season_id or "").strip(),
),
)
conn.commit()
updated = self.get_season(season_id)
if updated is None:
raise ValueError("invalid_card_id")
return updated
def list_season_announcements(self, season_id: str, *, limit: int = 20) -> list[dict[str, Any]]:
safe_id = str(season_id or "").strip()
if not safe_id:
return []
safe_limit = max(1, min(200, int(limit)))
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT id, season_id, title, content, created_by, created_at
FROM season_announcements
WHERE season_id=?
ORDER BY created_at DESC, id DESC
LIMIT ?
""",
(safe_id, safe_limit),
).fetchall()
return [dict(row) for row in rows]
def create_season_announcement(
self,
season_id: str,
*,
title: str,
content: str = "",
created_by: str = "",
) -> dict[str, Any]:
safe_id = str(season_id or "").strip()
if not safe_id or self.get_season(safe_id) is None:
raise ValueError("invalid_card_id")
safe_title = str(title or "").strip()[:120]
if not safe_title:
raise ValueError("invalid_card_payload")
safe_content = str(content or "").strip()[:2400]
safe_creator = str(created_by or "").strip()[:64]
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute(
"""
INSERT INTO season_announcements(season_id, title, content, created_by, created_at)
VALUES (?, ?, ?, ?, ?)
""",
(safe_id, safe_title, safe_content, safe_creator, now),
)
row_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
conn.commit()
row = conn.execute(
"""
SELECT id, season_id, title, content, created_by, created_at
FROM season_announcements
WHERE id=?
""",
(row_id,),
).fetchone()
return dict(row) if row else {}
def ensure_season_start_announcement(self, season_id: str) -> None:
safe_id = str(season_id or "").strip()
if not safe_id:
return
with self._lock, self._connect() as conn:
row = conn.execute(
"SELECT COUNT(*) AS c FROM season_announcements WHERE season_id=?",
(safe_id,),
).fetchone()
if int(dict(row or {}).get("c", 0)) > 0:
return
season = conn.execute("SELECT name, theme FROM seasons WHERE season_id=?", (safe_id,)).fetchone()
season_row = dict(season or {})
title = f"{season_row.get('name', '新赛季')} 开启公告"
content = (
f"欢迎进入 {season_row.get('name', '新赛季')}。\n"
f"主题:{season_row.get('theme', '未命名主题')}\n"
"每日刷新活动卡片已上线,完成任务可提升赛季等级并领取限定奖励。"
)
conn.execute(
"""
INSERT INTO season_announcements(season_id, title, content, created_by, created_at)
VALUES (?, ?, ?, 'system', ?)
""",
(safe_id, title[:120], content[:2400], utc_now_iso()),
)
conn.commit()
def _season_level_from_xp(self, xp_value: Any, *, xp_per_level: int = SEASON_XP_PER_LEVEL) -> int:
safe_xp = self._to_int(xp_value, 0, 0)
per_level = max(20, int(xp_per_level))
level = safe_xp // per_level + 1
return max(1, min(SEASON_MAX_LEVEL, level))
def _season_profile_payload(self, row: sqlite3.Row | dict[str, Any], *, season: dict[str, Any] | None = None) -> dict[str, Any]:
item = dict(row)
xp = self._to_int(item.get("xp", 0), 0, 0)
season_cfg = (season or {}).get("config", {}) if isinstance(season, dict) else {}
if not isinstance(season_cfg, dict):
season_cfg = {}
xp_per_level = self._to_int(season_cfg.get("xp_per_level", SEASON_XP_PER_LEVEL), SEASON_XP_PER_LEVEL, 20)
level = max(
self._to_int(item.get("level", 1), 1, 1),
self._season_level_from_xp(xp, xp_per_level=xp_per_level),
)
level_start_xp = (level - 1) * xp_per_level
next_level_xp = level * xp_per_level
return {
"user_id": str(item.get("user_id", "")).strip(),
"season_id": str(item.get("season_id", "")).strip(),
"pass_type": str(item.get("pass_type", "free")).strip() or "free",
"level": level,
"xp": xp,
"xp_per_level": xp_per_level,
"xp_in_current_level": max(0, xp - level_start_xp),
"xp_to_next_level": max(0, next_level_xp - xp),
"season_points": self._to_int(item.get("season_points", 0), 0, 0),
"total_claimed_cards": self._to_int(item.get("total_claimed_cards", 0), 0, 0),
"completed_tasks": self._to_int(item.get("completed_tasks", 0), 0, 0),
"last_refresh_key": str(item.get("last_refresh_key", "")).strip(),
"last_generated_at": str(item.get("last_generated_at", "")).strip(),
"created_at": str(item.get("created_at", "")).strip(),
"updated_at": str(item.get("updated_at", "")).strip(),
}
def get_or_create_user_season_profile(self, user_id: str, season_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
if not safe_user_id or not safe_season_id:
raise ValueError("user_not_found")
if self.get_user_by_id(safe_user_id) is None:
raise ValueError("user_not_found")
season = self.get_season(safe_season_id)
if season is None:
raise ValueError("invalid_card_id")
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute(
"""
INSERT INTO user_season_profiles(
user_id, season_id, pass_type, level, xp, season_points, total_claimed_cards,
completed_tasks, last_refresh_key, last_generated_at, created_at, updated_at
) VALUES (?, ?, 'free', 1, 0, 0, 0, 0, '', '', ?, ?)
ON CONFLICT(user_id, season_id) DO NOTHING
""",
(safe_user_id, safe_season_id, now, now),
)
row = conn.execute(
"""
SELECT user_id, season_id, pass_type, level, xp, season_points, total_claimed_cards,
completed_tasks, last_refresh_key, last_generated_at, created_at, updated_at
FROM user_season_profiles
WHERE user_id=? AND season_id=?
""",
(safe_user_id, safe_season_id),
).fetchone()
conn.commit()
if row is None:
raise ValueError("invalid_response")
return self._season_profile_payload(row, season=season)
def apply_user_season_gain(
self,
user_id: str,
season_id: str,
*,
xp_delta: int = 0,
season_points_delta: int = 0,
add_completed_task: bool = False,
add_claimed_card: bool = False,
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
season = self.get_season(safe_season_id)
if not safe_user_id or season is None:
raise ValueError("invalid_card_id")
profile = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
next_xp = self._to_int(profile.get("xp", 0), 0, 0) + self._to_int(xp_delta, 0, 0)
next_points = self._to_int(profile.get("season_points", 0), 0, 0) + self._to_int(season_points_delta, 0, 0)
next_completed = self._to_int(profile.get("completed_tasks", 0), 0, 0) + (1 if add_completed_task else 0)
next_claimed_cards = self._to_int(profile.get("total_claimed_cards", 0), 0, 0) + (1 if add_claimed_card else 0)
next_level = self._season_level_from_xp(
next_xp,
xp_per_level=self._to_int((season.get("config", {}) or {}).get("xp_per_level", SEASON_XP_PER_LEVEL), SEASON_XP_PER_LEVEL, 20),
)
with self._lock, self._connect() as conn:
conn.execute(
"""
UPDATE user_season_profiles
SET level=?, xp=?, season_points=?, completed_tasks=?, total_claimed_cards=?, updated_at=?
WHERE user_id=? AND season_id=?
""",
(
next_level,
next_xp,
next_points,
next_completed,
next_claimed_cards,
utc_now_iso(),
safe_user_id,
safe_season_id,
),
)
conn.commit()
return self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
def unlock_user_season_pass(self, user_id: str, season_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
season = self.get_season(safe_season_id)
if not safe_user_id or season is None:
raise ValueError("invalid_card_id")
profile = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
if str(profile.get("pass_type", "free")).strip() == "premium":
return profile
pass_price = self._to_int(season.get("pass_price", 680), 680, 0)
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
self._expire_point_lots_tx(conn, user_id=safe_user_id)
self._debit_points_tx(
conn,
safe_user_id,
pass_price,
tx_type="season_pass_unlock",
source="season",
related_id=safe_season_id,
note=f"解锁赛季通行证 {safe_season_id}",
count_spent=True,
)
conn.execute(
"""
UPDATE user_season_profiles
SET pass_type='premium', updated_at=?
WHERE user_id=? AND season_id=?
""",
(utc_now_iso(), safe_user_id, safe_season_id),
)
conn.commit()
return self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
def _build_season_task_template_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
item["target_value"] = self._to_int(item.get("target_value", 1), 1, 1)
item["weight"] = self._to_int(item.get("weight", 100), 100, 1)
item["reward_points"] = self._to_int(item.get("reward_points", 0), 0, 0)
item["reward_xp"] = self._to_int(item.get("reward_xp", 0), 0, 0)
item["reward_season_coin"] = self._to_int(item.get("reward_season_coin", 0), 0, 0)
item["sort_order"] = self._to_int(item.get("sort_order", 0), 0, 0)
item["constraints"] = self._json_object(item.get("constraints_json", "{}"))
item["reward_payload"] = self._json_object(item.get("reward_payload_json", "{}"))
item["is_active"] = bool(item.get("is_active", 0))
return item
def list_season_task_templates(
self,
season_id: str,
*,
task_type: str = "",
active_only: bool = False,
limit: int = 300,
) -> list[dict[str, Any]]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id:
return []
safe_task_type = str(task_type or "").strip().lower()
safe_limit = max(1, min(2000, int(limit)))
where = ["season_id=?"]
params: list[Any] = [safe_season_id]
if safe_task_type:
where.append("task_type=?")
params.append(safe_task_type)
if active_only:
where.append("is_active=1")
where_sql = " AND ".join(where)
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM season_task_templates
WHERE {where_sql}
ORDER BY sort_order ASC, updated_at DESC, created_at DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
return [self._build_season_task_template_payload(row) for row in rows]
def upsert_season_task_template(
self,
season_id: str,
payload: dict[str, Any],
*,
template_id: str = "",
) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id or self.get_season(safe_season_id) is None:
raise ValueError("invalid_card_id")
raw = payload if isinstance(payload, dict) else {}
safe_task_type = str(raw.get("task_type", "daily")).strip().lower()
if safe_task_type not in SEASON_TASK_TYPES:
safe_task_type = "daily"
safe_title = str(raw.get("title", "")).strip()[:84]
if not safe_title:
raise ValueError("invalid_card_payload")
safe_description = str(raw.get("description", "")).strip()[:240]
safe_event_key = str(raw.get("event_key", "")).strip().lower()[:48]
safe_target = self._to_int(raw.get("target_value", 1), 1, 1)
safe_difficulty = str(raw.get("difficulty", "normal")).strip().lower()[:24] or "normal"
safe_weight = self._to_int(raw.get("weight", 100), 100, 1)
safe_reward_points = self._to_int(raw.get("reward_points", 0), 0, 0)
safe_reward_xp = self._to_int(raw.get("reward_xp", 0), 0, 0)
safe_reward_coin = self._to_int(raw.get("reward_season_coin", 0), 0, 0)
safe_constraints = raw.get("constraints", {})
safe_reward_payload = raw.get("reward_payload", {})
safe_refresh_policy = str(raw.get("refresh_policy", "auto")).strip().lower()[:24] or "auto"
safe_active = bool(raw.get("is_active", True))
safe_sort = self._to_int(raw.get("sort_order", 0), 0, 0)
now = utc_now_iso()
safe_template_id = str(template_id or raw.get("template_id", "")).strip()[:80]
if not safe_template_id:
seed = re.sub(r"[^a-z0-9]+", "_", safe_title.lower()).strip("_")[:32] or "task"
safe_template_id = f"{safe_season_id}_{safe_task_type}_{seed}_{secrets.token_hex(3)}"[:80]
with self._lock, self._connect() as conn:
row = conn.execute(
"SELECT template_id FROM season_task_templates WHERE template_id=?",
(safe_template_id,),
).fetchone()
if row:
conn.execute(
"""
UPDATE season_task_templates
SET season_id=?, task_type=?, title=?, description=?, event_key=?, target_value=?,
difficulty=?, weight=?, reward_points=?, reward_xp=?, reward_season_coin=?,
reward_payload_json=?, constraints_json=?, refresh_policy=?, is_active=?, sort_order=?,
updated_at=?
WHERE template_id=?
""",
(
safe_season_id,
safe_task_type,
safe_title,
safe_description,
safe_event_key,
safe_target,
safe_difficulty,
safe_weight,
safe_reward_points,
safe_reward_xp,
safe_reward_coin,
json.dumps(safe_reward_payload if isinstance(safe_reward_payload, dict) else {}, ensure_ascii=False),
json.dumps(safe_constraints if isinstance(safe_constraints, dict) else {}, ensure_ascii=False),
safe_refresh_policy,
1 if safe_active else 0,
safe_sort,
now,
safe_template_id,
),
)
else:
conn.execute(
"""
INSERT INTO season_task_templates(
template_id, season_id, task_type, title, description, event_key, target_value,
difficulty, weight, reward_points, reward_xp, reward_season_coin,
reward_payload_json, constraints_json, refresh_policy, is_active, sort_order,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_template_id,
safe_season_id,
safe_task_type,
safe_title,
safe_description,
safe_event_key,
safe_target,
safe_difficulty,
safe_weight,
safe_reward_points,
safe_reward_xp,
safe_reward_coin,
json.dumps(safe_reward_payload if isinstance(safe_reward_payload, dict) else {}, ensure_ascii=False),
json.dumps(safe_constraints if isinstance(safe_constraints, dict) else {}, ensure_ascii=False),
safe_refresh_policy,
1 if safe_active else 0,
safe_sort,
now,
now,
),
)
conn.commit()
row = conn.execute(
"SELECT * FROM season_task_templates WHERE template_id=?",
(safe_template_id,),
).fetchone()
if row is None:
raise ValueError("invalid_response")
return self._build_season_task_template_payload(row)
def _season_daily_key(self, now_text: str) -> str:
parsed = self._parse_iso_datetime(now_text)
if parsed is None:
parsed = datetime.now(timezone.utc)
return parsed.strftime("%Y%m%d")
def _season_weekly_key(self, now_text: str) -> str:
parsed = self._parse_iso_datetime(now_text)
if parsed is None:
parsed = datetime.now(timezone.utc)
year, week, _ = parsed.isocalendar()
return f"{year}W{week:02d}"
def _weighted_sample_rows(self, rows: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
if count <= 0 or not rows:
return []
pool = list(rows)
selected: list[dict[str, Any]] = []
while pool and len(selected) < count:
total = sum(max(1, int(item.get("weight", 1))) for item in pool)
pick = secrets.randbelow(max(1, total)) + 1
cursor = 0
chosen_idx = 0
for idx, item in enumerate(pool):
cursor += max(1, int(item.get("weight", 1)))
if pick <= cursor:
chosen_idx = idx
break
selected.append(pool.pop(chosen_idx))
return selected
def _instantiate_user_tasks_tx(
self,
conn: sqlite3.Connection,
*,
user_id: str,
season_id: str,
task_type: str,
refresh_key: str,
count: int,
expires_at: str,
) -> int:
existing_row = conn.execute(
"""
SELECT COUNT(*) AS c
FROM user_season_tasks
WHERE user_id=? AND season_id=? AND task_type=? AND refresh_key=?
""",
(user_id, season_id, task_type, refresh_key),
).fetchone()
if int(dict(existing_row or {}).get("c", 0)) > 0:
return 0
template_rows = conn.execute(
"""
SELECT *
FROM season_task_templates
WHERE season_id=? AND task_type=? AND is_active=1
ORDER BY sort_order ASC, weight DESC, updated_at DESC
""",
(season_id, task_type),
).fetchall()
templates = [self._build_season_task_template_payload(row) for row in template_rows]
if not templates:
return 0
if task_type in {"season", "achievement"}:
selected = templates
else:
selected = self._weighted_sample_rows(templates, max(1, count))
now = utc_now_iso()
created = 0
for template in selected:
conn.execute(
"""
INSERT OR IGNORE INTO user_season_tasks(
user_id, season_id, template_id, task_type, refresh_key, title, description,
event_key, target_value, progress_value, status, reward_points, reward_xp, reward_season_coin,
reward_payload_json, constraints_json, expires_at, claimed_at, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 'active', ?, ?, ?, ?, ?, ?, '', ?, ?)
""",
(
user_id,
season_id,
str(template.get("template_id", "")),
task_type,
refresh_key,
str(template.get("title", "")).strip()[:84],
str(template.get("description", "")).strip()[:240],
str(template.get("event_key", "")).strip().lower()[:48],
self._to_int(template.get("target_value", 1), 1, 1),
self._to_int(template.get("reward_points", 0), 0, 0),
self._to_int(template.get("reward_xp", 0), 0, 0),
self._to_int(template.get("reward_season_coin", 0), 0, 0),
json.dumps(template.get("reward_payload", {}), ensure_ascii=False),
json.dumps(template.get("constraints", {}), ensure_ascii=False),
expires_at,
now,
now,
),
)
created += int(conn.total_changes > 0)
return created
def ensure_user_season_tasks(self, user_id: str, season_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
if not safe_user_id or not safe_season_id:
raise ValueError("user_not_found")
season = self.get_season(safe_season_id)
if season is None:
raise ValueError("invalid_card_id")
self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
now_iso = utc_now_iso()
day_key = self._season_daily_key(now_iso)
week_key = self._season_weekly_key(now_iso)
parsed_now = self._parse_iso_datetime(now_iso) or datetime.now(timezone.utc)
day_expire = (parsed_now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
week_expire = (parsed_now + timedelta(days=8)).replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
season_cfg = season.get("config", {})
if not isinstance(season_cfg, dict):
season_cfg = {}
daily_count = self._to_int(season_cfg.get("daily_task_count", 3), 3, 1)
weekly_count = self._to_int(season_cfg.get("weekly_task_count", 3), 3, 1)
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
daily_created = self._instantiate_user_tasks_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
task_type="daily",
refresh_key=day_key,
count=daily_count,
expires_at=day_expire,
)
weekly_created = self._instantiate_user_tasks_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
task_type="weekly",
refresh_key=week_key,
count=weekly_count,
expires_at=week_expire,
)
season_created = self._instantiate_user_tasks_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
task_type="season",
refresh_key=safe_season_id,
count=40,
expires_at=str(season.get("end_at", "")).strip(),
)
achievement_created = self._instantiate_user_tasks_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
task_type="achievement",
refresh_key=safe_season_id,
count=40,
expires_at="",
)
conn.execute(
"""
UPDATE user_season_tasks
SET status='expired', updated_at=?
WHERE user_id=? AND season_id=? AND status='active' AND expires_at<>'' AND expires_at<=?
""",
(utc_now_iso(), safe_user_id, safe_season_id, utc_now_iso()),
)
conn.commit()
return {
"daily_created": daily_created,
"weekly_created": weekly_created,
"season_created": season_created,
"achievement_created": achievement_created,
"day_key": day_key,
"week_key": week_key,
}
def _build_user_season_task_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
target = self._to_int(item.get("target_value", 1), 1, 1)
progress = self._to_int(item.get("progress_value", 0), 0, 0)
item["target_value"] = target
item["progress_value"] = min(progress, target)
item["reward_points"] = self._to_int(item.get("reward_points", 0), 0, 0)
item["reward_xp"] = self._to_int(item.get("reward_xp", 0), 0, 0)
item["reward_season_coin"] = self._to_int(item.get("reward_season_coin", 0), 0, 0)
item["constraints"] = self._json_object(item.get("constraints_json", "{}"))
item["reward_payload"] = self._json_object(item.get("reward_payload_json", "{}"))
item["progress_percent"] = min(100, int(round((item["progress_value"] / target) * 100))) if target > 0 else 0
return item
def list_user_season_tasks(
self,
user_id: str,
season_id: str,
*,
task_type: str = "",
include_claimed: bool = True,
limit: int = 500,
) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
if not safe_user_id or not safe_season_id:
return []
safe_task_type = str(task_type or "").strip().lower()
safe_limit = max(1, min(3000, int(limit)))
where = ["user_id=?", "season_id=?"]
params: list[Any] = [safe_user_id, safe_season_id]
if safe_task_type:
where.append("task_type=?")
params.append(safe_task_type)
if not include_claimed:
where.append("status<>'claimed'")
where_sql = " AND ".join(where)
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM user_season_tasks
WHERE {where_sql}
ORDER BY
CASE status WHEN 'completed' THEN 0 WHEN 'active' THEN 1 WHEN 'claimed' THEN 2 ELSE 3 END,
task_type ASC,
created_at DESC,
id DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
return [self._build_user_season_task_payload(row) for row in rows]
def _task_constraints_match(self, task_constraints: dict[str, Any], context: dict[str, Any]) -> bool:
if not task_constraints:
return True
for key, expected in task_constraints.items():
if key not in context:
return False
actual = context.get(key)
if isinstance(expected, list):
if actual not in expected:
return False
else:
if str(actual) != str(expected):
return False
return True
def progress_user_season_tasks(
self,
user_id: str,
season_id: str,
*,
event_key: str,
amount: int = 1,
context: dict[str, Any] | None = None,
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_event_key = str(event_key or "").strip().lower()
safe_amount = self._to_int(amount, 1, 1)
ctx = context if isinstance(context, dict) else {}
if not safe_user_id or not safe_season_id or not safe_event_key:
return {"updated": [], "completed_ids": []}
self.ensure_user_season_tasks(safe_user_id, safe_season_id)
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT *
FROM user_season_tasks
WHERE user_id=? AND season_id=? AND event_key=? AND status='active'
ORDER BY id ASC
""",
(safe_user_id, safe_season_id, safe_event_key),
).fetchall()
updated: list[dict[str, Any]] = []
completed_ids: list[int] = []
now = utc_now_iso()
for row in rows:
item = self._build_user_season_task_payload(row)
task_constraints = item.get("constraints", {})
if not isinstance(task_constraints, dict):
task_constraints = {}
if not self._task_constraints_match(task_constraints, ctx):
continue
before = self._to_int(item.get("progress_value", 0), 0, 0)
target = self._to_int(item.get("target_value", 1), 1, 1)
after = min(target, before + safe_amount)
next_status = "completed" if after >= target else "active"
conn.execute(
"""
UPDATE user_season_tasks
SET progress_value=?, status=?, updated_at=?
WHERE id=?
""",
(after, next_status, now, int(item.get("id", 0))),
)
fresh = dict(item)
fresh["progress_value"] = after
fresh["status"] = next_status
fresh["progress_percent"] = min(100, int(round((after / target) * 100))) if target > 0 else 0
if next_status == "completed":
completed_ids.append(int(item.get("id", 0)))
updated.append(fresh)
conn.commit()
return {"updated": updated, "completed_ids": completed_ids}
def _grant_reward_payload_tx(
self,
conn: sqlite3.Connection,
*,
user_id: str,
season_id: str,
reward_payload: dict[str, Any],
source: str,
) -> list[dict[str, Any]]:
payload = reward_payload if isinstance(reward_payload, dict) else {}
granted: list[dict[str, Any]] = []
now = utc_now_iso()
for title_id in payload.get("titles", []) if isinstance(payload.get("titles"), list) else []:
safe_id = str(title_id or "").strip()[:80]
if not safe_id:
continue
conn.execute(
"""
INSERT INTO user_decorations(
user_id, season_id, decoration_type, decoration_id, source, expires_at,
is_equipped, payload_json, created_at, updated_at
) VALUES (?, ?, 'title', ?, ?, '', 0, '{}', ?, ?)
ON CONFLICT(user_id, decoration_type, decoration_id)
DO UPDATE SET updated_at=excluded.updated_at
""",
(user_id, season_id, safe_id, source[:32], now, now),
)
granted.append({"decoration_type": "title", "decoration_id": safe_id})
for frame_id in payload.get("avatar_frames", []) if isinstance(payload.get("avatar_frames"), list) else []:
safe_id = str(frame_id or "").strip()[:80]
if not safe_id:
continue
conn.execute(
"""
INSERT INTO user_decorations(
user_id, season_id, decoration_type, decoration_id, source, expires_at,
is_equipped, payload_json, created_at, updated_at
) VALUES (?, ?, 'avatar_frame', ?, ?, '', 0, '{}', ?, ?)
ON CONFLICT(user_id, decoration_type, decoration_id)
DO UPDATE SET updated_at=excluded.updated_at
""",
(user_id, season_id, safe_id, source[:32], now, now),
)
granted.append({"decoration_type": "avatar_frame", "decoration_id": safe_id})
return granted
def claim_user_season_task(self, user_id: str, season_id: str, task_id: int) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_task_id = self._to_int(task_id, 0, 1)
if not safe_user_id or not safe_season_id or safe_task_id <= 0:
raise ValueError("invalid_card_payload")
profile_before = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"""
SELECT *
FROM user_season_tasks
WHERE id=? AND user_id=? AND season_id=?
""",
(safe_task_id, safe_user_id, safe_season_id),
).fetchone()
if row is None:
raise ValueError("invalid_card_id")
task = self._build_user_season_task_payload(row)
status = str(task.get("status", "active")).strip().lower()
if status == "claimed":
raise ValueError("shop_item_limit_reached")
if status != "completed":
raise ValueError("invalid_card_payload")
reward_points = self._to_int(task.get("reward_points", 0), 0, 0)
reward_xp = self._to_int(task.get("reward_xp", 0), 0, 0)
reward_coin = self._to_int(task.get("reward_season_coin", 0), 0, 0)
tx_payload = None
if reward_points > 0:
self._expire_point_lots_tx(conn, user_id=safe_user_id)
tx_payload = self._credit_points_tx(
conn,
safe_user_id,
reward_points,
tx_type="season_task_reward",
source="season",
related_id=f"{safe_season_id}:{safe_task_id}",
note=f"赛季任务奖励 #{safe_task_id}",
count_recharge=False,
create_lot=True,
allow_locked=True,
)
granted = self._grant_reward_payload_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
reward_payload=task.get("reward_payload", {}),
source="task",
)
claimed_at = utc_now_iso()
conn.execute(
"""
UPDATE user_season_tasks
SET status='claimed', claimed_at=?, updated_at=?
WHERE id=?
""",
(claimed_at, claimed_at, safe_task_id),
)
conn.commit()
next_profile = self.apply_user_season_gain(
safe_user_id,
safe_season_id,
xp_delta=reward_xp,
season_points_delta=reward_coin,
add_completed_task=True,
)
claimed_task = next((row for row in self.list_user_season_tasks(safe_user_id, safe_season_id, limit=800) if int(row.get("id", 0)) == safe_task_id), {})
return {
"task": claimed_task,
"profile_before": profile_before,
"profile_after": next_profile,
"transaction": (tx_payload or {}).get("tx", {}),
"granted_decorations": granted,
}
def claim_all_user_season_tasks(self, user_id: str, season_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
tasks = self.list_user_season_tasks(safe_user_id, safe_season_id, include_claimed=False, limit=1200)
completed_ids = [self._to_int(row.get("id", 0), 0, 1) for row in tasks if str(row.get("status", "")) == "completed"]
claimed: list[dict[str, Any]] = []
for task_id in completed_ids:
try:
claimed.append(self.claim_user_season_task(safe_user_id, safe_season_id, task_id))
except ValueError:
continue
profile = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
return {"claimed_count": len(claimed), "claimed": claimed, "profile": profile}
def _build_season_pass_reward_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
item["reward_level"] = self._to_int(item.get("reward_level", 1), 1, 1)
item["reward_points"] = self._to_int(item.get("reward_points", 0), 0, 0)
item["reward_xp"] = self._to_int(item.get("reward_xp", 0), 0, 0)
item["reward_season_coin"] = self._to_int(item.get("reward_season_coin", 0), 0, 0)
item["sort_order"] = self._to_int(item.get("sort_order", 0), 0, 0)
item["reward_payload"] = self._json_object(item.get("reward_payload_json", "{}"))
return item
def list_season_pass_rewards(self, season_id: str, *, limit: int = 800) -> list[dict[str, Any]]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id:
return []
safe_limit = max(1, min(4000, int(limit)))
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT *
FROM season_pass_rewards
WHERE season_id=?
ORDER BY reward_level ASC, reward_track ASC, sort_order ASC
LIMIT ?
""",
(safe_season_id, safe_limit),
).fetchall()
return [self._build_season_pass_reward_payload(row) for row in rows]
def upsert_season_pass_reward(
self,
season_id: str,
payload: dict[str, Any],
*,
reward_id: str = "",
) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id or self.get_season(safe_season_id) is None:
raise ValueError("invalid_card_id")
raw = payload if isinstance(payload, dict) else {}
safe_reward_level = self._to_int(raw.get("reward_level", 1), 1, 1)
safe_reward_track = str(raw.get("reward_track", "free")).strip().lower()
if safe_reward_track not in {"free", "premium"}:
safe_reward_track = "free"
safe_title = str(raw.get("title", f"{safe_reward_track} Lv.{safe_reward_level}")).strip()[:84]
safe_points = self._to_int(raw.get("reward_points", 0), 0, 0)
safe_xp = self._to_int(raw.get("reward_xp", 0), 0, 0)
safe_coin = self._to_int(raw.get("reward_season_coin", 0), 0, 0)
safe_sort = self._to_int(raw.get("sort_order", safe_reward_level), safe_reward_level, 0)
safe_payload = raw.get("reward_payload", {})
safe_id = str(reward_id or raw.get("reward_id", "")).strip()[:80]
if not safe_id:
safe_id = f"{safe_season_id}_{safe_reward_track}_{safe_reward_level}_{secrets.token_hex(2)}"[:80]
now = utc_now_iso()
with self._lock, self._connect() as conn:
exists = conn.execute("SELECT reward_id FROM season_pass_rewards WHERE reward_id=?", (safe_id,)).fetchone()
if exists:
conn.execute(
"""
UPDATE season_pass_rewards
SET season_id=?, reward_level=?, reward_track=?, title=?, reward_points=?, reward_xp=?,
reward_season_coin=?, reward_payload_json=?, sort_order=?, updated_at=?
WHERE reward_id=?
""",
(
safe_season_id,
safe_reward_level,
safe_reward_track,
safe_title,
safe_points,
safe_xp,
safe_coin,
json.dumps(safe_payload if isinstance(safe_payload, dict) else {}, ensure_ascii=False),
safe_sort,
now,
safe_id,
),
)
else:
conn.execute(
"""
INSERT INTO season_pass_rewards(
reward_id, season_id, reward_level, reward_track, title, reward_points, reward_xp,
reward_season_coin, reward_payload_json, sort_order, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_id,
safe_season_id,
safe_reward_level,
safe_reward_track,
safe_title,
safe_points,
safe_xp,
safe_coin,
json.dumps(safe_payload if isinstance(safe_payload, dict) else {}, ensure_ascii=False),
safe_sort,
now,
now,
),
)
conn.commit()
row = conn.execute("SELECT * FROM season_pass_rewards WHERE reward_id=?", (safe_id,)).fetchone()
if row is None:
raise ValueError("invalid_response")
return self._build_season_pass_reward_payload(row)
def list_user_season_reward_claims(self, user_id: str, season_id: str) -> set[str]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
if not safe_user_id or not safe_season_id:
return set()
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT reward_id
FROM user_season_reward_claims
WHERE user_id=? AND season_id=?
""",
(safe_user_id, safe_season_id),
).fetchall()
return {str(dict(row).get("reward_id", "")).strip() for row in rows if str(dict(row).get("reward_id", "")).strip()}
def claim_season_pass_reward(self, user_id: str, season_id: str, reward_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_reward_id = str(reward_id or "").strip()
if not safe_user_id or not safe_season_id or not safe_reward_id:
raise ValueError("invalid_card_payload")
profile = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
rewards = self.list_season_pass_rewards(safe_season_id, limit=2000)
reward = next((row for row in rewards if str(row.get("reward_id", "")) == safe_reward_id), None)
if reward is None:
raise ValueError("invalid_card_id")
if safe_reward_id in self.list_user_season_reward_claims(safe_user_id, safe_season_id):
raise ValueError("shop_item_limit_reached")
if self._to_int(profile.get("level", 1), 1, 1) < self._to_int(reward.get("reward_level", 1), 1, 1):
raise ValueError("invalid_card_payload")
if str(reward.get("reward_track", "free")).strip() == "premium" and str(profile.get("pass_type", "free")) != "premium":
raise ValueError("card_purchase_required")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
reward_points = self._to_int(reward.get("reward_points", 0), 0, 0)
reward_xp = self._to_int(reward.get("reward_xp", 0), 0, 0)
reward_coin = self._to_int(reward.get("reward_season_coin", 0), 0, 0)
tx_payload = None
if reward_points > 0:
self._expire_point_lots_tx(conn, user_id=safe_user_id)
tx_payload = self._credit_points_tx(
conn,
safe_user_id,
reward_points,
tx_type="season_pass_reward",
source="season",
related_id=f"{safe_season_id}:{safe_reward_id}",
note=f"赛季等级奖励 {safe_reward_id}",
count_recharge=False,
create_lot=True,
allow_locked=True,
)
granted = self._grant_reward_payload_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
reward_payload=reward.get("reward_payload", {}),
source="pass_reward",
)
claimed_at = utc_now_iso()
conn.execute(
"""
INSERT INTO user_season_reward_claims(user_id, season_id, reward_id, claimed_at)
VALUES (?, ?, ?, ?)
""",
(safe_user_id, safe_season_id, safe_reward_id, claimed_at),
)
conn.commit()
next_profile = self.apply_user_season_gain(
safe_user_id,
safe_season_id,
xp_delta=self._to_int(reward.get("reward_xp", 0), 0, 0),
season_points_delta=self._to_int(reward.get("reward_season_coin", 0), 0, 0),
)
return {
"reward": reward,
"profile": next_profile,
"transaction": (tx_payload or {}).get("tx", {}),
"granted_decorations": granted,
}
def _build_season_card_pool_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
item["target_value"] = self._to_int(item.get("target_value", 1), 1, 1)
item["reward_points"] = self._to_int(item.get("reward_points", 0), 0, 0)
item["reward_xp"] = self._to_int(item.get("reward_xp", 0), 0, 0)
item["weight"] = self._to_int(item.get("weight", 100), 100, 1)
item["min_level"] = self._to_int(item.get("min_level", 1), 1, 1)
item["sort_order"] = self._to_int(item.get("sort_order", 0), 0, 0)
item["payload"] = self._json_object(item.get("payload_json", "{}"))
item["is_active"] = bool(item.get("is_active", 0))
return item
def list_season_card_pools(
self,
season_id: str,
*,
active_only: bool = True,
limit: int = 200,
) -> list[dict[str, Any]]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id:
return []
safe_limit = max(1, min(2000, int(limit)))
where = "WHERE season_id=?"
if active_only:
where += " AND is_active=1"
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM season_card_pools
{where}
ORDER BY sort_order ASC, weight DESC, updated_at DESC
LIMIT ?
""",
(safe_season_id, safe_limit),
).fetchall()
return [self._build_season_card_pool_payload(row) for row in rows]
def upsert_season_card_pool(
self,
season_id: str,
payload: dict[str, Any],
*,
pool_id: str = "",
) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id or self.get_season(safe_season_id) is None:
raise ValueError("invalid_card_id")
raw = payload if isinstance(payload, dict) else {}
safe_card_type = str(raw.get("card_type", "rotating")).strip().lower()
if safe_card_type not in {"daily", "rotating", "surprise", "storyline"}:
safe_card_type = "rotating"
safe_title_hint = str(raw.get("title_hint", "")).strip()[:84]
if not safe_title_hint:
raise ValueError("invalid_card_payload")
safe_story_hook = str(raw.get("story_hook", "")).strip()[:360]
safe_objective = str(raw.get("objective_template", "")).strip()[:240]
safe_hint = str(raw.get("hint_template", "")).strip()[:240]
safe_event_key = str(raw.get("event_key", "complete_game")).strip().lower()[:48] or "complete_game"
safe_target = self._to_int(raw.get("target_value", 1), 1, 1)
safe_points = self._to_int(raw.get("reward_points", 0), 0, 0)
safe_xp = self._to_int(raw.get("reward_xp", 0), 0, 0)
safe_rarity = str(raw.get("rarity", "normal")).strip().lower()
if safe_rarity not in SEASON_CARD_RARITY_MULTIPLIER:
safe_rarity = "normal"
safe_weight = self._to_int(raw.get("weight", 100), 100, 1)
safe_min_level = self._to_int(raw.get("min_level", 1), 1, 1)
safe_payload = raw.get("payload", {})
safe_active = bool(raw.get("is_active", True))
safe_sort = self._to_int(raw.get("sort_order", 0), 0, 0)
safe_pool_id = str(pool_id or raw.get("pool_id", "")).strip()[:80]
if not safe_pool_id:
seed = re.sub(r"[^a-z0-9]+", "_", safe_title_hint.lower()).strip("_")[:32] or "pool"
safe_pool_id = f"{safe_season_id}_{safe_card_type}_{seed}_{secrets.token_hex(3)}"[:80]
now = utc_now_iso()
with self._lock, self._connect() as conn:
exists = conn.execute("SELECT pool_id FROM season_card_pools WHERE pool_id=?", (safe_pool_id,)).fetchone()
if exists:
conn.execute(
"""
UPDATE season_card_pools
SET season_id=?, card_type=?, title_hint=?, story_hook=?, objective_template=?, hint_template=?,
event_key=?, target_value=?, reward_points=?, reward_xp=?, rarity=?, weight=?, min_level=?,
payload_json=?, is_active=?, sort_order=?, updated_at=?
WHERE pool_id=?
""",
(
safe_season_id,
safe_card_type,
safe_title_hint,
safe_story_hook,
safe_objective,
safe_hint,
safe_event_key,
safe_target,
safe_points,
safe_xp,
safe_rarity,
safe_weight,
safe_min_level,
json.dumps(safe_payload if isinstance(safe_payload, dict) else {}, ensure_ascii=False),
1 if safe_active else 0,
safe_sort,
now,
safe_pool_id,
),
)
else:
conn.execute(
"""
INSERT INTO season_card_pools(
pool_id, season_id, card_type, title_hint, story_hook, objective_template, hint_template,
event_key, target_value, reward_points, reward_xp, rarity, weight, min_level, payload_json,
is_active, sort_order, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_pool_id,
safe_season_id,
safe_card_type,
safe_title_hint,
safe_story_hook,
safe_objective,
safe_hint,
safe_event_key,
safe_target,
safe_points,
safe_xp,
safe_rarity,
safe_weight,
safe_min_level,
json.dumps(safe_payload if isinstance(safe_payload, dict) else {}, ensure_ascii=False),
1 if safe_active else 0,
safe_sort,
now,
now,
),
)
conn.commit()
row = conn.execute("SELECT * FROM season_card_pools WHERE pool_id=?", (safe_pool_id,)).fetchone()
if row is None:
raise ValueError("invalid_response")
return self._build_season_card_pool_payload(row)
def _season_refresh_hours(self, season: dict[str, Any]) -> list[int]:
cfg = season.get("config", {})
if not isinstance(cfg, dict):
cfg = {}
raw_hours = cfg.get("refresh_hours", [0, 8, 18])
if not isinstance(raw_hours, list):
raw_hours = [0, 8, 18]
hours: list[int] = []
for row in raw_hours[:8]:
try:
value = int(row)
except (TypeError, ValueError):
continue
value = max(0, min(23, value))
if value not in hours:
hours.append(value)
hours.sort()
return hours or [0, 8, 18]
def _season_refresh_slot(self, now_dt: datetime, refresh_hours: list[int]) -> tuple[str, datetime]:
hours = sorted({max(0, min(23, int(row))) for row in refresh_hours}) if refresh_hours else [0, 8, 18]
if not hours:
hours = [0, 8, 18]
day = now_dt.replace(minute=0, second=0, microsecond=0)
slot_hour = -1
for hour in hours:
if now_dt.hour >= hour:
slot_hour = hour
if slot_hour < 0:
day = (day - timedelta(days=1)).replace(hour=0)
slot_hour = hours[-1]
slot_time = day.replace(hour=slot_hour)
refresh_key = slot_time.strftime("%Y%m%d%H")
next_slot: datetime | None = None
for hour in hours:
candidate = day.replace(hour=hour)
if candidate > slot_time:
next_slot = candidate
break
if next_slot is None:
next_slot = (day + timedelta(days=1)).replace(hour=hours[0])
return refresh_key, next_slot
def _seed_intel_pool(self, season: dict[str, Any], *, day_key: str) -> list[dict[str, str]]:
theme = str(season.get("theme", "")).strip()[:80] or "未知主题"
accent = str(season.get("accent_color", "#4f6dff")).strip()[:16] or "#4f6dff"
seed = str(season.get("story_seed", "")).strip()[:160]
outline = str(season.get("story_outline", "")).strip()[:220]
return [
{"intel_key": f"{day_key}:signal", "title": "电波摘录", "text": f"今日电波与“{theme}”共振,出现了新的异常频段。"},
{"intel_key": f"{day_key}:file", "title": "档案碎片", "text": seed or f"档案编号被抹去,只剩下与“{theme}”相关的暗示。"},
{"intel_key": f"{day_key}:map", "title": "街区地图", "text": outline or "地图上被标记的区域正在扩大,线索开始形成路径。"},
{"intel_key": f"{day_key}:tone", "title": "主题色注记", "text": f"记录员用 {accent} 标注关键节点,提示这是主线转折点。"},
]
def get_season_daily_game_card(self, season_id: str, *, day_key: str = "") -> dict[str, Any] | None:
safe_season_id = str(season_id or "").strip()
safe_day_key = str(day_key or "").strip()[:16] or _utc_day_key()
if not safe_season_id:
return None
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT *
FROM season_daily_game_cards
WHERE day_key=? AND season_id=?
""",
(safe_day_key, safe_season_id),
).fetchone()
if row is None:
return None
item = dict(row)
item["locked_intel"] = self._json_list(item.get("locked_intel_json", "[]"))
return item
def ensure_season_daily_game_card(
self,
season_id: str,
*,
card_payload: dict[str, Any] | None = None,
source: str = "ai",
) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
season = self.get_season(safe_season_id)
if season is None:
raise ValueError("invalid_card_id")
day_key = _utc_day_key()
existing = self.get_season_daily_game_card(safe_season_id, day_key=day_key)
if existing is not None:
return existing
payload = card_payload if isinstance(card_payload, dict) else {}
if payload:
card = self.create_game_card(
world_id=f"season:{safe_season_id}",
title=str(payload.get("title", "")),
subtitle=str(payload.get("subtitle", "")),
summary=str(payload.get("summary", "")),
cover_url=str(payload.get("cover_url", "")),
opening_text=str(payload.get("opening_text", "")),
starter_goals=payload.get("starter_goals", []),
state_schema=payload.get("state_schema", {}),
content_pack=payload.get("content_pack", {}),
creator_user_id="system",
)
else:
theme = str(season.get("theme", "")).strip()[:80] or "赛季主题"
season_name = str(season.get("name", "")).strip()[:80] or theme
title = season_name
subtitle = f"{theme} · {day_key}"
summary = (
str(season.get("description", "")).strip()[:200]
or f"{theme} 赛季关联剧本。"
)
opening_text = (
str(season.get("story_seed", "")).strip()[:600]
or str(season.get("story_outline", "")).strip()[:600]
or f"你踏入“{theme}”的世界,新的局势正悄然展开。"
)
starter_goals = ["确认世界观与角色定位", "完成一次关键抉择", "推进到阶段目标"]
cover_url = str(season.get("cover_url", "")).strip() or "/static/img/cards/fogharbor.svg"
accent = str(season.get("accent_color", "#4f6dff")).strip()[:16] or "#4f6dff"
state_schema = {
"ui": {
"cover_url": cover_url,
"room_background_url": cover_url,
"hero_status": {"key": "season_state", "label": "态势", "min": 0, "max": 100, "color": accent},
}
}
content_pack = {
"genre_tags": ["赛季关联"],
"feature_badges": ["赛季关联"],
"world_lore": str(season.get("story_seed", "")).strip()[:280],
"core_conflict": str(season.get("story_outline", "")).strip()[:280] or f"{theme} 的局势正在成形。",
"npc_roster": ["关键角色", "关键势力", "不明变量"],
"scene_blueprint": ["开场", "推进", "转折"],
"ending_catalog": ["完成阶段目标", "撤离并保留线索"],
}
card = self.create_game_card(
world_id=f"season:{safe_season_id}",
title=title,
subtitle=subtitle,
summary=summary,
cover_url=cover_url,
opening_text=opening_text,
starter_goals=starter_goals,
state_schema=state_schema,
content_pack=content_pack,
creator_user_id="system",
)
intel_pool = self._seed_intel_pool(season, day_key=day_key)
guide_text = f"赛季引导:每天会生成 1 张与“{str(season.get('theme', '')).strip()[:80] or '赛季主题'}”契合的赛季剧本卡。"
intro_text = "今天的资料已准备就绪:通关后可解锁其中一条。"
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"""
INSERT INTO season_daily_game_cards(
day_key, season_id, card_id, title, summary, guide_text, intro_text, locked_intel_json,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
day_key,
safe_season_id,
str(card.get("card_id", "")).strip(),
title,
summary,
guide_text,
intro_text,
json.dumps(intel_pool, ensure_ascii=False),
now,
now,
),
)
conn.commit()
row2 = conn.execute(
"SELECT * FROM season_daily_game_cards WHERE day_key=? AND season_id=?",
(day_key, safe_season_id),
).fetchone()
if row2 is None:
raise ValueError("invalid_response")
item2 = dict(row2)
item2["locked_intel"] = self._json_list(item2.get("locked_intel_json", "[]"))
return item2
def unlock_season_intel(self, user_id: str, season_id: str, *, day_key: str, intel_key: str = "") -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_day_key = str(day_key or "").strip()[:16]
if not safe_user_id or not safe_season_id or not safe_day_key:
raise ValueError("invalid_card_payload")
if self.get_user_by_id(safe_user_id) is None:
raise ValueError("user_not_found")
season = self.get_season(safe_season_id)
if season is None:
raise ValueError("invalid_card_id")
if not self.can_unlock_season_intel(safe_user_id, safe_season_id, day_key=safe_day_key):
raise ValueError("season_daily_card_not_completed")
with self._lock, self._connect() as conn:
row = conn.execute(
"SELECT locked_intel_json FROM season_daily_game_cards WHERE day_key=? AND season_id=?",
(safe_day_key, safe_season_id),
).fetchone()
if row is None:
raise ValueError("invalid_card_id")
pool = self._json_list(dict(row).get("locked_intel_json", "[]"))
candidates = [item for item in pool if isinstance(item, dict) and str(item.get("intel_key", "")).strip()]
if not candidates:
raise ValueError("invalid_card_payload")
chosen = candidates[0]
if intel_key:
override = next((c for c in candidates if str(c.get("intel_key", "")) == str(intel_key)), None)
if override is not None:
chosen = override
safe_intel_key = str(chosen.get("intel_key", "")).strip()[:80]
safe_title = str(chosen.get("title", "")).strip()[:64]
safe_text = str(chosen.get("text", "")).strip()[:900]
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"""
INSERT INTO user_season_intel_unlocks(
user_id, season_id, day_key, intel_key, intel_title, intel_text, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id, season_id, day_key, intel_key) DO NOTHING
""",
(safe_user_id, safe_season_id, safe_day_key, safe_intel_key, safe_title, safe_text, now),
)
conn.commit()
return {
"day_key": safe_day_key,
"intel_key": safe_intel_key,
"intel_title": safe_title,
"intel_text": safe_text,
}
def mark_season_daily_card_completed(
self,
user_id: str,
season_id: str,
*,
day_key: str,
room_id: str,
ending_state: str = "",
ending_summary: str = "",
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_day_key = str(day_key or "").strip()[:16]
safe_room_id = str(room_id or "").strip().upper()[:32]
if not safe_user_id or not safe_season_id or not safe_day_key or not safe_room_id:
raise ValueError("invalid_card_payload")
with self._lock, self._connect() as conn:
row = conn.execute(
"SELECT card_id FROM season_daily_game_cards WHERE season_id=? AND day_key=?",
(safe_season_id, safe_day_key),
).fetchone()
if row is None:
raise ValueError("invalid_card_id")
safe_card_id = str(dict(row).get("card_id", "")).strip()[:64]
if not safe_card_id:
raise ValueError("invalid_card_id")
safe_ending_state = str(ending_state or "").strip().lower()[:32]
safe_ending_summary = str(ending_summary or "").strip()[:1600]
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"""
INSERT INTO season_daily_card_completions(
user_id, season_id, day_key, room_id, card_id, ending_state, ending_summary, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id, season_id, day_key) DO NOTHING
""",
(
safe_user_id,
safe_season_id,
safe_day_key,
safe_room_id,
safe_card_id,
safe_ending_state,
safe_ending_summary,
now,
),
)
conn.commit()
return {
"user_id": safe_user_id,
"season_id": safe_season_id,
"day_key": safe_day_key,
"room_id": safe_room_id,
"card_id": safe_card_id,
"created_at": now,
}
def can_unlock_season_intel(self, user_id: str, season_id: str, *, day_key: str) -> bool:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_day_key = str(day_key or "").strip()[:16]
if not safe_user_id or not safe_season_id or not safe_day_key:
return False
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT id
FROM season_daily_card_completions
WHERE user_id=? AND season_id=? AND day_key=?
LIMIT 1
""",
(safe_user_id, safe_season_id, safe_day_key),
).fetchone()
return row is not None
def list_user_season_intel(self, user_id: str, season_id: str, *, limit: int = 200) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_limit = max(1, min(2000, int(limit)))
if not safe_user_id or not safe_season_id:
return []
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT day_key, intel_key, intel_title, intel_text, created_at
FROM user_season_intel_unlocks
WHERE user_id=? AND season_id=?
ORDER BY created_at ASC, id ASC
LIMIT ?
""",
(safe_user_id, safe_season_id, safe_limit),
).fetchall()
return [dict(r) for r in rows]
def build_season_chronicle_items(
self,
user_id: str,
season_id: str,
*,
limit_days: int = 60,
) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_limit = max(1, min(365, int(limit_days)))
if not safe_user_id or not safe_season_id:
raise ValueError("invalid_card_payload")
if self.get_user_by_id(safe_user_id) is None:
raise ValueError("user_not_found")
if self.get_season(safe_season_id) is None:
raise ValueError("season_not_found")
today_key = _utc_day_key()
with self._lock, self._connect() as conn:
daily_rows = conn.execute(
"""
SELECT day_key, title, summary, guide_text, intro_text, created_at
FROM season_daily_game_cards
WHERE season_id=?
ORDER BY day_key DESC
LIMIT ?
""",
(safe_season_id, safe_limit),
).fetchall()
intel_rows = conn.execute(
"""
SELECT day_key, intel_title, intel_text, created_at
FROM user_season_intel_unlocks
WHERE user_id=? AND season_id=?
ORDER BY created_at ASC, id ASC
LIMIT 2000
""",
(safe_user_id, safe_season_id),
).fetchall()
intel_by_day: dict[str, list[dict[str, Any]]] = {}
for row in intel_rows:
item = dict(row)
day_key = str(item.get("day_key", "")).strip()
if not day_key:
continue
intel_by_day.setdefault(day_key, []).append(
{
"intel_title": str(item.get("intel_title", "")).strip()[:64],
"intel_text": str(item.get("intel_text", "")).strip()[:900],
"created_at": str(item.get("created_at", "")).strip(),
}
)
items: list[dict[str, Any]] = []
for row in daily_rows:
item = dict(row)
day_key = str(item.get("day_key", "")).strip()
items.append(
{
"day_key": day_key,
"title": str(item.get("title", "")).strip()[:120],
"summary": str(item.get("summary", "")).strip()[:900],
"guide_text": str(item.get("guide_text", "")).strip()[:1800],
"intro_text": str(item.get("intro_text", "")).strip()[:1800],
"created_at": str(item.get("created_at", "")).strip(),
"is_today": bool(day_key == today_key),
"intel": intel_by_day.get(day_key, []),
}
)
return items
def _get_story_thread(self, user_id: str, season_id: str) -> dict[str, Any]:
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT id, user_id, season_id, thread_summary, thread_state_json, last_generated_at, updated_at
FROM season_story_threads
WHERE user_id=? AND season_id=?
""",
(user_id, season_id),
).fetchone()
if row is None:
return {
"id": 0,
"user_id": user_id,
"season_id": season_id,
"thread_summary": "",
"thread_state": {"chapter": 0, "last_event": "", "history": []},
"last_generated_at": "",
"updated_at": "",
}
item = dict(row)
state = self._json_object(item.get("thread_state_json", "{}"))
if not isinstance(state.get("history"), list):
state["history"] = []
state["chapter"] = self._to_int(state.get("chapter", 0), 0, 0)
state["last_event"] = str(state.get("last_event", "")).strip()[:240]
return {
"id": self._to_int(item.get("id", 0), 0, 0),
"user_id": str(item.get("user_id", "")).strip(),
"season_id": str(item.get("season_id", "")).strip(),
"thread_summary": str(item.get("thread_summary", "")).strip()[:2400],
"thread_state": state,
"last_generated_at": str(item.get("last_generated_at", "")).strip(),
"updated_at": str(item.get("updated_at", "")).strip(),
}
def _save_story_thread(self, user_id: str, season_id: str, thread_state: dict[str, Any], *, summary: str = "") -> None:
now = utc_now_iso()
safe_state = thread_state if isinstance(thread_state, dict) else {}
with self._lock, self._connect() as conn:
conn.execute(
"""
INSERT INTO season_story_threads(
user_id, season_id, thread_summary, thread_state_json, last_generated_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(user_id, season_id) DO UPDATE SET
thread_summary=excluded.thread_summary,
thread_state_json=excluded.thread_state_json,
last_generated_at=excluded.last_generated_at,
updated_at=excluded.updated_at
""",
(
str(user_id or "").strip(),
str(season_id or "").strip(),
str(summary or "").strip()[:2400],
json.dumps(safe_state, ensure_ascii=False),
now,
now,
),
)
conn.commit()
def _pick_pool_rows(self, pools: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
if count <= 0:
return []
daily = [row for row in pools if str(row.get("card_type", "")) == "daily"]
rotating = [row for row in pools if str(row.get("card_type", "")) == "rotating"]
surprise = [row for row in pools if str(row.get("card_type", "")) == "surprise"]
others = [row for row in pools if row not in daily and row not in rotating and row not in surprise]
selected: list[dict[str, Any]] = []
if daily:
selected.extend(self._weighted_sample_rows(daily, 1))
if len(selected) < count and rotating:
selected.extend(self._weighted_sample_rows(rotating, 1))
if len(selected) < count and surprise and secrets.randbelow(100) < 12:
selected.extend(self._weighted_sample_rows(surprise, 1))
remaining = [row for row in pools if row not in selected]
if len(selected) < count and remaining:
selected.extend(self._weighted_sample_rows(remaining, count - len(selected)))
if len(selected) < count:
selected.extend(self._weighted_sample_rows(others, count - len(selected)))
return selected[:count]
def _compose_season_card_payload(
self,
season: dict[str, Any],
thread_state: dict[str, Any],
pool: dict[str, Any],
*,
index: int,
) -> dict[str, Any]:
chapter = self._to_int(thread_state.get("chapter", 0), 0, 0) + 1
history = thread_state.get("history", [])
if not isinstance(history, list):
history = []
last_event = str(thread_state.get("last_event", "")).strip()
title_hint = str(pool.get("title_hint", "异象速递")).strip()[:64] or "异象速递"
opener = SEASON_STORY_OPENERS[(chapter + index) % len(SEASON_STORY_OPENERS)]
twist = SEASON_STORY_TWISTS[(chapter + index) % len(SEASON_STORY_TWISTS)]
hook = str(pool.get("story_hook", "")).strip()[:280]
theme = str(season.get("theme", "赛季主题")).strip()[:80] or "赛季主题"
if last_event:
continuation = f"延续昨日线索:{last_event[:70]}。"
else:
continuation = "故事刚刚开始,所有线索都处于散落状态。"
story_text = f"{opener} {continuation} {hook or twist} 主题“{theme}”的主线继续推进。"
objective_template = str(pool.get("objective_template", "")).strip()[:220]
objective = objective_template or f"完成 {self._to_int(pool.get('target_value', 1), 1, 1)} 次关键行动。"
hint_template = str(pool.get("hint_template", "")).strip()[:220]
hint = hint_template or SEASON_HINT_POOL[(chapter + index) % len(SEASON_HINT_POOL)]
rarity = str(pool.get("rarity", "normal")).strip().lower()
if rarity not in SEASON_CARD_RARITY_MULTIPLIER:
rarity = "normal"
ratio = float(SEASON_CARD_RARITY_MULTIPLIER.get(rarity, 1.0))
reward_points = int(round(self._to_int(pool.get("reward_points", 0), 0, 0) * ratio))
reward_xp = int(round(self._to_int(pool.get("reward_xp", 0), 0, 0) * ratio))
return {
"title": f"{title_hint} · 第{chapter}幕",
"story_text": story_text[:420],
"objective": objective[:220],
"hint_text": hint[:220],
"event_key": str(pool.get("event_key", "complete_game")).strip().lower()[:48] or "complete_game",
"target_value": self._to_int(pool.get("target_value", 1), 1, 1),
"reward_points": max(0, reward_points),
"reward_xp": max(0, reward_xp),
"rarity": rarity,
"card_type": str(pool.get("card_type", "rotating")).strip().lower() or "rotating",
"source_pool_id": str(pool.get("pool_id", "")).strip()[:80],
"payload": {
"theme": theme,
"chapter": chapter,
"continuity_prev": last_event[:120],
"history_tail": history[-5:],
"pool_payload": pool.get("payload", {}),
},
"next_event": f"{title_hint} 触发了新的异象回响",
}
def _build_user_season_card_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
item["reward_points"] = self._to_int(item.get("reward_points", 0), 0, 0)
item["reward_xp"] = self._to_int(item.get("reward_xp", 0), 0, 0)
item["target_value"] = self._to_int(item.get("target_value", 1), 1, 1)
item["progress_value"] = self._to_int(item.get("progress_value", 0), 0, 0)
item["payload"] = self._json_object(item.get("payload_json", "{}"))
target = max(1, item["target_value"])
item["progress_percent"] = min(100, int(round((min(item["progress_value"], target) / target) * 100)))
return item
def ensure_user_season_cards(self, user_id: str, season_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
season = self.get_season(safe_season_id)
if not safe_user_id or season is None:
raise ValueError("invalid_card_id")
profile = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
season_cfg = season.get("config", {})
if not isinstance(season_cfg, dict):
season_cfg = {}
now_dt = self._parse_iso_datetime(utc_now_iso()) or datetime.now(timezone.utc)
refresh_hours = self._season_refresh_hours(season)
refresh_key, next_refresh_at = self._season_refresh_slot(now_dt, refresh_hours)
card_count = self._to_int(season_cfg.get("daily_card_count", 3), 3, 1)
with self._lock, self._connect() as conn:
existing = conn.execute(
"""
SELECT *
FROM season_activity_cards
WHERE user_id=? AND season_id=? AND refresh_key=? AND status IN ('active', 'completed')
ORDER BY id ASC
""",
(safe_user_id, safe_season_id, refresh_key),
).fetchall()
if existing:
cards = [self._build_user_season_card_payload(row) for row in existing]
return {
"refresh_key": refresh_key,
"next_refresh_at": next_refresh_at.isoformat(),
"cards": cards,
"generated": False,
}
conn.execute("BEGIN IMMEDIATE")
conn.execute(
"""
UPDATE season_activity_cards
SET status='expired', updated_at=?
WHERE user_id=? AND season_id=? AND status IN ('active', 'completed') AND refresh_key<>?
""",
(utc_now_iso(), safe_user_id, safe_season_id, refresh_key),
)
pools = self.list_season_card_pools(safe_season_id, active_only=True, limit=600)
level = self._to_int(profile.get("level", 1), 1, 1)
pools = [row for row in pools if self._to_int(row.get("min_level", 1), 1, 1) <= level]
if not pools:
pools = [
{
"pool_id": f"{safe_season_id}_fallback",
"card_type": "daily",
"title_hint": "每日简报",
"story_hook": "系统未配置卡池,已启用兜底任务。",
"objective_template": "完成 1 局游戏。",
"hint_template": "先完成基础对局领取奖励。",
"event_key": "complete_game",
"target_value": 1,
"reward_points": 12,
"reward_xp": 28,
"rarity": "normal",
"payload": {},
"weight": 100,
}
]
picked = self._pick_pool_rows(pools, card_count)
thread = self._get_story_thread(safe_user_id, safe_season_id)
thread_state = thread.get("thread_state", {})
if not isinstance(thread_state, dict):
thread_state = {"chapter": 0, "last_event": "", "history": []}
history = thread_state.get("history", [])
if not isinstance(history, list):
history = []
inserted: list[dict[str, Any]] = []
for idx, pool in enumerate(picked):
card_payload = self._compose_season_card_payload(
season,
thread_state,
pool,
index=idx,
)
now_iso = utc_now_iso()
conn.execute(
"""
INSERT INTO season_activity_cards(
user_id, season_id, refresh_key, title, story_text, objective, hint_text,
reward_points, reward_xp, rarity, status, claimed_at, expires_at, payload_json,
created_at, updated_at, claimed_tx_id, event_key, target_value, progress_value, card_type, source_pool_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', '', ?, ?, ?, ?, '', ?, ?, 0, ?, ?)
""",
(
safe_user_id,
safe_season_id,
refresh_key,
card_payload["title"],
card_payload["story_text"],
card_payload["objective"],
card_payload["hint_text"],
card_payload["reward_points"],
card_payload["reward_xp"],
card_payload["rarity"],
next_refresh_at.isoformat(),
json.dumps(card_payload["payload"], ensure_ascii=False),
now_iso,
now_iso,
card_payload["event_key"],
card_payload["target_value"],
card_payload["card_type"],
card_payload["source_pool_id"],
),
)
row_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
created = conn.execute("SELECT * FROM season_activity_cards WHERE id=?", (row_id,)).fetchone()
if created:
inserted.append(self._build_user_season_card_payload(created))
thread_state["chapter"] = self._to_int(thread_state.get("chapter", 0), 0, 0) + 1
thread_state["last_event"] = card_payload.get("next_event", "")
history.append(card_payload["title"])
thread_state["history"] = history[-30:]
conn.execute(
"""
UPDATE user_season_profiles
SET last_refresh_key=?, last_generated_at=?, updated_at=?
WHERE user_id=? AND season_id=?
""",
(refresh_key, utc_now_iso(), utc_now_iso(), safe_user_id, safe_season_id),
)
conn.commit()
self._save_story_thread(
safe_user_id,
safe_season_id,
thread_state,
summary=f"已推进到第 {self._to_int(thread_state.get('chapter', 0), 0, 0)} 幕。",
)
return {
"refresh_key": refresh_key,
"next_refresh_at": next_refresh_at.isoformat(),
"cards": inserted,
"generated": True,
}
def list_user_season_cards(
self,
user_id: str,
season_id: str,
*,
refresh_key: str = "",
include_expired: bool = False,
limit: int = 200,
) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
if not safe_user_id or not safe_season_id:
return []
safe_refresh = str(refresh_key or "").strip()
safe_limit = max(1, min(1200, int(limit)))
where = ["user_id=?", "season_id=?"]
params: list[Any] = [safe_user_id, safe_season_id]
if safe_refresh:
where.append("refresh_key=?")
params.append(safe_refresh)
if not include_expired:
where.append("status<>'expired'")
where_sql = " AND ".join(where)
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM season_activity_cards
WHERE {where_sql}
ORDER BY id DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
return [self._build_user_season_card_payload(row) for row in rows]
def progress_user_season_cards(
self,
user_id: str,
season_id: str,
*,
event_key: str,
amount: int = 1,
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_event = str(event_key or "").strip().lower()
safe_amount = self._to_int(amount, 1, 1)
if not safe_user_id or not safe_season_id or not safe_event:
return {"updated": [], "completed_ids": []}
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT *
FROM season_activity_cards
WHERE user_id=? AND season_id=? AND event_key=? AND status='active'
ORDER BY id ASC
""",
(safe_user_id, safe_season_id, safe_event),
).fetchall()
updated: list[dict[str, Any]] = []
completed_ids: list[int] = []
now = utc_now_iso()
for row in rows:
item = self._build_user_season_card_payload(row)
target = self._to_int(item.get("target_value", 1), 1, 1)
before = self._to_int(item.get("progress_value", 0), 0, 0)
after = min(target, before + safe_amount)
next_status = "completed" if after >= target else "active"
conn.execute(
"""
UPDATE season_activity_cards
SET progress_value=?, status=?, updated_at=?
WHERE id=?
""",
(after, next_status, now, int(item.get("id", 0))),
)
item["progress_value"] = after
item["status"] = next_status
item["progress_percent"] = min(100, int(round((after / target) * 100))) if target > 0 else 0
if next_status == "completed":
completed_ids.append(int(item.get("id", 0)))
updated.append(item)
conn.commit()
return {"updated": updated, "completed_ids": completed_ids}
def claim_user_season_card(self, user_id: str, season_id: str, card_id: int) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_card_id = self._to_int(card_id, 0, 1)
if not safe_user_id or not safe_season_id or safe_card_id <= 0:
raise ValueError("invalid_card_payload")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"""
SELECT *
FROM season_activity_cards
WHERE id=? AND user_id=? AND season_id=?
""",
(safe_card_id, safe_user_id, safe_season_id),
).fetchone()
if row is None:
raise ValueError("invalid_card_id")
card = self._build_user_season_card_payload(row)
status = str(card.get("status", "active")).strip().lower()
if status == "claimed":
raise ValueError("shop_item_limit_reached")
if status != "completed":
raise ValueError("invalid_card_payload")
reward_points = self._to_int(card.get("reward_points", 0), 0, 0)
reward_xp = self._to_int(card.get("reward_xp", 0), 0, 0)
tx_payload = None
if reward_points > 0:
self._expire_point_lots_tx(conn, user_id=safe_user_id)
tx_payload = self._credit_points_tx(
conn,
safe_user_id,
reward_points,
tx_type="season_card_reward",
source="season",
related_id=f"{safe_season_id}:{safe_card_id}",
note=f"赛季卡片奖励 #{safe_card_id}",
count_recharge=False,
create_lot=True,
allow_locked=True,
)
granted = self._grant_reward_payload_tx(
conn,
user_id=safe_user_id,
season_id=safe_season_id,
reward_payload=card.get("payload", {}),
source="season_card",
)
claimed_at = utc_now_iso()
conn.execute(
"""
UPDATE season_activity_cards
SET status='claimed', claimed_at=?, updated_at=?
WHERE id=?
""",
(claimed_at, claimed_at, safe_card_id),
)
conn.commit()
next_profile = self.apply_user_season_gain(
safe_user_id,
safe_season_id,
xp_delta=reward_xp,
add_claimed_card=True,
)
cards = self.list_user_season_cards(safe_user_id, safe_season_id, include_expired=False, limit=200)
claimed_card = next((row for row in cards if int(row.get("id", 0)) == safe_card_id), {})
return {
"card": claimed_card,
"profile": next_profile,
"transaction": (tx_payload or {}).get("tx", {}),
"granted_decorations": granted,
}
def refresh_user_season_card(self, user_id: str, season_id: str, card_id: int, *, refresh_cost: int = 60) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_season_id = str(season_id or "").strip()
safe_card_id = self._to_int(card_id, 0, 1)
if not safe_user_id or not safe_season_id or safe_card_id <= 0:
raise ValueError("invalid_card_payload")
season = self.get_season(safe_season_id)
if season is None:
raise ValueError("invalid_card_id")
safe_cost = self._to_int(refresh_cost, 60, 0)
profile = self.get_or_create_user_season_profile(safe_user_id, safe_season_id)
pools = self.list_season_card_pools(safe_season_id, active_only=True, limit=800)
level = self._to_int(profile.get("level", 1), 1, 1)
pools = [row for row in pools if self._to_int(row.get("min_level", 1), 1, 1) <= level]
if not pools:
raise ValueError("invalid_card_payload")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"""
SELECT *
FROM season_activity_cards
WHERE id=? AND user_id=? AND season_id=?
""",
(safe_card_id, safe_user_id, safe_season_id),
).fetchone()
if row is None:
raise ValueError("invalid_card_id")
old_card = self._build_user_season_card_payload(row)
if str(old_card.get("status", "")).strip().lower() not in {"active", "completed"}:
raise ValueError("invalid_card_payload")
self._expire_point_lots_tx(conn, user_id=safe_user_id)
debit_payload = self._debit_points_tx(
conn,
safe_user_id,
safe_cost,
tx_type="season_card_refresh",
source="season",
related_id=f"{safe_season_id}:{safe_card_id}",
note=f"刷新赛季卡片 #{safe_card_id}",
count_spent=True,
)
refresh_key = str(old_card.get("refresh_key", "")).strip()
expires_at = str(old_card.get("expires_at", "")).strip()
pool = self._weighted_sample_rows(pools, 1)[0]
thread = self._get_story_thread(safe_user_id, safe_season_id)
thread_state = thread.get("thread_state", {})
if not isinstance(thread_state, dict):
thread_state = {"chapter": 0, "last_event": "", "history": []}
new_payload = self._compose_season_card_payload(season, thread_state, pool, index=0)
now_iso = utc_now_iso()
conn.execute(
"""
UPDATE season_activity_cards
SET status='expired', updated_at=?
WHERE id=?
""",
(now_iso, safe_card_id),
)
conn.execute(
"""
INSERT INTO season_activity_cards(
user_id, season_id, refresh_key, title, story_text, objective, hint_text,
reward_points, reward_xp, rarity, status, claimed_at, expires_at, payload_json, created_at, updated_at,
claimed_tx_id, event_key, target_value, progress_value, card_type, source_pool_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', '', ?, ?, ?, ?, '', ?, ?, 0, ?, ?)
""",
(
safe_user_id,
safe_season_id,
refresh_key,
new_payload["title"],
new_payload["story_text"],
new_payload["objective"],
new_payload["hint_text"],
new_payload["reward_points"],
new_payload["reward_xp"],
new_payload["rarity"],
expires_at,
json.dumps(new_payload["payload"], ensure_ascii=False),
now_iso,
now_iso,
new_payload["event_key"],
new_payload["target_value"],
new_payload["card_type"],
new_payload["source_pool_id"],
),
)
new_card_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
row_new = conn.execute("SELECT * FROM season_activity_cards WHERE id=?", (new_card_id,)).fetchone()
conn.commit()
thread_state["chapter"] = self._to_int(thread_state.get("chapter", 0), 0, 0) + 1
history = thread_state.get("history", [])
if not isinstance(history, list):
history = []
history.append(new_payload["title"])
thread_state["history"] = history[-30:]
thread_state["last_event"] = new_payload.get("next_event", "")
self._save_story_thread(
safe_user_id,
safe_season_id,
thread_state,
summary=f"手动刷新后推进到第 {self._to_int(thread_state.get('chapter', 0), 0, 0)} 幕。",
)
return {
"new_card": self._build_user_season_card_payload(row_new) if row_new else {},
"debit_transaction": debit_payload.get("tx", {}),
}
def list_season_titles(self, season_id: str, *, active_only: bool = False, limit: int = 200) -> list[dict[str, Any]]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id:
return []
safe_limit = max(1, min(1000, int(limit)))
where = "WHERE season_id=?"
if active_only:
where += " AND is_active=1"
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM season_titles
{where}
ORDER BY sort_order ASC, created_at ASC
LIMIT ?
""",
(safe_season_id, safe_limit),
).fetchall()
out: list[dict[str, Any]] = []
for row in rows:
item = dict(row)
item["is_active"] = bool(item.get("is_active", 0))
item["effect"] = self._json_object(item.get("effect_json", "{}"))
out.append(item)
return out
def upsert_season_title(
self,
season_id: str,
payload: dict[str, Any],
*,
title_id: str = "",
) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id or self.get_season(safe_season_id) is None:
raise ValueError("invalid_card_id")
raw = payload if isinstance(payload, dict) else {}
safe_name = str(raw.get("name", "")).strip()[:64]
if not safe_name:
raise ValueError("invalid_card_payload")
safe_id = str(title_id or raw.get("title_id", "")).strip()[:80]
if not safe_id:
seed = re.sub(r"[^a-z0-9]+", "_", safe_name.lower()).strip("_")[:32] or "title"
safe_id = f"{safe_season_id}_{seed}_{secrets.token_hex(2)}"[:80]
safe_color = str(raw.get("color_code", "#ffffff")).strip()[:16] or "#ffffff"
safe_rarity = str(raw.get("rarity", "normal")).strip().lower()[:24] or "normal"
safe_acquire = str(raw.get("acquire_type", "season_reward")).strip().lower()[:32] or "season_reward"
safe_image = str(raw.get("image_url", "")).strip()[:720]
safe_effect = raw.get("effect", {})
safe_active = bool(raw.get("is_active", True))
safe_sort = self._to_int(raw.get("sort_order", 0), 0, 0)
now = utc_now_iso()
with self._lock, self._connect() as conn:
exists = conn.execute("SELECT title_id FROM season_titles WHERE title_id=?", (safe_id,)).fetchone()
if exists:
conn.execute(
"""
UPDATE season_titles
SET season_id=?, name=?, color_code=?, rarity=?, acquire_type=?, image_url=?, effect_json=?,
is_active=?, sort_order=?, updated_at=?
WHERE title_id=?
""",
(
safe_season_id,
safe_name,
safe_color,
safe_rarity,
safe_acquire,
safe_image,
json.dumps(safe_effect if isinstance(safe_effect, dict) else {}, ensure_ascii=False),
1 if safe_active else 0,
safe_sort,
now,
safe_id,
),
)
else:
conn.execute(
"""
INSERT INTO season_titles(
title_id, season_id, name, color_code, rarity, acquire_type, image_url, effect_json,
is_active, sort_order, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_id,
safe_season_id,
safe_name,
safe_color,
safe_rarity,
safe_acquire,
safe_image,
json.dumps(safe_effect if isinstance(safe_effect, dict) else {}, ensure_ascii=False),
1 if safe_active else 0,
safe_sort,
now,
now,
),
)
conn.commit()
rows = self.list_season_titles(safe_season_id, active_only=False, limit=2000)
return next((row for row in rows if str(row.get("title_id", "")) == safe_id), {})
def list_season_avatar_frames(self, season_id: str, *, active_only: bool = False, limit: int = 200) -> list[dict[str, Any]]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id:
return []
safe_limit = max(1, min(1000, int(limit)))
where = "WHERE season_id=?"
if active_only:
where += " AND is_active=1"
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM season_avatar_frames
{where}
ORDER BY sort_order ASC, created_at ASC
LIMIT ?
""",
(safe_season_id, safe_limit),
).fetchall()
out: list[dict[str, Any]] = []
for row in rows:
item = dict(row)
item["is_active"] = bool(item.get("is_active", 0))
item["style"] = self._json_object(item.get("style_json", "{}"))
out.append(item)
return out
def upsert_season_avatar_frame(
self,
season_id: str,
payload: dict[str, Any],
*,
frame_id: str = "",
) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id or self.get_season(safe_season_id) is None:
raise ValueError("invalid_card_id")
raw = payload if isinstance(payload, dict) else {}
safe_name = str(raw.get("name", "")).strip()[:64]
if not safe_name:
raise ValueError("invalid_card_payload")
safe_id = str(frame_id or raw.get("frame_id", "")).strip()[:80]
if not safe_id:
seed = re.sub(r"[^a-z0-9]+", "_", safe_name.lower()).strip("_")[:32] or "frame"
safe_id = f"{safe_season_id}_{seed}_{secrets.token_hex(2)}"[:80]
safe_style = raw.get("style", {})
safe_rarity = str(raw.get("rarity", "normal")).strip().lower()[:24] or "normal"
safe_acquire = str(raw.get("acquire_type", "season_reward")).strip().lower()[:32] or "season_reward"
safe_preview = str(raw.get("preview_url", "")).strip()[:720]
safe_active = bool(raw.get("is_active", True))
safe_sort = self._to_int(raw.get("sort_order", 0), 0, 0)
now = utc_now_iso()
with self._lock, self._connect() as conn:
exists = conn.execute("SELECT frame_id FROM season_avatar_frames WHERE frame_id=?", (safe_id,)).fetchone()
if exists:
conn.execute(
"""
UPDATE season_avatar_frames
SET season_id=?, name=?, style_json=?, rarity=?, acquire_type=?, preview_url=?, is_active=?,
sort_order=?, updated_at=?
WHERE frame_id=?
""",
(
safe_season_id,
safe_name,
json.dumps(safe_style if isinstance(safe_style, dict) else {}, ensure_ascii=False),
safe_rarity,
safe_acquire,
safe_preview,
1 if safe_active else 0,
safe_sort,
now,
safe_id,
),
)
else:
conn.execute(
"""
INSERT INTO season_avatar_frames(
frame_id, season_id, name, style_json, rarity, acquire_type, preview_url,
is_active, sort_order, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_id,
safe_season_id,
safe_name,
json.dumps(safe_style if isinstance(safe_style, dict) else {}, ensure_ascii=False),
safe_rarity,
safe_acquire,
safe_preview,
1 if safe_active else 0,
safe_sort,
now,
now,
),
)
conn.commit()
rows = self.list_season_avatar_frames(safe_season_id, active_only=False, limit=2000)
return next((row for row in rows if str(row.get("frame_id", "")) == safe_id), {})
def list_user_decorations(self, user_id: str, *, decoration_type: str = "", limit: int = 600) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
safe_type = str(decoration_type or "").strip().lower()
safe_limit = max(1, min(3000, int(limit)))
if not safe_user_id:
return []
where = ["d.user_id=?"]
params: list[Any] = [safe_user_id]
if safe_type:
where.append("d.decoration_type=?")
params.append(safe_type)
where_sql = " AND ".join(where)
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT d.*
FROM user_decorations d
WHERE {where_sql}
ORDER BY d.is_equipped DESC, d.created_at DESC, d.id DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
out: list[dict[str, Any]] = []
for row in rows:
item = dict(row)
item["is_equipped"] = bool(item.get("is_equipped", 0))
item["payload"] = self._json_object(item.get("payload_json", "{}"))
out.append(item)
return out
def equip_user_decoration(self, user_id: str, decoration_type: str, decoration_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_type = str(decoration_type or "").strip().lower()
safe_id = str(decoration_id or "").strip()
if safe_type not in {"title", "avatar_frame"} or not safe_user_id:
raise ValueError("invalid_card_payload")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute(
"""
SELECT id
FROM user_decorations
WHERE user_id=? AND decoration_type=? AND decoration_id=?
""",
(safe_user_id, safe_type, safe_id),
).fetchone()
if row is None:
raise ValueError("invalid_card_id")
now = utc_now_iso()
conn.execute(
"""
UPDATE user_decorations
SET is_equipped=0, updated_at=?
WHERE user_id=? AND decoration_type=?
""",
(now, safe_user_id, safe_type),
)
conn.execute(
"""
UPDATE user_decorations
SET is_equipped=1, updated_at=?
WHERE user_id=? AND decoration_type=? AND decoration_id=?
""",
(now, safe_user_id, safe_type, safe_id),
)
conn.commit()
rows = self.list_user_decorations(safe_user_id, decoration_type=safe_type, limit=200)
return next((row for row in rows if str(row.get("decoration_id", "")) == safe_id), {})
def get_user_equipped_decorations(self, user_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return {"title": None, "avatar_frame": None}
rows = self.list_user_decorations(safe_user_id, limit=400)
title = next((row for row in rows if row.get("decoration_type") == "title" and row.get("is_equipped")), None)
frame = next((row for row in rows if row.get("decoration_type") == "avatar_frame" and row.get("is_equipped")), None)
return {"title": title, "avatar_frame": frame}
def apply_room_summary_season_progress(self, user_id: str, room_summary: dict[str, Any]) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
summary = room_summary if isinstance(room_summary, dict) else {}
room_id = str(summary.get("room_id", "")).strip().upper()
status = str(summary.get("status", "")).strip().lower()
if not safe_user_id or not room_id or status != "ended":
return {"applied": False, "events": [], "task_updates": [], "card_updates": [], "season_id": ""}
season = self.get_active_season()
if season is None:
return {"applied": False, "events": [], "task_updates": [], "card_updates": [], "season_id": ""}
season_id = str(season.get("season_id", "")).strip()
if not season_id:
return {"applied": False, "events": [], "task_updates": [], "card_updates": [], "season_id": ""}
day_key = ""
card_id = str(summary.get("game_card_id") or summary.get("card_id") or "").strip()
if card_id:
with self._lock, self._connect() as conn:
daily_row = conn.execute(
"SELECT day_key FROM season_daily_game_cards WHERE season_id=? AND card_id=?",
(season_id, card_id),
).fetchone()
if daily_row is not None:
day_key = str(dict(daily_row).get("day_key", "")).strip()[:16]
if not day_key:
try:
today = self.ensure_season_daily_game_card(season_id)
if str(today.get("card_id", "")).strip() == card_id:
day_key = str(today.get("day_key", "")).strip()[:16]
except ValueError:
day_key = ""
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT id
FROM season_room_progress_logs
WHERE user_id=? AND season_id=? AND room_id=?
""",
(safe_user_id, season_id, room_id),
).fetchone()
if row is not None:
return {"applied": False, "events": [], "task_updates": [], "card_updates": [], "season_id": season_id}
conn.execute(
"""
INSERT INTO season_room_progress_logs(user_id, season_id, room_id, created_at)
VALUES (?, ?, ?, ?)
""",
(safe_user_id, season_id, room_id, utc_now_iso()),
)
conn.commit()
events: list[tuple[str, int]] = [("complete_game", 1)]
ending_state = str(summary.get("ending_state", "")).strip().lower()
if ending_state in {"escaped", "victory", "success", "win", "pact", "sacrifice", "custom_end"}:
events.append(("win_game", 1))
actions = self._to_int(summary.get("total_actions", 0), 0, 0)
if actions > 0:
events.append(("submit_action", actions))
rounds = self._to_int(summary.get("rounds", 0), 0, 0)
if rounds > 0:
events.append(("round_count", rounds))
player_rows = summary.get("players", [])
if isinstance(player_rows, list):
for row in player_rows:
if not isinstance(row, dict):
continue
username = str(row.get("username", "")).strip()
if not username:
continue
if bool(row.get("is_mvp")):
events.append(("mvp_count", 1))
break
task_updates: list[dict[str, Any]] = []
card_updates: list[dict[str, Any]] = []
for event_key, event_amount in events:
task_res = self.progress_user_season_tasks(
safe_user_id,
season_id,
event_key=event_key,
amount=event_amount,
context={"room_id": room_id},
)
card_res = self.progress_user_season_cards(
safe_user_id,
season_id,
event_key=event_key,
amount=event_amount,
)
task_updates.append({"event_key": event_key, "updated": task_res.get("updated", []), "completed_ids": task_res.get("completed_ids", [])})
card_updates.append({"event_key": event_key, "updated": card_res.get("updated", []), "completed_ids": card_res.get("completed_ids", [])})
daily_completion = None
if day_key:
try:
daily_completion = self.mark_season_daily_card_completed(
safe_user_id,
season_id,
day_key=day_key,
room_id=room_id,
ending_state=ending_state,
ending_summary=str(summary.get("ending_summary", "")),
)
except ValueError:
daily_completion = None
return {
"applied": True,
"season_id": season_id,
"daily_card": {"day_key": day_key, "completed": bool(daily_completion)} if day_key else {"day_key": "", "completed": False},
"events": [{"event_key": key, "amount": amount} for key, amount in events],
"task_updates": task_updates,
"card_updates": card_updates,
}
def get_season_dashboard(self, season_id: str) -> dict[str, Any]:
safe_season_id = str(season_id or "").strip()
if not safe_season_id:
return {
"participants": 0,
"premium_users": 0,
"premium_conversion": 0.0,
"avg_level": 0.0,
"task_claimed": 0,
"card_claimed": 0,
}
with self._lock, self._connect() as conn:
participants_row = conn.execute(
"SELECT COUNT(*) AS c FROM user_season_profiles WHERE season_id=?",
(safe_season_id,),
).fetchone()
premium_row = conn.execute(
"SELECT COUNT(*) AS c FROM user_season_profiles WHERE season_id=? AND pass_type='premium'",
(safe_season_id,),
).fetchone()
level_row = conn.execute(
"SELECT COALESCE(AVG(level), 0) AS avg_level FROM user_season_profiles WHERE season_id=?",
(safe_season_id,),
).fetchone()
task_row = conn.execute(
"SELECT COUNT(*) AS c FROM user_season_tasks WHERE season_id=? AND status='claimed'",
(safe_season_id,),
).fetchone()
card_row = conn.execute(
"SELECT COUNT(*) AS c FROM season_activity_cards WHERE season_id=? AND status='claimed'",
(safe_season_id,),
).fetchone()
participants = self._to_int(dict(participants_row or {}).get("c", 0), 0, 0)
premium_users = self._to_int(dict(premium_row or {}).get("c", 0), 0, 0)
conversion = round((premium_users / participants) * 100, 2) if participants else 0.0
avg_level = round(float(dict(level_row or {}).get("avg_level", 0.0) or 0.0), 2)
task_claimed = self._to_int(dict(task_row or {}).get("c", 0), 0, 0)
card_claimed = self._to_int(dict(card_row or {}).get("c", 0), 0, 0)
return {
"participants": participants,
"premium_users": premium_users,
"premium_conversion": conversion,
"avg_level": avg_level,
"task_claimed": task_claimed,
"card_claimed": card_claimed,
}
def get_season_leaderboard(self, season_id: str, *, limit: int = 30) -> list[dict[str, Any]]:
safe_season_id = str(season_id or "").strip()
safe_limit = max(1, min(500, int(limit)))
if not safe_season_id:
return []
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT
p.user_id,
p.level,
p.xp,
p.season_points,
p.completed_tasks,
p.total_claimed_cards,
p.pass_type,
u.username,
u.display_name,
u.avatar_color,
u.avatar_url
FROM user_season_profiles p
JOIN users u ON u.user_id = p.user_id
WHERE p.season_id=?
ORDER BY p.season_points DESC, p.level DESC, p.xp DESC, p.completed_tasks DESC
LIMIT ?
""",
(safe_season_id, safe_limit),
).fetchall()
out: list[dict[str, Any]] = []
for idx, row in enumerate(rows, start=1):
item = dict(row)
item["rank"] = idx
item["level"] = self._to_int(item.get("level", 1), 1, 1)
item["xp"] = self._to_int(item.get("xp", 0), 0, 0)
item["season_points"] = self._to_int(item.get("season_points", 0), 0, 0)
item["completed_tasks"] = self._to_int(item.get("completed_tasks", 0), 0, 0)
item["total_claimed_cards"] = self._to_int(item.get("total_claimed_cards", 0), 0, 0)
out.append(item)
return out
def get_oauth_identity(self, provider: str, provider_user_id: str) -> dict[str, Any] | None:
safe_provider = str(provider or "").strip().lower()
safe_provider_user_id = str(provider_user_id or "").strip()
if not safe_provider or not safe_provider_user_id:
return None
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT provider, provider_user_id, user_id, username_snapshot, created_at, updated_at
FROM oauth_identities
WHERE provider=? AND provider_user_id=?
""",
(safe_provider, safe_provider_user_id),
).fetchone()
return dict(row) if row else None
def bind_oauth_identity(
self,
provider: str,
provider_user_id: str,
user_id: str,
*,
username_snapshot: str = "",
) -> None:
safe_provider = str(provider or "").strip().lower()
safe_provider_user_id = str(provider_user_id or "").strip()
safe_user_id = str(user_id or "").strip()
safe_snapshot = str(username_snapshot or "").strip()
if not safe_provider or not safe_provider_user_id or not safe_user_id:
raise ValueError("oauth_identity_invalid")
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute(
"""
INSERT INTO oauth_identities(provider, provider_user_id, user_id, username_snapshot, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(provider, provider_user_id) DO UPDATE SET
user_id=excluded.user_id,
username_snapshot=excluded.username_snapshot,
updated_at=excluded.updated_at
""",
(safe_provider, safe_provider_user_id, safe_user_id, safe_snapshot, now, now),
)
conn.commit()
def create_invitation(
self,
*,
from_user_id: str,
to_user_id: str,
room_id: str,
message: str = "",
) -> dict[str, Any]:
safe_from = str(from_user_id or "").strip()
safe_to = str(to_user_id or "").strip()
safe_room_id = str(room_id or "").strip().upper()
safe_message = str(message or "").strip()[:200]
if not safe_from or not safe_to or not safe_room_id:
raise ValueError("invite_invalid")
now = utc_now_iso()
with self._lock, self._connect() as conn:
existing = conn.execute(
"""
SELECT id FROM invitations
WHERE from_user_id=? AND to_user_id=? AND room_id=? AND status='pending'
ORDER BY id DESC
LIMIT 1
""",
(safe_from, safe_to, safe_room_id),
).fetchone()
if existing:
conn.execute(
"""
UPDATE invitations
SET message=?, updated_at=?
WHERE id=?
""",
(safe_message, now, int(existing["id"])),
)
conn.commit()
row = conn.execute(
"SELECT * FROM invitations WHERE id=?",
(int(existing["id"]),),
).fetchone()
return dict(row) if row else {}
conn.execute(
"""
INSERT INTO invitations(
from_user_id, to_user_id, room_id, message, status, created_at, updated_at
) VALUES (?, ?, ?, ?, 'pending', ?, ?)
""",
(safe_from, safe_to, safe_room_id, safe_message, now, now),
)
invite_id = int(conn.execute("SELECT last_insert_rowid()").fetchone()[0])
conn.commit()
row = conn.execute("SELECT * FROM invitations WHERE id=?", (invite_id,)).fetchone()
return dict(row) if row else {}
def list_user_invitations(
self,
to_user_id: str,
*,
status: str = "pending",
limit: int = 20,
) -> list[dict[str, Any]]:
safe_to = str(to_user_id or "").strip()
safe_status = str(status or "").strip().lower() or "pending"
try:
safe_limit = max(1, min(100, int(limit)))
except (TypeError, ValueError):
safe_limit = 20
if not safe_to:
return []
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT
i.id,
i.from_user_id,
i.to_user_id,
i.room_id,
i.message,
i.status,
i.created_at,
i.updated_at,
i.responded_at,
fu.username AS from_username,
fu.display_name AS from_display_name,
fu.avatar_color AS from_avatar_color,
fu.avatar_url AS from_avatar_url,
r.status AS room_status,
r.is_public AS room_is_public,
r.game_card_id AS room_game_card_id
FROM invitations i
JOIN users fu ON fu.user_id = i.from_user_id
LEFT JOIN rooms r ON r.room_id = i.room_id
WHERE i.to_user_id=? AND i.status=?
ORDER BY i.created_at DESC
LIMIT ?
""",
(safe_to, safe_status, safe_limit),
).fetchall()
result = []
for row in rows:
item = dict(row)
item["room_is_public"] = bool(item.get("room_is_public", 0))
result.append(item)
return result
def get_pending_invitation(self, invite_id: int, to_user_id: str) -> dict[str, Any] | None:
safe_to = str(to_user_id or "").strip()
try:
safe_id = int(invite_id)
except (TypeError, ValueError):
return None
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT *
FROM invitations
WHERE id=? AND to_user_id=? AND status='pending'
""",
(safe_id, safe_to),
).fetchone()
return dict(row) if row else None
def update_invitation_status(self, invite_id: int, to_user_id: str, status: str) -> dict[str, Any]:
safe_to = str(to_user_id or "").strip()
safe_status = str(status or "").strip().lower()
if safe_status not in {"accepted", "dismissed"}:
raise ValueError("invite_status_invalid")
try:
safe_id = int(invite_id)
except (TypeError, ValueError):
raise ValueError("invite_not_found")
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT *
FROM invitations
WHERE id=? AND to_user_id=? AND status='pending'
""",
(safe_id, safe_to),
).fetchone()
if row is None:
raise ValueError("invite_not_found")
now = utc_now_iso()
conn.execute(
"""
UPDATE invitations
SET status=?, responded_at=?, updated_at=?
WHERE id=?
""",
(safe_status, now, now, safe_id),
)
conn.commit()
out = conn.execute("SELECT * FROM invitations WHERE id=?", (safe_id,)).fetchone()
return dict(out) if out else {}
def list_users(self, limit: int = 500) -> list[dict[str, Any]]:
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT
u.user_id,
u.username,
u.display_name,
u.role,
u.avatar_color,
u.avatar_url,
u.bio,
u.created_at,
u.updated_at,
u.theme,
u.active_badge,
COALESCE(p.balance, 0) AS points_balance,
COALESCE(p.total_recharged, 0) AS points_total_recharged,
COALESCE(p.total_spent, 0) AS points_total_spent,
COALESCE(p.frozen_balance, 0) AS points_frozen_balance,
COALESCE(p.is_locked, 0) AS points_is_locked
FROM users u
LEFT JOIN user_points_accounts p ON p.user_id = u.user_id
ORDER BY u.created_at DESC
LIMIT ?
""",
(limit,),
).fetchall()
result = []
for row in rows:
item = dict(row)
item["points_is_locked"] = bool(item.get("points_is_locked", 0))
result.append(item)
return result
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)
item["world_id"] = str(item.get("world_id", "global")).strip()[:64] or "global"
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, *, world_id: str = "") -> list[dict[str, Any]]:
safe_world_id = str(world_id or "").strip()[:64]
with self._lock, self._connect() as conn:
if safe_world_id:
rows = conn.execute(
"""
SELECT * FROM game_cards
WHERE world_id=?
ORDER BY is_official DESC, created_at DESC
""",
(safe_world_id,),
).fetchall()
else:
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,
*,
world_id: str = "global",
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]:
safe_world_id = str(world_id or "global").strip()[:64] or "global"
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, world_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,
safe_world_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
def _build_module_payload(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
try:
item["patch_json"] = json.loads(item.get("patch_json", "{}"))
except json.JSONDecodeError:
item["patch_json"] = {}
if not isinstance(item.get("patch_json"), dict):
item["patch_json"] = {}
item["is_official"] = bool(item.get("is_official", 0))
return item
def list_card_modules(self, base_card_id: str, *, status: str = "published") -> list[dict[str, Any]]:
safe_base = str(base_card_id or "").strip()
safe_status = str(status or "").strip().lower() or "published"
if not safe_base:
return []
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT
m.module_id,
m.base_card_id,
m.title,
m.summary,
m.patch_json,
m.creator_user_id,
m.is_official,
m.status,
m.created_at,
m.updated_at,
u.username AS creator_username,
u.display_name AS creator_display_name
FROM card_modules m
LEFT JOIN users u ON u.user_id = m.creator_user_id
WHERE m.base_card_id=? AND m.status=?
ORDER BY m.is_official DESC, m.created_at DESC
""",
(safe_base, safe_status),
).fetchall()
return [self._build_module_payload(row) for row in rows]
def list_card_modules_admin(self, limit: int = 500) -> list[dict[str, Any]]:
safe_limit = max(1, min(1000, int(limit)))
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT
m.module_id,
m.base_card_id,
m.title,
m.summary,
m.patch_json,
m.creator_user_id,
m.is_official,
m.status,
m.created_at,
m.updated_at,
u.username AS creator_username,
u.display_name AS creator_display_name,
c.title AS base_card_title
FROM card_modules m
LEFT JOIN users u ON u.user_id = m.creator_user_id
LEFT JOIN game_cards c ON c.card_id = m.base_card_id
ORDER BY m.updated_at DESC, m.created_at DESC
LIMIT ?
""",
(safe_limit,),
).fetchall()
return [self._build_module_payload(row) for row in rows]
def get_card_module(self, module_id: str) -> dict[str, Any] | None:
safe_id = str(module_id or "").strip()
if not safe_id:
return None
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT
m.module_id,
m.base_card_id,
m.title,
m.summary,
m.patch_json,
m.creator_user_id,
m.is_official,
m.status,
m.created_at,
m.updated_at,
u.username AS creator_username,
u.display_name AS creator_display_name
FROM card_modules m
LEFT JOIN users u ON u.user_id = m.creator_user_id
WHERE m.module_id=?
""",
(safe_id,),
).fetchone()
if row is None:
return None
return self._build_module_payload(row)
def create_card_module(
self,
*,
base_card_id: str,
title: str,
summary: str,
patch_json: dict[str, Any] | None,
creator_user_id: str,
) -> dict[str, Any]:
safe_base = str(base_card_id or "").strip()
safe_title = str(title or "").strip()[:84]
safe_summary = str(summary or "").strip()[:240]
safe_creator = str(creator_user_id or "").strip()
if not safe_base or not safe_title or not safe_creator:
raise ValueError("invalid_card_payload")
if self.get_game_card(safe_base) is None:
raise ValueError("invalid_card_id")
if self.get_user_by_id(safe_creator) is None:
raise ValueError("user_not_found")
base_slug = re.sub(r"[^a-z0-9]+", "-", safe_title.lower()).strip("-") or "module"
module_id = f"{safe_base}-{base_slug}"[:52]
suffix = 2
while self.get_card_module(module_id) is not None:
module_id = f"{safe_base}-{base_slug[:40]}-{suffix}"[:60]
suffix += 1
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute(
"""
INSERT INTO card_modules(
module_id, base_card_id, title, summary, patch_json, creator_user_id,
is_official, status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, 0, 'published', ?, ?)
""",
(
module_id,
safe_base,
safe_title,
safe_summary,
json.dumps(patch_json or {}, ensure_ascii=False),
safe_creator,
now,
now,
),
)
conn.commit()
module = self.get_card_module(module_id)
if module is None:
raise ValueError("invalid_response")
return module
def update_card_module(
self,
module_id: str,
*,
title: str | None = None,
summary: str | None = None,
patch_json: dict[str, Any] | None = None,
actor_user_id: str,
as_admin: bool = False,
) -> dict[str, Any]:
existing = self.get_card_module(module_id)
if existing is None:
raise ValueError("invalid_card_id")
if not as_admin and str(existing.get("creator_user_id") or "") != str(actor_user_id):
raise ValueError("card_edit_forbidden")
next_title = str(title if title is not None else existing.get("title", "")).strip()[:84]
next_summary = str(summary if summary is not None else existing.get("summary", "")).strip()[:240]
next_patch = patch_json if patch_json is not None else existing.get("patch_json", {})
if not isinstance(next_patch, dict):
next_patch = {}
if not next_title:
raise ValueError("invalid_card_payload")
with self._lock, self._connect() as conn:
conn.execute(
"""
UPDATE card_modules
SET title=?, summary=?, patch_json=?, updated_at=?
WHERE module_id=?
""",
(
next_title,
next_summary,
json.dumps(next_patch, ensure_ascii=False),
utc_now_iso(),
str(module_id or "").strip(),
),
)
conn.commit()
updated = self.get_card_module(module_id)
if updated is None:
raise ValueError("invalid_card_id")
return updated
def delete_card_module(self, module_id: str, actor_user_id: str, *, as_admin: bool = False) -> None:
existing = self.get_card_module(module_id)
if existing is None:
raise ValueError("invalid_card_id")
if not as_admin and str(existing.get("creator_user_id") or "") != str(actor_user_id):
raise ValueError("card_delete_forbidden")
with self._lock, self._connect() as conn:
conn.execute("DELETE FROM card_modules WHERE module_id=?", (str(module_id or "").strip(),))
conn.commit()
def _to_int(self, value: Any, default: int = 0, minimum: int = 0) -> int:
try:
out = int(value)
except (TypeError, ValueError):
out = int(default)
return max(int(minimum), out)
def _safe_json_loads(self, raw_value: Any, fallback: Any) -> Any:
if not isinstance(raw_value, str):
return fallback
try:
parsed = json.loads(raw_value)
except json.JSONDecodeError:
return fallback
return parsed
def _ensure_points_account_tx(self, conn: sqlite3.Connection, user_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
raise ValueError("user_not_found")
row = conn.execute(
"""
SELECT user_id, balance, total_recharged, total_spent, frozen_balance, is_locked, created_at, updated_at
FROM user_points_accounts
WHERE user_id=?
""",
(safe_user_id,),
).fetchone()
if row:
return dict(row)
if self.get_user_by_id(safe_user_id) is None:
raise ValueError("user_not_found")
now = utc_now_iso()
conn.execute(
"""
INSERT INTO user_points_accounts(
user_id, balance, total_recharged, total_spent, frozen_balance, is_locked, created_at, updated_at
) VALUES (?, 0, 0, 0, 0, 0, ?, ?)
""",
(safe_user_id, now, now),
)
return {
"user_id": safe_user_id,
"balance": 0,
"total_recharged": 0,
"total_spent": 0,
"frozen_balance": 0,
"is_locked": 0,
"created_at": now,
"updated_at": now,
}
def _update_points_account_tx(
self,
conn: sqlite3.Connection,
user_id: str,
*,
balance: int,
total_recharged: int | None = None,
total_spent: int | None = None,
frozen_balance: int | None = None,
is_locked: bool | None = None,
) -> dict[str, Any]:
current = self._ensure_points_account_tx(conn, user_id)
next_total_recharged = (
self._to_int(total_recharged, current.get("total_recharged", 0), 0)
if total_recharged is not None
else self._to_int(current.get("total_recharged", 0), 0, 0)
)
next_total_spent = (
self._to_int(total_spent, current.get("total_spent", 0), 0)
if total_spent is not None
else self._to_int(current.get("total_spent", 0), 0, 0)
)
next_frozen = (
self._to_int(frozen_balance, current.get("frozen_balance", 0), 0)
if frozen_balance is not None
else self._to_int(current.get("frozen_balance", 0), 0, 0)
)
safe_balance = self._to_int(balance, current.get("balance", 0), 0)
next_frozen = min(next_frozen, safe_balance)
locked_flag = bool(is_locked) if is_locked is not None else bool(current.get("is_locked", 0))
now = utc_now_iso()
conn.execute(
"""
UPDATE user_points_accounts
SET balance=?, total_recharged=?, total_spent=?, frozen_balance=?, is_locked=?, updated_at=?
WHERE user_id=?
""",
(
safe_balance,
next_total_recharged,
next_total_spent,
next_frozen,
1 if locked_flag else 0,
now,
str(user_id or "").strip(),
),
)
return {
"user_id": str(user_id or "").strip(),
"balance": safe_balance,
"total_recharged": next_total_recharged,
"total_spent": next_total_spent,
"frozen_balance": next_frozen,
"is_locked": locked_flag,
"updated_at": now,
}
def _append_point_tx_tx(
self,
conn: sqlite3.Connection,
user_id: str,
*,
change_amount: int,
balance_after: int,
tx_type: str,
source: str = "",
related_id: str = "",
note: str = "",
expires_at: str = "",
status: str = "applied",
operator_user_id: str = "",
) -> dict[str, Any]:
now = utc_now_iso()
conn.execute(
"""
INSERT INTO user_point_transactions(
user_id, change_amount, balance_after, tx_type, source, related_id, note, expires_at, status,
operator_user_id, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
str(user_id or "").strip(),
int(change_amount),
int(balance_after),
str(tx_type or "").strip()[:48] or "unknown",
str(source or "").strip()[:64],
str(related_id or "").strip()[:120],
str(note or "").strip()[:300],
str(expires_at or "").strip()[:40],
str(status or "applied").strip()[:24],
str(operator_user_id or "").strip()[:64],
now,
),
)
tx_id = int(conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"])
return {
"tx_id": tx_id,
"user_id": str(user_id or "").strip(),
"change_amount": int(change_amount),
"balance_after": int(balance_after),
"tx_type": str(tx_type or "").strip()[:48] or "unknown",
"source": str(source or "").strip()[:64],
"related_id": str(related_id or "").strip()[:120],
"note": str(note or "").strip()[:300],
"expires_at": str(expires_at or "").strip()[:40],
"status": str(status or "applied").strip()[:24],
"operator_user_id": str(operator_user_id or "").strip()[:64],
"created_at": now,
}
def _insert_point_lot_tx(
self,
conn: sqlite3.Connection,
user_id: str,
*,
source_tx_id: int | None,
amount: int,
expires_at: str = "",
) -> None:
safe_amount = self._to_int(amount, 0, 0)
if safe_amount <= 0:
return
now = utc_now_iso()
conn.execute(
"""
INSERT INTO user_point_lots(
user_id, source_tx_id, original_amount, remaining_amount, expires_at, status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, 'active', ?, ?)
""",
(
str(user_id or "").strip(),
int(source_tx_id) if source_tx_id is not None else None,
safe_amount,
safe_amount,
str(expires_at or "").strip()[:40],
now,
now,
),
)
def _consume_point_lots_tx(self, conn: sqlite3.Connection, user_id: str, amount: int) -> int:
remain = self._to_int(amount, 0, 0)
if remain <= 0:
return 0
rows = conn.execute(
"""
SELECT lot_id, remaining_amount, expires_at
FROM user_point_lots
WHERE user_id=? AND status='active' AND remaining_amount > 0
ORDER BY
CASE WHEN expires_at='' THEN 1 ELSE 0 END ASC,
CASE WHEN expires_at='' THEN '' ELSE expires_at END ASC,
created_at ASC,
lot_id ASC
""",
(str(user_id or "").strip(),),
).fetchall()
used = 0
now = utc_now_iso()
for row in rows:
if remain <= 0:
break
lot_id = int(row["lot_id"])
available = self._to_int(row["remaining_amount"], 0, 0)
if available <= 0:
continue
spend = min(available, remain)
next_amount = available - spend
next_status = "active" if next_amount > 0 else "consumed"
conn.execute(
"""
UPDATE user_point_lots
SET remaining_amount=?, status=?, updated_at=?
WHERE lot_id=?
""",
(next_amount, next_status, now, lot_id),
)
remain -= spend
used += spend
return used
def _expire_point_lots_tx(
self,
conn: sqlite3.Connection,
*,
user_id: str | None = None,
) -> int:
now = utc_now_iso()
params: list[Any] = [now]
where_user = ""
if user_id:
where_user = "AND user_id=?"
params.append(str(user_id).strip())
rows = conn.execute(
f"""
SELECT lot_id, user_id, remaining_amount, expires_at
FROM user_point_lots
WHERE status='active' AND remaining_amount>0 AND expires_at<>'' AND expires_at<=?
{where_user}
ORDER BY expires_at ASC, lot_id ASC
""",
tuple(params),
).fetchall()
if not rows:
return 0
expired_count = 0
for row in rows:
safe_user_id = str(row["user_id"])
remain_amount = self._to_int(row.get("remaining_amount", 0), 0, 0)
if remain_amount <= 0:
continue
account = self._ensure_points_account_tx(conn, safe_user_id)
current_balance = self._to_int(account.get("balance", 0), 0, 0)
next_balance = max(0, current_balance - remain_amount)
updated = self._update_points_account_tx(
conn,
safe_user_id,
balance=next_balance,
total_recharged=self._to_int(account.get("total_recharged", 0), 0, 0),
total_spent=self._to_int(account.get("total_spent", 0), 0, 0),
frozen_balance=self._to_int(account.get("frozen_balance", 0), 0, 0),
is_locked=bool(account.get("is_locked", 0)),
)
self._append_point_tx_tx(
conn,
safe_user_id,
change_amount=-remain_amount,
balance_after=self._to_int(updated.get("balance", 0), 0, 0),
tx_type="expire",
source="points_expire",
related_id=str(row.get("lot_id", "")),
note="积分过期自动扣减",
status="applied",
)
conn.execute(
"""
UPDATE user_point_lots
SET remaining_amount=0, status='expired', updated_at=?
WHERE lot_id=?
""",
(now, int(row["lot_id"])),
)
expired_count += 1
return expired_count
def ensure_points_account(self, user_id: str) -> dict[str, Any]:
with self._lock, self._connect() as conn:
row = self._ensure_points_account_tx(conn, user_id)
conn.commit()
row["is_locked"] = bool(row.get("is_locked", 0))
return row
def get_points_account(self, user_id: str) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return {
"user_id": "",
"balance": 0,
"total_recharged": 0,
"total_spent": 0,
"frozen_balance": 0,
"available_balance": 0,
"is_locked": False,
}
with self._lock, self._connect() as conn:
self._ensure_points_account_tx(conn, safe_user_id)
self._expire_point_lots_tx(conn, user_id=safe_user_id)
row = conn.execute(
"""
SELECT user_id, balance, total_recharged, total_spent, frozen_balance, is_locked, created_at, updated_at
FROM user_points_accounts
WHERE user_id=?
""",
(safe_user_id,),
).fetchone()
conn.commit()
item = dict(row) if row else {}
balance = self._to_int(item.get("balance", 0), 0, 0)
frozen = self._to_int(item.get("frozen_balance", 0), 0, 0)
is_locked = bool(item.get("is_locked", 0))
return {
"user_id": safe_user_id,
"balance": balance,
"total_recharged": self._to_int(item.get("total_recharged", 0), 0, 0),
"total_spent": self._to_int(item.get("total_spent", 0), 0, 0),
"frozen_balance": min(frozen, balance),
"available_balance": max(0, balance - frozen),
"is_locked": is_locked,
"created_at": str(item.get("created_at", "")).strip(),
"updated_at": str(item.get("updated_at", "")).strip(),
}
def _credit_points_tx(
self,
conn: sqlite3.Connection,
user_id: str,
amount: int,
*,
tx_type: str,
source: str,
related_id: str = "",
note: str = "",
expires_at: str = "",
operator_user_id: str = "",
count_recharge: bool = False,
create_lot: bool = True,
allow_locked: bool = False,
) -> dict[str, Any]:
safe_amount = self._to_int(amount, 0, 1)
account = self._ensure_points_account_tx(conn, user_id)
if bool(account.get("is_locked", 0)) and not allow_locked:
raise ValueError("points_locked")
next_balance = self._to_int(account.get("balance", 0), 0, 0) + safe_amount
total_recharged = self._to_int(account.get("total_recharged", 0), 0, 0)
if count_recharge:
total_recharged += safe_amount
updated = self._update_points_account_tx(
conn,
user_id,
balance=next_balance,
total_recharged=total_recharged,
total_spent=self._to_int(account.get("total_spent", 0), 0, 0),
frozen_balance=self._to_int(account.get("frozen_balance", 0), 0, 0),
is_locked=bool(account.get("is_locked", 0)),
)
tx = self._append_point_tx_tx(
conn,
user_id,
change_amount=safe_amount,
balance_after=self._to_int(updated.get("balance", 0), 0, 0),
tx_type=tx_type,
source=source,
related_id=related_id,
note=note,
expires_at=expires_at,
status="applied",
operator_user_id=operator_user_id,
)
if create_lot:
self._insert_point_lot_tx(
conn,
user_id,
source_tx_id=tx.get("tx_id"),
amount=safe_amount,
expires_at=expires_at,
)
return {"account": updated, "tx": tx}
def _debit_points_tx(
self,
conn: sqlite3.Connection,
user_id: str,
amount: int,
*,
tx_type: str,
source: str,
related_id: str = "",
note: str = "",
operator_user_id: str = "",
count_spent: bool = False,
allow_locked: bool = False,
allow_negative_check: bool = False,
) -> dict[str, Any]:
safe_amount = self._to_int(amount, 0, 1)
account = self._ensure_points_account_tx(conn, user_id)
if bool(account.get("is_locked", 0)) and not allow_locked:
raise ValueError("points_locked")
balance = self._to_int(account.get("balance", 0), 0, 0)
frozen = min(balance, self._to_int(account.get("frozen_balance", 0), 0, 0))
available = max(0, balance - frozen)
if not allow_negative_check and safe_amount > available:
raise ValueError("points_insufficient")
next_balance = max(0, balance - safe_amount)
total_spent = self._to_int(account.get("total_spent", 0), 0, 0)
if count_spent:
total_spent += safe_amount
updated = self._update_points_account_tx(
conn,
user_id,
balance=next_balance,
total_recharged=self._to_int(account.get("total_recharged", 0), 0, 0),
total_spent=total_spent,
frozen_balance=self._to_int(account.get("frozen_balance", 0), 0, 0),
is_locked=bool(account.get("is_locked", 0)),
)
self._consume_point_lots_tx(conn, user_id, safe_amount)
tx = self._append_point_tx_tx(
conn,
user_id,
change_amount=-safe_amount,
balance_after=self._to_int(updated.get("balance", 0), 0, 0),
tx_type=tx_type,
source=source,
related_id=related_id,
note=note,
status="applied",
operator_user_id=operator_user_id,
)
return {"account": updated, "tx": tx}
def credit_user_points(
self,
user_id: str,
amount: int,
*,
tx_type: str = "reward",
source: str = "system",
related_id: str = "",
note: str = "",
expires_at: str = "",
operator_user_id: str = "",
count_recharge: bool = False,
create_lot: bool = True,
allow_locked: bool = False,
) -> dict[str, Any]:
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
self._expire_point_lots_tx(conn, user_id=str(user_id or "").strip())
payload = self._credit_points_tx(
conn,
user_id,
amount,
tx_type=tx_type,
source=source,
related_id=related_id,
note=note,
expires_at=expires_at,
operator_user_id=operator_user_id,
count_recharge=count_recharge,
create_lot=create_lot,
allow_locked=allow_locked,
)
conn.commit()
account = self.get_points_account(str(user_id or "").strip())
return {"account": account, "transaction": payload["tx"]}
def debit_user_points(
self,
user_id: str,
amount: int,
*,
tx_type: str = "consume",
source: str = "system",
related_id: str = "",
note: str = "",
operator_user_id: str = "",
count_spent: bool = True,
allow_locked: bool = False,
) -> dict[str, Any]:
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
self._expire_point_lots_tx(conn, user_id=str(user_id or "").strip())
payload = self._debit_points_tx(
conn,
user_id,
amount,
tx_type=tx_type,
source=source,
related_id=related_id,
note=note,
operator_user_id=operator_user_id,
count_spent=count_spent,
allow_locked=allow_locked,
)
conn.commit()
account = self.get_points_account(str(user_id or "").strip())
return {"account": account, "transaction": payload["tx"]}
def expire_user_points(self, user_id: str | None = None) -> int:
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
count = self._expire_point_lots_tx(conn, user_id=str(user_id).strip() if user_id else None)
conn.commit()
return int(count)
def list_user_point_transactions(
self,
user_id: str,
*,
limit: int = 100,
tx_type: str = "",
) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return []
safe_limit = max(1, min(500, int(limit)))
safe_type = str(tx_type or "").strip()
with self._lock, self._connect() as conn:
if safe_type:
rows = conn.execute(
"""
SELECT tx_id, user_id, change_amount, balance_after, tx_type, source, related_id, note,
expires_at, status, operator_user_id, created_at
FROM user_point_transactions
WHERE user_id=? AND tx_type=?
ORDER BY tx_id DESC
LIMIT ?
""",
(safe_user_id, safe_type, safe_limit),
).fetchall()
else:
rows = conn.execute(
"""
SELECT tx_id, user_id, change_amount, balance_after, tx_type, source, related_id, note,
expires_at, status, operator_user_id, created_at
FROM user_point_transactions
WHERE user_id=?
ORDER BY tx_id DESC
LIMIT ?
""",
(safe_user_id, safe_limit),
).fetchall()
return [dict(row) for row in rows]
def list_point_transactions_admin(
self,
*,
user_id: str = "",
limit: int = 300,
) -> list[dict[str, Any]]:
safe_limit = max(1, min(1000, int(limit)))
safe_user_id = str(user_id or "").strip()
with self._lock, self._connect() as conn:
if safe_user_id:
rows = conn.execute(
"""
SELECT t.*, u.username
FROM user_point_transactions t
LEFT JOIN users u ON u.user_id = t.user_id
WHERE t.user_id=?
ORDER BY t.tx_id DESC
LIMIT ?
""",
(safe_user_id, safe_limit),
).fetchall()
else:
rows = conn.execute(
"""
SELECT t.*, u.username
FROM user_point_transactions t
LEFT JOIN users u ON u.user_id = t.user_id
ORDER BY t.tx_id DESC
LIMIT ?
""",
(safe_limit,),
).fetchall()
return [dict(row) for row in rows]
def update_points_lock(
self,
user_id: str,
*,
is_locked: bool | None = None,
frozen_balance: int | None = None,
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
raise ValueError("user_not_found")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
account = self._ensure_points_account_tx(conn, safe_user_id)
updated = self._update_points_account_tx(
conn,
safe_user_id,
balance=self._to_int(account.get("balance", 0), 0, 0),
total_recharged=self._to_int(account.get("total_recharged", 0), 0, 0),
total_spent=self._to_int(account.get("total_spent", 0), 0, 0),
frozen_balance=self._to_int(
frozen_balance if frozen_balance is not None else account.get("frozen_balance", 0),
0,
0,
),
is_locked=bool(is_locked) if is_locked is not None else bool(account.get("is_locked", 0)),
)
conn.commit()
return self.get_points_account(safe_user_id)
def admin_adjust_user_points(
self,
user_id: str,
delta: int,
*,
note: str = "",
operator_user_id: str = "",
count_recharge: bool = False,
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
raise ValueError("user_not_found")
safe_delta = int(delta)
if safe_delta == 0:
raise ValueError("invalid_points_amount")
if safe_delta > 0:
return self.credit_user_points(
safe_user_id,
safe_delta,
tx_type="admin_adjust",
source="admin",
note=note or "后台调增积分",
operator_user_id=operator_user_id,
count_recharge=count_recharge,
create_lot=True,
allow_locked=True,
)
return self.debit_user_points(
safe_user_id,
abs(safe_delta),
tx_type="admin_adjust",
source="admin",
note=note or "后台调减积分",
operator_user_id=operator_user_id,
count_spent=False,
allow_locked=True,
)
def _record_activation_attempt_tx(
self,
conn: sqlite3.Connection,
*,
ip_address: str,
user_id: str,
code: str,
success: bool,
error_code: str = "",
) -> None:
conn.execute(
"""
INSERT INTO activation_code_attempts(ip_address, user_id, code, success, error_code, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
str(ip_address or "").strip()[:96],
str(user_id or "").strip()[:64],
str(code or "").strip()[:80],
1 if success else 0,
str(error_code or "").strip()[:64],
utc_now_iso(),
),
)
def _check_activation_rate_limit_tx(
self,
conn: sqlite3.Connection,
*,
ip_address: str,
user_id: str,
window_seconds: int = 60,
ip_limit: int = 12,
user_limit: int = 10,
) -> None:
safe_window = max(10, min(3600, int(window_seconds)))
threshold = (datetime.now(timezone.utc) - timedelta(seconds=safe_window)).isoformat()
safe_ip = str(ip_address or "").strip()[:96]
safe_user = str(user_id or "").strip()[:64]
if safe_ip:
ip_row = conn.execute(
"SELECT COUNT(*) AS c FROM activation_code_attempts WHERE ip_address=? AND created_at>=?",
(safe_ip, threshold),
).fetchone()
if ip_row and self._to_int(ip_row.get("c", 0), 0, 0) >= max(1, int(ip_limit)):
raise ValueError("redeem_rate_limited")
if safe_user:
user_row = conn.execute(
"SELECT COUNT(*) AS c FROM activation_code_attempts WHERE user_id=? AND created_at>=?",
(safe_user, threshold),
).fetchone()
if user_row and self._to_int(user_row.get("c", 0), 0, 0) >= max(1, int(user_limit)):
raise ValueError("redeem_rate_limited")
def _refresh_activation_code_status_tx(self, conn: sqlite3.Connection) -> None:
now = utc_now_iso()
conn.execute(
"""
UPDATE activation_codes
SET status='expired', updated_at=?
WHERE status='unused' AND expires_at<>'' AND expires_at<?
""",
(now, now),
)
def _generate_activation_code(self, prefix: str = "ET", length: int = 16) -> str:
alphabet = string.ascii_uppercase + string.digits
safe_length = max(10, min(28, int(length)))
core = "".join(secrets.choice(alphabet) for _ in range(safe_length))
safe_prefix = re.sub(r"[^A-Z0-9]+", "", str(prefix or "").upper())[:8]
return f"{safe_prefix}{core}" if safe_prefix else core
def generate_activation_codes(
self,
*,
points: int,
quantity: int,
expires_at: str = "",
batch_no: str = "",
remark: str = "",
created_by: str = "",
code_prefix: str = "ET",
) -> list[dict[str, Any]]:
safe_points = self._to_int(points, 0, 1)
safe_quantity = max(1, min(5000, int(quantity)))
safe_batch = str(batch_no or "").strip()[:64] or datetime.now(timezone.utc).strftime("BATCH%Y%m%d%H%M%S")
safe_remark = str(remark or "").strip()[:180]
safe_expires = str(expires_at or "").strip()[:40]
safe_created_by = str(created_by or "").strip()[:64]
created_rows: list[dict[str, Any]] = []
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
now = utc_now_iso()
for _ in range(safe_quantity):
code = self._generate_activation_code(prefix=code_prefix, length=14)
retry = 0
while conn.execute("SELECT 1 FROM activation_codes WHERE code=?", (code,)).fetchone():
retry += 1
if retry > 12:
code = f"{self._generate_activation_code(prefix=code_prefix, length=10)}{uuid.uuid4().hex[:4].upper()}"
break
code = self._generate_activation_code(prefix=code_prefix, length=14)
conn.execute(
"""
INSERT INTO activation_codes(
code, points, status, batch_no, remark, expires_at, created_by, created_at, updated_at
) VALUES (?, ?, 'unused', ?, ?, ?, ?, ?, ?)
""",
(
code,
safe_points,
safe_batch,
safe_remark,
safe_expires,
safe_created_by,
now,
now,
),
)
created_rows.append(
{
"code": code,
"points": safe_points,
"status": "unused",
"batch_no": safe_batch,
"remark": safe_remark,
"expires_at": safe_expires,
"created_by": safe_created_by,
"created_at": now,
"updated_at": now,
"used_by_user_id": "",
"used_at": "",
"used_ip": "",
}
)
conn.commit()
return created_rows
def list_activation_codes(
self,
*,
status: str = "",
batch_no: str = "",
keyword: str = "",
limit: int = 300,
) -> list[dict[str, Any]]:
safe_status = str(status or "").strip().lower()
safe_batch = str(batch_no or "").strip()
safe_keyword = str(keyword or "").strip()
safe_limit = max(1, min(3000, int(limit)))
with self._lock, self._connect() as conn:
self._refresh_activation_code_status_tx(conn)
conn.commit()
where: list[str] = []
params: list[Any] = []
if safe_status:
where.append("c.status=?")
params.append(safe_status)
if safe_batch:
where.append("c.batch_no=?")
params.append(safe_batch)
if safe_keyword:
where.append("(c.code LIKE ? OR c.remark LIKE ?)")
like = f"%{safe_keyword}%"
params.extend([like, like])
where_sql = f"WHERE {' AND '.join(where)}" if where else ""
rows = conn.execute(
f"""
SELECT c.*, u.username AS used_by_username
FROM activation_codes c
LEFT JOIN users u ON u.user_id = c.used_by_user_id
{where_sql}
ORDER BY c.created_at DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
return [dict(row) for row in rows]
def update_activation_code(
self,
code: str,
*,
status: str | None = None,
expires_at: str | None = None,
remark: str | None = None,
) -> dict[str, Any]:
safe_code = str(code or "").strip().upper()
if not safe_code:
raise ValueError("activation_code_not_found")
allowed_status = {"unused", "void", "expired", "used"}
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute("SELECT * FROM activation_codes WHERE code=?", (safe_code,)).fetchone()
if not row:
raise ValueError("activation_code_not_found")
current = dict(row)
next_status = str(status if status is not None else current.get("status", "unused")).strip().lower()
if next_status not in allowed_status:
raise ValueError("invalid_activation_code_status")
if current.get("status") == "used" and next_status != "used":
raise ValueError("activation_code_used")
next_expires = (
str(expires_at).strip()[:40] if expires_at is not None else str(current.get("expires_at", "")).strip()
)
next_remark = str(remark).strip()[:180] if remark is not None else str(current.get("remark", "")).strip()[:180]
now = utc_now_iso()
conn.execute(
"""
UPDATE activation_codes
SET status=?, expires_at=?, remark=?, updated_at=?
WHERE code=?
""",
(next_status, next_expires, next_remark, now, safe_code),
)
updated = conn.execute("SELECT * FROM activation_codes WHERE code=?", (safe_code,)).fetchone()
conn.commit()
return dict(updated) if updated else {}
def redeem_activation_code(
self,
user_id: str,
code: str,
*,
ip_address: str = "",
ip_limit: int = 12,
user_limit: int = 10,
window_seconds: int = 60,
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_code = str(code or "").strip().upper()
if not safe_user_id:
raise ValueError("auth_required")
if not safe_code:
raise ValueError("activation_code_invalid")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
self._expire_point_lots_tx(conn, user_id=safe_user_id)
self._check_activation_rate_limit_tx(
conn,
ip_address=ip_address,
user_id=safe_user_id,
window_seconds=window_seconds,
ip_limit=ip_limit,
user_limit=user_limit,
)
self._refresh_activation_code_status_tx(conn)
row = conn.execute("SELECT * FROM activation_codes WHERE code=?", (safe_code,)).fetchone()
if not row:
self._record_activation_attempt_tx(
conn,
ip_address=ip_address,
user_id=safe_user_id,
code=safe_code,
success=False,
error_code="activation_code_invalid",
)
conn.commit()
raise ValueError("activation_code_invalid")
code_row = dict(row)
status = str(code_row.get("status", "")).strip().lower()
if status != "unused":
self._record_activation_attempt_tx(
conn,
ip_address=ip_address,
user_id=safe_user_id,
code=safe_code,
success=False,
error_code="activation_code_used" if status == "used" else "activation_code_unavailable",
)
conn.commit()
raise ValueError("activation_code_used" if status == "used" else "activation_code_unavailable")
expires = str(code_row.get("expires_at", "")).strip()
if expires and expires < utc_now_iso():
conn.execute(
"UPDATE activation_codes SET status='expired', updated_at=? WHERE code=?",
(utc_now_iso(), safe_code),
)
self._record_activation_attempt_tx(
conn,
ip_address=ip_address,
user_id=safe_user_id,
code=safe_code,
success=False,
error_code="activation_code_expired",
)
conn.commit()
raise ValueError("activation_code_expired")
now = utc_now_iso()
result = conn.execute(
"""
UPDATE activation_codes
SET status='used', used_by_user_id=?, used_at=?, used_ip=?, updated_at=?
WHERE code=? AND status='unused'
""",
(safe_user_id, now, str(ip_address or "").strip()[:96], now, safe_code),
)
if result.rowcount != 1:
self._record_activation_attempt_tx(
conn,
ip_address=ip_address,
user_id=safe_user_id,
code=safe_code,
success=False,
error_code="activation_code_used",
)
conn.commit()
raise ValueError("activation_code_used")
points = self._to_int(code_row.get("points", 0), 0, 1)
payload = self._credit_points_tx(
conn,
safe_user_id,
points,
tx_type="recharge",
source="activation_code",
related_id=safe_code,
note=f"激活码充值 {safe_code}",
operator_user_id="",
count_recharge=True,
create_lot=True,
allow_locked=False,
)
self._record_activation_attempt_tx(
conn,
ip_address=ip_address,
user_id=safe_user_id,
code=safe_code,
success=True,
error_code="",
)
conn.commit()
account = self.get_points_account(safe_user_id)
return {
"code": safe_code,
"points": points,
"account": account,
"transaction": payload["tx"],
}
def _parse_shop_item(self, row: sqlite3.Row | dict[str, Any]) -> dict[str, Any]:
item = dict(row)
metadata = self._safe_json_loads(item.get("metadata_json", "{}"), {})
if not isinstance(metadata, dict):
metadata = {}
item["metadata"] = metadata
item["price"] = self._to_int(item.get("price", 0), 0, 0)
item["stock"] = int(item.get("stock", -1))
item["purchase_limit"] = self._to_int(item.get("purchase_limit", 0), 0, 0)
item["sort_order"] = int(item.get("sort_order", 0))
item["is_recommended"] = bool(item.get("is_recommended", 0))
return item
def create_shop_item(
self,
*,
name: str,
description: str = "",
image_url: str = "",
item_type: str = "decoration",
subtype: str = "",
price: int = 0,
stock: int = -1,
purchase_limit: int = 0,
status: str = "active",
sort_order: int = 0,
is_recommended: bool = False,
metadata: dict[str, Any] | None = None,
item_id: str = "",
) -> dict[str, Any]:
safe_name = str(name or "").strip()[:84]
if not safe_name:
raise ValueError("invalid_shop_item")
safe_item_type = str(item_type or "decoration").strip().lower()[:32] or "decoration"
safe_status = str(status or "active").strip().lower()
if safe_status not in {"active", "inactive"}:
safe_status = "inactive"
safe_item_id = str(item_id or "").strip().lower()
if safe_item_id:
safe_item_id = re.sub(r"[^a-z0-9_-]+", "_", safe_item_id).strip("_")[:64]
if not safe_item_id:
base = re.sub(r"[^a-z0-9_-]+", "_", safe_name.lower()).strip("_")[:40] or "item"
safe_item_id = f"{safe_item_type}_{base}_{uuid.uuid4().hex[:6]}"
try:
stock_value = int(stock)
except (TypeError, ValueError):
stock_value = -1
stock_value = stock_value if stock_value >= -1 else -1
now = utc_now_iso()
with self._lock, self._connect() as conn:
conn.execute(
"""
INSERT INTO shop_items(
item_id, name, description, image_url, item_type, subtype, price, stock, purchase_limit, status,
sort_order, is_recommended, metadata_json, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_item_id,
safe_name,
str(description or "").strip()[:360],
str(image_url or "").strip()[:720],
safe_item_type,
str(subtype or "").strip()[:48],
self._to_int(price, 0, 0),
stock_value,
self._to_int(purchase_limit, 0, 0),
safe_status,
int(sort_order),
1 if bool(is_recommended) else 0,
json.dumps(metadata or {}, ensure_ascii=False),
now,
now,
),
)
conn.commit()
row = conn.execute("SELECT * FROM shop_items WHERE item_id=?", (safe_item_id,)).fetchone()
return self._parse_shop_item(row) if row else {}
def update_shop_item(self, item_id: str, patch: dict[str, Any]) -> dict[str, Any]:
safe_item_id = str(item_id or "").strip().lower()
if not safe_item_id:
raise ValueError("shop_item_not_found")
allowed_fields = {
"name",
"description",
"image_url",
"item_type",
"subtype",
"price",
"stock",
"purchase_limit",
"status",
"sort_order",
"is_recommended",
"metadata",
}
payload = patch if isinstance(patch, dict) else {}
with self._lock, self._connect() as conn:
row = conn.execute("SELECT * FROM shop_items WHERE item_id=?", (safe_item_id,)).fetchone()
if not row:
raise ValueError("shop_item_not_found")
current = self._parse_shop_item(row)
merged = dict(current)
for key, value in payload.items():
if key not in allowed_fields:
continue
if key == "name":
merged[key] = str(value or "").strip()[:84]
elif key == "description":
merged[key] = str(value or "").strip()[:360]
elif key == "image_url":
merged[key] = str(value or "").strip()[:720]
elif key == "item_type":
merged[key] = str(value or "").strip().lower()[:32] or merged.get("item_type", "decoration")
elif key == "subtype":
merged[key] = str(value or "").strip()[:48]
elif key == "price":
merged[key] = self._to_int(value, merged.get("price", 0), 0)
elif key == "stock":
try:
stock_value = int(value)
except (TypeError, ValueError):
stock_value = int(merged.get("stock", -1))
merged[key] = stock_value if stock_value >= -1 else -1
elif key == "purchase_limit":
merged[key] = self._to_int(value, merged.get("purchase_limit", 0), 0)
elif key == "status":
text = str(value or "").strip().lower()
merged[key] = text if text in {"active", "inactive"} else merged.get("status", "inactive")
elif key == "sort_order":
try:
merged[key] = int(value)
except (TypeError, ValueError):
merged[key] = int(merged.get("sort_order", 0))
elif key == "is_recommended":
merged[key] = bool(value)
elif key == "metadata":
merged[key] = value if isinstance(value, dict) else merged.get("metadata", {})
if not str(merged.get("name", "")).strip():
raise ValueError("invalid_shop_item")
now = utc_now_iso()
conn.execute(
"""
UPDATE shop_items
SET name=?, description=?, image_url=?, item_type=?, subtype=?, price=?, stock=?, purchase_limit=?,
status=?, sort_order=?, is_recommended=?, metadata_json=?, updated_at=?
WHERE item_id=?
""",
(
str(merged.get("name", "")).strip()[:84],
str(merged.get("description", "")).strip()[:360],
str(merged.get("image_url", "")).strip()[:720],
str(merged.get("item_type", "decoration")).strip().lower()[:32],
str(merged.get("subtype", "")).strip()[:48],
self._to_int(merged.get("price", 0), 0, 0),
int(merged.get("stock", -1)),
self._to_int(merged.get("purchase_limit", 0), 0, 0),
str(merged.get("status", "inactive")).strip().lower(),
int(merged.get("sort_order", 0)),
1 if bool(merged.get("is_recommended", False)) else 0,
json.dumps(merged.get("metadata", {}) if isinstance(merged.get("metadata"), dict) else {}, ensure_ascii=False),
now,
safe_item_id,
),
)
conn.commit()
updated = conn.execute("SELECT * FROM shop_items WHERE item_id=?", (safe_item_id,)).fetchone()
return self._parse_shop_item(updated) if updated else {}
def get_shop_item(self, item_id: str) -> dict[str, Any] | None:
safe_item_id = str(item_id or "").strip().lower()
if not safe_item_id:
return None
with self._lock, self._connect() as conn:
row = conn.execute("SELECT * FROM shop_items WHERE item_id=?", (safe_item_id,)).fetchone()
if not row:
return None
return self._parse_shop_item(row)
def list_shop_items(
self,
*,
status: str = "active",
item_type: str = "",
keyword: str = "",
limit: int = 200,
) -> list[dict[str, Any]]:
safe_limit = max(1, min(3000, int(limit)))
safe_status = str(status or "").strip().lower()
safe_type = str(item_type or "").strip().lower()
safe_keyword = str(keyword or "").strip()
where: list[str] = []
params: list[Any] = []
if safe_status:
where.append("status=?")
params.append(safe_status)
if safe_type:
where.append("item_type=?")
params.append(safe_type)
if safe_keyword:
where.append("(name LIKE ? OR description LIKE ?)")
like = f"%{safe_keyword}%"
params.extend([like, like])
where_sql = f"WHERE {' AND '.join(where)}" if where else ""
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT *
FROM shop_items
{where_sql}
ORDER BY is_recommended DESC, sort_order DESC, created_at DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
return [self._parse_shop_item(row) for row in rows]
def delete_shop_item(self, item_id: str) -> None:
safe_item_id = str(item_id or "").strip().lower()
if not safe_item_id:
raise ValueError("shop_item_not_found")
with self._lock, self._connect() as conn:
row = conn.execute("SELECT item_id FROM shop_items WHERE item_id=?", (safe_item_id,)).fetchone()
if not row:
raise ValueError("shop_item_not_found")
conn.execute("DELETE FROM shop_items WHERE item_id=?", (safe_item_id,))
conn.commit()
def list_user_entitlements(
self,
user_id: str,
*,
entitlement_type: str = "",
status: str = "active",
limit: int = 300,
) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return []
safe_limit = max(1, min(2000, int(limit)))
safe_type = str(entitlement_type or "").strip()
safe_status = str(status or "").strip()
with self._lock, self._connect() as conn:
where = ["user_id=?"]
params: list[Any] = [safe_user_id]
if safe_type:
where.append("entitlement_type=?")
params.append(safe_type)
if safe_status:
where.append("status=?")
params.append(safe_status)
rows = conn.execute(
f"""
SELECT *
FROM user_entitlements
WHERE {' AND '.join(where)}
ORDER BY created_at DESC, id DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
out = []
for row in rows:
item = dict(row)
payload = self._safe_json_loads(item.get("payload_json", "{}"), {})
item["payload"] = payload if isinstance(payload, dict) else {}
out.append(item)
return out
def has_game_card_access(self, user_id: str, card_id: str) -> bool:
safe_user_id = str(user_id or "").strip()
safe_card_id = str(card_id or "").strip()
if not safe_user_id or not safe_card_id:
return False
card = self.get_game_card(safe_card_id)
if card and bool(card.get("is_official", False)):
return True
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT id
FROM user_entitlements
WHERE user_id=? AND entitlement_type='game_card' AND ref_id=? AND status='active'
LIMIT 1
""",
(safe_user_id, safe_card_id),
).fetchone()
return bool(row)
def get_game_card_price_map_for_user(self, user_id: str | None) -> dict[str, dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT item_id, name, price, metadata_json
FROM shop_items
WHERE status='active' AND item_type IN ('game_card', 'card') AND price>0
ORDER BY is_recommended DESC, sort_order DESC, created_at DESC
"""
).fetchall()
owned_card_ids: set[str] = set()
if safe_user_id:
own_rows = conn.execute(
"""
SELECT ref_id
FROM user_entitlements
WHERE user_id=? AND entitlement_type='game_card' AND status='active'
""",
(safe_user_id,),
).fetchall()
owned_card_ids = {str(row["ref_id"]).strip() for row in own_rows}
out: dict[str, dict[str, Any]] = {}
for row in rows:
item = dict(row)
metadata = self._safe_json_loads(item.get("metadata_json", "{}"), {})
if not isinstance(metadata, dict):
metadata = {}
card_id = str(metadata.get("card_id") or metadata.get("ref_id") or "").strip()
if not card_id:
continue
price = self._to_int(item.get("price", 0), 0, 0)
if card_id not in out or price < self._to_int(out[card_id].get("price", 0), 0, 0):
out[card_id] = {
"item_id": str(item.get("item_id", "")).strip(),
"item_name": str(item.get("name", "")).strip(),
"price": price,
"owned": bool(card_id in owned_card_ids),
}
return out
def _grant_entitlement_tx(
self,
conn: sqlite3.Connection,
*,
user_id: str,
entitlement_type: str,
ref_id: str,
source_order_id: str,
payload: dict[str, Any] | None = None,
expires_at: str = "",
) -> None:
now = utc_now_iso()
conn.execute(
"""
INSERT INTO user_entitlements(
user_id, entitlement_type, ref_id, source_order_id, payload_json, status, expires_at, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, 'active', ?, ?, ?)
""",
(
str(user_id or "").strip(),
str(entitlement_type or "").strip()[:36] or "item",
str(ref_id or "").strip()[:120],
str(source_order_id or "").strip()[:64],
json.dumps(payload or {}, ensure_ascii=False),
str(expires_at or "").strip()[:40],
now,
now,
),
)
def purchase_shop_item(
self,
user_id: str,
item_id: str,
*,
quantity: int = 1,
note: str = "",
) -> dict[str, Any]:
safe_user_id = str(user_id or "").strip()
safe_item_id = str(item_id or "").strip().lower()
safe_qty = max(1, min(99, int(quantity)))
if not safe_user_id:
raise ValueError("auth_required")
if not safe_item_id:
raise ValueError("shop_item_not_found")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
self._expire_point_lots_tx(conn, user_id=safe_user_id)
row = conn.execute("SELECT * FROM shop_items WHERE item_id=?", (safe_item_id,)).fetchone()
if not row:
raise ValueError("shop_item_not_found")
item = self._parse_shop_item(row)
if str(item.get("status", "")).strip().lower() != "active":
raise ValueError("shop_item_unavailable")
price = self._to_int(item.get("price", 0), 0, 0)
if price <= 0:
raise ValueError("shop_item_unavailable")
stock = int(item.get("stock", -1))
if stock >= 0 and stock < safe_qty:
raise ValueError("shop_item_out_of_stock")
purchase_limit = self._to_int(item.get("purchase_limit", 0), 0, 0)
if purchase_limit > 0:
bought_row = conn.execute(
"""
SELECT COALESCE(SUM(quantity), 0) AS qty
FROM shop_orders
WHERE user_id=? AND item_id=? AND status='paid'
""",
(safe_user_id, safe_item_id),
).fetchone()
bought = self._to_int((dict(bought_row) if bought_row else {}).get("qty", 0), 0, 0)
if bought + safe_qty > purchase_limit:
raise ValueError("shop_item_limit_reached")
total_cost = price * safe_qty
order_id = uuid.uuid4().hex
debit_payload = self._debit_points_tx(
conn,
safe_user_id,
total_cost,
tx_type="shop_purchase",
source="shop_order",
related_id=order_id,
note=str(note or "").strip()[:180] or f"购买商品 {item.get('name', safe_item_id)}",
count_spent=True,
)
now = utc_now_iso()
item_snapshot = {
"item_id": safe_item_id,
"name": str(item.get("name", "")).strip(),
"item_type": str(item.get("item_type", "")).strip(),
"subtype": str(item.get("subtype", "")).strip(),
"price": price,
"metadata": item.get("metadata", {}) if isinstance(item.get("metadata"), dict) else {},
}
conn.execute(
"""
INSERT INTO shop_orders(
order_id, user_id, item_id, item_snapshot, price, quantity, total_cost, status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'paid', ?, ?)
""",
(
order_id,
safe_user_id,
safe_item_id,
json.dumps(item_snapshot, ensure_ascii=False),
price,
safe_qty,
total_cost,
now,
now,
),
)
if stock >= 0:
conn.execute(
"UPDATE shop_items SET stock=?, updated_at=? WHERE item_id=?",
(max(0, stock - safe_qty), now, safe_item_id),
)
meta = item.get("metadata", {}) if isinstance(item.get("metadata"), dict) else {}
item_type = str(item.get("item_type", "item")).strip().lower()
entitlement_type = {
"game_card": "game_card",
"card": "game_card",
"module": "module",
"decoration": "decoration",
"privilege": "privilege",
"bundle": "bundle",
}.get(item_type, item_type or "item")
ref_id = str(
meta.get("ref_id")
or meta.get("card_id")
or meta.get("module_id")
or safe_item_id
).strip()[:120]
expires_at = str(meta.get("expires_at", "")).strip()[:40]
entitlement_payload = {"quantity": safe_qty, "item": item_snapshot}
self._grant_entitlement_tx(
conn,
user_id=safe_user_id,
entitlement_type=entitlement_type,
ref_id=ref_id,
source_order_id=order_id,
payload=entitlement_payload,
expires_at=expires_at,
)
conn.commit()
account = self.get_points_account(safe_user_id)
return {
"order_id": order_id,
"item_id": safe_item_id,
"quantity": safe_qty,
"total_cost": total_cost,
"account": account,
"transaction": debit_payload["tx"],
}
def get_shop_order(self, order_id: str) -> dict[str, Any] | None:
safe_order_id = str(order_id or "").strip()
if not safe_order_id:
return None
with self._lock, self._connect() as conn:
row = conn.execute(
"""
SELECT o.*, u.username, i.name AS item_name
FROM shop_orders o
LEFT JOIN users u ON u.user_id = o.user_id
LEFT JOIN shop_items i ON i.item_id = o.item_id
WHERE o.order_id=?
""",
(safe_order_id,),
).fetchone()
if not row:
return None
item = dict(row)
snapshot = self._safe_json_loads(item.get("item_snapshot", "{}"), {})
item["item_snapshot_payload"] = snapshot if isinstance(snapshot, dict) else {}
return item
def list_user_shop_orders(self, user_id: str, *, limit: int = 120) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return []
safe_limit = max(1, min(1200, int(limit)))
with self._lock, self._connect() as conn:
rows = conn.execute(
"""
SELECT o.*, i.name AS item_name
FROM shop_orders o
LEFT JOIN shop_items i ON i.item_id = o.item_id
WHERE o.user_id=?
ORDER BY o.created_at DESC
LIMIT ?
""",
(safe_user_id, safe_limit),
).fetchall()
out = []
for row in rows:
item = dict(row)
snapshot = self._safe_json_loads(item.get("item_snapshot", "{}"), {})
item["item_snapshot_payload"] = snapshot if isinstance(snapshot, dict) else {}
out.append(item)
return out
def list_shop_orders_admin(
self,
*,
status: str = "",
user_id: str = "",
limit: int = 400,
) -> list[dict[str, Any]]:
safe_limit = max(1, min(5000, int(limit)))
safe_status = str(status or "").strip().lower()
safe_user_id = str(user_id or "").strip()
where: list[str] = []
params: list[Any] = []
if safe_status:
where.append("o.status=?")
params.append(safe_status)
if safe_user_id:
where.append("o.user_id=?")
params.append(safe_user_id)
where_sql = f"WHERE {' AND '.join(where)}" if where else ""
with self._lock, self._connect() as conn:
rows = conn.execute(
f"""
SELECT o.*, u.username, i.name AS item_name
FROM shop_orders o
LEFT JOIN users u ON u.user_id = o.user_id
LEFT JOIN shop_items i ON i.item_id = o.item_id
{where_sql}
ORDER BY o.created_at DESC
LIMIT ?
""",
tuple(params + [safe_limit]),
).fetchall()
out = []
for row in rows:
item = dict(row)
snapshot = self._safe_json_loads(item.get("item_snapshot", "{}"), {})
item["item_snapshot_payload"] = snapshot if isinstance(snapshot, dict) else {}
out.append(item)
return out
def refund_shop_order(
self,
order_id: str,
*,
operator_user_id: str = "",
note: str = "",
) -> dict[str, Any]:
safe_order_id = str(order_id or "").strip()
if not safe_order_id:
raise ValueError("shop_order_not_found")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
row = conn.execute("SELECT * FROM shop_orders WHERE order_id=?", (safe_order_id,)).fetchone()
if not row:
raise ValueError("shop_order_not_found")
order = dict(row)
if str(order.get("status", "")).strip().lower() != "paid":
raise ValueError("shop_order_not_refundable")
user_id = str(order.get("user_id", "")).strip()
total_cost = self._to_int(order.get("total_cost", 0), 0, 0)
quantity = self._to_int(order.get("quantity", 1), 1, 1)
item_id = str(order.get("item_id", "")).strip().lower()
self._credit_points_tx(
conn,
user_id,
total_cost,
tx_type="shop_refund",
source="shop_order",
related_id=safe_order_id,
note=str(note or "").strip()[:180] or "订单退款返还积分",
operator_user_id=operator_user_id,
count_recharge=False,
create_lot=True,
allow_locked=True,
)
now = utc_now_iso()
conn.execute(
"""
UPDATE shop_orders
SET status='refunded', refund_note=?, operator_user_id=?, refunded_at=?, updated_at=?
WHERE order_id=?
""",
(
str(note or "").strip()[:180],
str(operator_user_id or "").strip()[:64],
now,
now,
safe_order_id,
),
)
conn.execute(
"""
UPDATE user_entitlements
SET status='revoked', updated_at=?
WHERE source_order_id=? AND status='active'
""",
(now, safe_order_id),
)
item_row = conn.execute("SELECT stock FROM shop_items WHERE item_id=?", (item_id,)).fetchone()
if item_row:
stock = int(dict(item_row).get("stock", -1))
if stock >= 0:
conn.execute(
"UPDATE shop_items SET stock=?, updated_at=? WHERE item_id=?",
(stock + quantity, now, item_id),
)
conn.commit()
return self.get_shop_order(safe_order_id) or {}
def tip_user_points(
self,
from_user_id: str,
to_user_id: str,
amount: int,
*,
room_id: str = "",
message: str = "",
) -> dict[str, Any]:
safe_from = str(from_user_id or "").strip()
safe_to = str(to_user_id or "").strip()
safe_amount = self._to_int(amount, 0, 1)
if not safe_from or not safe_to:
raise ValueError("user_not_found")
if safe_from == safe_to:
raise ValueError("tip_self_forbidden")
if self.get_user_by_id(safe_to) is None:
raise ValueError("user_not_found")
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
self._expire_point_lots_tx(conn, user_id=safe_from)
debit_payload = self._debit_points_tx(
conn,
safe_from,
safe_amount,
tx_type="tip_send",
source="tip",
related_id=safe_to,
note=f"打赏给用户 {safe_to}",
count_spent=True,
)
self._credit_points_tx(
conn,
safe_to,
safe_amount,
tx_type="tip_receive",
source="tip",
related_id=safe_from,
note=f"收到来自 {safe_from} 的打赏",
count_recharge=False,
create_lot=True,
)
conn.execute(
"""
INSERT INTO user_tips(from_user_id, to_user_id, amount, room_id, message, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
safe_from,
safe_to,
safe_amount,
str(room_id or "").strip()[:24],
str(message or "").strip()[:180],
utc_now_iso(),
),
)
conn.commit()
return {
"from_user_id": safe_from,
"to_user_id": safe_to,
"amount": safe_amount,
"sender_account": self.get_points_account(safe_from),
"sender_tx": debit_payload["tx"],
}
def evaluate_achievement_point_rewards(self, user_id: str) -> list[dict[str, Any]]:
safe_user_id = str(user_id or "").strip()
if not safe_user_id:
return []
account = self.get_points_account(safe_user_id)
metrics = {
"recharged": self._to_int(account.get("total_recharged", 0), 0, 0),
"spent": self._to_int(account.get("total_spent", 0), 0, 0),
}
granted: list[dict[str, Any]] = []
with self._lock, self._connect() as conn:
conn.execute("BEGIN IMMEDIATE")
for track in ACHIEVEMENT_POINT_REWARD_TRACKS:
metric_key = str(track.get("metric", "")).strip()
metric_value = self._to_int(metrics.get(metric_key, 0), 0, 0)
for tier in track.get("tiers", []):
threshold = self._to_int(tier.get("threshold", 0), 0, 1)
reward_points = self._to_int(tier.get("points", 0), 0, 1)
if metric_value < threshold or reward_points <= 0:
continue
reward_key = f"{track.get('id', 'track')}:{metric_key}:{threshold}"
exists = conn.execute(
"""
SELECT id
FROM achievement_point_rewards
WHERE user_id=? AND reward_key=?
""",
(safe_user_id, reward_key),
).fetchone()
if exists:
continue
conn.execute(
"""
INSERT INTO achievement_point_rewards(user_id, reward_key, points, created_at)
VALUES (?, ?, ?, ?)
""",
(safe_user_id, reward_key, reward_points, utc_now_iso()),
)
payload = self._credit_points_tx(
conn,
safe_user_id,
reward_points,
tx_type="achievement_reward",
source="achievement",
related_id=reward_key,
note=f"成就里程碑奖励({metric_key} >= {threshold})",
count_recharge=False,
create_lot=True,
allow_locked=True,
)
granted.append(payload["tx"])
conn.commit()
return granted
def get_finance_summary(self, *, days: int = 30) -> dict[str, Any]:
safe_days = max(1, min(365, int(days)))
threshold = (datetime.now(timezone.utc) - timedelta(days=safe_days)).isoformat()
with self._lock, self._connect() as conn:
code_total_row = conn.execute("SELECT COUNT(*) AS c FROM activation_codes").fetchone()
code_used_row = conn.execute("SELECT COUNT(*) AS c FROM activation_codes WHERE status='used'").fetchone()
recharge_row = conn.execute(
"""
SELECT COALESCE(SUM(change_amount), 0) AS total
FROM user_point_transactions
WHERE source='activation_code' AND change_amount>0 AND created_at>=?
""",
(threshold,),
).fetchone()
spend_row = conn.execute(
"""
SELECT COALESCE(SUM(-change_amount), 0) AS total
FROM user_point_transactions
WHERE change_amount<0 AND tx_type IN ('shop_purchase', 'tip_send', 'room_create_fee') AND created_at>=?
""",
(threshold,),
).fetchone()
top_items = conn.execute(
"""
SELECT item_id, SUM(quantity) AS qty, SUM(total_cost) AS amount
FROM shop_orders
WHERE status='paid' AND created_at>=?
GROUP BY item_id
ORDER BY qty DESC, amount DESC
LIMIT 10
""",
(threshold,),
).fetchall()
trends = conn.execute(
"""
SELECT
substr(created_at, 1, 10) AS day,
SUM(CASE WHEN source='activation_code' AND change_amount>0 THEN change_amount ELSE 0 END) AS recharge,
SUM(CASE WHEN change_amount<0 AND tx_type IN ('shop_purchase', 'tip_send', 'room_create_fee')
THEN -change_amount ELSE 0 END) AS spend
FROM user_point_transactions
WHERE created_at>=?
GROUP BY substr(created_at, 1, 10)
ORDER BY day ASC
""",
(threshold,),
).fetchall()
return {
"code_total": self._to_int((dict(code_total_row) if code_total_row else {}).get("c", 0), 0, 0),
"code_used": self._to_int((dict(code_used_row) if code_used_row else {}).get("c", 0), 0, 0),
"recent_recharge_total": self._to_int((dict(recharge_row) if recharge_row else {}).get("total", 0), 0, 0),
"recent_spend_total": self._to_int((dict(spend_row) if spend_row else {}).get("total", 0), 0, 0),
"top_items": [dict(row) for row in top_items],
"trends": [dict(row) for row in trends],
}