import re from collections import Counter from datetime import datetime, timezone from io import BytesIO from html import escape from urllib.parse import quote from docx import Document from fastapi import APIRouter, Depends, HTTPException, Query from fastapi.responses import StreamingResponse from sqlalchemy.orm import Session from database import get_db from models.classroom import ClassroomActivity from models.user import User from schemas.classroom import ( ClassroomActivityCreate, ClassroomAnalysisOut, ClassroomActivityOut, ClassroomActivityUpdate, ClassroomResponseCreate, ) from services.auth import get_current_user router = APIRouter(prefix="/api/classroom", tags=["课堂工具"]) STOP_WORDS = { "这个", "我们", "老师", "学生", "因为", "所以", "可以", "需要", "没有", "已经", "the", "and", "that", "with", "this", "from", "have", "will", } def _format_answer(answer) -> str: if isinstance(answer, str): return answer.strip() if answer is None: return "" return str(answer) def _percent(count: int, total: int) -> int: return round((count / total) * 100) if total else 0 def _extract_keywords(text: str) -> list[dict]: words = re.findall(r"[\u4e00-\u9fa5]{2,}|[a-zA-Z0-9]{2,}", text) counter = Counter(word for word in words if word.lower() not in STOP_WORDS) return [{"keyword": word, "count": count} for word, count in counter.most_common(10)] def _infer_correct_answer(activity: ClassroomActivity) -> str: config = activity.config or {} for key in ("answer", "correct_answer", "correctAnswer"): value = config.get(key) if value: return str(value).strip() return "" def _build_mastery_level(activity_type: str, response_count: int, accuracy: int | None, option_stats: list[dict]) -> str: if response_count <= 0: return "暂无数据" if accuracy is not None: if accuracy >= 85: return "整体掌握较好" if accuracy >= 60: return "基础掌握，仍需巩固" return "薄弱点明显，建议重点讲评" if option_stats: top = max(option_stats, key=lambda item: item["count"]) if top["percent"] >= 70: return f"倾向集中：{top['option']}" if top["percent"] >= 45: return f"主要倾向：{top['option']}" return "分布分散，需要追问原因" if activity_type in {"qa", "feedback"}: return "已收集开放反馈" return "已收集课堂数据" def _build_diagnosis( activity: ClassroomActivity, response_count: int, participant_count: int, option_stats: list[dict], keywords: list[dict], accuracy: int | None, ) -> list[str]: if response_count <= 0: return ["尚未收到学生提交，建议先发布活动并提醒学生扫码参与。"] diagnosis = [f"本次共收到 {response_count} 条提交，覆盖 {participant_count} 名学生。"] if accuracy is not None: diagnosis.append(f"按正确答案统计，当前正确率为 {accuracy}%。") if option_stats: top = max(option_stats, key=lambda item: item["count"]) diagnosis.append(f"选择最多的是“{top['option']}”，占比 {top['percent']}%。") weak_options = [item for item in option_stats if item["count"] and item["percent"] >= 20 and item is not top] if weak_options: names = "、".join(item["option"] for item in weak_options[:3]) diagnosis.append(f"仍有较多学生选择“{names}”，建议追问选择原因。") if keywords: diagnosis.append("开放回答中高频词包括：" + "、".join(item["keyword"] for item in keywords[:5]) + "。") if activity.activity_type == "checkin": diagnosis.append("该活动更适合作为参与记录，可结合后续测验或反馈判断掌握情况。") return diagnosis def _build_teaching_suggestions( activity: ClassroomActivity, response_count: int, option_stats: list[dict], accuracy: int | None, keywords: list[dict], ) -> list[str]: if response_count <= 0: return ["先用投放链接收集至少 5 条反馈，再生成更可靠的讲评建议。"] suggestions = [] if accuracy is not None and accuracy < 60: suggestions.append("用 5 分钟回到概念源头，先讲清关键定义或公式来源，再做同类例题。") suggestions.append("安排 3 道低门槛变式题，确认学生能独立完成基本步骤。") elif accuracy is not None and accuracy < 85: suggestions.append("挑选一个典型错选项进行对比讲评，让学生说出错误思路。") suggestions.append("追加 2 到 3 道变式练习，覆盖易错条件和表达规范。") elif accuracy is not None: suggestions.append("整体掌握较好，可进入拓展任务或让学生互相解释解题依据。") elif option_stats: top = max(option_stats, key=lambda item: item["count"]) suggestions.append(f"围绕“{top['option']}”这个主流反馈追问原因，确认学生是理解到位还是凭感觉选择。") suggestions.append("让不同选择的学生各举一个理由，形成板书对比后再归纳共识。") else: suggestions.append("把高频回答聚类成 2 到 3 类，在下一轮讲评中分别回应。") if keywords: suggestions.append(f"优先处理“{keywords[0]['keyword']}”相关问题，并设计一个即时追问。") suggestions.append("课末再投放一次简短反馈，比较讲评前后的变化。") return suggestions[:5] def _build_followup_prompt(activity: ClassroomActivity, analysis: dict) -> str: question = analysis.get("question") or activity.title diagnosis = "；".join(analysis.get("diagnosis") or []) material_context = str((activity.config or {}).get("material_context") or "").strip() material_part = f"参考素材：{material_context}。" if material_context else "" return ( f"基于课堂数据回收《{activity.title}》生成一套二次巩固练习。" f"回收题目：{question}。" f"{material_part}" f"学情诊断：{diagnosis}。" "要求题目有梯度，覆盖高频错误点，附答案和解析。" ) def _safe_filename(name: str, suffix: str) -> str: stem = re.sub(r"[\\/:*?\"<>|\r\n]+", "_", name).strip(" .") or "课堂诊断报告" return f"{stem[:80]}{suffix}" def _excel_cell(value, cell_type: str = "String") -> str: if value is None: value = "" if isinstance(value, datetime): value = value.strftime("%Y-%m-%d %H:%M:%S") text = escape(str(value), quote=False) return f"{text}" def _excel_row(values: list) -> str: return "" + "".join(_excel_cell(value) for value in values) + "" def _excel_sheet(name: str, rows: list[list]) -> str: safe_name = escape(name[:31] or "Sheet", quote=True) body = "\n".join(_excel_row(row) for row in rows) return f"{body}
" def _build_activity_workbook(activity: ClassroomActivity, analysis: dict) -> bytes: responses = list(activity.responses or []) detail_rows = [ ["活动ID", "活动标题", "活动类型", "学生姓名", "回答", "提交时间", "来源"], *[ [ activity.id, activity.title, activity.activity_type, res.get("student_name") or "匿名", _format_answer(res.get("answer")), res.get("submitted_at") or "", (res.get("meta") or {}).get("source", ""), ] for res in responses ], ] stats_rows = [ ["指标", "数值"], ["提交数", analysis["response_count"]], ["参与学生数", analysis["participant_count"]], ["掌握水平", analysis["mastery_level"]], ["正确率", "" if analysis["accuracy"] is None else f"{analysis['accuracy']}%"], ["回收题目", analysis["question"]], ] if analysis["option_stats"]: stats_rows.extend([[], ["选项", "人数", "占比"]]) stats_rows.extend([[item["option"], item["count"], f"{item['percent']}%"] for item in analysis["option_stats"]]) if analysis["keywords"]: stats_rows.extend([[], ["关键词", "次数"]]) stats_rows.extend([[item["keyword"], item["count"]] for item in analysis["keywords"]]) diagnosis_rows = [ ["诊断结论"], *[[item] for item in analysis["diagnosis"]], [], ["讲评建议"], *[[item] for item in analysis["teaching_suggestions"]], [], ["二次练习提示词"], [analysis["followup_prompt"]], ] workbook = f""" {_excel_sheet("提交明细", detail_rows)} {_excel_sheet("统计汇总", stats_rows)} {_excel_sheet("诊断建议", diagnosis_rows)} """ return workbook.encode("utf-8") def _analysis_payload(activity: ClassroomActivity) -> dict: responses = list(activity.responses or []) response_count = len(responses) participants = { str(res.get("student_name") or "匿名").strip() or "匿名" for res in responses } participant_count = len(participants) answers = [_format_answer(res.get("answer")) for res in responses] answer_text = " ".join(answers) config = activity.config or {} options = [str(option).strip() for option in config.get("options", []) if str(option).strip()] option_stats = [] if options: counter = Counter(answers) option_stats = [ {"option": option, "count": counter.get(option, 0), "percent": _percent(counter.get(option, 0), response_count)} for option in options ] correct_answer = _infer_correct_answer(activity) accuracy = None if correct_answer and response_count: accuracy = _percent(sum(1 for answer in answers if answer == correct_answer), response_count) common_answers = [answer for answer, _ in Counter(answer for answer in answers if answer).most_common(6)] keywords = _extract_keywords(answer_text) mastery_level = _build_mastery_level(activity.activity_type, response_count, accuracy, option_stats) diagnosis = _build_diagnosis(activity, response_count, participant_count, option_stats, keywords, accuracy) teaching_suggestions = _build_teaching_suggestions(activity, response_count, option_stats, accuracy, keywords) payload = { "activity_id": activity.id, "title": activity.title, "activity_type": activity.activity_type, "question": config.get("question") or activity.description or activity.title, "response_count": response_count, "participant_count": participant_count, "option_stats": option_stats, "keywords": keywords, "common_answers": common_answers, "mastery_level": mastery_level, "accuracy": accuracy, "diagnosis": diagnosis, "teaching_suggestions": teaching_suggestions, "generated_at": datetime.now(timezone.utc), } payload["followup_prompt"] = _build_followup_prompt(activity, payload) return payload def _public_activity_payload(activity: ClassroomActivity) -> ClassroomActivity: config = dict(activity.config or {}) for key in ("answer", "correct_answer", "correctAnswer", "material_context"): config.pop(key, None) setattr(activity, "config", config) return activity def _add_bullets(document: Document, items: list[str]) -> None: for item in items: document.add_paragraph(str(item), style="List Bullet") @router.post("/activities", response_model=ClassroomActivityOut, status_code=201) def create_activity( data: ClassroomActivityCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = ClassroomActivity(user_id=current_user.id, **data.model_dump()) db.add(activity) db.commit() db.refresh(activity) return activity @router.get("/activities", response_model=list[ClassroomActivityOut]) def list_activities( activity_type: str = "", skip: int = Query(0, ge=0), limit: int = Query(20, ge=1, le=100), current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): query = db.query(ClassroomActivity).filter(ClassroomActivity.user_id == current_user.id) if activity_type: query = query.filter(ClassroomActivity.activity_type == activity_type) return query.order_by(ClassroomActivity.updated_at.desc()).offset(skip).limit(limit).all() @router.get("/activities/{activity_id}", response_model=ClassroomActivityOut) def get_activity( activity_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter( ClassroomActivity.id == activity_id, ClassroomActivity.user_id == current_user.id, ).first() if not activity: raise HTTPException(status_code=404, detail="课堂活动不存在") return activity @router.get("/activities/{activity_id}/analysis", response_model=ClassroomAnalysisOut) def analyze_activity( activity_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter( ClassroomActivity.id == activity_id, ClassroomActivity.user_id == current_user.id, ).first() if not activity: raise HTTPException(status_code=404, detail="课堂活动不存在") return _analysis_payload(activity) @router.get("/activities/{activity_id}/analysis/export") def export_activity_analysis( activity_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter( ClassroomActivity.id == activity_id, ClassroomActivity.user_id == current_user.id, ).first() if not activity: raise HTTPException(status_code=404, detail="课堂活动不存在") analysis = _analysis_payload(activity) document = Document() document.add_heading(f"课堂学情诊断报告：{activity.title}", level=1) document.add_paragraph(f"活动类型：{activity.activity_type}") document.add_paragraph(f"回收题目：{analysis['question']}") document.add_paragraph(f"生成时间：{analysis['generated_at'].strftime('%Y-%m-%d %H:%M:%S UTC')}") document.add_heading("核心指标", level=2) document.add_paragraph(f"提交数：{analysis['response_count']}") document.add_paragraph(f"参与学生数：{analysis['participant_count']}") document.add_paragraph(f"掌握水平：{analysis['mastery_level']}") if analysis["accuracy"] is not None: document.add_paragraph(f"正确率：{analysis['accuracy']}%") if analysis["option_stats"]: document.add_heading("选项分布", level=2) table = document.add_table(rows=1, cols=3) table.style = "Table Grid" hdr = table.rows[0].cells hdr[0].text = "选项" hdr[1].text = "人数" hdr[2].text = "占比" for item in analysis["option_stats"]: row = table.add_row().cells row[0].text = item["option"] row[1].text = str(item["count"]) row[2].text = f"{item['percent']}%" if analysis["keywords"]: document.add_heading("高频关键词", level=2) document.add_paragraph("、".join(f"{item['keyword']}({item['count']})" for item in analysis["keywords"])) if analysis["common_answers"]: document.add_heading("常见回答", level=2) _add_bullets(document, analysis["common_answers"]) document.add_heading("诊断结论", level=2) _add_bullets(document, analysis["diagnosis"]) document.add_heading("讲评建议", level=2) _add_bullets(document, analysis["teaching_suggestions"]) document.add_heading("二次练习提示词", level=2) document.add_paragraph(analysis["followup_prompt"]) buffer = BytesIO() document.save(buffer) buffer.seek(0) filename = _safe_filename(f"{activity.title}-学情诊断报告", ".docx") return StreamingResponse( buffer, media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(filename)}"}, ) @router.get("/activities/{activity_id}/responses/export") def export_activity_responses( activity_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter( ClassroomActivity.id == activity_id, ClassroomActivity.user_id == current_user.id, ).first() if not activity: raise HTTPException(status_code=404, detail="课堂活动不存在") analysis = _analysis_payload(activity) buffer = BytesIO(_build_activity_workbook(activity, analysis)) filename = _safe_filename(f"{activity.title}-课堂数据回收", ".xls") return StreamingResponse( buffer, media_type="application/vnd.ms-excel", headers={"Content-Disposition": f"attachment; filename*=UTF-8''{quote(filename)}"}, ) @router.get("/public/activities/{activity_id}", response_model=ClassroomActivityOut) def get_public_activity( activity_id: int, db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter(ClassroomActivity.id == activity_id).first() if not activity or activity.status not in {"active", "closed"}: raise HTTPException(status_code=404, detail="课堂活动不存在") return _public_activity_payload(activity) @router.put("/activities/{activity_id}", response_model=ClassroomActivityOut) def update_activity( activity_id: int, data: ClassroomActivityUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter( ClassroomActivity.id == activity_id, ClassroomActivity.user_id == current_user.id, ).first() if not activity: raise HTTPException(status_code=404, detail="课堂活动不存在") for key, value in data.model_dump(exclude_unset=True).items(): setattr(activity, key, value) db.commit() db.refresh(activity) return activity @router.post("/activities/{activity_id}/responses", response_model=dict, status_code=201) def submit_response( activity_id: int, data: ClassroomResponseCreate, db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter(ClassroomActivity.id == activity_id).first() if not activity or activity.status == "archived": raise HTTPException(status_code=404, detail="课堂活动不存在") if activity.status != "active": raise HTTPException(status_code=409, detail="课堂活动未发布或已结束") responses = list(activity.responses or []) responses.append({ "student_name": data.student_name, "answer": data.answer, "meta": data.meta, "submitted_at": datetime.now(timezone.utc).isoformat(), }) activity.responses = responses db.commit() return {"success": True, "count": len(responses)} @router.delete("/activities/{activity_id}", status_code=204) def delete_activity( activity_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): activity = db.query(ClassroomActivity).filter( ClassroomActivity.id == activity_id, ClassroomActivity.user_id == current_user.id, ).first() if not activity: raise HTTPException(status_code=404, detail="课堂活动不存在") db.delete(activity) db.commit()