from flask import Flask, render_template, request, redirect, url_for, flash, session, jsonify from flask_sqlalchemy import SQLAlchemy from werkzeug.security import generate_password_hash, check_password_hash from datetime import datetime, date import base64 import time import requests import pymysql from backend.config import get_settings from backend.database import engine, Base, ensure_user_profile_columns import backend.models # noqa: F401 - 确保 backend 模型注册到 Base.metadata，供 create_all 创建表 from neo4j import GraphDatabase from backend.services.llm_service import expand_with_llm from backend.services.question_parser import analyze_question from backend.services.mastery_service import update_mastery_from_answers, mark_daily_recommendations_done from backend.services.recommendation import ( get_or_generate_daily_recommendations, get_mastery_matrix, get_motivation_summary, get_question_analysis, get_diagnostic_questions, ) from backend.services.teacher_report_service import get_teacher_class_report # 修正 MySQL 驱动兼容性问题（如果遇到报错可以尝试加上） pymysql.install_as_MySQLdb() app = Flask(__name__) settings = get_settings() # --- 数据库配置 --- # 格式：mysql+用户名:密码@localhost:3306/数据库名 # 当前使用 MySQL 用户 maple@localhost，密码 123456，数据库 ai_system app.config['SQLALCHEMY_DATABASE_URI'] = settings.MYSQL_URI app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SECRET_KEY'] = 'dev-key-please-change-in-production' # 用于Session加密 """数据库初始化""" db = SQLAlchemy(app) # --- Neo4j 配置（用于知识图谱） --- neo4j_driver = GraphDatabase.driver( settings.NEO4J_URI, auth=(settings.NEO4J_USER, settings.NEO4J_PASSWORD), ) # --- 百度语音合成（TTS）全局配置与简单缓存 --- _BAIDU_TTS_TOKEN_CACHE = { "access_token": None, "expires_at": 0, } def _get_baidu_tts_access_token(): """ 获取百度语音合成 access_token，简单内存缓存，避免每次都去请求。 """ from backend.config import get_settings as _get_settings settings = _get_settings() api_key = settings.BAIDU_TTS_API_KEY secret_key = settings.BAIDU_TTS_SECRET_KEY if not api_key or not secret_key: raise RuntimeError("百度语音合成 API Key/Secret 未配置") now = time.time() if ( _BAIDU_TTS_TOKEN_CACHE["access_token"] and now < _BAIDU_TTS_TOKEN_CACHE["expires_at"] - 60 ): return _BAIDU_TTS_TOKEN_CACHE["access_token"] token_url = "https://aip.baidubce.com/oauth/2.0/token" resp = requests.get( token_url, params={ "grant_type": "client_credentials", "client_id": api_key, "client_secret": secret_key, }, timeout=10, ) data = resp.json() access_token = data.get("access_token") if not access_token: raise RuntimeError(f"获取百度 TTS access_token 失败: {data}") expires_in = int(data.get("expires_in", 2592000)) _BAIDU_TTS_TOKEN_CACHE["access_token"] = access_token _BAIDU_TTS_TOKEN_CACHE["expires_at"] = now + expires_in return access_token # --- 定义数据模型 --- class User(db.Model): """用户表，对应初中化学教学系统中的学生/教师账号""" __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) password_hash = db.Column(db.String(255), nullable=False) role = db.Column(db.String(20), default='student') # 个人信息字段（与 FastAPI 后端对齐，均为可选） full_name = db.Column(db.String(50), nullable=True) # 真实姓名 student_id = db.Column(db.String(50), nullable=True) # 学号 class_name = db.Column(db.String(100), nullable=True) # 班级，如“初二(3)班” grade = db.Column(db.String(20), nullable=True) # 年级，如“初一”“初二” email = db.Column(db.String(120), nullable=True) phone = db.Column(db.String(20), nullable=True) created_at = db.Column(db.DateTime, default=datetime.utcnow) def set_password(self, password): self.password_hash = generate_password_hash(password) def check_password(self, password): return check_password_hash(self.password_hash, password) class Question(db.Model): """化学每日训练题库""" __tablename__ = 'questions' id = db.Column(db.Integer, primary_key=True) content = db.Column(db.Text, nullable=False) # 题干 option_a = db.Column(db.String(255), nullable=False) option_b = db.Column(db.String(255), nullable=False) option_c = db.Column(db.String(255), nullable=False) option_d = db.Column(db.String(255), nullable=False) correct_option = db.Column(db.String(1), nullable=False) # A/B/C/D knowledge_point = db.Column(db.String(100), nullable=True) # 所属知识点，如“电解质” difficulty = db.Column(db.Integer, default=1) # 1-5 难度 created_at = db.Column(db.DateTime, default=datetime.utcnow) class AnswerRecord(db.Model): """答题记录，用于学情分析与错题集""" __tablename__ = 'answer_records' id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False) question_id = db.Column(db.Integer, db.ForeignKey('questions.id'), nullable=False) selected_option = db.Column(db.String(1), nullable=False) is_correct = db.Column(db.Boolean, default=False) answered_at = db.Column(db.DateTime, default=datetime.utcnow) user = db.relationship('User', backref=db.backref('answer_records', lazy='dynamic')) question = db.relationship('Question', backref=db.backref('answer_records', lazy='dynamic')) class KnowledgePoint(db.Model): """知识图谱：知识点实体（初中化学）""" __tablename__ = 'knowledge_points' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(100), unique=True, nullable=False) # 如：电解质 description = db.Column(db.Text, nullable=True) # 知识点描述 chapter = db.Column(db.String(80), nullable=True) # 所属章节 created_at = db.Column(db.DateTime, default=datetime.utcnow) class KnowledgeRelation(db.Model): """知识图谱：三元组关系（主体-谓词-客体）""" __tablename__ = 'knowledge_relations' id = db.Column(db.Integer, primary_key=True) subject = db.Column(db.String(120), nullable=False) predicate = db.Column(db.String(80), nullable=False) obj = db.Column(db.String(120), nullable=False) created_at = db.Column(db.DateTime, default=datetime.utcnow) class CurriculumTopic(db.Model): """课程主题/知识主题，如：物质分类、原子结构等""" __tablename__ = 'curriculum_topics' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(50), unique=True, nullable=False) phase = db.Column(db.String(20), default='初中') # 学段：初中/高中等 chapter = db.Column(db.String(80), nullable=True) # 教材章节 description = db.Column(db.Text, nullable=True) class ChemistryQuestionBank(db.Model): """多题型统一题库，支撑 AI 数字人教学""" __tablename__ = 'chemistry_question_bank' question_id = db.Column(db.Integer, primary_key=True) type = db.Column(db.Enum('choice', 'equation', 'experiment', 'calculation', 'inference', 'diagram', name='question_type'), nullable=False) difficulty = db.Column(db.Enum('easy', 'medium', 'hard', name='question_difficulty'), nullable=False, default='easy') topic = db.Column( db.String(50), db.ForeignKey('curriculum_topics.name'), nullable=True, comment='如：物质分类、酸碱盐、金属等' ) unit_no = db.Column( db.Integer, nullable=True, comment='教材单元编号，如 1-12' ) knowledge_tags = db.Column( db.String(255), nullable=True, comment='知识点标签，逗号分隔（与 schema.sql 一致）', ) content = db.Column(db.Text, nullable=False, comment='题干，可包含 HTML/Markdown') answer_schema = db.Column( db.JSON, nullable=False, comment='结构化答案：选项/方程式/步骤/推断链等' ) hint = db.Column(db.Text, nullable=True) created_at = db.Column(db.DateTime, default=datetime.utcnow) topic_rel = db.relationship( 'CurriculumTopic', primaryjoin='ChemistryQuestionBank.topic == foreign(CurriculumTopic.name)', # 注意：此处不能使用 lazy='dynamic'，因为 SQLAlchemy 会将该关系 # 视为一对一 / many-to-one（外键指向唯一列），dynamic 只支持一对多集合关系。 # 使用默认的 select 加载方式即可。 backref=db.backref('questions', lazy='select') ) with app.app_context(): # 确保定义的表在数据库中存在（需要先手动创建 ai_system 数据库） try: db.create_all() # 创建 backend 模型对应的表（qa_sessions、student_profiles 等）， # 即使只启动 Flask 不启动 FastAPI，这些表也会被自动创建 Base.metadata.create_all(bind=engine) # 补充 FastAPI 侧新增的用户个人信息字段，避免表结构不一致 try: ensure_user_profile_columns() except Exception as e: print(f"[WARN] ensure_user_profile_columns() failed: {e}") except Exception as e: # 开发期常见：MySQL 未启动/账号密码不对/库不存在 # 不阻塞 Web 服务启动，方便先联调前后端；具体接口用到 DB 时仍会报错并提示。 print(f"[WARN] db.create_all() failed: {e}") # --- 路由功能 --- @app.route('/') # 1：根路径直接重定向到登录页 def index(): return redirect(url_for('login')) @app.route('/login', methods=['GET', 'POST']) def login(): # ... (保持之前的登录逻辑不变) ... if request.method == 'POST': username = request.form['username'] password = request.form['password'] user = User.query.filter_by(username=username).first() if user and user.check_password(password): session['user_id'] = user.id session['username'] = user.username flash("登录成功！", "success") return redirect(url_for('dashboard')) # 登录成功后跳转到 dashboard else: flash("用户名或密码错误。", "error") return render_template('login.html') @app.route('/register', methods=['GET', 'POST']) def register(): # ... (保持之前的注册逻辑不变) ... if request.method == 'POST': username = request.form['username'] password = request.form['password'] if not username or not password: flash("用户名和密码不能为空", "error") return redirect(url_for('register')) if User.query.filter_by(username=username).first(): flash("用户名已存在，请更换一个。", "error") return redirect(url_for('register')) new_user = User(username=username) new_user.set_password(password) try: db.session.add(new_user) db.session.commit() flash("注册成功！请登录。", "success") return redirect(url_for('login')) except Exception as e: db.session.rollback() flash(f"发生错误: {e}", "error") return render_template('register.html') @app.route('/logout') def logout(): session.clear() flash("您已退出登录。", "info") return redirect(url_for('login')) # --- 主系统路由 (需要增加登录保护) --- # 定义一个装饰器，用于检查用户是否登录 from functools import wraps def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: flash("请先登录。", "info") return redirect(url_for('login')) return f(*args, **kwargs) return decorated_function @app.route('/dashboard') @login_required def dashboard(): return render_template( 'dashboard.html', active_page='home', questions=None, training_result=None, chart_data=[] ) @app.route('/ai_assistant') @login_required # 添加保护 def ai_assistant(): return redirect("http://localhost:8880") def _get_user_proficiency(user_id): """按知识点统计用户熟练度：正确题数/总题数""" from sqlalchemy import func, case stats = db.session.query( Question.knowledge_point, func.sum(case((AnswerRecord.is_correct == True, 1), else_=0)).label('correct'), func.count(AnswerRecord.id).label('total') ).join(AnswerRecord, Question.id == AnswerRecord.question_id).filter( AnswerRecord.user_id == user_id, Question.knowledge_point.isnot(None), Question.knowledge_point != '' ).group_by(Question.knowledge_point).all() return {s.knowledge_point: int(s.correct / s.total * 100) if s.total else 0 for s in stats} def _get_wrong_question_ids(user_id, limit=50): """获取用户错题集中的题目ID列表（去重，最近优先）""" from sqlalchemy import desc records = AnswerRecord.query.filter_by(user_id=user_id, is_correct=False)\ .order_by(desc(AnswerRecord.answered_at)).all() seen = set() ids = [] for r in records: if r.question_id not in seen: seen.add(r.question_id) ids.append(r.question_id) if len(ids) >= limit: break return ids def _get_weak_knowledge_points(user_id, top_n=3): """获取用户薄弱知识点（熟练度最低的）""" prof = _get_user_proficiency(user_id) if not prof: return [] sorted_prof = sorted(prof.items(), key=lambda x: x[1]) return [p[0] for p in sorted_prof[:top_n]] def _get_recommended_questions(user_id, count=5): """习题测评推荐算法：优先推荐错题相关、薄弱知识点题目""" from sqlalchemy.sql import func wrong_ids = _get_wrong_question_ids(user_id, limit=20) weak_points = _get_weak_knowledge_points(user_id, top_n=3) recommended = [] # 1. 优先从薄弱知识点补充 if wrong_ids: qs = Question.query.filter(Question.id.in_(wrong_ids)).order_by(func.rand()).limit(count).all() recommended = [q.id for q in qs] # 2. 不足时从错题补充 exclude_ids = list(set(recommended + wrong_ids)) if len(recommended) < count and weak_points: extra = Question.query.filter( Question.knowledge_point.in_(weak_points) ) if exclude_ids: extra = extra.filter(~Question.id.in_(exclude_ids)) extra = extra.order_by(func.rand()).limit(count - len(recommended)).all() recommended.extend([q.id for q in extra]) exclude_ids = list(set(recommended + wrong_ids)) # 3. 仍不足则随机补充 if len(recommended) < count: fallback = Question.query if exclude_ids: fallback = fallback.filter(~Question.id.in_(exclude_ids)) fallback = fallback.order_by(func.rand()).limit(count - len(recommended)).all() recommended.extend([q.id for q in fallback]) if not recommended: return [] return Question.query.filter(Question.id.in_(recommended[:count])).all() @app.route('/daily_training', methods=['GET', 'POST']) @login_required def daily_training(): from sqlalchemy.sql import func if request.method == 'POST': question_ids = request.form.getlist('question_ids') correct_count = 0 total = 0 for qid in question_ids: selected = request.form.get(f"q_{qid}") if not selected: continue question = Question.query.get(int(qid)) if not question: continue is_correct = (selected == question.correct_option) total += 1 if is_correct: correct_count += 1 record = AnswerRecord( user_id=session['user_id'], question_id=question.id, selected_option=selected, is_correct=is_correct ) db.session.add(record) if total > 0: db.session.commit() accuracy = int(round(correct_count / total * 100)) if total else 0 training_result = {'total': total, 'correct': correct_count, 'accuracy': accuracy} # 知识点题目推荐：答题后按推荐算法抽取下一组 questions = _get_recommended_questions(session['user_id'], count=5) return render_template( 'dashboard.html', active_page='training', questions=questions, training_result=training_result, chart_data=[] ) # GET：优先使用推荐算法，无历史数据时随机出题 questions = _get_recommended_questions(session['user_id'], count=5) if not questions: questions = Question.query.order_by(func.rand()).limit(5).all() return render_template( 'dashboard.html', active_page='training', questions=questions, training_result=None, chart_data=[] ) @app.route('/wrong_questions') @login_required def wrong_questions(): """错题集：展示用户答错的题目，可重做巩固""" wrong_ids = _get_wrong_question_ids(session['user_id'], limit=50) questions = Question.query.filter(Question.id.in_(wrong_ids)).all() if wrong_ids else [] # 保持错题集中的题目顺序 id_order = {qid: i for i, qid in enumerate(wrong_ids)} questions = sorted(questions, key=lambda q: id_order.get(q.id, 999)) return render_template( 'dashboard.html', active_page='wrong_questions', questions=questions, training_result=None, chart_data=[] ) @app.route('/knowledge_qa') @login_required def knowledge_qa(): """知识点询问：大模型+知识图谱+数字人答疑""" return render_template('knowledge_qa.html') def _search_knowledge_graph(query_text, limit=15): """ 知识图谱检索（Flask 版）： 复用当前可视化模块使用的 Neo4j 结构和字段， 使用 app.py 中的 neo4j_driver，避免影响 backend.services 里的已有逻辑。 """ q = (query_text or "").strip() if not q: return [] triples = [] # 1. 优先使用 Neo4j：与 /api/v1/knowledge/graph 相同的一套节点展示名规则 try: with neo4j_driver.session() as session: part_s = "coalesce(" + ", ".join([ "s.name", "s.formula", "s.symbol", "s.description", "CASE WHEN s.text IS NOT NULL AND size(s.text) > 0 THEN substring(s.text, 0, 80) + '…' ELSE NULL END", "s.type", "s.category", "s.state", "CASE WHEN s.chunk IS NOT NULL AND size(toString(s.chunk)) > 0 THEN substring(toString(s.chunk), 0, 80) + '…' ELSE NULL END", ]) + ")" part_o = "coalesce(" + ", ".join([ "o.name", "o.formula", "o.symbol", "o.description", "CASE WHEN o.text IS NOT NULL AND size(o.text) > 0 THEN substring(o.text, 0, 80) + '…' ELSE NULL END", "o.type", "o.category", "o.state", "CASE WHEN o.chunk IS NOT NULL AND size(toString(o.chunk)) > 0 THEN substring(toString(o.chunk), 0, 80) + '…' ELSE NULL END", ]) + ")" cypher = f""" MATCH (s)-[r]->(o) WITH s, r, o, {part_s} AS subjDisplay, {part_o} AS objDisplay WHERE subjDisplay IS NOT NULL AND objDisplay IS NOT NULL WITH coalesce(toString(subjDisplay), '') AS subject, coalesce(r.name, toString(type(r))) AS predicate, coalesce(toString(objDisplay), '') AS obj WHERE subject <> '' AND obj <> '' AND ( subject CONTAINS $q OR obj CONTAINS $q OR predicate CONTAINS $q ) RETURN DISTINCT subject, predicate, obj LIMIT $limit """ records = session.run(cypher, q=q, limit=limit).data() for rec in records: subject = (rec.get("subject") or "").strip() obj = (rec.get("obj") or "").strip() predicate = (rec.get("predicate") or "关联").strip() if not subject or not obj: continue triples.append({"subject": subject, "predicate": predicate, "obj": obj}) if triples: return triples except Exception as e: print(f"[WARN] Neo4j _search_knowledge_graph failed: {e}") # 2. 回退到原有 MySQL 知识图谱表（保持兼容） rels = KnowledgeRelation.query.filter( db.or_( KnowledgeRelation.subject.contains(q), KnowledgeRelation.predicate.contains(q), KnowledgeRelation.obj.contains(q), ) ).limit(limit).all() if rels: for r in rels: triples.append( {"subject": r.subject, "predicate": r.predicate, "obj": r.obj} ) return triples # 3. 若无直接匹配，再按知识点名称检索 MySQL 中的关系 kps = KnowledgePoint.query.filter(KnowledgePoint.name.contains(q)).limit(5).all() for kp in kps: rels = ( KnowledgeRelation.query.filter( db.or_( KnowledgeRelation.subject.contains(kp.name), KnowledgeRelation.obj.contains(kp.name), ) ) .limit(3) .all() ) for r in rels: triples.append( {"subject": r.subject, "predicate": r.predicate, "obj": r.obj} ) if len(triples) >= limit: break return triples[:limit] def _expand_with_llm(query_text, kg_context, meta=None): """大模型扩充回答：复用 backend.services.llm_service 的 RAG + DeepSeek 逻辑""" # kg_context 现在优先是 Neo4j/SQL 三元组 dict 列表 triples = [] for r in kg_context or []: if isinstance(r, dict): triples.append( { "subject": r.get("subject", ""), "predicate": r.get("predicate", ""), "obj": r.get("obj", ""), } ) else: # 兼容旧的 SQLAlchemy 模型对象 triples.append( { "subject": getattr(r, "subject", ""), "predicate": getattr(r, "predicate", ""), "obj": getattr(r, "obj", ""), } ) if meta is None: try: meta = analyze_question(query_text) except Exception: meta = None # 优先调用统一的大模型服务（DeepSeek / Qwen 等） try: return expand_with_llm(query_text, triples, meta=meta) except Exception as e: print(f"[WARN] expand_with_llm fallback due to error: {e}") # 若大模型不可用，退回到本地兜底逻辑 context_str = "\n".join( [f"{t['subject']} {t['predicate']} {t['obj']}" for t in triples] ) if context_str: return ( "根据初中化学知识图谱，相关内容如下：\n\n" f"{context_str}\n\n" "建议结合教材和习题进一步巩固理解。" ) return ( f"暂无与「{query_text}」直接相关的图谱条目，" "建议从教材相关章节或习题中查找，或尝试更换关键词。" ) @app.route('/api/v1/tts/baidu', methods=['POST']) @login_required def api_v1_tts_baidu(): """ 百度语音合成接口：接收文本，返回 base64 编码的 MP3 音频数据。 前端调用示例： POST /api/v1/tts/baidu { "text": "你好，同学", "spd": 5, "pit": 5, "vol": 5, "per": 5118 } """ data = request.get_json(silent=True) or {} text = (data.get('text') or '').strip() if not text: return {'detail': 'text 不能为空'}, 400 # 百度 TTS 对 tex 参数长度有严格限制（按字节计数）， # 日志中出现 "Normal invalid text length" / "tex param err" 即代表超出范围。 # 这里做一次兜底截断，避免长回答导致整个 TTS 调用失败。 # 官方常见限制为 1024 字节，这里按 1024 字节做保守截断。 max_tex_bytes = 1024 text_bytes = text.encode('utf-8') if len(text_bytes) > max_tex_bytes: cut = text_bytes[:max_tex_bytes] # 避免截断在 UTF-8 多字节字符的中间，向前回退到合法边界 while cut and (cut[-1] & 0xC0) == 0x80: cut = cut[:-1] text = cut.decode('utf-8', errors='ignore') # 语速、音调、音量、发音人可选；使用百度默认区间 0-15 spd = int(data.get('spd', 5)) pit = int(data.get('pit', 5)) vol = int(data.get('vol', 5)) per = int(data.get('per', 5118)) # 5118 为度小萌，中文女声，比较自然 try: token = _get_baidu_tts_access_token() except Exception as e: return {'detail': f'获取百度 TTS token 失败: {e}'}, 500 settings = get_settings() cuid = settings.BAIDU_TTS_APP_ID or 'ai-teaching-system' tts_url = "https://tsn.baidu.com/text2audio" try: resp = requests.post( tts_url, data={ "tex": text, "tok": token, "cuid": cuid, "ctp": 1, "lan": "zh", "spd": spd, "pit": pit, "vol": vol, "per": per, }, headers={"Content-Type": "application/x-www-form-urlencoded"}, timeout=15, ) except Exception as e: return {'detail': f'调用百度 TTS 接口失败: {e}'}, 502 content_type = resp.headers.get("Content-Type", "") if content_type.startswith("application/json") or content_type.startswith("text/"): # 百度错误时会返回 JSON 文本 try: err_data = resp.json() except Exception: err_data = {"raw": resp.text} return {'detail': f'百度 TTS 返回错误: {err_data}'}, 502 audio_bytes = resp.content audio_base64 = base64.b64encode(audio_bytes).decode('ascii') return jsonify({ "audio_base64": audio_base64, "format": "mp3", }) @app.route('/knowledge_graph') @login_required def knowledge_graph(): """知识图谱可视化""" rels = KnowledgeRelation.query.limit(50).all() nodes, links = set(), [] for r in rels: nodes.add(r.subject) nodes.add(r.obj) links.append({'source': r.subject, 'target': r.obj, 'value': r.predicate}) graph_data = { 'nodes': [{'name': n} for n in nodes], 'links': links } return render_template('knowledge_graph.html', graph_data=graph_data) @app.route('/api/ask', methods=['POST']) @login_required def api_ask(): """智能答疑 API：意图识别 -> 图谱检索 -> 大模型扩充 -> 返回文本（供数字人展示）""" data = request.get_json() or {} question = (data.get('question') or '').strip() if not question: return {'ok': False, 'error': '请输入问题'} # 1. 图谱检索 rels = _search_knowledge_graph(question) # 2. 大模型扩充（可替换为真实 API 调用） answer = _expand_with_llm(question, rels) return {'ok': True, 'answer': answer} @app.route('/analysis') @login_required def analysis(): """学情分析：基于答题记录生成知识点掌握情况""" user_id = session['user_id'] proficiency = _get_user_proficiency(user_id) chart_data = [{'name': k, 'value': v} for k, v in proficiency.items()] if proficiency else [] return render_template( 'dashboard.html', active_page='analysis', questions=None, training_result=None, chart_data=chart_data or [] ) # ----------------------------- # API v1: 给前端（Vite/React）调用的 JSON 接口 # ----------------------------- def _api_require_login(): user_id = session.get('user_id') if not user_id: return None, ({'ok': False, 'error': '未登录'}, 401) return int(user_id), None @app.route('/api/v1/auth/login', methods=['POST']) def api_v1_login(): """ 前端登录接口：POST JSON { username, password } 成功后写入 session，并返回前端期望的结构： { access_token, user_id, username, role } """ data = request.get_json(silent=True) or {} username = (data.get('username') or '').strip() password = (data.get('password') or '').strip() if not username or not password: return {'detail': '用户名或密码不能为空'}, 400 user = User.query.filter_by(username=username).first() if not user or not user.check_password(password): return {'detail': '用户名或密码错误'}, 401 session['user_id'] = user.id session['username'] = user.username # 这里 access_token 只是占位，当前后端通过 Session 识别用户 return { 'access_token': 'session', 'user_id': user.id, 'username': user.username, 'role': user.role, } @app.route('/api/v1/auth/register', methods=['POST']) def api_v1_register(): """ 前端注册接口：POST JSON { username, password, role? } 创建用户并直接登录，返回 { access_token, user_id, username, role } """ data = request.get_json(silent=True) or {} username = (data.get('username') or '').strip() password = (data.get('password') or '').strip() role = (data.get('role') or 'student').strip() or 'student' if not username or not password: return {'detail': '用户名和密码不能为空'}, 400 if User.query.filter_by(username=username).first(): return {'detail': '用户名已存在'}, 400 user = User(username=username, role=role) user.set_password(password) db.session.add(user) db.session.commit() session['user_id'] = user.id session['username'] = user.username return { 'access_token': 'session', 'user_id': user.id, 'username': user.username, 'role': user.role, }, 201 @app.route('/api/v1/auth/logout', methods=['POST']) def api_v1_logout(): session.clear() return {'ok': True} @app.route('/api/v1/auth/me', methods=['GET']) def api_v1_me(): user_id, err = _api_require_login() if err: return err user = User.query.get(user_id) if not user: session.clear() return {'detail': '用户不存在'}, 401 return {'user_id': user.id, 'username': user.username, 'role': user.role} @app.route('/api/v1/users/me', methods=['GET']) def api_v1_users_me_get(): """ 个人信息页：获取当前登录用户的详细资料。 返回字段与 FastAPI /api/v1/users/me 保持一致，便于前端统一使用。 """ user_id, err = _api_require_login() if err: return err user = User.query.get(user_id) if not user: session.clear() return {'detail': '用户不存在'}, 401 return { 'id': user.id, 'username': user.username, 'role': user.role, 'full_name': user.full_name, 'student_id': user.student_id, 'class_name': user.class_name, 'grade': user.grade, 'email': user.email, 'phone': user.phone, 'created_at': user.created_at.isoformat() if user.created_at else None, } @app.route('/api/v1/users/me', methods=['PUT']) def api_v1_users_me_update(): """ 个人信息页：更新当前登录用户的个人资料。 接收可选字段：full_name, student_id, class_name, grade, email, phone。 """ user_id, err = _api_require_login() if err: return err user = User.query.get(user_id) if not user: session.clear() return {'detail': '用户不存在'}, 401 data = request.get_json(silent=True) or {} updatable_fields = ['full_name', 'student_id', 'class_name', 'grade', 'email', 'phone'] for field in updatable_fields: if field in data: value = data.get(field) if isinstance(value, str): value = value.strip() setattr(user, field, value) try: db.session.add(user) db.session.commit() except Exception as e: db.session.rollback() return {'detail': f'更新失败: {e}'}, 500 return { 'id': user.id, 'username': user.username, 'role': user.role, 'full_name': user.full_name, 'student_id': user.student_id, 'class_name': user.class_name, 'grade': user.grade, 'email': user.email, 'phone': user.phone, 'created_at': user.created_at.isoformat() if user.created_at else None, } @app.route('/api/v1/recommendation/proficiency', methods=['GET']) def api_v1_proficiency(): user_id, err = _api_require_login() if err: return err proficiency = _get_user_proficiency(user_id) # 前端直接期望拿到 { 知识点: 正确率 } 这样的字典 return proficiency or {} @app.route('/api/v1/recommendation/wrong_questions', methods=['GET']) def api_v1_wrong_questions(): user_id, err = _api_require_login() if err: return err wrong_ids = _get_wrong_question_ids(user_id, limit=50) if not wrong_ids: return [] questions = Question.query.filter(Question.id.in_(wrong_ids)).all() id_order = {qid: i for i, qid in enumerate(wrong_ids)} questions = sorted(questions, key=lambda q: id_order.get(q.id, 999)) return [ { 'id': q.id, 'content': q.content, 'option_a': q.option_a, 'option_b': q.option_b, 'option_c': q.option_c, 'option_d': q.option_d, } for q in questions ] @app.route('/api/v1/recommendation/recommended_questions', methods=['GET']) def api_v1_recommended_questions(): user_id, err = _api_require_login() if err: return err qs = _get_recommended_questions(user_id, count=5) or [] return [ { 'id': q.id, 'content': q.content, 'option_a': q.option_a, 'option_b': q.option_b, 'option_c': q.option_c, 'option_d': q.option_d, } for q in qs ] @app.route('/api/v1/recommendation/daily_practice', methods=['GET']) def api_v1_daily_practice(): """每日一练：按弱点权重生成今日推荐题，若今日已有则直接返回""" user_id, err = _api_require_login() if err: return err count = request.args.get('count', 10, type=int) try: qs = get_or_generate_daily_recommendations(db.session, user_id, count=count) except Exception as e: print(f"[WARN] get_or_generate_daily_recommendations failed: {e}") qs = _get_recommended_questions(user_id, count=count) or [] return [ { 'id': q.id, 'content': q.content, 'option_a': q.option_a, 'option_b': q.option_b, 'option_c': q.option_c, 'option_d': q.option_d, 'knowledge_point': getattr(q, 'knowledge_point', None), } for q in qs ] @app.route('/api/v1/recommendation/mastery_matrix', methods=['GET']) def api_v1_mastery_matrix(): """知识点掌握度矩阵：用于前端颜色分级、雷达图等可视化""" user_id, err = _api_require_login() if err: return err try: return get_mastery_matrix(db.session, user_id) except Exception as e: print(f"[WARN] get_mastery_matrix failed: {e}") return [] @app.route('/api/v1/recommendation/motivation_summary', methods=['GET']) def api_v1_motivation_summary(): """激励信息汇总：连续打卡天数、全绿模块、章节平均掌握度""" user_id, err = _api_require_login() if err: return err try: return get_motivation_summary(db.session, user_id) except Exception as e: print(f"[WARN] get_motivation_summary failed: {e}") return {} @app.route('/api/v1/recommendation/question_analysis', methods=['GET']) def api_v1_question_analysis(): """题目解析：从 chemistry_question_bank.analysis 读取。""" user_id, err = _api_require_login() if err: return err _ = user_id question_id = request.args.get('question_id', type=int) if not question_id: return {'detail': 'question_id 不能为空'}, 400 try: analysis = get_question_analysis(db.session, question_id) return {'question_id': question_id, 'analysis': analysis} except Exception as e: print(f"[WARN] get_question_analysis failed: {e}") return {'question_id': question_id, 'analysis': '暂无解析'} @app.route('/api/v1/recommendation/diagnostic_questions', methods=['GET']) def api_v1_diagnostic_questions(): """诊断测评：覆盖广、难度适中（用于弱点更新）""" user_id, err = _api_require_login() if err: return err count = request.args.get('count', 10, type=int) try: qs = get_diagnostic_questions(db.session, user_id, count=count, session_kind='diagnostic') except Exception as e: print(f"[WARN] get_diagnostic_questions(diagnostic) failed: {e}") qs = [] return [ { 'id': q.id, 'content': q.content, 'option_a': q.option_a, 'option_b': q.option_b, 'option_c': q.option_c, 'option_d': q.option_d, 'knowledge_point': getattr(q, 'knowledge_point', None), } for q in qs ] @app.route('/api/v1/recommendation/re_evaluate_questions', methods=['GET']) def api_v1_re_evaluate_questions(): """再测评：在测评后/训练后重新抽取题目""" user_id, err = _api_require_login() if err: return err count = request.args.get('count', 10, type=int) try: qs = get_diagnostic_questions(db.session, user_id, count=count, session_kind='re_evaluate') except Exception as e: print(f"[WARN] get_diagnostic_questions(re_evaluate) failed: {e}") qs = [] return [ { 'id': q.id, 'content': q.content, 'option_a': q.option_a, 'option_b': q.option_b, 'option_c': q.option_c, 'option_d': q.option_d, 'knowledge_point': getattr(q, 'knowledge_point', None), } for q in qs ] def _api_submit_diagnostic_like(session_kind: str = "diagnostic"): """ 诊断/再测评批量提交： - 写入 answer_records - 更新掌握度 - 生成（或复用）每日一练 """ user_id, err = _api_require_login() if err: return err data = request.get_json(silent=True) or {} answers = data.get('answers') if not isinstance(answers, list): return {'detail': '参数格式错误'}, 400 train_count = data.get('train_count', 10) regenerate_daily = bool(data.get('regenerate_daily', False)) include_diagnostic_in_mastery = bool(data.get('include_diagnostic_in_mastery', False)) correct_count = 0 total = 0 processed_answers = [] for item in answers: try: qid = int(item.get('question_id')) except (TypeError, ValueError): continue selected = (item.get('selected_option') or '').strip() if not selected: continue question = Question.query.get(qid) if not question: continue is_correct = (selected == question.correct_option) total += 1 if is_correct: correct_count += 1 db.session.add(AnswerRecord( user_id=user_id, question_id=question.id, selected_option=selected, is_correct=is_correct, )) processed_answers.append({"question_id": qid, "selected_option": selected}) if total == 0: return {'ok': False, 'detail': '未提交有效答题'}, 400 db.session.commit() try: update_mastery_from_answers( db.session, user_id, processed_answers, session_kind=session_kind, include_diagnostic_in_mastery=include_diagnostic_in_mastery, ) db.session.commit() except Exception as e: print(f"[WARN] update_mastery_from_answers in diagnostic submit failed: {e}") if regenerate_daily: try: from backend.models import DailyRecommendation daily_query = db.session.query(DailyRecommendation).filter( DailyRecommendation.user_id == user_id, DailyRecommendation.recommend_date == date.today(), ) if session_kind == "diagnostic": # 诊断提交后按最新测评结果重算“今日10题”，清空今日旧推荐（含已完成）。 daily_query.delete(synchronize_session=False) else: # 再测评保留已完成记录，仅替换未完成推荐，避免影响已完成闭环。 daily_query.filter(DailyRecommendation.is_done == 0).delete(synchronize_session=False) db.session.commit() except Exception as e: print(f"[WARN] regenerate_daily delete failed: {e}") try: daily_qs = get_or_generate_daily_recommendations(db.session, user_id, count=int(train_count)) except Exception as e: print(f"[WARN] get_or_generate_daily_recommendations in diagnostic submit failed: {e}") daily_qs = [] accuracy = int(round(correct_count / total * 100)) if total else 0 return { 'ok': True, 'diagnostic': {'total': total, 'correct': correct_count, 'accuracy': accuracy}, 'daily_practice': [ { 'id': q.id, 'content': q.content, 'option_a': q.option_a, 'option_b': q.option_b, 'option_c': q.option_c, 'option_d': q.option_d, 'knowledge_point': getattr(q, 'knowledge_point', None), } for q in daily_qs ], } @app.route('/api/v1/recommendation/diagnostic_submit_batch', methods=['POST']) def api_v1_diagnostic_submit_batch(): return _api_submit_diagnostic_like(session_kind="diagnostic") @app.route('/api/v1/recommendation/re_evaluate_submit_batch', methods=['POST']) def api_v1_re_evaluate_submit_batch(): return _api_submit_diagnostic_like(session_kind="re_evaluate") @app.route('/api/v1/teacher/class_report', methods=['GET']) def api_v1_teacher_class_report(): """ 教师端班级学情报表： - 仅教师可访问 - 前端对齐路径：/api/v1/teacher/class_report """ user_id, err = _api_require_login() if err: return err user = User.query.get(user_id) if not user: session.clear() return {'detail': '用户不存在'}, 401 if user.role != 'teacher': return {'detail': '仅教师可访问该接口'}, 403 days = request.args.get('days', 14, type=int) weak_top = request.args.get('weak_top', 5, type=int) wrong_top = request.args.get('wrong_top', 10, type=int) try: return get_teacher_class_report( db.session, teacher_id=user_id, days=days, weak_top=weak_top, wrong_top=wrong_top, ) except ValueError as e: return {'detail': str(e)}, 400 except Exception as e: print(f"[WARN] get_teacher_class_report failed: {e}") return {'detail': '获取教师班级报表失败'}, 500 @app.route('/api/v1/recommendation/submit_batch', methods=['POST']) def api_v1_submit_batch(): """ 批量提交答题结果：计算本次正确率并记录到 answer_records 请求体（前端目前会多包一层 answers）： { "answers": [ { "question_id": 1, "selected_option": "A" }, ... ] } 或 { "answers": { "answers": [ ... ] } } """ user_id, err = _api_require_login() if err: return err data = request.get_json(silent=True) or {} answers = data.get('answers') # 兼容前端传入 {answers: {answers: [...]}} if isinstance(answers, dict) and 'answers' in answers: answers = answers['answers'] if not isinstance(answers, list): return {'detail': '参数格式错误'}, 400 correct_count = 0 total = 0 processed_answers = [] for item in answers: try: qid = int(item.get('question_id')) except (TypeError, ValueError): continue selected = (item.get('selected_option') or '').strip() if not selected: continue question = Question.query.get(qid) if not question: continue is_correct = (selected == question.correct_option) total += 1 if is_correct: correct_count += 1 db.session.add(AnswerRecord( user_id=user_id, question_id=question.id, selected_option=selected, is_correct=is_correct, )) processed_answers.append({"question_id": qid, "selected_option": selected}) if total > 0: try: update_mastery_from_answers( db.session, user_id, processed_answers, session_kind="practice" ) mark_daily_recommendations_done( db.session, user_id, [a["question_id"] for a in processed_answers], today=date.today(), ) except Exception as e: print(f"[WARN] update_mastery_from_answers failed: {e}") db.session.commit() accuracy = int(round(correct_count / total * 100)) if total else 0 return {'total': total, 'correct': correct_count, 'accuracy': accuracy} # 节点展示名仅用：name/formula/symbol/description/text/type/category/state/chunk，不用 id/标签 # 视为“无意义”的展示名（纯标签或占位），不展示 _BAD_DISPLAY_NAMES = frozenset({ "element", "chunk", "document", "__entity__", "nodea", "nodeb", "entity", "doc", "kgbuilder", "__kgbuilder__" }) @app.route('/api/v1/knowledge/graph', methods=['GET']) def api_v1_knowledge_graph(): """知识图谱 JSON 版本，优先从 Neo4j 读取，失败时回退到内置 MySQL 表。 展示名严格使用 formula/description/symbol/name/text/type/category/state/chunk，不显示 id 或标签。""" nodes, links = set(), [] limit = request.args.get("limit", 2000, type=int) limit = min(max(limit, 100), 10000) # 1）优先从 Neo4j 查询 try: with neo4j_driver.session() as session: # 展示名仅从具体属性取：name, formula, symbol, description, text, type, category, state, chunk（长文本截断） # 不用 id、不用 labels，避免出现 elementId 或 "Element"/"Chunk" part_s = "coalesce(" + ", ".join([ "s.name", "s.formula", "s.symbol", "s.description", "CASE WHEN s.text IS NOT NULL AND size(s.text) > 0 THEN substring(s.text, 0, 80) + '…' ELSE NULL END", "s.type", "s.category", "s.state", "CASE WHEN s.chunk IS NOT NULL AND size(toString(s.chunk)) > 0 THEN substring(toString(s.chunk), 0, 80) + '…' ELSE NULL END", ]) + ")" part_o = "coalesce(" + ", ".join([ "o.name", "o.formula", "o.symbol", "o.description", "CASE WHEN o.text IS NOT NULL AND size(o.text) > 0 THEN substring(o.text, 0, 80) + '…' ELSE NULL END", "o.type", "o.category", "o.state", "CASE WHEN o.chunk IS NOT NULL AND size(toString(o.chunk)) > 0 THEN substring(toString(o.chunk), 0, 80) + '…' ELSE NULL END", ]) + ")" cypher = f""" MATCH (s)-[r]->(o) WITH s, r, o, {part_s} AS subjDisplay, {part_o} AS objDisplay WHERE subjDisplay IS NOT NULL AND objDisplay IS NOT NULL WITH coalesce(toString(subjDisplay), '') AS subject, coalesce(r.name, toString(type(r))) AS predicate, coalesce(toString(objDisplay), '') AS obj WHERE subject <> '' AND obj <> '' RETURN DISTINCT subject, predicate, obj LIMIT $limit """ records = session.run(cypher, limit=limit).data() for rec in records: subject = (rec.get("subject") or "").strip() obj = (rec.get("obj") or "").strip() predicate = (rec.get("predicate") or "关联").strip() if not subject or not obj: continue # 排除像 elementId 的字符串（含 : 且含 - 且较长） def looks_like_id(s): if len(s) < 20: return False return ":" in s and "-" in s if looks_like_id(subject) or looks_like_id(obj): continue # 排除纯标签/占位名 low_subj, low_obj = subject.lower(), obj.lower() if low_subj in _BAD_DISPLAY_NAMES or low_obj in _BAD_DISPLAY_NAMES: continue nodes.add(subject) nodes.add(obj) links.append({"source": subject, "target": obj, "value": predicate}) except Exception as e: # Neo4j 不可用时打印告警，继续使用 MySQL 数据 print(f"[WARN] Neo4j knowledge graph query failed: {e}") # 2）如果 Neo4j 没有查到有效关系，则回退到原来的 MySQL 表 if not links: rels = KnowledgeRelation.query.limit(50).all() for r in rels: nodes.add(r.subject) nodes.add(r.obj) links.append({"source": r.subject, "target": r.obj, "value": r.predicate}) return { "nodes": [{"name": n} for n in nodes], "links": links, } @app.route('/api/v1/qa/ask', methods=['POST']) def api_v1_qa_ask(): """知识点问答 JSON 接口，对齐前端 qa.ask""" data = request.get_json(silent=True) or {} question = (data.get('question') or '').strip() if not question: return {'detail': '请输入问题'}, 400 # 先进行问题解析，提取关键词与检索词 try: analysis = analyze_question(question) except Exception: analysis = {"clean_text": question, "kg_terms": [question]} kg_terms = analysis.get("kg_terms") or [analysis.get("clean_text") or question] # 基于解析得到的多个检索词进行图谱检索，并去重合并 triples = [] seen = set() for term in kg_terms: results = _search_knowledge_graph(term) for r in results: key = (r.get("subject"), r.get("predicate"), r.get("obj")) if key in seen: continue seen.add(key) triples.append(r) answer = _expand_with_llm(question, triples, meta=analysis) return {'answer': answer} if __name__ == '__main__': app.run(debug=True, port=5000)