Files
Echo-Tavern/gm_platform.py
T

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