137 lines
7.9 KiB
Python
137 lines
7.9 KiB
Python
from html import escape
|
||
from pathlib import Path
|
||
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill
|
||
from sqlalchemy import select
|
||
from weasyprint import HTML
|
||
|
||
from .config import get_settings
|
||
from .models import (CriterionScore, ExportTask, Project, ScoreEvidence, ScoringBatch, Student, SubmissionFile)
|
||
from .rules import totals
|
||
|
||
|
||
def project_result(db, batch: ScoringBatch, project: Project):
|
||
scores = db.scalars(select(CriterionScore).where(CriterionScore.batch_id == batch.id,
|
||
CriterionScore.project_id == project.id)).all()
|
||
by_id = {score.criterion_id: score for score in scores}
|
||
criteria = batch.rubric_snapshot["criteria"]
|
||
totals_ = totals({key: score.final_score for key, score in by_id.items()}, criteria, batch.rubric_snapshot["weights"])
|
||
rows = []
|
||
for criterion in criteria:
|
||
score = by_id.get(criterion["criterion_id"])
|
||
evidence = db.scalars(select(ScoreEvidence).where(ScoreEvidence.score_id == score.id)).all() if score else []
|
||
rows.append({"criterion": criterion, "score_id": score.id if score else None,
|
||
"model_score": score.model_score if score else None, "teacher_score": score.teacher_score if score else None,
|
||
"final_score": score.final_score if score else None,
|
||
"confidence": score.confidence if score else None,
|
||
"model_comment": score.model_comment if score else "", "teacher_comment": score.teacher_comment if score else None,
|
||
"missing_evidence": score.missing_evidence if score else [], "suggestion": score.suggestion if score else "",
|
||
"status": score.status if score else "pending", "error": score.error if score else "",
|
||
"evidence": [{"page": item.page, "quote": item.quote, "file_id": item.submission_file_id,
|
||
"document": db.get(SubmissionFile, item.submission_file_id).relative_path if item.submission_file_id else "课程资料",
|
||
"image_url": f"/api/assets/{item.submission_file_id}/{item.page}" if item.image_path else None} for item in evidence]})
|
||
student = db.get(Student, project.student_id)
|
||
return {"project_id": project.id, "student_no": student.student_no, "name": student.name, "topic": project.topic,
|
||
"totals": totals_, "rows": rows, "facts": project.facts}
|
||
|
||
|
||
def batch_projects(db, batch: ScoringBatch):
|
||
if batch.project_ids is None:
|
||
return db.scalars(select(Project).where(Project.course_version_id == batch.course_version_id).order_by(Project.id)).all()
|
||
if not batch.project_ids:
|
||
return []
|
||
by_id = {project.id: project for project in db.scalars(select(Project).where(Project.id.in_(batch.project_ids))).all()}
|
||
return [by_id[project_id] for project_id in batch.project_ids if project_id in by_id]
|
||
|
||
|
||
def export_excel(db, batch_id: int):
|
||
batch = db.get(ScoringBatch, batch_id)
|
||
if not batch:
|
||
raise ValueError("评分批次不存在")
|
||
root = get_settings().exports_path
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
task = ExportTask(batch_id=batch_id, kind="excel", status="running")
|
||
db.add(task)
|
||
db.flush()
|
||
path = root / f"batch-{batch_id}-scores-{task.id}.xlsx"
|
||
try:
|
||
book = Workbook()
|
||
sheet = book.active
|
||
sheet.title = "成绩"
|
||
final_criteria = [c for c in batch.rubric_snapshot["criteria"] if c["category"] == "final"]
|
||
sheet.append(["学号", "姓名", "选题", "阶段报告1", "阶段报告2", "阶段报告3", *[c["name"] for c in final_criteria], "期末成绩", "总评", "状态", "批次", "规则版本"])
|
||
for project in batch_projects(db, batch):
|
||
result = project_result(db, batch, project)
|
||
scores = {row["criterion"]["criterion_id"]: row["final_score"] for row in result["rows"]}
|
||
totals_ = result["totals"]
|
||
sheet.append([result["student_no"], result["name"], result["topic"], *[totals_[key] for key in ("stage1", "stage2", "stage3")],
|
||
*[scores.get(c["criterion_id"]) for c in final_criteria], totals_["final"], totals_["overall"],
|
||
"完成" if totals_["overall"] is not None else "未完成/待人工处理", batch.id, batch.rubric_version_id])
|
||
sheet.freeze_panes = "D2"
|
||
sheet.auto_filter.ref = sheet.dimensions
|
||
for cell in sheet[1]:
|
||
cell.font = Font(bold=True, color="FFFFFF")
|
||
cell.fill = PatternFill("solid", fgColor="153C65")
|
||
for column in sheet.columns:
|
||
letter = column[0].column_letter
|
||
sheet.column_dimensions[letter].width = min(50, max(12, max(len(str(cell.value or "")) for cell in column) + 2))
|
||
book.save(path)
|
||
task.path, task.status = str(path), "completed"
|
||
except Exception as exc:
|
||
task.status, task.error = "failed", str(exc)[:1000]
|
||
db.commit()
|
||
raise
|
||
db.commit()
|
||
return task
|
||
|
||
|
||
def _html_report(result, batch):
|
||
esc = lambda value: escape(str(value or ""))
|
||
rows = []
|
||
for row in result["rows"]:
|
||
criterion = row["criterion"]
|
||
evidences = "".join(
|
||
f"<li>{esc(e['document'])} · 第{e['page']}{'张幻灯片' if str(e['document']).lower().endswith('.pptx') else '页'}：{esc(e['quote'])}</li>"
|
||
for e in row["evidence"]
|
||
)
|
||
missing = "、".join(row["missing_evidence"] or [])
|
||
rows.append(f"<section><h3>{esc(criterion['name'])}（{esc(row['final_score'])} / {criterion['max_score']}）</h3>"
|
||
f"<p>模型原始分：{esc(row['model_score'])}；教师修改分：{esc(row['teacher_score'])}；状态：{esc(row['status'])}</p>"
|
||
f"<p>模型评语：{esc(row['model_comment'])}</p><p>教师评语：{esc(row['teacher_comment'])}</p>"
|
||
f"<p>缺失证据：{esc(missing)}</p><p>建议：{esc(row['suggestion'])}</p><p>异常：{esc(row['error'])}</p>"
|
||
f"<ul>{evidences}</ul></section>")
|
||
totals_ = result["totals"]
|
||
return f"""<!doctype html><html lang="zh"><meta charset="utf-8"><style>
|
||
body {{ font-family: 'Noto Sans CJK SC', sans-serif; font-size: 11px; color: #172b4d; }}
|
||
h1 {{ color: #153c65; }} section {{ border-bottom: 1px solid #ccd7e0; break-inside: avoid; }}
|
||
h3 {{ margin-bottom: 3px; }} p {{ margin: 3px 0; }} li {{ margin: 3px 0; }}
|
||
</style><body><h1>WiseGrade 智评 · 学生反馈</h1>
|
||
<p>学号：{esc(result['student_no'])} 姓名：{esc(result['name'])} 选题：{esc(result['topic'])}</p>
|
||
<p>评分批次：{batch.id} 课程版本：{batch.course_version_id} 规则版本：{batch.rubric_version_id}</p>
|
||
<p>模型：{esc(batch.model_config_snapshot.get('model'))} 嵌入模型：{esc(batch.embedding_model_name)} ({esc(batch.embedding_model_version)})</p>
|
||
<h2>成绩</h2><p>阶段1：{esc(totals_['stage1'])} 阶段2：{esc(totals_['stage2'])} 阶段3：{esc(totals_['stage3'])} 期末：{esc(totals_['final'])} 总评：{esc(totals_['overall'])}</p>
|
||
{''.join(rows)}</body></html>"""
|
||
|
||
|
||
def export_project_pdf(db, batch_id: int, project_id: int):
|
||
batch = db.get(ScoringBatch, batch_id)
|
||
project = db.get(Project, project_id)
|
||
if not batch or not project or project.course_version_id != batch.course_version_id or (batch.project_ids is not None and project_id not in batch.project_ids):
|
||
raise ValueError("批次或项目不存在")
|
||
root = get_settings().exports_path
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
task = ExportTask(batch_id=batch_id, project_id=project_id, kind="pdf", status="running")
|
||
db.add(task)
|
||
db.flush()
|
||
path = root / f"batch-{batch_id}-project-{project_id}-{task.id}.pdf"
|
||
try:
|
||
HTML(string=_html_report(project_result(db, batch, project), batch)).write_pdf(path)
|
||
task.path, task.status = str(path), "completed"
|
||
except Exception as exc:
|
||
task.status, task.error = "failed", str(exc)[:1000]
|
||
db.commit()
|
||
raise
|
||
db.commit()
|
||
return task
|