simon
b6df028839
变更摘要: - [fix] 清理根目录 index.html 遗留代码（questionnaire-error + retryLoad 移除） - [fix] 副标题同步更新为自估591分 + 候选志愿546条数据校准 - [fix] 页脚注入版本号 v1.0.0 - [fix] 部署版 index.html: 取消家长标签，统一 filler-tianheng - [fix] Dashboard 文案修正（4份→2份线上问卷） - [fix] 清理冗余部署脚本（v1-v4归档至 archived_scripts/） - [fix] 等效位次工具纳入 deploy 目录 - [chore] .gitignore 初始化 - [init] Git 仓库初始化
675 lines
22 KiB
Python
675 lines
22 KiB
Python
"""
|
||
高考志愿家庭门户 — Flask 后端
|
||
功能: 问卷 API + SQLite 数据存储 + 多成员结果汇总
|
||
|
||
数据库表:
|
||
- questionnaires: 问卷定义（标题、描述、题目 JSON）
|
||
- responses: 成员提交的答卷（成员名、问卷ID、答案 JSON、提交时间）
|
||
"""
|
||
import sqlite3
|
||
import json
|
||
import os
|
||
from datetime import datetime, timezone, timedelta
|
||
from flask import Flask, request, jsonify
|
||
from flask_cors import CORS
|
||
|
||
# ==================== 应用初始化 ====================
|
||
app = Flask(__name__)
|
||
CORS(app) # 允许前端跨域请求（Nginx 反向代理场景也需要）
|
||
|
||
# 数据库路径：挂载卷 /app/data/gaokao.db
|
||
DB_PATH = os.path.join(os.path.dirname(__file__), "data", "gaokao.db")
|
||
|
||
# 上海时区
|
||
CST = timezone(timedelta(hours=8))
|
||
|
||
|
||
def get_db():
|
||
"""获取数据库连接（每次请求新建，自动提交）"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row # 让查询结果可以用列名访问
|
||
conn.execute("PRAGMA journal_mode=WAL") # WAL 模式提升并发性能
|
||
conn.execute("PRAGMA foreign_keys=ON")
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
"""初始化数据库表（幂等操作——表不存在才创建）"""
|
||
conn = get_db()
|
||
conn.executescript("""
|
||
CREATE TABLE IF NOT EXISTS questionnaires (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
title TEXT NOT NULL, -- 问卷标题
|
||
description TEXT, -- 问卷说明
|
||
questions TEXT NOT NULL, -- 题目列表（JSON 数组）
|
||
target_user TEXT DEFAULT '天恒', -- 目标填写人：天恒 / 父母
|
||
created_at TEXT DEFAULT (datetime('now', 'localtime'))
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS responses (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
questionnaire_id INTEGER NOT NULL, -- 关联问卷
|
||
user_name TEXT NOT NULL, -- 填写人姓名（如"爸爸""天恒"）
|
||
answers TEXT NOT NULL, -- 答案（JSON 对象: {题号: 答案}）
|
||
created_at TEXT DEFAULT (datetime('now', 'localtime')),
|
||
FOREIGN KEY (questionnaire_id) REFERENCES questionnaires(id)
|
||
);
|
||
""")
|
||
conn.commit()
|
||
|
||
# 迁移：为已有表添加 target_user 列（如果不存在）
|
||
try:
|
||
conn.execute("ALTER TABLE questionnaires ADD COLUMN target_user TEXT DEFAULT '天恒'")
|
||
conn.commit()
|
||
except sqlite3.OperationalError:
|
||
pass # 列已存在
|
||
|
||
# 独立测评结果表（霍兰德/MBTI 自评HTML提交）
|
||
conn.executescript("""
|
||
CREATE TABLE IF NOT EXISTS standalone_results (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
test_name TEXT NOT NULL, -- 'holland' 或 'mbti'
|
||
user_name TEXT NOT NULL DEFAULT '天恒',
|
||
result TEXT NOT NULL, -- JSON: 霍兰德三字母代码 或 MBTI四字母类型
|
||
full_data TEXT, -- JSON: 完整结果数据
|
||
created_at TEXT DEFAULT (datetime('now', 'localtime'))
|
||
);
|
||
""")
|
||
|
||
conn.close()
|
||
|
||
|
||
# ==================== API 路由 ====================
|
||
|
||
@app.route("/api/health")
|
||
def health():
|
||
"""健康检查（用于确认服务启动成功）"""
|
||
return jsonify({"status": "ok", "time": datetime.now(CST).isoformat()})
|
||
|
||
|
||
@app.route("/api/questionnaires", methods=["GET"])
|
||
def list_questionnaires():
|
||
"""获取所有问卷列表"""
|
||
conn = get_db()
|
||
rows = conn.execute(
|
||
"SELECT id, title, description, target_user, created_at FROM questionnaires ORDER BY id"
|
||
).fetchall()
|
||
conn.close()
|
||
return jsonify([dict(r) for r in rows])
|
||
|
||
|
||
@app.route("/api/questionnaire/<int:qid>", methods=["GET"])
|
||
def get_questionnaire(qid):
|
||
"""获取单个问卷的完整内容（含所有题目）"""
|
||
conn = get_db()
|
||
row = conn.execute(
|
||
"SELECT * FROM questionnaires WHERE id = ?", (qid,)
|
||
).fetchone()
|
||
conn.close()
|
||
|
||
if not row:
|
||
return jsonify({"error": "问卷不存在"}), 404
|
||
|
||
result = dict(row)
|
||
# questions 在数据库中存储为 JSON 字符串，解析后返回
|
||
result["questions"] = json.loads(result["questions"])
|
||
return jsonify(result)
|
||
|
||
|
||
@app.route("/api/questionnaire/<int:qid>/submit", methods=["POST"])
|
||
def submit_response(qid):
|
||
"""
|
||
提交问卷答案
|
||
请求体 JSON: {"user_name": "爸爸", "answers": {"1": "A", "2": "B", ...}}
|
||
|
||
设计考量:
|
||
- 同一用户对同一问卷可以多次提交（保留最新+历史记录）
|
||
- user_name 用于区分不同家庭成员
|
||
"""
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({"error": "请求体为空"}), 400
|
||
|
||
user_name = data.get("user_name", "").strip()
|
||
answers = data.get("answers", {})
|
||
|
||
if not user_name:
|
||
return jsonify({"error": "请填写您的称呼（如：爸爸、妈妈、天恒）"}), 400
|
||
if not answers:
|
||
return jsonify({"error": "请至少回答一题"}), 400
|
||
|
||
# 验证问卷是否存在
|
||
conn = get_db()
|
||
q = conn.execute("SELECT id FROM questionnaires WHERE id = ?", (qid,)).fetchone()
|
||
if not q:
|
||
conn.close()
|
||
return jsonify({"error": "问卷不存在"}), 404
|
||
|
||
# 插入答卷（JSON 序列化答案对象）
|
||
conn.execute(
|
||
"INSERT INTO responses (questionnaire_id, user_name, answers) VALUES (?, ?, ?)",
|
||
(qid, user_name, json.dumps(answers, ensure_ascii=False))
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return jsonify({"success": True, "message": f"{user_name} 的答卷已保存"})
|
||
|
||
|
||
@app.route("/api/results/<int:qid>", methods=["GET"])
|
||
def view_results(qid):
|
||
"""
|
||
查看某问卷的所有答卷（按时间倒序）
|
||
返回包含解析后的 answers 字段
|
||
"""
|
||
conn = get_db()
|
||
rows = conn.execute(
|
||
"""SELECT id, questionnaire_id, user_name, answers, created_at
|
||
FROM responses
|
||
WHERE questionnaire_id = ?
|
||
ORDER BY created_at DESC""",
|
||
(qid,)
|
||
).fetchall()
|
||
conn.close()
|
||
|
||
results = []
|
||
for r in rows:
|
||
d = dict(r)
|
||
d["answers"] = json.loads(d["answers"])
|
||
results.append(d)
|
||
|
||
return jsonify(results)
|
||
|
||
|
||
@app.route("/api/results/<int:qid>/summary", methods=["GET"])
|
||
def view_summary(qid):
|
||
"""
|
||
查看某问卷的汇总分析
|
||
对于选择题问卷（如 Holland 评估），按选项计算得分/分布
|
||
对于偏好问卷，展示每个成员的答案对比
|
||
"""
|
||
conn = get_db()
|
||
qrow = conn.execute(
|
||
"SELECT questions FROM questionnaires WHERE id = ?", (qid,)
|
||
).fetchone()
|
||
if not qrow:
|
||
conn.close()
|
||
return jsonify({"error": "问卷不存在"}), 404
|
||
|
||
questions = json.loads(qrow["questions"])
|
||
|
||
rows = conn.execute(
|
||
"""SELECT user_name, answers, created_at
|
||
FROM responses
|
||
WHERE questionnaire_id = ?
|
||
ORDER BY created_at DESC""",
|
||
(qid,)
|
||
).fetchall()
|
||
conn.close()
|
||
|
||
# 构建汇总：每个成员 × 每道题
|
||
members = {}
|
||
for r in rows:
|
||
name = r["user_name"]
|
||
answers = json.loads(r["answers"])
|
||
members[name] = {
|
||
"answers": answers,
|
||
"submitted_at": r["created_at"]
|
||
}
|
||
|
||
return jsonify({
|
||
"questionnaire_id": qid,
|
||
"question_count": len(questions),
|
||
"member_count": len(members),
|
||
"members": members
|
||
})
|
||
|
||
|
||
# ==================== 报告生成 API ====================
|
||
|
||
@app.route("/api/report/<int:qid>", methods=["GET"])
|
||
def generate_report(qid):
|
||
"""
|
||
生成问卷分析报告（核心功能）
|
||
- Holland评估：生成六维雷达图数据 + 职业推荐
|
||
- 偏好问卷：生成差异分析 + 共识点
|
||
- 支持导出为后续分析可用的JSON格式
|
||
"""
|
||
conn = get_db()
|
||
qrow = conn.execute(
|
||
"SELECT * FROM questionnaires WHERE id = ?", (qid,)
|
||
).fetchone()
|
||
if not qrow:
|
||
conn.close()
|
||
return jsonify({"error": "问卷不存在"}), 404
|
||
|
||
questions = json.loads(qrow["questions"])
|
||
rows = conn.execute(
|
||
"""SELECT user_name, answers, created_at
|
||
FROM responses
|
||
WHERE questionnaire_id = ?
|
||
ORDER BY created_at DESC""",
|
||
(qid,)
|
||
).fetchall()
|
||
conn.close()
|
||
|
||
if not rows:
|
||
return jsonify({"error": "暂无答卷数据", "message": "请先填写问卷"}), 400
|
||
|
||
# 解析所有答卷
|
||
members = {}
|
||
for r in rows:
|
||
name = r["user_name"]
|
||
answers = json.loads(r["answers"])
|
||
members[name] = {
|
||
"answers": answers,
|
||
"submitted_at": r["created_at"]
|
||
}
|
||
|
||
# 根据问卷类型生成报告
|
||
report = {
|
||
"questionnaire_id": qid,
|
||
"questionnaire_title": qrow["title"],
|
||
"generated_at": datetime.now(CST).isoformat(),
|
||
"member_count": len(members),
|
||
"report_type": None,
|
||
"data": {},
|
||
"conclusions": [],
|
||
"next_input": {} # 供后续分析使用的结构化数据
|
||
}
|
||
|
||
# Holland 评估报告（问卷ID=1）
|
||
if qid == 1:
|
||
report["report_type"] = "holland"
|
||
report["data"], report["conclusions"], report["next_input"] = _generate_holland_report(members, questions)
|
||
# 院校偏好问卷报告（问卷ID=2）
|
||
elif qid == 2:
|
||
report["report_type"] = "preference"
|
||
report["data"], report["conclusions"], report["next_input"] = _generate_preference_report(members, questions)
|
||
# 家庭期望问卷报告（问卷ID=3）
|
||
elif qid == 3:
|
||
report["report_type"] = "expectation"
|
||
report["data"], report["conclusions"], report["next_input"] = _generate_expectation_report(members, questions)
|
||
else:
|
||
report["report_type"] = "generic"
|
||
report["data"] = {"members": list(members.keys())}
|
||
|
||
return jsonify(report)
|
||
|
||
|
||
def _generate_holland_report(members, questions):
|
||
"""生成 Holland 职业兴趣报告"""
|
||
# RIASEC 六维定义
|
||
riasec_dims = ["R", "I", "A", "S", "E", "C"]
|
||
riasec_names = {
|
||
"R": "现实型", "I": "研究型", "A": "艺术型",
|
||
"S": "社会型", "E": "企业型", "C": "常规型"
|
||
}
|
||
|
||
# 计算每个成员的维度得分
|
||
results = {}
|
||
for name, data in members.items():
|
||
answers = data["answers"]
|
||
scores = {dim: 0 for dim in riasec_dims}
|
||
count = {dim: 0 for dim in riasec_dims}
|
||
|
||
for q_id, answer in answers.items():
|
||
q_num = int(q_id.replace("q", ""))
|
||
# 题目1-4: R, 5-8: I, 9-12: A, 13-16: S, 17-20: E, 21-24: C
|
||
if 1 <= q_num <= 4:
|
||
dim = "R"
|
||
elif 5 <= q_num <= 8:
|
||
dim = "I"
|
||
elif 9 <= q_num <= 12:
|
||
dim = "A"
|
||
elif 13 <= q_num <= 16:
|
||
dim = "S"
|
||
elif 17 <= q_num <= 20:
|
||
dim = "E"
|
||
else:
|
||
dim = "C"
|
||
scores[dim] += int(answer) if answer.isdigit() else 3
|
||
count[dim] += 1
|
||
|
||
# 计算平均分
|
||
avg_scores = {
|
||
dim: round(scores[dim] / count[dim], 1) if count[dim] > 0 else 0
|
||
for dim in riasec_dims
|
||
}
|
||
results[name] = avg_scores
|
||
|
||
# 差异分析
|
||
all_dims_avg = {dim: 0 for dim in riasec_dims}
|
||
for dim in riasec_dims:
|
||
total = sum(results[m][dim] for m in results)
|
||
all_dims_avg[dim] = round(total / len(results), 1) if results else 0
|
||
|
||
# 职业推荐（取Top3维度）
|
||
conclusions = []
|
||
for name, scores in results.items():
|
||
sorted_dims = sorted(scores.items(), key=lambda x: x[1], reverse=True)
|
||
top3 = [f"{d[0]}({riasec_names[d[0]]}:{d[1]})" for d in sorted_dims[:3]]
|
||
conclusions.append({
|
||
"member": name,
|
||
"top_dims": top3,
|
||
"career_type": "".join([d[0] for d in sorted_dims[:3]])
|
||
})
|
||
|
||
# 后续分析输入
|
||
next_input = {
|
||
"holland_code": conclusions[0]["career_type"] if conclusions else "",
|
||
"primary_type": riasec_names.get(conclusions[0]["career_type"][0], "") if conclusions else "",
|
||
"dimension_scores": results,
|
||
"family_avg": all_dims_avg
|
||
}
|
||
|
||
return {"dimension_scores": results, "family_avg": all_dims_avg}, conclusions, next_input
|
||
|
||
|
||
def _generate_preference_report(members, questions):
|
||
"""生成院校偏好报告"""
|
||
# 分析每个问题的答案分布
|
||
question_analysis = {}
|
||
for q_id, data in members.items():
|
||
answers = data["answers"]
|
||
for q, ans in answers.items():
|
||
if q not in question_analysis:
|
||
question_analysis[q] = {}
|
||
question_analysis[q][q_id] = ans
|
||
|
||
# 差异点识别
|
||
conflicts = []
|
||
consensus = []
|
||
for q, ans_dict in question_analysis.items():
|
||
unique_ans = set(ans_dict.values())
|
||
if len(unique_ans) > 1:
|
||
conflicts.append({"question": q, "answers": ans_dict})
|
||
else:
|
||
consensus.append({"question": q, "answer": list(unique_ans)[0]})
|
||
|
||
conclusions = [
|
||
{"type": "conflict", "count": len(conflicts), "details": conflicts[:3]},
|
||
{"type": "consensus", "count": len(consensus), "details": consensus[:3]}
|
||
]
|
||
|
||
# 后续分析输入
|
||
next_input = {
|
||
"conflict_count": len(conflicts),
|
||
"consensus_count": len(consensus),
|
||
"key_conflicts": [c["question"] for c in conflicts[:3]]
|
||
}
|
||
|
||
return {"conflicts": conflicts, "consensus": consensus}, conclusions, next_input
|
||
|
||
|
||
def _generate_expectation_report(members, questions):
|
||
"""生成家庭期望对齐报告"""
|
||
# 类似偏好报告，但强调开放式问题
|
||
text_answers = {}
|
||
choice_answers = {}
|
||
|
||
for name, data in members.items():
|
||
answers = data["answers"]
|
||
for q, ans in answers.items():
|
||
if len(str(ans)) > 50: # 开放式长文本
|
||
text_answers.setdefault(q, {})[name] = ans
|
||
else:
|
||
choice_answers.setdefault(q, {})[name] = ans
|
||
|
||
# 核心差异
|
||
conflicts = []
|
||
for q, ans_dict in choice_answers.items():
|
||
if len(set(ans_dict.values())) > 1:
|
||
conflicts.append({"question": q, "answers": ans_dict})
|
||
|
||
conclusions = [
|
||
{"type": "text_response", "count": len(text_answers)},
|
||
{"type": "choice_conflict", "count": len(conflicts), "details": conflicts[:3]}
|
||
]
|
||
|
||
next_input = {
|
||
"text_count": len(text_answers),
|
||
"choice_conflicts": len(conflicts),
|
||
"requires_discussion": [c["question"] for c in conflicts[:3]]
|
||
}
|
||
|
||
return {"text": text_answers, "choices": choice_answers}, conclusions, next_input
|
||
|
||
|
||
@app.route("/api/export/<int:qid>", methods=["GET"])
|
||
def export_report(qid):
|
||
"""
|
||
导出报告为JSON（供后续分析使用）
|
||
"""
|
||
import io
|
||
conn = get_db()
|
||
qrow = conn.execute("SELECT * FROM questionnaires WHERE id = ?", (qid,)).fetchone()
|
||
if not qrow:
|
||
conn.close()
|
||
return jsonify({"error": "问卷不存在"}), 404
|
||
|
||
rows = conn.execute(
|
||
"SELECT user_name, answers, created_at FROM responses WHERE questionnaire_id = ?",
|
||
(qid,)
|
||
).fetchall()
|
||
conn.close()
|
||
|
||
members = {}
|
||
for r in rows:
|
||
members[r["user_name"]] = {
|
||
"answers": json.loads(r["answers"]),
|
||
"submitted_at": r["created_at"]
|
||
}
|
||
|
||
# 导出结构
|
||
export_data = {
|
||
"questionnaire": dict(qrow),
|
||
"responses": members,
|
||
"exported_at": datetime.now(CST).isoformat(),
|
||
"version": "1.0"
|
||
}
|
||
|
||
# 返回JSON下载
|
||
return jsonify(export_data)
|
||
|
||
|
||
# ==================== 独立测评提交 API（霍兰德/MBTI） ====================
|
||
|
||
@app.route("/api/standalone/submit", methods=["POST"])
|
||
def submit_standalone():
|
||
"""
|
||
接收独立HTML自评测试的结果提交
|
||
请求体 JSON: {"test_name": "holland", "user_name": "天恒", "result": "RIC", "full_data": {...}}
|
||
"""
|
||
data = request.get_json()
|
||
if not data:
|
||
return jsonify({"error": "请求体为空"}), 400
|
||
|
||
test_name = data.get("test_name", "").strip()
|
||
user_name = data.get("user_name", "天恒").strip()
|
||
result = data.get("result", "").strip()
|
||
full_data = data.get("full_data", {})
|
||
|
||
if test_name not in ("holland", "mbti"):
|
||
return jsonify({"error": "test_name 必须为 holland 或 mbti"}), 400
|
||
if not result:
|
||
return jsonify({"error": "请提供 result"}), 400
|
||
|
||
conn = get_db()
|
||
# 同一用户同一测试的最新结果（替换旧结果）
|
||
conn.execute(
|
||
"INSERT INTO standalone_results (test_name, user_name, result, full_data) VALUES (?, ?, ?, ?)",
|
||
(test_name, user_name, result, json.dumps(full_data, ensure_ascii=False))
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
return jsonify({"success": True, "message": f"{test_name} 结果已保存"})
|
||
|
||
|
||
@app.route("/api/standalone/status", methods=["GET"])
|
||
def standalone_status():
|
||
"""查询独立测评完成状态"""
|
||
conn = get_db()
|
||
rows = conn.execute(
|
||
"""SELECT test_name, user_name, result, created_at
|
||
FROM standalone_results
|
||
ORDER BY created_at DESC"""
|
||
).fetchall()
|
||
conn.close()
|
||
|
||
status = {"holland": None, "mbti": None}
|
||
for r in rows:
|
||
name = r["test_name"]
|
||
if status[name] is None: # 取最新一条
|
||
status[name] = {
|
||
"completed": True,
|
||
"user_name": r["user_name"],
|
||
"result": r["result"],
|
||
"completed_at": r["created_at"]
|
||
}
|
||
|
||
for k in status:
|
||
if status[k] is None:
|
||
status[k] = {"completed": False}
|
||
|
||
return jsonify(status)
|
||
|
||
|
||
# ==================== 完成状态总览 API ====================
|
||
|
||
@app.route("/api/completion-status", methods=["GET"])
|
||
def completion_status():
|
||
"""
|
||
返回所有6个测评的完成状态总览
|
||
"""
|
||
conn = get_db()
|
||
|
||
# 1. 4份API问卷状态
|
||
q_rows = conn.execute(
|
||
"""SELECT q.id, q.title, q.target_user,
|
||
(SELECT COUNT(*) FROM responses r WHERE r.questionnaire_id = q.id) as response_count
|
||
FROM questionnaires q
|
||
ORDER BY q.id"""
|
||
).fetchall()
|
||
|
||
questionnaires_status = []
|
||
for qr in q_rows:
|
||
q = dict(qr)
|
||
# 取最新答卷人
|
||
latest = conn.execute(
|
||
"SELECT user_name, created_at FROM responses WHERE questionnaire_id = ? ORDER BY created_at DESC LIMIT 1",
|
||
(q["id"],)
|
||
).fetchone()
|
||
q["latest_respondent"] = dict(latest) if latest else None
|
||
q["completed"] = q["response_count"] > 0
|
||
questionnaires_status.append(q)
|
||
|
||
# 2. 独立测评状态
|
||
standalone_rows = conn.execute(
|
||
"""SELECT test_name, user_name, result, created_at
|
||
FROM standalone_results
|
||
ORDER BY created_at DESC"""
|
||
).fetchall()
|
||
|
||
standalone = {"holland": {"completed": False}, "mbti": {"completed": False}}
|
||
for r in standalone_rows:
|
||
name = r["test_name"]
|
||
if not standalone[name]["completed"]:
|
||
standalone[name] = {
|
||
"completed": True,
|
||
"user_name": r["user_name"],
|
||
"result": r["result"],
|
||
"completed_at": r["created_at"]
|
||
}
|
||
|
||
conn.close()
|
||
|
||
# 3. 汇总统计
|
||
api_completed = sum(1 for q in questionnaires_status if q["completed"])
|
||
standalone_completed = sum(1 for s in standalone.values() if s["completed"])
|
||
total = 6 # 4 API + 2 standalone
|
||
all_completed = (api_completed + standalone_completed) == total
|
||
|
||
return jsonify({
|
||
"total": total,
|
||
"completed_count": api_completed + standalone_completed,
|
||
"all_completed": all_completed,
|
||
"questionnaires": questionnaires_status,
|
||
"standalone": standalone,
|
||
"updated_at": datetime.now(CST).isoformat()
|
||
})
|
||
|
||
|
||
# ==================== 汇总报告 API ====================
|
||
|
||
@app.route("/api/summary", methods=["GET"])
|
||
def generate_summary():
|
||
"""
|
||
汇总所有已完成问卷和测评的结果，生成可下载的结构化数据
|
||
"""
|
||
conn = get_db()
|
||
|
||
# API问卷结果
|
||
q_rows = conn.execute(
|
||
"""SELECT q.id, q.title, q.target_user, q.description
|
||
FROM questionnaires q ORDER BY q.id"""
|
||
).fetchall()
|
||
|
||
questionnaires = []
|
||
for qr in q_rows:
|
||
q = dict(qr)
|
||
responses = conn.execute(
|
||
"SELECT user_name, answers, created_at FROM responses WHERE questionnaire_id = ? ORDER BY created_at DESC",
|
||
(q["id"],)
|
||
).fetchall()
|
||
q["responses"] = []
|
||
for r in responses:
|
||
rd = dict(r)
|
||
rd["answers"] = json.loads(rd["answers"])
|
||
q["responses"].append(rd)
|
||
questionnaires.append(q)
|
||
|
||
# 独立测评结果
|
||
standalone_rows = conn.execute(
|
||
"""SELECT test_name, user_name, result, full_data, created_at
|
||
FROM standalone_results ORDER BY created_at DESC"""
|
||
).fetchall()
|
||
|
||
standalone = {"holland": None, "mbti": None}
|
||
for r in standalone_rows:
|
||
name = r["test_name"]
|
||
if standalone[name] is None:
|
||
standalone[name] = {
|
||
"user_name": r["user_name"],
|
||
"result": r["result"],
|
||
"full_data": json.loads(r["full_data"]) if r["full_data"] else {},
|
||
"completed_at": r["created_at"]
|
||
}
|
||
|
||
conn.close()
|
||
|
||
# 完成统计
|
||
q_completed = sum(1 for q in questionnaires if len(q["responses"]) > 0)
|
||
s_completed = sum(1 for v in standalone.values() if v is not None)
|
||
|
||
return jsonify({
|
||
"generated_at": datetime.now(CST).isoformat(),
|
||
"status": {
|
||
"total": 6,
|
||
"completed": q_completed + s_completed,
|
||
"questionnaires_done": q_completed,
|
||
"standalone_done": s_completed
|
||
},
|
||
"questionnaires": questionnaires,
|
||
"standalone_tests": standalone,
|
||
"note": "此数据可用于生成志愿填报分析报告。如独立测评（霍兰德/MBTI）未提交，请通过测试页面的提交按钮上传结果。"
|
||
})
|
||
|
||
|
||
# ==================== 启动入口 ====================
|
||
if __name__ == "__main__":
|
||
init_db()
|
||
# Flask 开发服务器，生产环境建议用 gunicorn（但 Alpine + 轻量场景 flask run 足够）
|
||
app.run(host="0.0.0.0", port=5000, debug=False)
|