wolfydw/xsk-education-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add summary registration and deletion review

Browse Source
This commit is contained in:
Codex
2026-06-15 16:23:06 +08:00
parent cc46f651dc
commit 854e4942ba
10 changed files with 597 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/data.py
+335
View File
@@ -42,6 +42,7 @@ PAYMENT_LINE_RE = re.compile(r"^(?P<student>.+?)-(?P<date>\d{4}-\d{2}-\d{2}):(?P
TIME_RANGE_RE = re.compile(r"^(?P<sh>\d{1,2}):(?P<sm>\d{2})-(?P<eh>\d{1,2}):(?P<em>\d{2})$")
COURSE_SUMMARY_HEADING_RE = re.compile(r"^###\s+(?P<title>.+)$", re.M)
COURSE_SUMMARY_DATE_RE = re.compile(r"(?P<date>\d{4}[.-]\d{1,2}[.-]\d{1,2})")
SUMMARY_FIELD_RE = re.compile(r"^(?P<label>学生|学员|日期|上课日期|时间|上课时间|老师|教师|科目|课程|班级|分组|正文|内容|小结)[:：]\s*(?P<value>.*)$")
DATE_RANGE_SEPARATOR = r"(?:到|至|－|-|~|—|–)"
CHINESE_DATE_RANGE_RE = re.compile(
rf"(?:(?P<sy>\d{{4}})\s*年\s*)?"
@@ -579,6 +580,35 @@ def submit_correction_tasks(tasks_path: Path, items: list[dict]) -> dict:
return {"submitted": len(created), "items": [task_to_dict(task) for task in created]}
def submit_deletion_tasks(tasks_path: Path, items: list[dict]) -> dict:
if not items:
raise ValueError("提交审核的删除记录不能为空")
tasks = read_admin_tasks(tasks_path)
now = datetime.now().isoformat(timespec="seconds")
created: list[dict] = []
for item in items:
original_line = str(item.get("original_line", "")).strip()
if not original_line:
raise ValueError("删除审核记录缺少原记录")
original = parse_class_record_line(original_line)
task = {
"id": int(tasks["next_id"]),
"type": "class_record_deletion",
"status": "pending",
"created_at": now,
"updated_at": now,
"original_line": class_record_to_line(original),
"original": record_to_dict(original),
"student": original.student,
"reasons": ["申请删除课程记录"],
}
tasks["next_id"] = int(tasks["next_id"]) + 1
tasks["items"].append(task)
created.append(task)
write_admin_tasks(tasks_path, tasks)
return {"submitted": len(created), "items": [task_to_dict(task) for task in created]}
def list_admin_tasks(tasks_path: Path, status_filter: str = "", task_type: str = "") -> dict:
tasks = read_admin_tasks(tasks_path)
items = tasks["items"]
@@ -606,6 +636,16 @@ def replace_class_record_line(original_text: str, original_line: str, corrected_
return "\n".join(lines) + trailing_newline
def delete_class_record_line(original_text: str, original_line: str) -> str:
lines = original_text.splitlines()
matched = [index for index, line in enumerate(lines) if line.strip() == original_line]
if not matched:
raise ValueError("原上课记录在正式文件中不存在，可能已被修改")
del lines[matched[0]]
trailing_newline = "\n" if original_text.endswith("\n") and lines else ""
return "\n".join(lines) + trailing_newline
def mark_admin_task(tasks_path: Path, task_id: int, status: str, message: str = "") -> dict:
tasks = read_admin_tasks(tasks_path)
task = find_admin_task(tasks, task_id)
@@ -680,6 +720,73 @@ def approve_correction_task(tasks_path: Path, classnotes_path: Path, task_id: in
return {"task": task_to_dict(task), "backup_id": backup_dir.name}
def approve_deletion_task(
tasks_path: Path,
classnotes_path: Path,
accounts_path: Path,
task_id: int,
) -> dict:
tasks = read_admin_tasks(tasks_path)
task = find_admin_task(tasks, task_id)
if task.get("type") != "class_record_deletion":
raise ValueError("该任务不是上课记录删除")
if task.get("status") not in {"pending", "conflict"}:
raise ValueError("该任务已处理，不能重复批准")
original_line = str(task.get("original_line", "")).strip()
original = parse_class_record_line(original_line)
original_classnotes = classnotes_path.read_text(encoding="utf-8")
try:
new_classnotes = delete_class_record_line(original_classnotes, original_line)
accounts = read_accounts(accounts_path)
updated_accounts = list(accounts)
account_index = find_account_index(updated_accounts, original.student)
updated_accounts[account_index] = update_account_remaining(updated_accounts[account_index], original.duration_hours)
except ValueError as exc:
task["status"] = "conflict"
task["updated_at"] = datetime.now().isoformat(timespec="seconds")
task["message"] = str(exc)
write_admin_tasks(tasks_path, tasks)
raise
original_accounts = accounts_path.read_text(encoding="utf-8")
original_tasks = json.dumps(tasks, ensure_ascii=False, indent=2) + "\n"
now = datetime.now().isoformat(timespec="seconds")
task["status"] = "approved"
task["updated_at"] = now
task["reviewed_at"] = now
task["deleted_line"] = original_line
task["restored_hours"] = original.duration_hours
updated_accounts_by_id = {updated_accounts[account_index].student_id: updated_accounts[account_index]}
new_accounts = replace_account_lines(original_accounts, updated_accounts_by_id)
new_tasks = json.dumps(tasks, ensure_ascii=False, indent=2) + "\n"
backup_dir = create_data_backup(
"admin-approve-deletion",
{
accounts_path: original_accounts,
classnotes_path: original_classnotes,
tasks_path: original_tasks,
},
[original_line],
)
try:
atomic_write_text(accounts_path, new_accounts)
atomic_write_text(classnotes_path, new_classnotes)
atomic_write_text(tasks_path, new_tasks)
except Exception:
atomic_write_text(accounts_path, original_accounts)
atomic_write_text(classnotes_path, original_classnotes)
atomic_write_text(tasks_path, original_tasks)
raise
try:
prune_data_backups(backup_dir.parent)
except OSError:
pass
return {"task": task_to_dict(task), "backup_id": backup_dir.name}
def safe_filename_part(value: object) -> str:
text = str(value or "").strip()
text = re.sub(r"[\\/:*?\"<>|\s]+", "_", text)
@@ -755,6 +862,147 @@ def duration_minutes_from_summary(summary: dict) -> int | None:
return duration_minutes_from_time_range(str(summary.get("time_range") or ""))
def normalize_date_text(value: str) -> str:
text = value.strip().replace(".", "-").replace("/", "-")
match = re.search(r"(?P<y>\d{4})-(?P<m>\d{1,2})-(?P<d>\d{1,2})", text)
if not match:
raise ValueError(f"无法识别日期: {value}")
return f"{int(match.group('y')):04d}-{int(match.group('m')):02d}-{int(match.group('d')):02d}"
def extract_course_summary_from_text(text: str, index: int = 0, known_students: list[str] | None = None) -> dict:
raw = text.strip()
if not raw:
raise ValueError("课程小结内容不能为空")
fields: dict[str, str] = {}
body_lines: list[str] = []
in_body = False
label_map = {
"学生": "student",
"学员": "student",
"日期": "date_iso",
"上课日期": "date_iso",
"时间": "time_range",
"上课时间": "time_range",
"老师": "teacher",
"教师": "teacher",
"科目": "subject",
"课程": "subject",
"班级": "group",
"分组": "group",
"正文": "body",
"内容": "body",
"小结": "body",
}
for line in raw.splitlines():
stripped = line.strip()
if not stripped:
if in_body:
body_lines.append("")
continue
match = SUMMARY_FIELD_RE.match(stripped)
if match:
key = label_map[match.group("label")]
value = match.group("value").strip()
if key == "body":
in_body = True
if value:
body_lines.append(value)
else:
fields[key] = value
in_body = False
continue
if in_body:
body_lines.append(line.rstrip())
else:
body_lines.append(line.rstrip())
first_line = raw.splitlines()[0].strip()
try:
record = parse_class_record_line(first_line)
except ValueError:
record = None
if record:
fields.setdefault("date_iso", record.date.replace(".", "-"))
fields.setdefault("time_range", record.time)
fields.setdefault("student", record.student)
fields.setdefault("duration_minutes", str(int(round(record.duration_hours * 60))))
fields.setdefault("teacher", record.teacher)
fields.setdefault("subject", record.subject)
if "date_iso" not in fields:
date_match = COURSE_SUMMARY_DATE_RE.search(raw)
if date_match:
fields["date_iso"] = date_match.group("date")
if "time_range" not in fields:
time_match = TIME_RANGE_RE.search(raw)
if time_match:
fields["time_range"] = time_match.group(0)
if "student" not in fields:
for student in known_students or []:
if student and student in raw:
fields["student"] = student
break
if "subject" not in fields:
for subject in SUBJECTS:
if subject in raw:
fields["subject"] = subject
break
if "teacher" not in fields:
teacher_match = re.search(r"(?P<teacher>[\u4e00-\u9fa5A-Za-z0-9]{1,8}老师)", raw)
if teacher_match:
fields["teacher"] = teacher_match.group("teacher")
if "date_iso" in fields:
fields["date_iso"] = normalize_date_text(fields["date_iso"])
if "time_range" in fields:
time_match = TIME_RANGE_RE.search(fields["time_range"])
if time_match:
fields["time_range"] = time_match.group(0)
body = "\n".join(body_lines).strip() or raw
source_id = f"manual:{sha1_text(raw, 24)}"
return {
**fields,
"source_id": source_id,
"body": body,
"recognition_source": "manual_admin",
"confidence": "manual",
"teacher_trusted": True,
"sender": "管理后台",
"local_id": str(index + 1),
}
def manual_review_summary(raw: dict, error: Exception) -> dict:
body = str(raw.get("body") or raw.get("content") or "").strip()
source_id = str(raw.get("source_id") or "").strip() or f"manual:{sha1_text(body, 24)}"
return {
"source_id": source_id,
"student": canonical_name(str(raw.get("student") or "").strip()) or "待核对学生",
"date_iso": str(raw.get("date_iso") or raw.get("date") or "").strip(),
"time_range": str(raw.get("time_range") or raw.get("time") or "").strip(),
"duration_minutes": raw.get("duration_minutes"),
"duration": str(raw.get("duration") or "").strip(),
"teacher": canonical_name(str(raw.get("teacher") or "").strip()),
"subject": parse_subject_code(str(raw.get("subject") or "").strip()),
"group": str(raw.get("group") or "").strip(),
"sender": str(raw.get("sender") or "管理后台").strip(),
"sender_id": str(raw.get("sender_id") or "").strip(),
"message_time": str(raw.get("message_time") or "").strip(),
"message_date": str(raw.get("message_date") or "").strip(),
"db": str(raw.get("db") or "").strip(),
"local_id": str(raw.get("local_id") or "").strip(),
"title": str(raw.get("title") or "手工登记课程小结").strip(),
"body": body or str(error),
"recognition_source": "manual_admin",
"confidence": "manual_review",
"teacher_trusted": True,
"remark": f"手工登记待审核：{error}",
}
def normalize_course_summary(raw: dict) -> dict:
student = canonical_name(str(raw.get("student") or "").strip())
teacher = canonical_name(str(raw.get("teacher") or "").strip())
@@ -1240,11 +1488,98 @@ def approve_admin_task(
task = find_admin_task(read_admin_tasks(tasks_path), task_id)
if task.get("type") == "class_record_correction":
return approve_correction_task(tasks_path, classnotes_path, task_id)
if task.get("type") == "class_record_deletion":
return approve_deletion_task(tasks_path, classnotes_path, accounts_path, task_id)
if task.get("type") == "course_summary_review":
return approve_course_summary_task(tasks_path, classnotes_path, accounts_path, task_id)
raise ValueError("不支持的审核任务类型")
def register_course_summary_texts(
*,
classnotes_path: Path,
accounts_path: Path,
tasks_path: Path,
summaries_root: Path,
state_path: Path,
operation_logs_path: Path,
lines: list[str] | None = None,
line: str | None = None,
) -> dict:
texts = normalize_lines(lines=lines, line=line)
now = datetime.now().isoformat(timespec="seconds")
accounts = read_accounts(accounts_path)
known_students = [account.student for account in accounts]
summaries: list[dict] = []
manual_review_items: list[dict] = []
result = {
"received": len(texts),
"saved": 0,
"auto_registered": 0,
"review_pending": 0,
"duplicates": 0,
"rejected": 0,
"operation_log_ids": [],
"items": [],
}
for index, text in enumerate(texts):
raw = extract_course_summary_from_text(text, index, known_students=known_students)
try:
normalize_course_summary(raw)
summaries.append(raw)
except ValueError as exc:
summary = manual_review_summary(raw, exc)
saved = save_course_summary_markdown(summaries_root, summary)
task = create_course_summary_review_task(
tasks_path,
summary,
"",
[str(exc), "手工登记课程小结需人工补全"],
saved_path=str(saved.get("path") or ""),
)
log_id = append_operation_log(
operation_logs_path,
"course_summary_ingest",
"review",
batch_id=f"manual-{datetime.now().strftime('%Y%m%d%H%M%S')}",
source_id=summary["source_id"],
student=summary["student"],
reasons=[str(exc), "手工登记课程小结需人工补全"],
task_id=task.get("id"),
saved_path=str(saved.get("path") or ""),
)
result["saved"] += 1 if saved.get("added") else 0
result["review_pending"] += 1
result["operation_log_ids"].append(log_id)
result["items"].append(
{
"source_id": summary["source_id"],
"status": "review",
"task_id": task.get("id"),
"reasons": [str(exc), "手工登记课程小结需人工补全"],
}
)
if not summaries:
return result
ingest_result = ingest_course_summaries(
classnotes_path=classnotes_path,
accounts_path=accounts_path,
tasks_path=tasks_path,
summaries_root=summaries_root,
state_path=state_path,
operation_logs_path=operation_logs_path,
batch_id=f"manual-{datetime.now().strftime('%Y%m%d%H%M%S')}-{sha1_text('|'.join(texts), 8)}",
window={"source": "admin_register", "submitted_at": now},
students=[],
summaries=summaries,
)
for key in ("saved", "auto_registered", "review_pending", "duplicates", "rejected"):
result[key] += int(ingest_result.get(key) or 0)
result["operation_log_ids"].extend(ingest_result.get("operation_log_ids") or [])
result["items"].extend(ingest_result.get("items") or [])
return result
def ingest_course_summaries(
*,
classnotes_path: Path,