308 lines
11 KiB
Python
308 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
管理后台 API（/api/admin/*）
|
||
============================
|
||
功能：
|
||
1. 登录认证（Bearer Token，密码存 site_config，默认 admin123）
|
||
2. 通用数据 CRUD：teams / players / games / game_player_stats / standings / news / persons / sports / leagues
|
||
3. 站点配置管理：开场白、快捷问题、站点名称、页脚等（site_config 表）
|
||
安全：
|
||
- 表名白名单 + 字段名白名单（PRAGMA 校验），防 SQL 注入
|
||
- 登录失败限速（10 分钟 5 次锁定）
|
||
- Token 有效期 12 小时
|
||
"""
|
||
import json
|
||
import logging
|
||
import secrets
|
||
import time
|
||
from functools import wraps
|
||
|
||
from flask import jsonify, request
|
||
|
||
from db import query, query_one, execute, executemany
|
||
|
||
log = logging.getLogger("admin")
|
||
|
||
# ------------------------------------------------------------------ 常量
|
||
TABLES = {
|
||
"teams": "球队", "players": "球员", "games": "比赛",
|
||
"game_player_stats": "比赛技术统计", "standings": "排名",
|
||
"news": "新闻/百科", "persons": "人物",
|
||
"sports": "运动项目", "leagues": "联赛",
|
||
}
|
||
# 不允许前端直接改的字段（系统维护）
|
||
RO_FIELDS = {"id", "created_at"}
|
||
|
||
_tokens = {} # token -> expire_ts
|
||
TOKEN_TTL = 12 * 3600
|
||
MAX_FAIL = 5
|
||
FAIL_WINDOW = 600 # 10 分钟
|
||
_fails = [] # [(ip, ts), ...]
|
||
|
||
DEFAULT_CONFIG = {
|
||
"site_name": "NBA球迷大全",
|
||
"site_subtitle": "比赛 · 球员 · 球队 · 资讯 · 人物 · 百科",
|
||
"welcome_text": "你好，我是**NBA球迷大全**助手！可以问我任何关于比赛、球员、球队、新闻、人物的问题，我会基于数据库给你准确答案～",
|
||
"welcome_hint": "试试：",
|
||
"suggestions": json.dumps([
|
||
"最近一场比赛结果", "湖人本赛季战绩怎么样", "库里本赛季场均数据",
|
||
"2026年总决赛谁赢了", "SGA拿了什么荣誉", "NBA工资帽是什么",
|
||
"介绍一下波波维奇", "今天有什么新闻", "西部排名",
|
||
"雷霆和凯尔特人总决赛G6数据",
|
||
], ensure_ascii=False),
|
||
"footer_text": "NBA球迷大全 · 数据为模拟演示数据（2025-26 赛季） · LLM: DeepSeek · 向量: Chroma + bge-large-zh",
|
||
"admin_password": "admin123",
|
||
"entity_mark_mode": "first", # 实体标记：first=只标记首次出现 / all=全部标记
|
||
"suggestion_count": "3", # 对话中底部快捷问题预测个数（默认3）
|
||
}
|
||
|
||
SEARCHABLE = { # 每个表可搜索的 TEXT 字段
|
||
"teams": ("name", "name_en", "code", "city"),
|
||
"players": ("name", "name_en", "country", "bio"),
|
||
"games": ("round_name", "game_time", "venue"),
|
||
"game_player_stats": (),
|
||
"standings": ("conference",),
|
||
"news": ("title", "author", "source", "tags", "content"),
|
||
"persons": ("name", "name_en", "role", "role_cn", "title"),
|
||
"sports": ("code", "name"),
|
||
"leagues": ("code", "name", "country"),
|
||
}
|
||
|
||
|
||
# ------------------------------------------------------------------ 初始化
|
||
def init_defaults():
|
||
"""写入默认配置（已存在的跳过）"""
|
||
for k, v in DEFAULT_CONFIG.items():
|
||
if not query_one("SELECT 1 FROM site_config WHERE key=?", (k,)):
|
||
execute("INSERT INTO site_config(key, value) VALUES(?,?)", (k, v))
|
||
|
||
|
||
def get_config():
|
||
rows = query("SELECT key, value FROM site_config")
|
||
return {r["key"]: r["value"] for r in rows}
|
||
|
||
|
||
def set_config(key, value):
|
||
execute("""INSERT INTO site_config(key, value, updated_at) VALUES(?,?,datetime('now','localtime'))
|
||
ON CONFLICT(key) DO UPDATE SET value=excluded.value,
|
||
updated_at=datetime('now','localtime')""", (key, value))
|
||
|
||
|
||
# ------------------------------------------------------------------ 认证
|
||
def _check_pw(pw):
|
||
cfg = get_config()
|
||
return bool(pw) and pw == cfg.get("admin_password", "")
|
||
|
||
|
||
def _too_many_fails(ip):
|
||
now = time.time()
|
||
_fails[:] = [(i, t) for i, t in _fails if now - t < FAIL_WINDOW]
|
||
return sum(1 for i, _ in _fails if i == ip) >= MAX_FAIL
|
||
|
||
|
||
def login():
|
||
body = request.get_json(force=True, silent=True) or {}
|
||
pw = body.get("password") or ""
|
||
ip = request.remote_addr or "?"
|
||
if _too_many_fails(ip):
|
||
return jsonify({"error": "失败次数过多，请 10 分钟后再试"}), 429
|
||
if not _check_pw(pw):
|
||
_fails.append((ip, time.time()))
|
||
return jsonify({"error": "密码错误"}), 401
|
||
token = secrets.token_hex(24)
|
||
_tokens[token] = time.time() + TOKEN_TTL
|
||
return jsonify({"token": token, "expires_in": TOKEN_TTL,
|
||
"site_name": get_config().get("site_name", "NBA球迷大全")})
|
||
|
||
|
||
def logout():
|
||
auth = request.headers.get("Authorization", "")
|
||
token = auth.replace("Bearer ", "").strip()
|
||
_tokens.pop(token, None)
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
def require_admin(fn):
|
||
@wraps(fn)
|
||
def wrapper(*a, **kw):
|
||
auth = request.headers.get("Authorization", "")
|
||
token = auth.replace("Bearer ", "").strip()
|
||
if token not in _tokens or _tokens[token] < time.time():
|
||
return jsonify({"error": "未登录或登录已过期"}), 401
|
||
return fn(*a, **kw)
|
||
return wrapper
|
||
|
||
|
||
# ------------------------------------------------------------------ 工具
|
||
def _check_table(table):
|
||
if table not in TABLES:
|
||
raise ValueError(f"未知数据表: {table}")
|
||
|
||
|
||
def _columns(table):
|
||
"""返回表字段 [{name, type, pk}]（白名单）"""
|
||
_check_table(table)
|
||
rows = query(f"PRAGMA table_info({table})")
|
||
return [{"name": r["name"], "type": r["type"], "pk": bool(r["pk"])} for r in rows]
|
||
|
||
|
||
def _valid_field(table, field):
|
||
return field in {c["name"] for c in _columns(table)}
|
||
|
||
|
||
def _clean_payload(table, data):
|
||
"""过滤非法字段、剔除系统字段；TEXT 空字符串 → None 可留空"""
|
||
cols = {c["name"]: c["type"] for c in _columns(table)}
|
||
out = {}
|
||
for k, v in (data or {}).items():
|
||
if k in RO_FIELDS or k not in cols:
|
||
continue
|
||
if v == "":
|
||
v = None
|
||
if cols[k] in ("INTEGER", "REAL") and v is not None:
|
||
try:
|
||
v = int(v) if cols[k] == "INTEGER" else float(v)
|
||
except (TypeError, ValueError):
|
||
v = None
|
||
out[k] = v
|
||
return out
|
||
|
||
|
||
def _fmt_row(table, row):
|
||
"""JSON 友好化：TEXT 字段里的 JSON 字符串（如 news.tags）原样返回，前端解析"""
|
||
return row
|
||
|
||
|
||
# ------------------------------------------------------------------ 统计
|
||
def stats():
|
||
return jsonify({
|
||
"tables": {t: {"cn": cn, "count": query_one(f"SELECT COUNT(*) AS c FROM {t}")["c"]}
|
||
for t, cn in TABLES.items()},
|
||
"config": len(query("SELECT 1 FROM site_config")),
|
||
})
|
||
|
||
|
||
# ------------------------------------------------------------------ 通用 CRUD
|
||
def list_rows(table):
|
||
try:
|
||
_check_table(table)
|
||
except ValueError as e:
|
||
return jsonify({"error": str(e)}), 400
|
||
page = max(1, int(request.args.get("page", 1)))
|
||
size = min(100, max(1, int(request.args.get("size", 20))))
|
||
q = (request.args.get("q") or "").strip()
|
||
# 排序：字段白名单校验（防注入），默认 id 降序
|
||
cols = _columns(table)
|
||
valid_fields = {c["name"] for c in cols}
|
||
sort = request.args.get("sort", "") or "id"
|
||
if sort not in valid_fields:
|
||
sort = "id"
|
||
order = (request.args.get("order", "") or "desc").lower()
|
||
if order not in ("asc", "desc"):
|
||
order = "desc"
|
||
where, args = "", []
|
||
if q:
|
||
fields = SEARCHABLE.get(table) or ()
|
||
if fields:
|
||
like = f"%{q}%"
|
||
where = "WHERE " + " OR ".join(f"{f} LIKE ? ESCAPE '\\'" for f in fields)
|
||
args = [like] * len(fields)
|
||
total = query_one(f"SELECT COUNT(*) AS c FROM {table} {where}", args)["c"]
|
||
rows = query(f"SELECT * FROM {table} {where} ORDER BY {sort} {order.upper()}, id {order.upper()} LIMIT ? OFFSET ?",
|
||
args + [size, (page - 1) * size])
|
||
return jsonify({"table": table, "cn": TABLES[table], "columns": cols,
|
||
"total": total, "page": page, "size": size, "rows": rows,
|
||
"sort": sort, "order": order})
|
||
|
||
|
||
def get_row(table, rid):
|
||
try:
|
||
_check_table(table)
|
||
except ValueError as e:
|
||
return jsonify({"error": str(e)}), 400
|
||
row = query_one(f"SELECT * FROM {table} WHERE id=?", (rid,))
|
||
if not row:
|
||
return jsonify({"error": "记录不存在"}), 404
|
||
return jsonify(row)
|
||
|
||
|
||
def create_row(table):
|
||
try:
|
||
_check_table(table)
|
||
except ValueError as e:
|
||
return jsonify({"error": str(e)}), 400
|
||
data = _clean_payload(table, request.get_json(force=True, silent=True) or {})
|
||
if not data:
|
||
return jsonify({"error": "没有可写入的字段"}), 400
|
||
cols = list(data.keys())
|
||
marks = ",".join("?" for _ in cols)
|
||
nid = execute(f"INSERT INTO {table}({','.join(cols)}) VALUES({marks})",
|
||
[data[c] for c in cols])
|
||
_after_write(table, nid, data)
|
||
return jsonify({"ok": True, "id": nid})
|
||
|
||
|
||
def update_row(table, rid):
|
||
try:
|
||
_check_table(table)
|
||
except ValueError as e:
|
||
return jsonify({"error": str(e)}), 400
|
||
if not query_one(f"SELECT 1 FROM {table} WHERE id=?", (rid,)):
|
||
return jsonify({"error": "记录不存在"}), 404
|
||
data = _clean_payload(table, request.get_json(force=True, silent=True) or {})
|
||
if not data:
|
||
return jsonify({"error": "没有可更新的字段"}), 400
|
||
sets = ",".join(f"{c}=?" for c in data)
|
||
execute(f"UPDATE {table} SET {sets} WHERE id=?", [data[c] for c in data] + [rid])
|
||
_after_write(table, rid, data)
|
||
return jsonify({"ok": True, "id": rid})
|
||
|
||
|
||
def delete_row(table, rid):
|
||
try:
|
||
_check_table(table)
|
||
except ValueError as e:
|
||
return jsonify({"error": str(e)}), 400
|
||
if not query_one(f"SELECT 1 FROM {table} WHERE id=?", (rid,)):
|
||
return jsonify({"error": "记录不存在"}), 404
|
||
# 级联清理：删除比赛时清理技术统计；删除球队时清理其球员/新闻归属
|
||
if table == "games":
|
||
execute("DELETE FROM game_player_stats WHERE game_id=?", (rid,))
|
||
if table == "teams":
|
||
execute("UPDATE players SET team_id=NULL WHERE team_id=?", (rid,))
|
||
execute("UPDATE news SET team_id=NULL WHERE team_id=?", (rid,))
|
||
execute(f"DELETE FROM {table} WHERE id=?", (rid,))
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
def _after_write(table, rid, data):
|
||
"""写入后的联动维护：比赛改比分/球队时同步球队得分（可选扩展）"""
|
||
pass
|
||
|
||
|
||
# ------------------------------------------------------------------ 配置
|
||
def get_config_api():
|
||
cfg = get_config()
|
||
out = dict(cfg)
|
||
out["_protected"] = ["admin_password"]
|
||
return jsonify(out)
|
||
|
||
|
||
def update_config():
|
||
body = request.get_json(force=True, silent=True) or {}
|
||
if not isinstance(body, dict):
|
||
return jsonify({"error": "参数格式错误"}), 400
|
||
for k, v in body.items():
|
||
if k.startswith("_") or not isinstance(v, str):
|
||
continue
|
||
if k == "admin_password" and not v.strip():
|
||
continue # 空密码 = 不修改
|
||
set_config(k, v)
|
||
return jsonify({"ok": True, "config": get_config()})
|
||
|
||
|
||
# ------------------------------------------------------------------ 恢复默认快捷问题
|
||
def reset_suggestions():
|
||
set_config("suggestions", DEFAULT_CONFIG["suggestions"])
|
||
return jsonify({"ok": True, "suggestions": json.loads(DEFAULT_CONFIG["suggestions"])})
|