181 lines
10 KiB
Python
181 lines
10 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 .knowledge_pages import evidence_links
|
||
from .models import (CriterionScore, ExportTask, Project, ScoreEvidence, ScoringBatch, Student, SubmissionFile)
|
||
from .rules import totals
|
||
|
||
#: Teacher-facing wording for a criterion status, shared by exports and the API.
|
||
STATUS_LABELS = {
|
||
"pending": "待评分",
|
||
"scored": "模型已评分",
|
||
"reviewed": "教师已复核",
|
||
"needs_review": "待教师复核",
|
||
"unscorable": "材料缺失或解析失败",
|
||
"failed": "模型调用失败",
|
||
}
|
||
PAGE_TYPE_LABELS = {
|
||
"cover": "封面", "abstract": "摘要", "toc": "目录", "body": "正文",
|
||
"appendix": "附录", "references": "参考文献", "unknown": "识别不确定",
|
||
"mixed": "目录与正文混合页",
|
||
}
|
||
#: Batch status wording. ``review_required`` means grading finished but some criteria
|
||
#: still need a teacher decision, which is not the same as an incomplete batch.
|
||
BATCH_STATUS_LABELS = {
|
||
"created": "待启动", "queued": "排队中", "running": "运行中",
|
||
"completed": "已完成", "review_required": "已完成，有待复核项",
|
||
"partial": "部分完成", "failed": "失败",
|
||
}
|
||
|
||
|
||
def page_type_label(value) -> str:
|
||
return PAGE_TYPE_LABELS.get(str(value or "unknown"), "未分类页面")
|
||
|
||
|
||
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 = []
|
||
def evidence_dict(item, links):
|
||
return {"page": item.page, "quote": item.quote, "file_id": item.submission_file_id,
|
||
"document_id": item.document_id,
|
||
"target_id": item.target_id or "", "target_label": item.target_label or "",
|
||
"page_type": item.page_type or links.get("page_type") or "unknown",
|
||
"source_role": item.source_role or "content",
|
||
# A quotation saved before the evidence policy existed is labelled as a
|
||
# historical citation and never presented as report body evidence.
|
||
"historical": bool(item.historical),
|
||
"document": db.get(SubmissionFile, item.submission_file_id).relative_path if item.submission_file_id else "课程资料",
|
||
**links}
|
||
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_gaps": (score.evidence_gaps or []) if score else [],
|
||
"evidence": [evidence_dict(item, evidence_links(db, item)) 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 '页'}"
|
||
f"（{esc(page_type_label(e.get('page_type')))}{'，历史评分引用' if e.get('historical') else ''}）"
|
||
f"{' 支持：' + esc(e['target_label']) if e.get('target_label') else ''}：{esc(e['quote'])}</li>"
|
||
for e in row["evidence"]
|
||
)
|
||
missing = "、".join(row["missing_evidence"] or [])
|
||
gaps = "；".join(f"{gap.get('target_label')}（检索范围：{'、'.join(page_type_label(t) for t in gap.get('searched_page_types') or []) or '正文'}）"
|
||
for gap in row.get("evidence_gaps") or [])
|
||
review = "本项缺少正文证据，已转教师复核。" if row["status"] == "needs_review" else ""
|
||
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(STATUS_LABELS.get(row['status'], row['status']))}</p>"
|
||
f"<p>模型评语：{esc(row['model_comment'])}</p><p>教师评语：{esc(row['teacher_comment'])}</p>"
|
||
f"<p>{esc(review)}</p>"
|
||
f"<p>缺失证据：{esc(missing)}</p><p>缺正文证据的目标：{esc(gaps)}</p>"
|
||
f"<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
|