jiahong
678 lines
22 KiB
Python
678 lines
22 KiB
Python
# 自然写教学数据分析与学情诊断系统软件 V1.0
|
|
# report/report_generator.py - 学情报告生成引擎
|
|
|
|
import logging
|
|
import json
|
|
import hashlib
|
|
from typing import Any, Dict, List, Optional
|
|
from datetime import datetime, date, timedelta
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
|
|
logger = logging.getLogger("writech.analytics.report")
|
|
|
|
|
|
# ============================================================
|
|
# 报告类型与模型
|
|
# ============================================================
|
|
|
|
class ReportType(str, Enum):
|
|
"""报告类型枚举"""
|
|
STUDENT_WEEKLY = "student_weekly" # 学生周报
|
|
STUDENT_MONTHLY = "student_monthly" # 学生月报
|
|
CLASS_WEEKLY = "class_weekly" # 班级周报
|
|
CLASS_MONTHLY = "class_monthly" # 班级月报
|
|
EXAM_ANALYSIS = "exam_analysis" # 考试分析报告
|
|
WRITING_GROWTH = "writing_growth" # 书写成长报告
|
|
PARENT_PUSH = "parent_push" # 家长推送报告
|
|
|
|
|
|
class ReportFormat(str, Enum):
|
|
"""报告输出格式"""
|
|
JSON = "json"
|
|
PDF = "pdf"
|
|
HTML = "html"
|
|
|
|
|
|
@dataclass
|
|
class ReportSection:
|
|
"""报告章节"""
|
|
title: str
|
|
section_type: str # summary/chart/table/text/recommendation
|
|
content: Dict[str, Any] = field(default_factory=dict)
|
|
order: int = 0
|
|
|
|
|
|
@dataclass
|
|
class ReportConfig:
|
|
"""报告生成配置"""
|
|
report_type: ReportType
|
|
target_id: str # 学生ID或班级ID
|
|
start_date: str
|
|
end_date: str
|
|
output_format: ReportFormat = ReportFormat.JSON
|
|
include_charts: bool = True
|
|
include_recommendations: bool = True
|
|
language: str = "zh-CN"
|
|
|
|
|
|
@dataclass
|
|
class GeneratedReport:
|
|
"""生成的报告"""
|
|
report_id: str
|
|
report_type: ReportType
|
|
target_id: str
|
|
title: str
|
|
period: str
|
|
sections: List[ReportSection]
|
|
summary: str = ""
|
|
generated_at: str = ""
|
|
file_path: Optional[str] = None
|
|
|
|
def to_json(self) -> Dict[str, Any]:
|
|
"""序列化为JSON"""
|
|
return {
|
|
"report_id": self.report_id,
|
|
"report_type": self.report_type.value,
|
|
"target_id": self.target_id,
|
|
"title": self.title,
|
|
"period": self.period,
|
|
"summary": self.summary,
|
|
"sections": [
|
|
{
|
|
"title": s.title,
|
|
"type": s.section_type,
|
|
"content": s.content,
|
|
"order": s.order,
|
|
}
|
|
for s in self.sections
|
|
],
|
|
"generated_at": self.generated_at,
|
|
"file_path": self.file_path,
|
|
}
|
|
|
|
|
|
# ============================================================
|
|
# 报告生成引擎
|
|
# ============================================================
|
|
|
|
class ReportGenerator:
|
|
"""
|
|
学情报告生成引擎
|
|
|
|
支持生成:
|
|
1. 学生周报/月报（个人学情概览+各科分析+书写能力+建议）
|
|
2. 班级周报/月报（班级统计+分数分布+薄弱知识点）
|
|
3. 考试分析报告（成绩分析+区分度+难度系数）
|
|
4. 书写成长报告（书写质量趋势+笔顺进步+对比）
|
|
5. 家长推送报告（简化版个人学情+学习建议）
|
|
|
|
输出格式: JSON / PDF / HTML
|
|
"""
|
|
|
|
def __init__(self, output_dir: str, template_dir: str):
|
|
"""初始化报告引擎"""
|
|
self.output_dir = output_dir
|
|
self.template_dir = template_dir
|
|
logger.info("报告引擎初始化: output=%s", output_dir)
|
|
|
|
async def generate_report(
|
|
self, config: ReportConfig
|
|
) -> GeneratedReport:
|
|
"""
|
|
根据配置生成报告
|
|
|
|
流程:
|
|
1. 从ClickHouse/MySQL查询原始数据
|
|
2. 调用对应报告类型的分析逻辑
|
|
3. 组装报告章节
|
|
4. 输出为指定格式
|
|
"""
|
|
logger.info(
|
|
"开始生成报告: type=%s, target=%s, period=%s~%s",
|
|
config.report_type.value,
|
|
config.target_id,
|
|
config.start_date,
|
|
config.end_date,
|
|
)
|
|
|
|
# 根据报告类型分发
|
|
generator_map = {
|
|
ReportType.STUDENT_WEEKLY: self._gen_student_report,
|
|
ReportType.STUDENT_MONTHLY: self._gen_student_report,
|
|
ReportType.CLASS_WEEKLY: self._gen_class_report,
|
|
ReportType.CLASS_MONTHLY: self._gen_class_report,
|
|
ReportType.EXAM_ANALYSIS: self._gen_exam_report,
|
|
ReportType.WRITING_GROWTH: self._gen_writing_report,
|
|
ReportType.PARENT_PUSH: self._gen_parent_report,
|
|
}
|
|
|
|
gen_func = generator_map.get(config.report_type)
|
|
if not gen_func:
|
|
raise ValueError(f"不支持的报告类型: {config.report_type}")
|
|
|
|
report = await gen_func(config)
|
|
|
|
# 输出为指定格式
|
|
if config.output_format == ReportFormat.PDF:
|
|
await self._export_pdf(report)
|
|
elif config.output_format == ReportFormat.HTML:
|
|
await self._export_html(report)
|
|
|
|
logger.info(
|
|
"报告生成完成: id=%s, title=%s",
|
|
report.report_id, report.title,
|
|
)
|
|
|
|
return report
|
|
|
|
async def _gen_student_report(
|
|
self, config: ReportConfig
|
|
) -> GeneratedReport:
|
|
"""
|
|
生成学生个人学情报告（周报/月报）
|
|
|
|
章节结构:
|
|
1. 总体概览（综合评分、排名、趋势）
|
|
2. 各科目分析（分数、掌握知识点、薄弱点）
|
|
3. 作业完成情况
|
|
4. 书写能力评估
|
|
5. 学习习惯分析
|
|
6. 个性化建议
|
|
"""
|
|
report_id = self._gen_report_id(config)
|
|
period_label = f"{config.start_date} ~ {config.end_date}"
|
|
is_weekly = config.report_type == ReportType.STUDENT_WEEKLY
|
|
|
|
sections: List[ReportSection] = []
|
|
|
|
# 第1节: 总体概览
|
|
# overview_data = await self._query_student_overview(
|
|
# config.target_id, config.start_date, config.end_date
|
|
# )
|
|
sections.append(ReportSection(
|
|
title="总体学情概览",
|
|
section_type="summary",
|
|
content={
|
|
"overall_score": 0,
|
|
"rank_in_class": 0,
|
|
"rank_change": 0, # 与上期对比排名变化
|
|
"trend": "stable",
|
|
"highlight": "", # 亮点描述
|
|
},
|
|
order=1,
|
|
))
|
|
|
|
# 第2节: 各科目分析
|
|
sections.append(ReportSection(
|
|
title="各科目学情分析",
|
|
section_type="chart",
|
|
content={
|
|
"chart_type": "radar", # 雷达图
|
|
"subjects": [], # [{name, score, class_avg, grade_avg}]
|
|
"detail": [], # 各科详细分析
|
|
},
|
|
order=2,
|
|
))
|
|
|
|
# 第3节: 作业完成情况
|
|
sections.append(ReportSection(
|
|
title="作业完成统计",
|
|
section_type="table",
|
|
content={
|
|
"total_homework": 0,
|
|
"completed": 0,
|
|
"on_time": 0,
|
|
"avg_score": 0,
|
|
"completion_rate": 0,
|
|
"detail_list": [], # 各科作业明细
|
|
},
|
|
order=3,
|
|
))
|
|
|
|
# 第4节: 书写能力评估
|
|
sections.append(ReportSection(
|
|
title="书写能力评估",
|
|
section_type="chart",
|
|
content={
|
|
"chart_type": "line", # 折线图展示趋势
|
|
"stroke_order_accuracy": 0,
|
|
"writing_quality": 0,
|
|
"writing_speed": 0,
|
|
"trend_data": [], # 时序数据点
|
|
"improvement": "",
|
|
},
|
|
order=4,
|
|
))
|
|
|
|
# 第5节: 学习习惯
|
|
sections.append(ReportSection(
|
|
title="学习习惯分析",
|
|
section_type="chart",
|
|
content={
|
|
"chart_type": "bar", # 柱状图展示每日时长
|
|
"avg_daily_minutes": 0,
|
|
"peak_hour": 0,
|
|
"weekly_pattern": [], # 周一~日时长
|
|
"consistency": 0,
|
|
},
|
|
order=5,
|
|
))
|
|
|
|
# 第6节: 个性化建议
|
|
if config.include_recommendations:
|
|
recommendations = self._generate_recommendations(
|
|
student_id=config.target_id,
|
|
sections=sections,
|
|
)
|
|
sections.append(ReportSection(
|
|
title="个性化学习建议",
|
|
section_type="recommendation",
|
|
content={
|
|
"recommendations": recommendations,
|
|
},
|
|
order=6,
|
|
))
|
|
|
|
# 生成摘要
|
|
summary = self._generate_summary(sections, "student")
|
|
|
|
return GeneratedReport(
|
|
report_id=report_id,
|
|
report_type=config.report_type,
|
|
target_id=config.target_id,
|
|
title=f"学生{'周' if is_weekly else '月'}学情报告",
|
|
period=period_label,
|
|
sections=sections,
|
|
summary=summary,
|
|
generated_at=datetime.now().isoformat(),
|
|
)
|
|
|
|
async def _gen_class_report(
|
|
self, config: ReportConfig
|
|
) -> GeneratedReport:
|
|
"""
|
|
生成班级学情报告
|
|
|
|
章节: 班级概览、成绩分布、薄弱知识点、优秀/进步学生、教学建议
|
|
"""
|
|
report_id = self._gen_report_id(config)
|
|
sections: List[ReportSection] = []
|
|
|
|
# 班级概览
|
|
sections.append(ReportSection(
|
|
title="班级学情概览",
|
|
section_type="summary",
|
|
content={
|
|
"student_count": 0,
|
|
"avg_score": 0,
|
|
"median_score": 0,
|
|
"pass_rate": 0,
|
|
"excellent_rate": 0,
|
|
},
|
|
order=1,
|
|
))
|
|
|
|
# 成绩分布
|
|
sections.append(ReportSection(
|
|
title="成绩分布分析",
|
|
section_type="chart",
|
|
content={
|
|
"chart_type": "histogram",
|
|
"distribution": {}, # 分数段人数分布
|
|
"comparison": {}, # 与上期对比
|
|
},
|
|
order=2,
|
|
))
|
|
|
|
# 薄弱知识点
|
|
sections.append(ReportSection(
|
|
title="班级薄弱知识点",
|
|
section_type="table",
|
|
content={
|
|
"weak_points": [], # [{知识点, 正确率, 涉及人数}]
|
|
},
|
|
order=3,
|
|
))
|
|
|
|
# 优秀/进步学生
|
|
sections.append(ReportSection(
|
|
title="优秀与进步学生",
|
|
section_type="table",
|
|
content={
|
|
"top_students": [], # 前10名
|
|
"most_improved": [], # 进步最大的学生
|
|
"need_attention": [], # 需关注的学生
|
|
},
|
|
order=4,
|
|
))
|
|
|
|
# 教学建议
|
|
sections.append(ReportSection(
|
|
title="教学改进建议",
|
|
section_type="recommendation",
|
|
content={
|
|
"recommendations": [
|
|
"针对薄弱知识点加强集中讲解和专项练习",
|
|
"关注成绩下滑学生，及时进行个别辅导",
|
|
"利用分层作业满足不同水平学生需求",
|
|
],
|
|
},
|
|
order=5,
|
|
))
|
|
|
|
return GeneratedReport(
|
|
report_id=report_id,
|
|
report_type=config.report_type,
|
|
target_id=config.target_id,
|
|
title="班级学情分析报告",
|
|
period=f"{config.start_date} ~ {config.end_date}",
|
|
sections=sections,
|
|
generated_at=datetime.now().isoformat(),
|
|
)
|
|
|
|
async def _gen_exam_report(
|
|
self, config: ReportConfig
|
|
) -> GeneratedReport:
|
|
"""生成考试分析报告（成绩分布+题目区分度+难度系数）"""
|
|
report_id = self._gen_report_id(config)
|
|
|
|
sections = [
|
|
ReportSection(
|
|
title="考试基本信息",
|
|
section_type="summary",
|
|
content={"exam_name": "", "subject": "", "total_score": 100},
|
|
order=1,
|
|
),
|
|
ReportSection(
|
|
title="成绩统计",
|
|
section_type="chart",
|
|
content={
|
|
"avg": 0, "median": 0, "max": 0, "min": 0,
|
|
"std_dev": 0, "pass_rate": 0,
|
|
"distribution": {},
|
|
},
|
|
order=2,
|
|
),
|
|
ReportSection(
|
|
title="题目分析",
|
|
section_type="table",
|
|
content={
|
|
"questions": [], # 每题的得分率、区分度、难度系数
|
|
},
|
|
order=3,
|
|
),
|
|
]
|
|
|
|
return GeneratedReport(
|
|
report_id=report_id,
|
|
report_type=config.report_type,
|
|
target_id=config.target_id,
|
|
title="考试分析报告",
|
|
period=config.start_date,
|
|
sections=sections,
|
|
generated_at=datetime.now().isoformat(),
|
|
)
|
|
|
|
async def _gen_writing_report(
|
|
self, config: ReportConfig
|
|
) -> GeneratedReport:
|
|
"""生成书写成长报告"""
|
|
report_id = self._gen_report_id(config)
|
|
|
|
sections = [
|
|
ReportSection(
|
|
title="书写能力总评",
|
|
section_type="summary",
|
|
content={
|
|
"overall_level": "",
|
|
"stroke_accuracy": 0,
|
|
"quality_score": 0,
|
|
"speed": 0,
|
|
},
|
|
order=1,
|
|
),
|
|
ReportSection(
|
|
title="成长趋势",
|
|
section_type="chart",
|
|
content={
|
|
"chart_type": "line",
|
|
"data_points": [], # 按周/月的评分趋势
|
|
},
|
|
order=2,
|
|
),
|
|
ReportSection(
|
|
title="常见书写问题",
|
|
section_type="table",
|
|
content={
|
|
"issues": [], # 笔顺错误、结构问题等
|
|
},
|
|
order=3,
|
|
),
|
|
]
|
|
|
|
return GeneratedReport(
|
|
report_id=report_id,
|
|
report_type=config.report_type,
|
|
target_id=config.target_id,
|
|
title="书写成长报告",
|
|
period=f"{config.start_date} ~ {config.end_date}",
|
|
sections=sections,
|
|
generated_at=datetime.now().isoformat(),
|
|
)
|
|
|
|
async def _gen_parent_report(
|
|
self, config: ReportConfig
|
|
) -> GeneratedReport:
|
|
"""
|
|
生成家长推送报告（简化版）
|
|
|
|
家长端报告简洁明了：
|
|
- 本周学习概况（评分、排名变化）
|
|
- 学习时长统计
|
|
- 需要关注的科目
|
|
- 家长配合建议
|
|
"""
|
|
report_id = self._gen_report_id(config)
|
|
|
|
sections = [
|
|
ReportSection(
|
|
title="本周学习概况",
|
|
section_type="summary",
|
|
content={
|
|
"overall_score": 0,
|
|
"rank_change": 0,
|
|
"homework_completed": 0,
|
|
"total_homework": 0,
|
|
"study_minutes": 0,
|
|
},
|
|
order=1,
|
|
),
|
|
ReportSection(
|
|
title="需要关注",
|
|
section_type="text",
|
|
content={
|
|
"attention_subjects": [],
|
|
"weak_points": [],
|
|
},
|
|
order=2,
|
|
),
|
|
ReportSection(
|
|
title="家长建议",
|
|
section_type="recommendation",
|
|
content={
|
|
"recommendations": [
|
|
"建议督促孩子按时完成作业",
|
|
"建议每天安排15-20分钟练字时间",
|
|
"多鼓励孩子在薄弱科目上的进步",
|
|
],
|
|
},
|
|
order=3,
|
|
),
|
|
]
|
|
|
|
return GeneratedReport(
|
|
report_id=report_id,
|
|
report_type=config.report_type,
|
|
target_id=config.target_id,
|
|
title="孩子本周学情报告",
|
|
period=f"{config.start_date} ~ {config.end_date}",
|
|
sections=sections,
|
|
generated_at=datetime.now().isoformat(),
|
|
)
|
|
|
|
def _generate_recommendations(
|
|
self,
|
|
student_id: str,
|
|
sections: List[ReportSection],
|
|
) -> List[str]:
|
|
"""基于各章节数据生成个性化学习建议"""
|
|
recommendations: List[str] = []
|
|
|
|
# 根据作业完成情况生成建议
|
|
for section in sections:
|
|
if section.title == "作业完成统计":
|
|
rate = section.content.get("completion_rate", 0)
|
|
if rate < 80:
|
|
recommendations.append(
|
|
"作业完成率偏低，建议养成当天作业当天完成的习惯"
|
|
)
|
|
|
|
if section.title == "书写能力评估":
|
|
quality = section.content.get("writing_quality", 0)
|
|
if quality < 60:
|
|
recommendations.append(
|
|
"书写规范性有待提高，建议每天坚持15分钟字帖练习"
|
|
)
|
|
|
|
if section.title == "学习习惯分析":
|
|
consistency = section.content.get("consistency", 0)
|
|
if consistency < 0.5:
|
|
recommendations.append(
|
|
"学习时间不够规律，建议制定固定的学习作息计划"
|
|
)
|
|
|
|
if not recommendations:
|
|
recommendations.append("继续保持良好的学习习惯，争取更大进步！")
|
|
|
|
return recommendations
|
|
|
|
def _generate_summary(
|
|
self,
|
|
sections: List[ReportSection],
|
|
report_target: str,
|
|
) -> str:
|
|
"""根据报告章节自动生成文字摘要"""
|
|
if report_target == "student":
|
|
return "本报告汇总了该学生在报告周期内的学业表现、书写能力和学习习惯分析。"
|
|
elif report_target == "class":
|
|
return "本报告汇总了班级在报告周期内的整体学情、成绩分布和教学建议。"
|
|
return ""
|
|
|
|
def _gen_report_id(self, config: ReportConfig) -> str:
|
|
"""生成唯一报告ID"""
|
|
raw = (
|
|
f"{config.report_type.value}_{config.target_id}_"
|
|
f"{config.start_date}_{config.end_date}"
|
|
)
|
|
return hashlib.md5(raw.encode()).hexdigest()[:16]
|
|
|
|
async def _export_pdf(self, report: GeneratedReport) -> None:
|
|
"""
|
|
将报告导出为PDF文件
|
|
|
|
使用ReportLab/WeasyPrint渲染PDF:
|
|
- 页眉: 自然写logo + 报告标题
|
|
- 正文: 各章节内容（图表使用ECharts渲染为图片）
|
|
- 页脚: 页码 + 生成时间
|
|
"""
|
|
# from weasyprint import HTML
|
|
# html_content = self._render_html_template(report)
|
|
# pdf_path = f"{self.output_dir}/{report.report_id}.pdf"
|
|
# HTML(string=html_content).write_pdf(pdf_path)
|
|
# report.file_path = pdf_path
|
|
logger.info("PDF导出: %s", report.report_id)
|
|
|
|
async def _export_html(self, report: GeneratedReport) -> None:
|
|
"""将报告导出为HTML文件"""
|
|
# html_path = f"{self.output_dir}/{report.report_id}.html"
|
|
# with open(html_path, "w", encoding="utf-8") as f:
|
|
# f.write(self._render_html_template(report))
|
|
# report.file_path = html_path
|
|
logger.info("HTML导出: %s", report.report_id)
|
|
|
|
|
|
# ============================================================
|
|
# 定时报告生成调度
|
|
# ============================================================
|
|
|
|
class ReportScheduler:
|
|
"""
|
|
报告定时生成调度器
|
|
|
|
支持:
|
|
- 每日凌晨生成前一天的学生日报
|
|
- 每周一生成上周的学生周报和班级周报
|
|
- 每月1日生成上月的月报
|
|
"""
|
|
|
|
def __init__(self, generator: ReportGenerator):
|
|
self.generator = generator
|
|
logger.info("报告调度器初始化")
|
|
|
|
async def run_daily_reports(self) -> int:
|
|
"""执行每日报告生成任务"""
|
|
yesterday = (date.today() - timedelta(days=1)).isoformat()
|
|
logger.info("执行每日报告生成: date=%s", yesterday)
|
|
|
|
generated_count = 0
|
|
# 查询所有活跃学生ID
|
|
# student_ids = await get_active_student_ids()
|
|
# for sid in student_ids:
|
|
# config = ReportConfig(
|
|
# report_type=ReportType.PARENT_PUSH,
|
|
# target_id=sid,
|
|
# start_date=yesterday,
|
|
# end_date=yesterday,
|
|
# )
|
|
# await self.generator.generate_report(config)
|
|
# generated_count += 1
|
|
|
|
logger.info("每日报告生成完成: 共%d份", generated_count)
|
|
return generated_count
|
|
|
|
async def run_weekly_reports(self) -> int:
|
|
"""执行每周报告生成任务"""
|
|
end_date = date.today() - timedelta(days=1)
|
|
start_date = end_date - timedelta(days=6)
|
|
logger.info(
|
|
"执行每周报告: %s ~ %s",
|
|
start_date.isoformat(),
|
|
end_date.isoformat(),
|
|
)
|
|
|
|
generated_count = 0
|
|
# 生成学生周报和班级周报
|
|
# ...
|
|
|
|
logger.info("每周报告生成完成: 共%d份", generated_count)
|
|
return generated_count
|
|
|
|
async def run_monthly_reports(self) -> int:
|
|
"""执行月度报告生成任务"""
|
|
today = date.today()
|
|
end_date = today.replace(day=1) - timedelta(days=1)
|
|
start_date = end_date.replace(day=1)
|
|
logger.info(
|
|
"执行月度报告: %s ~ %s",
|
|
start_date.isoformat(),
|
|
end_date.isoformat(),
|
|
)
|
|
|
|
generated_count = 0
|
|
# 生成学生月报、班级月报、书写成长报告
|
|
# ...
|
|
|
|
logger.info("月度报告生成完成: 共%d份", generated_count)
|
|
return generated_count
|