Archived
836 lines
34 KiB
Python
836 lines
34 KiB
Python
"""
|
||
app.py — Flask 后端主文件
|
||
=========================
|
||
作用：整个 MVP 的"大脑"，负责三件事：
|
||
1. 提供 API 接口（视频列表、AI 对话、行为记录）
|
||
2. 管理 SQLite 数据库（用户、视频、行为日志）
|
||
3. 对接 DeepSeek 大模型 API
|
||
|
||
产品概念说明：
|
||
- 这里用的是 "单体应用" 架构：一个 app.py 把路由+数据库+API 全包了。
|
||
适合 MVP 阶段快速验证，后续用户量起来再拆分为微服务。
|
||
- 路由 (Route)：浏览器访问不同网址时，Flask 把请求分发给对应的函数处理。
|
||
类比：路由 = 你走进商场，问服务台"女装在三楼"，服务台就是路由层的功能。
|
||
"""
|
||
|
||
import gzip
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
from datetime import timedelta
|
||
|
||
import requests
|
||
from flask import Flask, render_template, request, jsonify, send_from_directory
|
||
|
||
# ---- 初始化 Flask 应用 ----
|
||
app = Flask(__name__)
|
||
# 静态资源缓存 7 天（模板引用带 ?v= 版本号，更新时递增即可刷新）
|
||
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = timedelta(days=7)
|
||
|
||
# 配置路径 (使用绝对路径以避免 sqlite 找不到文件)
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
DATA_DIR = os.path.join(BASE_DIR, "data").replace("\\", "/")
|
||
# 修改数据库文件名，绕过之前的旧文件锁定问题
|
||
DB_PATH = os.path.join(DATA_DIR, "miniapp_v3.db").replace("\\", "/")
|
||
VIDEOS_PATH = os.path.join(DATA_DIR, "seed_videos.json").replace("\\", "/")
|
||
STORY_PATH = os.path.join(DATA_DIR, "seed_story.json").replace("\\", "/")
|
||
MATH_TEEN_STORY_PATH = os.path.join(DATA_DIR, "数学少年线_story.json").replace("\\", "/")
|
||
SIMULATOR_PATH = os.path.join(DATA_DIR, "seed_simulator.json").replace("\\", "/")
|
||
MATHBTI_PATH = os.path.join(DATA_DIR, "seed_mathbti.json").replace("\\", "/")
|
||
ALUMNI_PATH = os.path.join(DATA_DIR, "seed_alumni.json").replace("\\", "/")
|
||
|
||
# 确保 data 目录存在
|
||
os.makedirs(DATA_DIR, exist_ok=True)
|
||
|
||
|
||
# 内容型 GET API 短缓存（种子数据极少变动）
|
||
_API_CACHE_SECONDS = {
|
||
'/api/videos': 300,
|
||
'/api/knowledge': 300,
|
||
'/api/story': 300,
|
||
'/api/simulator': 300,
|
||
'/api/alumni': 300,
|
||
'/api/math_personality': 300,
|
||
}
|
||
|
||
|
||
@app.after_request
|
||
def compress_response(response):
|
||
"""gzip 压缩文本响应（HTML/CSS/JS/JSON >500B）+ 内容型 API 短缓存"""
|
||
if request.method == 'GET' and request.path in _API_CACHE_SECONDS:
|
||
response.headers['Cache-Control'] = f"private, max-age={_API_CACHE_SECONDS[request.path]}"
|
||
if (response.status_code != 200
|
||
or 'Content-Encoding' in response.headers):
|
||
return response
|
||
if 'gzip' not in request.headers.get('Accept-Encoding', ''):
|
||
return response
|
||
mimetype = response.mimetype or ''
|
||
if not (mimetype.startswith('text/') or mimetype in ('application/json', 'application/javascript')):
|
||
return response
|
||
if response.direct_passthrough:
|
||
response.direct_passthrough = False # 静态文件需读入内存才能压缩
|
||
data = response.get_data()
|
||
if len(data) < 500:
|
||
return response
|
||
compressed = gzip.compress(data, compresslevel=6)
|
||
response.set_data(compressed)
|
||
response.headers['Content-Encoding'] = 'gzip'
|
||
response.headers['Content-Length'] = len(compressed)
|
||
response.headers['Vary'] = 'Accept-Encoding'
|
||
return response
|
||
|
||
# DeepSeek API 配置 — 需要用户填入自己的 API Key
|
||
DEEPSEEK_API_KEY = "sk-0cd8b068c4a3439c9fc6d67b08865c21"
|
||
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
|
||
|
||
|
||
# ============================================================
|
||
# 模块 1: 数据库初始化
|
||
# ============================================================
|
||
def init_db():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS behavior_log (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id TEXT,
|
||
action TEXT,
|
||
target TEXT,
|
||
detail TEXT,
|
||
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
# 旧库迁移：补充 detail 列
|
||
try:
|
||
c.execute("ALTER TABLE behavior_log ADD COLUMN detail TEXT")
|
||
except Exception:
|
||
pass
|
||
# 知识卡片表
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS knowledge_cards (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
title TEXT NOT NULL,
|
||
question TEXT,
|
||
description TEXT,
|
||
detail TEXT,
|
||
color TEXT DEFAULT '#3B82F6',
|
||
icon TEXT DEFAULT '📖',
|
||
topic TEXT,
|
||
read_time TEXT DEFAULT '5分钟阅读',
|
||
sort_order INTEGER DEFAULT 0
|
||
)
|
||
""")
|
||
# 收藏表
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS favorites (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id TEXT NOT NULL,
|
||
target_type TEXT NOT NULL,
|
||
target_id INTEGER NOT NULL,
|
||
title TEXT,
|
||
cover TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE(user_id, target_type, target_id)
|
||
)
|
||
""")
|
||
# 观看历史表
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS watch_history (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id TEXT NOT NULL,
|
||
target_type TEXT NOT NULL,
|
||
target_id INTEGER NOT NULL,
|
||
title TEXT,
|
||
cover TEXT,
|
||
watched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
# 用户表
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
user_id TEXT UNIQUE NOT NULL,
|
||
nickname TEXT DEFAULT '游客用户',
|
||
avatar_emoji TEXT DEFAULT 'π',
|
||
avatar_color TEXT DEFAULT '#5B6AF0',
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
# 学生表（B端看板）
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS students (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
student_id TEXT UNIQUE,
|
||
name TEXT,
|
||
class_name TEXT,
|
||
grade TEXT
|
||
)
|
||
""")
|
||
# 常用查询索引
|
||
c.execute("CREATE INDEX IF NOT EXISTS idx_favorites_user ON favorites(user_id)")
|
||
c.execute("CREATE INDEX IF NOT EXISTS idx_history_user ON watch_history(user_id)")
|
||
c.execute("CREATE INDEX IF NOT EXISTS idx_behavior_user ON behavior_log(user_id)")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def load_seed_knowledge():
|
||
"""加载知识卡片种子数据"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("SELECT COUNT(*) FROM knowledge_cards")
|
||
if c.fetchone()[0] > 0:
|
||
conn.close()
|
||
return
|
||
|
||
cards = [
|
||
{"title": "被平均数骗过的所有人", "question": "你和马云的平均财富是多少？", "desc": "平均数、中位数、众数——这三兄弟让你看到的「真相」可能截然相反。", "detail": "平均数是最常用的统计量，但它也是最容易被误解的。当数据中有极端值时，平均数会被拉偏。比如你和马云的平均财富可能是千亿级别，但中位数可能只有几万。了解平均数、中位数、众数的区别，是统计学入门的第一课。", "color": "#EF4444", "icon": "📊", "topic": "统计陷阱", "read_time": "5分钟阅读", "sort": 1},
|
||
{"title": "彩票：一场精心设计的数学税", "question": "2元博500万，值不值？", "desc": "你中头奖的概率比你连续被雷劈三次还低。但为什么每天都有人买？", "detail": "双色球头奖概率约1/1772万。对比：被雷劈的概率约1/100万。彩票的期望值永远是负的——这是彩票公司盈利的数学基础。人们购买彩票不是因为他们不懂数学，而是因为损失厌恶和过度乐观的认知偏差。", "color": "#F59E0B", "icon": "🎲", "topic": "概率思维", "read_time": "5分钟阅读", "sort": 2},
|
||
{"title": "为什么所有地图都在骗你", "question": "格陵兰岛有非洲大吗？", "desc": "把球面摊平一定会变形——高斯发现的绝妙定理决定了你看到的世界。", "detail": "墨卡托投影让格陵兰岛看起来和非洲差不多大，实际上非洲面积是格陵兰的14倍。球面无法完美展开成平面，这是高斯的'绝妙定理'（Theorema Egregium）决定的。每种地图投影都有取舍：要么保面积，要么保角度，要么保方向——你永远无法三者兼得。", "color": "#3B82F6", "icon": "🗺️", "topic": "几何直觉", "read_time": "5分钟阅读", "sort": 3},
|
||
{"title": "大自然最爱的数列", "question": "兔子繁殖与斐波那契", "desc": "从向日葵的花瓣到鹦鹉螺的壳，斐波那契数列无处不在。这是巧合吗？", "detail": "斐波那契数列：1, 1, 2, 3, 5, 8, 13, 21... 后一项等于前两项之和。这个简单的规则在大自然中反复出现：松果的螺旋数、花瓣的排列、菠萝的鳞片——都是斐波那契数。这不是巧合，而是植物在生长过程中最优空间利用的结果。", "color": "#8B5CF6", "icon": "🔢", "topic": "数列之美", "read_time": "6分钟阅读", "sort": 4},
|
||
{"title": "博弈论：竞争的最优解", "question": "为什么奶茶店总是开在隔壁？", "desc": "纳什均衡解释了一个奇怪的现象：竞争对手为什么总是扎堆？", "detail": "两个冰淇淋摊在海滩上竞争，最后都会摆到海滩正中间——这就是霍特林模型。纳什均衡告诉我们：当每个人都做出对自己最优的选择时，结果可能对所有人都不好。这个原理适用于从商业选址到交通拥堵的各种场景。", "color": "#10B981", "icon": "⚖️", "topic": "博弈策略", "read_time": "5分钟阅读", "sort": 5},
|
||
{"title": "混沌理论：看不见的秩序", "question": "一只蝴蝶能引起台风吗？", "desc": "天气预报为什么不准？非线性动力学如何揭示混乱中的规律。", "detail": "洛伦兹在1961年发现：输入数据小数点后三位的微小差异，会导致完全不同的天气预报结果。这就是'蝴蝶效应'。混沌理论最迷人的地方在于：混沌不是随机——它是确定的系统产生不可预测的行为。这种'确定性的随机'挑战了我们对因果关系的直觉。", "color": "#EC4899", "icon": "🦋", "topic": "混沌理论", "read_time": "7分钟阅读", "sort": 6},
|
||
{"title": "蒙提霍尔：三门悖论", "question": "主持人打开一扇空门后，你该换吗？", "desc": "一个让数学家和大众都吵翻天的概率谜题。", "detail": "游戏节目有三扇门：一扇后面是汽车，两扇后面是山羊。你选择一扇后，主持人打开另一扇有山羊的门。现在你可以坚持原来的选择或换到剩下那扇——换门的中奖概率是2/3，不是1/2。这个结论让无数人（包括数学家）抓狂，但它是对的。关键：主持人的行为提供了信息。", "color": "#06B6D4", "icon": "🚪", "topic": "概率思维", "read_time": "4分钟阅读", "sort": 7},
|
||
{"title": "本福特定律：数字会说谎", "question": "财务报表里1开头的数字占比是多少？", "desc": "审计师用这条数学定律抓财务造假，准确率高得惊人。", "detail": "自然界里，首位数字为1的概率约为30%，不是直觉中的1/9。河流长度、人口数量、股票价格——都遵循本福特定律。聪明的会计师如果不知道这条定律，造出的假账一眼就会被看穿。美国国税局就用它来筛选审计目标。", "color": "#F97316", "icon": "🔍", "topic": "统计陷阱", "read_time": "5分钟阅读", "sort": 8},
|
||
]
|
||
for c_data in cards:
|
||
c.execute("""
|
||
INSERT INTO knowledge_cards (title, question, description, detail, color, icon, topic, read_time, sort_order)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (c_data["title"], c_data["question"], c_data["desc"], c_data["detail"],
|
||
c_data["color"], c_data["icon"], c_data["topic"], c_data["read_time"], c_data["sort"]))
|
||
conn.commit()
|
||
conn.close()
|
||
print(f"[完成] 已导入 {len(cards)} 条知识卡片")
|
||
|
||
|
||
def load_seed_videos():
|
||
"""
|
||
从 JSON 文件加载预设视频数据写入数据库。
|
||
"""
|
||
if not os.path.exists(VIDEOS_PATH):
|
||
print(f"[提示] 种子数据文件不存在: {VIDEOS_PATH}，跳过导入")
|
||
return
|
||
|
||
with open(VIDEOS_PATH, "r", encoding="utf-8") as f:
|
||
videos = json.load(f)
|
||
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
|
||
# 重建视频表（确保新增字段生效）
|
||
c.execute("DROP TABLE IF EXISTS videos")
|
||
c.execute("""
|
||
CREATE TABLE videos (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
title TEXT NOT NULL,
|
||
module TEXT NOT NULL,
|
||
sub_category TEXT,
|
||
discipline TEXT,
|
||
discipline_icon TEXT,
|
||
video_url TEXT,
|
||
cover_url TEXT,
|
||
description TEXT,
|
||
duration_min INTEGER DEFAULT 5,
|
||
author TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
|
||
for v in videos:
|
||
c.execute("""
|
||
INSERT INTO videos (title, module, sub_category, discipline, discipline_icon, video_url, cover_url, description, duration_min, author)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (v["title"], v["module"],
|
||
v.get("sub_category", ""), v.get("discipline", ""), v.get("discipline_icon", ""),
|
||
v.get("video_url", ""), v.get("cover_url", ""),
|
||
v.get("description", ""), v.get("duration_min", 5), v.get("author", "")))
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
print(f"[完成] 已导入 {len(videos)} 条视频数据")
|
||
|
||
|
||
# ============================================================
|
||
# 模块 2: 路由定义（API 接口）
|
||
# ============================================================
|
||
|
||
@app.route("/")
|
||
def index():
|
||
"""
|
||
主页路由 — 返回 SPA（单页应用）的 HTML。
|
||
前端所有页面切换都在这一个 HTML 里完成，不刷新浏览器。
|
||
"""
|
||
return render_template("index.html")
|
||
|
||
|
||
@app.route("/demo")
|
||
def demo():
|
||
"""返回独立静态 Demo 页面"""
|
||
return send_from_directory(BASE_DIR, "mathbrain-demo.html")
|
||
|
||
|
||
@app.route("/web")
|
||
def web_desktop():
|
||
"""桌面端网页版 — 左侧栏 + 横向布局"""
|
||
return render_template("desktop.html")
|
||
|
||
|
||
@app.route("/data/<path:filename>")
|
||
def data_files(filename):
|
||
"""本地开发时提供 data/ 下的公开 JSON 数据"""
|
||
if not filename.lower().endswith(".json"):
|
||
return jsonify({"code": -1, "msg": "not allowed"}), 404
|
||
return send_from_directory(DATA_DIR, filename)
|
||
|
||
|
||
@app.route("/math_teen")
|
||
def math_teen():
|
||
"""数学少年线互动剧本独立播放页"""
|
||
return render_template("math_teen.html")
|
||
|
||
|
||
@app.route("/api/videos")
|
||
def api_videos():
|
||
"""
|
||
获取视频列表 API。
|
||
支持 module 和 discipline 参数筛选：
|
||
/api/videos?module=数人
|
||
/api/videos?module=数人&discipline=计算机与AI
|
||
"""
|
||
module = request.args.get("module", "")
|
||
discipline = request.args.get("discipline", "")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
|
||
query = "SELECT * FROM videos WHERE 1=1"
|
||
params = []
|
||
if module:
|
||
query += " AND module=?"
|
||
params.append(module)
|
||
if discipline:
|
||
query += " AND discipline=?"
|
||
params.append(discipline)
|
||
query += " ORDER BY module, id DESC"
|
||
|
||
c.execute(query, params)
|
||
rows = [dict(r) for r in c.fetchall()]
|
||
|
||
# 同时返回当前模块的学科列表（用于前端渲染分类筛选）
|
||
if module == "数人":
|
||
c.execute("SELECT DISTINCT discipline, discipline_icon FROM videos WHERE module='数人' AND discipline != '' ORDER BY discipline")
|
||
disciplines = [dict(r) for r in c.fetchall()]
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": rows, "disciplines": disciplines})
|
||
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": rows})
|
||
|
||
|
||
@app.route("/api/chat", methods=["POST"])
|
||
def api_chat():
|
||
"""
|
||
AI 对话接口 — 转发用户消息给 DeepSeek，支持上下文感知。
|
||
|
||
请求格式: { "message": "学数学对学医有什么用？", "context": {"type": "video", "title": "xxx"} }
|
||
"""
|
||
data = request.get_json()
|
||
user_message = data.get("message", "")
|
||
student_id = data.get("student_id", "")
|
||
context = data.get("context") or {}
|
||
ctx_type = context.get("type", "")
|
||
ctx_title = context.get("title", "")
|
||
|
||
if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == "your-deepseek-api-key-here":
|
||
return jsonify({
|
||
"code": -1,
|
||
"reply": "DeepSeek API Key 尚未配置，请在 app.py 中填入你的 Key。\n申请地址: https://platform.deepseek.com/",
|
||
"is_fallback": True
|
||
})
|
||
|
||
# 根据上下文构建 system prompt
|
||
system_prompt = (
|
||
"你是'葫芦数学'的AI助教，面向中小学生和数学爱好者。你的核心任务是：\n"
|
||
"1. 用生活化的语言解释数学在各学科中的基础作用\n"
|
||
"2. 引导学生发现'原来数学这么有用'\n"
|
||
"3. 回答尽量简短、有趣，控制在200字以内\n"
|
||
"4. 不要出现超出初中生理解范围的专业术语"
|
||
)
|
||
if ctx_type == "video" and ctx_title:
|
||
system_prompt = (
|
||
f"用户正在观看视频《{ctx_title}》。你是'葫芦数学'的AI助教。\n"
|
||
f"请围绕这个视频主题，用生活化的语言帮助用户理解其中的数学概念。\n"
|
||
"回答简短有趣，控制在200字以内。"
|
||
)
|
||
elif ctx_type == "knowledge" and ctx_title:
|
||
system_prompt = (
|
||
f"用户正在阅读知识卡片《{ctx_title}》。你是'葫芦数学'的AI助教。\n"
|
||
f"请围绕这个主题深入浅出地解释，可以补充有趣的例子和冷知识。\n"
|
||
"回答简短有趣，控制在200字以内。"
|
||
)
|
||
elif ctx_type == "simulator":
|
||
system_prompt = (
|
||
"用户正在玩'数学专业模拟器'——这是模拟大学数学系4年学习生涯的互动游戏。\n"
|
||
"你是'葫芦数学'的AI助教，请扮演大学数学系的学长/学姐角色，\n"
|
||
"用鼓励、有趣的方式回答用户关于数学专业的问题。\n"
|
||
"可以分享一些大学数学系的真实体验和趣事。回答在200字以内。"
|
||
)
|
||
|
||
try:
|
||
resp = requests.post(
|
||
DEEPSEEK_API_URL,
|
||
headers={
|
||
"Authorization": f"Bearer {DEEPSEEK_API_KEY}",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"model": "deepseek-chat",
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_message}
|
||
],
|
||
"temperature": 0.7,
|
||
"max_tokens": 500
|
||
},
|
||
timeout=15
|
||
)
|
||
result = resp.json()
|
||
reply = result["choices"][0]["message"]["content"]
|
||
is_fallback = False
|
||
|
||
except Exception as e:
|
||
reply = f"[助手暂时离线] API 调用失败: {str(e)[:100]}"
|
||
is_fallback = True
|
||
|
||
# 记录行为日志
|
||
_log_behavior(student_id, "send_message", detail=user_message)
|
||
|
||
return jsonify({"code": 0, "reply": reply, "is_fallback": is_fallback, "context_type": ctx_type})
|
||
|
||
|
||
@app.route("/api/log", methods=["POST"])
|
||
def api_log():
|
||
data = request.json
|
||
user_id = data.get("user_id", "guest")
|
||
action = data.get("action", "")
|
||
target = data.get("target", "")
|
||
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute(
|
||
"INSERT INTO behavior_log (user_id, action, target) VALUES (?, ?, ?)",
|
||
(user_id, action, target)
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return jsonify({"code": 0, "msg": "logged"})
|
||
|
||
|
||
@app.route("/api/students")
|
||
def api_students():
|
||
"""获取预设学生列表（B端看板用）"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
c.execute("SELECT * FROM students ORDER BY grade, class_name")
|
||
rows = [dict(r) for r in c.fetchall()]
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": rows})
|
||
|
||
|
||
@app.route("/api/story")
|
||
def api_story():
|
||
"""获取数学武林故事数据"""
|
||
if not os.path.exists(STORY_PATH):
|
||
return jsonify({"code": -1, "msg": "故事数据文件不存在"})
|
||
with open(STORY_PATH, "r", encoding="utf-8") as f:
|
||
story = json.load(f)
|
||
return jsonify({"code": 0, "data": story})
|
||
|
||
|
||
@app.route("/api/story/math_teen")
|
||
def api_story_math_teen():
|
||
"""获取数学少年线互动剧情数据"""
|
||
if not os.path.exists(MATH_TEEN_STORY_PATH):
|
||
return jsonify({"code": -1, "msg": "数学少年线故事数据文件不存在"})
|
||
with open(MATH_TEEN_STORY_PATH, "r", encoding="utf-8") as f:
|
||
story = json.load(f)
|
||
return jsonify({"code": 0, "data": story})
|
||
|
||
|
||
@app.route("/api/alumni")
|
||
def api_alumni():
|
||
"""获取数学校友访谈与四年度模拟预留数据"""
|
||
if not os.path.exists(ALUMNI_PATH):
|
||
return jsonify({"code": 0, "data": [], "meta": {"status": "reserved"}})
|
||
with open(ALUMNI_PATH, "r", encoding="utf-8") as f:
|
||
payload = json.load(f)
|
||
return jsonify({"code": 0, "data": payload.get("alumni", []), "meta": payload.get("meta", {})})
|
||
|
||
|
||
@app.route("/api/simulator")
|
||
def api_simulator():
|
||
"""获取数学专业模拟器数据"""
|
||
if not os.path.exists(SIMULATOR_PATH):
|
||
return jsonify({"code": -1, "msg": "模拟器数据文件不存在"})
|
||
with open(SIMULATOR_PATH, "r", encoding="utf-8") as f:
|
||
sim = json.load(f)
|
||
return jsonify({"code": 0, "data": sim})
|
||
|
||
|
||
@app.route("/api/simulator/save", methods=["POST"])
|
||
def api_simulator_save():
|
||
"""保存模拟器进度"""
|
||
data = request.get_json()
|
||
user_id = data.get("user_id", "guest")
|
||
save_data = json.dumps(data.get("state", {}))
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS simulator_saves (
|
||
user_id TEXT PRIMARY KEY,
|
||
state TEXT,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
""")
|
||
c.execute("INSERT OR REPLACE INTO simulator_saves (user_id, state) VALUES (?, ?)", (user_id, save_data))
|
||
conn.commit()
|
||
conn.close()
|
||
return jsonify({"code": 0, "msg": "saved"})
|
||
|
||
|
||
@app.route("/api/simulator/save", methods=["GET"])
|
||
def api_simulator_load():
|
||
"""加载模拟器进度"""
|
||
user_id = request.args.get("user_id", "guest")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("CREATE TABLE IF NOT EXISTS simulator_saves (user_id TEXT PRIMARY KEY, state TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)")
|
||
c.execute("SELECT state FROM simulator_saves WHERE user_id=?", (user_id,))
|
||
row = c.fetchone()
|
||
conn.close()
|
||
if row:
|
||
return jsonify({"code": 0, "data": json.loads(row[0])})
|
||
return jsonify({"code": 0, "data": None})
|
||
|
||
|
||
@app.route("/api/knowledge")
|
||
def api_knowledge():
|
||
"""获取知识卡片列表"""
|
||
topic = request.args.get("topic", "")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
query = "SELECT * FROM knowledge_cards WHERE 1=1"
|
||
params = []
|
||
if topic:
|
||
query += " AND topic=?"
|
||
params.append(topic)
|
||
query += " ORDER BY sort_order"
|
||
c.execute(query, params)
|
||
rows = [dict(r) for r in c.fetchall()]
|
||
# 获取所有话题
|
||
c.execute("SELECT DISTINCT topic FROM knowledge_cards WHERE topic != '' ORDER BY topic")
|
||
topics = [r["topic"] for r in c.fetchall()]
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": rows, "topics": topics})
|
||
|
||
|
||
@app.route("/api/knowledge/<int:card_id>")
|
||
def api_knowledge_detail(card_id):
|
||
"""获取单个知识卡片详情"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
c.execute("SELECT * FROM knowledge_cards WHERE id=?", (card_id,))
|
||
row = c.fetchone()
|
||
conn.close()
|
||
if row:
|
||
return jsonify({"code": 0, "data": dict(row)})
|
||
return jsonify({"code": -1, "msg": "卡片不存在"})
|
||
|
||
|
||
@app.route("/api/favorites", methods=["GET", "POST", "DELETE"])
|
||
def api_favorites():
|
||
"""收藏管理"""
|
||
user_id = request.args.get("user_id") or (request.get_json() or {}).get("user_id", "guest")
|
||
|
||
if request.method == "GET":
|
||
target_type = request.args.get("type", "")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
if target_type:
|
||
c.execute("SELECT * FROM favorites WHERE user_id=? AND target_type=? ORDER BY created_at DESC", (user_id, target_type))
|
||
else:
|
||
c.execute("SELECT * FROM favorites WHERE user_id=? ORDER BY created_at DESC", (user_id,))
|
||
rows = [dict(r) for r in c.fetchall()]
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": rows})
|
||
|
||
elif request.method == "POST":
|
||
data = request.get_json()
|
||
target_type = data.get("target_type", "video")
|
||
target_id = data.get("target_id", 0)
|
||
title = data.get("title", "")
|
||
cover = data.get("cover", "")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
try:
|
||
c.execute(
|
||
"INSERT OR IGNORE INTO favorites (user_id, target_type, target_id, title, cover) VALUES (?, ?, ?, ?, ?)",
|
||
(user_id, target_type, target_id, title, cover)
|
||
)
|
||
conn.commit()
|
||
added = c.rowcount > 0
|
||
except Exception:
|
||
added = False
|
||
conn.close()
|
||
return jsonify({"code": 0, "msg": "collected" if added else "already_exists"})
|
||
|
||
elif request.method == "DELETE":
|
||
data = request.get_json()
|
||
target_type = data.get("target_type", "video")
|
||
target_id = data.get("target_id", 0)
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("DELETE FROM favorites WHERE user_id=? AND target_type=? AND target_id=?", (user_id, target_type, target_id))
|
||
conn.commit()
|
||
conn.close()
|
||
return jsonify({"code": 0, "msg": "removed"})
|
||
|
||
|
||
@app.route("/api/favorites/check")
|
||
def api_favorites_check():
|
||
"""批量检查收藏状态"""
|
||
user_id = request.args.get("user_id", "guest")
|
||
target_type = request.args.get("type", "video")
|
||
ids_str = request.args.get("ids", "")
|
||
if not ids_str:
|
||
return jsonify({"code": 0, "data": {}})
|
||
ids = [int(x) for x in ids_str.split(",") if x.strip().isdigit()]
|
||
if not ids:
|
||
return jsonify({"code": 0, "data": {}})
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
placeholders = ",".join("?" * len(ids))
|
||
c.execute(f"SELECT target_id FROM favorites WHERE user_id=? AND target_type=? AND target_id IN ({placeholders})",
|
||
[user_id, target_type] + ids)
|
||
favorited = {row[0]: True for row in c.fetchall()}
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": favorited})
|
||
|
||
|
||
@app.route("/api/history", methods=["GET", "POST"])
|
||
def api_history():
|
||
"""观看历史"""
|
||
user_id = request.args.get("user_id") or (request.get_json() or {}).get("user_id", "guest")
|
||
|
||
if request.method == "GET":
|
||
target_type = request.args.get("type", "")
|
||
limit = int(request.args.get("limit", 20))
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
if target_type:
|
||
c.execute("SELECT * FROM watch_history WHERE user_id=? AND target_type=? ORDER BY watched_at DESC LIMIT ?", (user_id, target_type, limit))
|
||
else:
|
||
c.execute("SELECT * FROM watch_history WHERE user_id=? ORDER BY watched_at DESC LIMIT ?", (user_id, limit))
|
||
rows = [dict(r) for r in c.fetchall()]
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": rows})
|
||
|
||
elif request.method == "POST":
|
||
data = request.get_json()
|
||
target_type = data.get("target_type", "video")
|
||
target_id = data.get("target_id", 0)
|
||
title = data.get("title", "")
|
||
cover = data.get("cover", "")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute(
|
||
"INSERT INTO watch_history (user_id, target_type, target_id, title, cover) VALUES (?, ?, ?, ?, ?)",
|
||
(user_id, target_type, target_id, title, cover)
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
return jsonify({"code": 0, "msg": "recorded"})
|
||
|
||
|
||
@app.route("/api/user/profile", methods=["GET", "POST"])
|
||
def api_user_profile():
|
||
"""用户个人资料"""
|
||
user_id = request.args.get("user_id") or (request.get_json() or {}).get("user_id", "guest")
|
||
|
||
if request.method == "GET":
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
c.execute("SELECT * FROM users WHERE user_id=?", (user_id,))
|
||
row = c.fetchone()
|
||
if not row:
|
||
c.execute("INSERT INTO users (user_id) VALUES (?)", (user_id,))
|
||
conn.commit()
|
||
c.execute("SELECT * FROM users WHERE user_id=?", (user_id,))
|
||
row = c.fetchone()
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": dict(row)})
|
||
|
||
elif request.method == "POST":
|
||
data = request.get_json()
|
||
nickname = data.get("nickname", "")
|
||
avatar_emoji = data.get("avatar_emoji", "")
|
||
avatar_color = data.get("avatar_color", "")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("INSERT INTO users (user_id) VALUES (?) ON CONFLICT(user_id) DO NOTHING", (user_id,))
|
||
if nickname:
|
||
c.execute("UPDATE users SET nickname=? WHERE user_id=?", (nickname, user_id))
|
||
if avatar_emoji:
|
||
c.execute("UPDATE users SET avatar_emoji=? WHERE user_id=?", (avatar_emoji, user_id))
|
||
if avatar_color:
|
||
c.execute("UPDATE users SET avatar_color=? WHERE user_id=?", (avatar_color, user_id))
|
||
conn.commit()
|
||
conn.close()
|
||
return jsonify({"code": 0, "msg": "updated"})
|
||
|
||
|
||
@app.route("/api/user/stats")
|
||
def api_user_stats():
|
||
"""用户统计数据"""
|
||
user_id = request.args.get("user_id", "guest")
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("SELECT COUNT(*) FROM watch_history WHERE user_id=?", (user_id,))
|
||
watch_count = c.fetchone()[0] or 0
|
||
c.execute("SELECT COUNT(*) FROM favorites WHERE user_id=?", (user_id,))
|
||
fav_count = c.fetchone()[0] or 0
|
||
conn.close()
|
||
return jsonify({"code": 0, "data": {"watch_count": watch_count, "favorite_count": fav_count}})
|
||
|
||
|
||
# ---- 工具函数 ----
|
||
def _log_behavior(student_id, action_type, target_id="", detail=""):
|
||
"""内部函数：写行为日志"""
|
||
try:
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute(
|
||
"INSERT INTO behavior_log (user_id, action, target, detail) VALUES (?, ?, ?, ?)",
|
||
(student_id, action_type, target_id, detail)
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
except Exception:
|
||
pass # 日志写入失败不阻塞主流程
|
||
|
||
|
||
# ---- 预设学生数据（模拟学校导入） ----
|
||
def seed_students():
|
||
"""插入预设学生，模拟'三下乡'时学校提供的名单"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
c = conn.cursor()
|
||
c.execute("SELECT COUNT(*) FROM students")
|
||
if c.fetchone()[0] == 0:
|
||
students = [
|
||
("2024001", "张三", "七年级(1)班", "7"),
|
||
("2024002", "李四", "七年级(1)班", "7"),
|
||
("2024003", "王五", "七年级(2)班", "7"),
|
||
("2024004", "赵六", "八年级(1)班", "8"),
|
||
("2024005", "陈七", "八年级(1)班", "8"),
|
||
("2024006", "刘八", "九年级(1)班", "9"),
|
||
]
|
||
c.executemany(
|
||
"INSERT INTO students (student_id, name, class_name, grade) VALUES (?, ?, ?, ?)",
|
||
students
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
# ============================================================
|
||
# MathBTI — 数学人格测试
|
||
# ============================================================
|
||
|
||
@app.route("/mathbti")
|
||
def mathbti_page():
|
||
"""MathBTI 独立页面 — 可分享链接"""
|
||
return render_template("mathbti.html")
|
||
|
||
|
||
@app.route("/api/math_personality")
|
||
def api_math_personality():
|
||
"""获取数学人格测试题目数据"""
|
||
if not os.path.exists(MATHBTI_PATH):
|
||
return jsonify({"code": -1, "msg": "测试数据文件不存在"})
|
||
with open(MATHBTI_PATH, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
return jsonify({"code": 0, "data": data})
|
||
|
||
|
||
@app.route("/api/math_personality/calculate", methods=["POST"])
|
||
def api_math_personality_calculate():
|
||
"""
|
||
根据用户答案计算结果
|
||
请求格式: { "answers": [0,3,2,1,0,3,2,1,0,3,2,1] } # 12个得分值
|
||
返回: { code: 0, data: { binary_code: "0101", result: {...}, axis_scores: {...} } }
|
||
"""
|
||
req_data = request.get_json()
|
||
answers = req_data.get("answers", []) if req_data else []
|
||
|
||
if not os.path.exists(MATHBTI_PATH):
|
||
return jsonify({"code": -1, "msg": "测试数据文件不存在"})
|
||
|
||
with open(MATHBTI_PATH, "r", encoding="utf-8") as f:
|
||
test_data = json.load(f)
|
||
|
||
if len(answers) != 12:
|
||
return jsonify({"code": -1, "msg": "需要12个答案"})
|
||
|
||
questions = test_data.get("questions", [])
|
||
axes = test_data.get("scoring", {}).get("axes", ["style", "purpose", "era", "social"])
|
||
cutoff = test_data.get("scoring", {}).get("cutoff", 4)
|
||
|
||
# 计算每轴总分
|
||
axis_scores = {ax: 0 for ax in axes}
|
||
for i, q in enumerate(questions):
|
||
ax = q.get("axis", "")
|
||
score = answers[i] if i < len(answers) else 0
|
||
if ax in axis_scores:
|
||
axis_scores[ax] += score
|
||
|
||
# 生成二进制码
|
||
binary_code = ""
|
||
for ax in axes:
|
||
total = axis_scores.get(ax, 0)
|
||
binary_code += "1" if total > cutoff else "0"
|
||
|
||
# 获取结果
|
||
results = test_data.get("results", {})
|
||
result = results.get(binary_code, results.get("0000", {}))
|
||
|
||
return jsonify({
|
||
"code": 0,
|
||
"data": {
|
||
"binary_code": binary_code,
|
||
"result": result,
|
||
"axis_scores": axis_scores
|
||
}
|
||
})
|
||
|
||
|
||
# ============================================================
|
||
# 模块 3: 应用启动入口
|
||
# ============================================================
|
||
if __name__ == "__main__":
|
||
print("=" * 50)
|
||
print(" 数学桥梁计划 · MVP 后端启动中...")
|
||
print("=" * 50)
|
||
init_db()
|
||
load_seed_videos()
|
||
load_seed_knowledge()
|
||
seed_students()
|
||
print(f"\n 访问地址: http://127.0.0.1:5000\n")
|
||
app.run(debug=False, host="0.0.0.0", port=5000)
|