Codex
1759 lines
66 KiB
Python
1759 lines
66 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
from dataclasses import replace
|
||
from datetime import date, datetime, timedelta
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
from .domain import (
|
||
ACCOUNT_STATUSES,
|
||
AUTO_RECOGNITION_SOURCES,
|
||
Account,
|
||
ClassRecord,
|
||
DuplicateRecordError,
|
||
FALLBACK_ALIASES,
|
||
HIGH_CONFIDENCE_VALUES,
|
||
Payment,
|
||
QuerySpec,
|
||
ROLE_WORDS,
|
||
SUBJECTS,
|
||
SUBJECT_ALIASES,
|
||
UNKNOWN_SUBJECTS,
|
||
UNKNOWN_TEACHERS,
|
||
WEEKDAYS,
|
||
)
|
||
from .storage import atomic_write_text, create_data_backup, prune_data_backups
|
||
|
||
|
||
CLASSNOTE_RE = re.compile(
|
||
r"^(?P<date>\d{4}\.\d{2}\.\d{2})-"
|
||
r"(?P<weekday>星期[一二三四五六日])-"
|
||
r"(?P<time>\d{1,2}:\d{2}-\d{1,2}:\d{2})-"
|
||
r"(?P<student>[^-]+)-"
|
||
r"(?P<duration>\d+小时\d+分)-"
|
||
r"(?P<teacher>[^-]+)-"
|
||
r"(?P<subject>.+)$"
|
||
)
|
||
PAYMENT_LINE_RE = re.compile(r"^(?P<student>.+?)-(?P<date>\d{4}-\d{2}-\d{2}):(?P<hours>\d+(?:\.\d+)?)$")
|
||
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})")
|
||
DATE_RANGE_SEPARATOR = r"(?:到|至|－|-|~|—|–)"
|
||
CHINESE_DATE_RANGE_RE = re.compile(
|
||
rf"(?:(?P<sy>\d{{4}})\s*年\s*)?"
|
||
rf"(?P<sm>\d{{1,2}})\s*月\s*(?P<sd>\d{{1,2}})\s*[日号]?\s*"
|
||
rf"{DATE_RANGE_SEPARATOR}\s*"
|
||
rf"(?:(?P<ey>\d{{4}})\s*年\s*)?"
|
||
rf"(?:(?P<em>\d{{1,2}})\s*月\s*)?"
|
||
rf"(?P<ed>\d{{1,2}})\s*[日号]?"
|
||
)
|
||
NUMERIC_DATE_RANGE_RE = re.compile(
|
||
rf"(?<!\d)"
|
||
rf"(?:(?P<sy>\d{{4}})[./-])?"
|
||
rf"(?P<sm>\d{{1,2}})[./-](?P<sd>\d{{1,2}})\s*"
|
||
rf"{DATE_RANGE_SEPARATOR}\s*"
|
||
rf"(?:(?P<ey>\d{{4}})[./-])?"
|
||
rf"(?:(?P<em>\d{{1,2}})[./-])?"
|
||
rf"(?P<ed>\d{{1,2}})"
|
||
rf"(?!\d)"
|
||
)
|
||
|
||
|
||
def canonical_name(name: str) -> str:
|
||
text = name.strip()
|
||
return FALLBACK_ALIASES.get(text, text)
|
||
|
||
|
||
def parse_hours_text(text: str) -> float:
|
||
match = re.fullmatch(r"(\d+)小时(\d+)分", text.strip())
|
||
if not match:
|
||
raise ValueError(f"无法解析时长: {text}")
|
||
hours, minutes = map(int, match.groups())
|
||
return hours + minutes / 60.0
|
||
|
||
|
||
def format_number(value: float) -> str:
|
||
if float(value).is_integer():
|
||
return str(int(value))
|
||
return f"{value:.2f}".rstrip("0").rstrip(".")
|
||
|
||
|
||
def format_payment(payment: Payment) -> str:
|
||
return f"{payment.date}:{format_number(payment.hours)}"
|
||
|
||
|
||
def format_account_row(account: Account) -> str:
|
||
payments = ",".join(format_payment(payment) for payment in account.payments)
|
||
return (
|
||
f"| {account.student_id} | {account.student} | {payments} | "
|
||
f"{format_number(account.remaining)} | {account.account_status} | {account.note} |"
|
||
)
|
||
|
||
|
||
def parse_payments(text: str) -> list[Payment]:
|
||
payments: list[Payment] = []
|
||
if not text.strip():
|
||
return payments
|
||
for item in text.split(","):
|
||
item = item.strip()
|
||
if not item or ":" not in item:
|
||
continue
|
||
paid_date, hours_text = item.split(":", 1)
|
||
payments.append(Payment(date=paid_date.strip(), hours=float(hours_text.strip())))
|
||
return payments
|
||
|
||
|
||
def validate_payment(payment: Payment) -> Payment:
|
||
try:
|
||
datetime.strptime(payment.date, "%Y-%m-%d")
|
||
except ValueError as exc:
|
||
raise ValueError(f"缴费日期不存在: {payment.date}") from exc
|
||
return Payment(date=payment.date, hours=round(float(payment.hours), 2))
|
||
|
||
|
||
def validate_account(account: Account) -> Account:
|
||
student_id = account.student_id.strip()
|
||
student = canonical_name(account.student)
|
||
if not re.fullmatch(r"XS\d{3}", student_id):
|
||
raise ValueError("学生ID格式应为 XS001 这样的三位编号")
|
||
if not student:
|
||
raise ValueError("学生姓名不能为空")
|
||
if account.account_status not in ACCOUNT_STATUSES:
|
||
raise ValueError("账户状态必须是 正常、预警、欠费、结课、退费")
|
||
return Account(
|
||
student_id=student_id,
|
||
student=student,
|
||
payments=[validate_payment(payment) for payment in account.payments],
|
||
remaining=round(float(account.remaining), 2),
|
||
account_status=account.account_status,
|
||
note=account.note.strip(),
|
||
)
|
||
|
||
|
||
def read_classnotes(path: Path) -> list[ClassRecord]:
|
||
records: list[ClassRecord] = []
|
||
with path.open("r", encoding="utf-8") as handle:
|
||
for line_number, raw_line in enumerate(handle, start=1):
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
match = CLASSNOTE_RE.fullmatch(line)
|
||
if not match:
|
||
raise ValueError(f"{path}:{line_number} 无法解析课程记录行: {line}")
|
||
duration = match.group("duration")
|
||
records.append(
|
||
ClassRecord(
|
||
date=match.group("date"),
|
||
weekday=match.group("weekday"),
|
||
time=match.group("time"),
|
||
student=canonical_name(match.group("student")),
|
||
duration=duration,
|
||
duration_hours=parse_hours_text(duration),
|
||
teacher=match.group("teacher").strip(),
|
||
subject=match.group("subject").strip(),
|
||
)
|
||
)
|
||
return records
|
||
|
||
|
||
def read_accounts(path: Path) -> list[Account]:
|
||
accounts: list[Account] = []
|
||
with path.open("r", encoding="utf-8") as handle:
|
||
for line_number, raw_line in enumerate(handle, start=1):
|
||
line = raw_line.strip()
|
||
if not line.startswith("|") or line.startswith("|---") or "学生ID" in line:
|
||
continue
|
||
parts = [part.strip() for part in line.strip("|").split("|")]
|
||
if len(parts) < 5:
|
||
continue
|
||
try:
|
||
remaining = float(parts[3])
|
||
except ValueError as exc:
|
||
raise ValueError(f"{path}:{line_number} 无法解析剩余课时: {parts[3]}") from exc
|
||
accounts.append(
|
||
Account(
|
||
student_id=parts[0],
|
||
student=canonical_name(parts[1]),
|
||
payments=parse_payments(parts[2]),
|
||
remaining=remaining,
|
||
account_status=parts[4],
|
||
note=parts[5] if len(parts) > 5 else "",
|
||
)
|
||
)
|
||
return accounts
|
||
|
||
|
||
def parse_class_record_line(line: str) -> ClassRecord:
|
||
raw = line.strip()
|
||
match = CLASSNOTE_RE.fullmatch(raw)
|
||
if not match:
|
||
raise ValueError(f"上课记录格式错误: {line}")
|
||
|
||
record_date = parse_record_date(match.group("date"))
|
||
expected_weekday = WEEKDAYS[record_date.weekday()]
|
||
if match.group("weekday") != expected_weekday:
|
||
raise ValueError(f"上课记录星期错误: {line}，应为 {expected_weekday}")
|
||
|
||
duration = match.group("duration")
|
||
duration_hours = parse_hours_text(duration)
|
||
time_hours = parse_time_range_hours(match.group("time"))
|
||
if abs(duration_hours - time_hours) > 0.01:
|
||
raise ValueError(f"上课记录时长与时间段不一致: {line}")
|
||
|
||
student = canonical_name(match.group("student"))
|
||
teacher = match.group("teacher").strip()
|
||
subject = match.group("subject").strip()
|
||
if not student or not teacher or not subject:
|
||
raise ValueError(f"上课记录学生、老师、科目不能为空: {line}")
|
||
|
||
return ClassRecord(
|
||
date=match.group("date"),
|
||
weekday=match.group("weekday"),
|
||
time=match.group("time"),
|
||
student=student,
|
||
duration=duration,
|
||
duration_hours=duration_hours,
|
||
teacher=teacher,
|
||
subject=subject,
|
||
)
|
||
|
||
|
||
def parse_time_range_hours(text: str) -> float:
|
||
match = TIME_RANGE_RE.fullmatch(text.strip())
|
||
if not match:
|
||
raise ValueError(f"无法解析时间段: {text}")
|
||
start_hour = int(match.group("sh"))
|
||
start_minute = int(match.group("sm"))
|
||
end_hour = int(match.group("eh"))
|
||
end_minute = int(match.group("em"))
|
||
if start_hour > 23 or end_hour > 23:
|
||
raise ValueError(f"时间段小时超出范围: {text}")
|
||
if start_minute > 59 or end_minute > 59:
|
||
raise ValueError(f"时间段分钟超出范围: {text}")
|
||
start_total = start_hour * 60 + start_minute
|
||
end_total = end_hour * 60 + end_minute
|
||
if end_total <= start_total:
|
||
raise ValueError(f"结束时间必须晚于开始时间: {text}")
|
||
return (end_total - start_total) / 60.0
|
||
|
||
|
||
def class_record_to_line(record: ClassRecord) -> str:
|
||
return (
|
||
f"{record.date}-{record.weekday}-{record.time}-{record.student}-"
|
||
f"{record.duration}-{record.teacher}-{record.subject}"
|
||
)
|
||
|
||
|
||
def parse_payment_line(line: str) -> tuple[str, Payment]:
|
||
raw = line.strip()
|
||
match = PAYMENT_LINE_RE.fullmatch(raw)
|
||
if not match:
|
||
raise ValueError(f"缴费记录格式错误: {line}")
|
||
student = canonical_name(match.group("student"))
|
||
payment_date = match.group("date")
|
||
try:
|
||
datetime.strptime(payment_date, "%Y-%m-%d")
|
||
except ValueError as exc:
|
||
raise ValueError(f"缴费日期不存在: {payment_date}") from exc
|
||
hours = float(match.group("hours"))
|
||
if hours <= 0:
|
||
raise ValueError(f"缴费课时必须大于 0: {line}")
|
||
return student, Payment(date=payment_date, hours=hours)
|
||
|
||
|
||
def recalc_account_status(account: Account) -> str:
|
||
if account.account_status in {"结课", "退费"}:
|
||
return account.account_status
|
||
if account.remaining < 0:
|
||
return "欠费"
|
||
if account.remaining < 10:
|
||
return "预警"
|
||
return "正常"
|
||
|
||
|
||
def find_account_index(accounts: list[Account], student: str) -> int:
|
||
canonical_student = canonical_name(student)
|
||
for index, account in enumerate(accounts):
|
||
if account.student == canonical_student or account.student_id == canonical_student:
|
||
return index
|
||
raise ValueError(f"未找到学生账户: {student}")
|
||
|
||
|
||
def update_account_remaining(account: Account, delta_hours: float) -> Account:
|
||
updated = replace(account, remaining=round(account.remaining + delta_hours, 2))
|
||
return replace(updated, account_status=recalc_account_status(updated))
|
||
|
||
|
||
def append_account_payment(account: Account, payment: Payment) -> Account:
|
||
updated = replace(
|
||
account,
|
||
payments=[*account.payments, payment],
|
||
remaining=round(account.remaining + payment.hours, 2),
|
||
)
|
||
return replace(updated, account_status=recalc_account_status(updated))
|
||
|
||
|
||
def replace_account_lines(original_text: str, accounts_by_id: dict[str, Account]) -> str:
|
||
lines = original_text.splitlines()
|
||
output: list[str] = []
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if stripped.startswith("|") and not stripped.startswith("|---") and "学生ID" not in stripped:
|
||
parts = [part.strip() for part in stripped.strip("|").split("|")]
|
||
if parts and parts[0] in accounts_by_id:
|
||
output.append(format_account_row(accounts_by_id[parts[0]]))
|
||
continue
|
||
output.append(line)
|
||
trailing_newline = "\n" if original_text.endswith("\n") else ""
|
||
return "\n".join(output) + trailing_newline
|
||
|
||
|
||
def append_account_line(original_text: str, account: Account) -> str:
|
||
lines = original_text.splitlines()
|
||
insert_at = len(lines)
|
||
for index, line in enumerate(lines):
|
||
stripped = line.strip()
|
||
if stripped.startswith("|") and not stripped.startswith("|---") and "学生ID" not in stripped:
|
||
insert_at = index + 1
|
||
lines.insert(insert_at, format_account_row(account))
|
||
trailing_newline = "\n" if original_text.endswith("\n") else ""
|
||
return "\n".join(lines) + trailing_newline
|
||
|
||
|
||
def replace_single_account_line(original_text: str, old_student_id: str, account: Account) -> str:
|
||
lines = original_text.splitlines()
|
||
replaced = False
|
||
output: list[str] = []
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if stripped.startswith("|") and not stripped.startswith("|---") and "学生ID" not in stripped:
|
||
parts = [part.strip() for part in stripped.strip("|").split("|")]
|
||
if parts and parts[0] == old_student_id:
|
||
output.append(format_account_row(account))
|
||
replaced = True
|
||
continue
|
||
output.append(line)
|
||
if not replaced:
|
||
raise ValueError(f"未找到学生账户: {old_student_id}")
|
||
trailing_newline = "\n" if original_text.endswith("\n") else ""
|
||
return "\n".join(output) + trailing_newline
|
||
|
||
|
||
def next_student_id(accounts: list[Account]) -> str:
|
||
values = []
|
||
for account in accounts:
|
||
match = re.fullmatch(r"XS(\d{3})", account.student_id)
|
||
if match:
|
||
values.append(int(match.group(1)))
|
||
return f"XS{(max(values) if values else 0) + 1:03d}"
|
||
|
||
|
||
def write_accounts(path: Path, accounts: list[Account]) -> None:
|
||
original_text = path.read_text(encoding="utf-8")
|
||
accounts_by_id = {account.student_id: account for account in accounts}
|
||
atomic_write_text(path, replace_account_lines(original_text, accounts_by_id))
|
||
|
||
|
||
def create_account(path: Path, account: Account) -> dict:
|
||
accounts = read_accounts(path)
|
||
account = validate_account(replace(account, student_id=next_student_id(accounts)))
|
||
if any(item.student_id == account.student_id for item in accounts):
|
||
raise ValueError(f"学生ID已存在: {account.student_id}")
|
||
original_accounts = path.read_text(encoding="utf-8")
|
||
backup_dir = create_data_backup("admin-create-account", {path: original_accounts}, [format_account_row(account)])
|
||
atomic_write_text(path, append_account_line(original_accounts, account))
|
||
try:
|
||
prune_data_backups(backup_dir.parent)
|
||
except OSError:
|
||
pass
|
||
return {"account": account_to_dict(account), "backup_id": backup_dir.name}
|
||
|
||
|
||
def update_account(path: Path, old_student_id: str, account: Account) -> dict:
|
||
old_student_id = old_student_id.strip()
|
||
accounts = read_accounts(path)
|
||
account = validate_account(account)
|
||
if not any(item.student_id == old_student_id for item in accounts):
|
||
raise ValueError(f"未找到学生账户: {old_student_id}")
|
||
if account.student_id != old_student_id and any(item.student_id == account.student_id for item in accounts):
|
||
raise ValueError(f"学生ID已存在: {account.student_id}")
|
||
original_accounts = path.read_text(encoding="utf-8")
|
||
backup_dir = create_data_backup("admin-update-account", {path: original_accounts}, [old_student_id, format_account_row(account)])
|
||
atomic_write_text(path, replace_single_account_line(original_accounts, old_student_id, account))
|
||
try:
|
||
prune_data_backups(backup_dir.parent)
|
||
except OSError:
|
||
pass
|
||
return {"account": account_to_dict(account), "backup_id": backup_dir.name}
|
||
|
||
|
||
def render_accounts_text(path: Path, accounts: list[Account]) -> str:
|
||
original_text = path.read_text(encoding="utf-8")
|
||
accounts_by_id = {account.student_id: account for account in accounts}
|
||
return replace_account_lines(original_text, accounts_by_id)
|
||
|
||
|
||
def render_accounts_text_from_text(original_text: str, accounts: list[Account]) -> str:
|
||
accounts_by_id = {account.student_id: account for account in accounts}
|
||
return replace_account_lines(original_text, accounts_by_id)
|
||
|
||
|
||
def normalize_lines(lines: list[str] | None = None, line: str | None = None) -> list[str]:
|
||
values: list[str] = []
|
||
if line is not None:
|
||
values.append(line)
|
||
if lines is not None:
|
||
values.extend(lines)
|
||
result = [item.strip() for item in values if item and item.strip()]
|
||
if not result:
|
||
raise ValueError("登记内容不能为空")
|
||
return result
|
||
|
||
|
||
def register_class_record_lines(
|
||
classnotes_path: Path,
|
||
accounts_path: Path,
|
||
lines: list[str] | None = None,
|
||
line: str | None = None,
|
||
) -> dict:
|
||
input_lines = normalize_lines(lines=lines, line=line)
|
||
records = [parse_class_record_line(item) for item in input_lines]
|
||
record_lines = [class_record_to_line(record) for record in records]
|
||
if len(record_lines) != len(set(record_lines)):
|
||
raise DuplicateRecordError("本次提交包含重复上课记录")
|
||
|
||
existing_lines = {raw.strip() for raw in classnotes_path.read_text(encoding="utf-8").splitlines()}
|
||
duplicates = [item for item in record_lines if item in existing_lines]
|
||
if duplicates:
|
||
raise DuplicateRecordError(f"上课记录已存在: {duplicates[0]}")
|
||
|
||
accounts = read_accounts(accounts_path)
|
||
updated_accounts = list(accounts)
|
||
updated_account_ids: set[str] = set()
|
||
for record in records:
|
||
account_index = find_account_index(updated_accounts, record.student)
|
||
updated_accounts[account_index] = update_account_remaining(
|
||
updated_accounts[account_index],
|
||
-record.duration_hours,
|
||
)
|
||
updated_account_ids.add(updated_accounts[account_index].student_id)
|
||
|
||
original_classnotes = classnotes_path.read_text(encoding="utf-8")
|
||
original_accounts = accounts_path.read_text(encoding="utf-8")
|
||
separator = "" if not original_classnotes or original_classnotes.endswith("\n") else "\n"
|
||
new_classnotes = f"{original_classnotes}{separator}" + "".join(f"{item}\n" for item in record_lines)
|
||
updated_accounts_by_id = {
|
||
account.student_id: account
|
||
for account in updated_accounts
|
||
if account.student_id in updated_account_ids
|
||
}
|
||
new_accounts = replace_account_lines(original_accounts, updated_accounts_by_id)
|
||
backup_dir = create_data_backup(
|
||
"register-class-records",
|
||
{
|
||
accounts_path: original_accounts,
|
||
classnotes_path: original_classnotes,
|
||
},
|
||
record_lines,
|
||
)
|
||
try:
|
||
atomic_write_text(accounts_path, new_accounts)
|
||
atomic_write_text(classnotes_path, new_classnotes)
|
||
except Exception:
|
||
atomic_write_text(accounts_path, original_accounts)
|
||
atomic_write_text(classnotes_path, original_classnotes)
|
||
raise
|
||
try:
|
||
prune_data_backups(backup_dir.parent)
|
||
except OSError:
|
||
pass
|
||
return {"registered": len(record_lines), "lines": record_lines, "backup_id": backup_dir.name}
|
||
|
||
|
||
def register_payment_lines(
|
||
accounts_path: Path,
|
||
lines: list[str] | None = None,
|
||
line: str | None = None,
|
||
) -> dict:
|
||
input_lines = normalize_lines(lines=lines, line=line)
|
||
payments = [parse_payment_line(item) for item in input_lines]
|
||
accounts = read_accounts(accounts_path)
|
||
updated_accounts = list(accounts)
|
||
updated_account_ids: set[str] = set()
|
||
registered: list[str] = []
|
||
for student, payment in payments:
|
||
account_index = find_account_index(updated_accounts, student)
|
||
updated_accounts[account_index] = append_account_payment(updated_accounts[account_index], payment)
|
||
updated_account_ids.add(updated_accounts[account_index].student_id)
|
||
registered.append(f"{student}-{format_payment(payment)}")
|
||
original_accounts = accounts_path.read_text(encoding="utf-8")
|
||
updated_accounts_by_id = {
|
||
account.student_id: account
|
||
for account in updated_accounts
|
||
if account.student_id in updated_account_ids
|
||
}
|
||
backup_dir = create_data_backup(
|
||
"register-payments",
|
||
{accounts_path: original_accounts},
|
||
registered,
|
||
)
|
||
atomic_write_text(accounts_path, replace_account_lines(original_accounts, updated_accounts_by_id))
|
||
try:
|
||
prune_data_backups(backup_dir.parent)
|
||
except OSError:
|
||
pass
|
||
return {"registered": len(registered), "lines": registered, "backup_id": backup_dir.name}
|
||
|
||
|
||
def default_admin_tasks() -> dict:
|
||
return {"version": 1, "next_id": 1, "items": []}
|
||
|
||
|
||
def read_admin_tasks(path: Path) -> dict:
|
||
if not path.exists():
|
||
return default_admin_tasks()
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"管理任务文件 JSON 格式错误: {path}") from exc
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("管理任务文件必须是 JSON 对象")
|
||
payload.setdefault("version", 1)
|
||
payload.setdefault("next_id", 1)
|
||
payload.setdefault("items", [])
|
||
if not isinstance(payload["items"], list):
|
||
raise ValueError("管理任务 items 必须是数组")
|
||
return payload
|
||
|
||
|
||
def write_admin_tasks(path: Path, tasks: dict) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
atomic_write_text(path, json.dumps(tasks, ensure_ascii=False, indent=2) + "\n")
|
||
|
||
|
||
def task_to_dict(task: dict) -> dict:
|
||
return dict(task)
|
||
|
||
|
||
def find_admin_task(tasks: dict, task_id: int) -> dict:
|
||
for task in tasks["items"]:
|
||
if int(task.get("id", 0)) == task_id:
|
||
return task
|
||
raise ValueError(f"未找到管理任务: {task_id}")
|
||
|
||
|
||
def submit_correction_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()
|
||
corrected_line = str(item.get("corrected_line", "")).strip()
|
||
if not original_line or not corrected_line:
|
||
raise ValueError("纠错审核记录缺少原记录或修改后记录")
|
||
original = parse_class_record_line(original_line)
|
||
corrected = parse_class_record_line(corrected_line)
|
||
if original_line == corrected_line:
|
||
raise ValueError("原记录和修改后记录相同，无需提交审核")
|
||
task = {
|
||
"id": int(tasks["next_id"]),
|
||
"type": "class_record_correction",
|
||
"status": "pending",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"original_line": class_record_to_line(original),
|
||
"corrected_line": class_record_to_line(corrected),
|
||
"original": record_to_dict(original),
|
||
"corrected": record_to_dict(corrected),
|
||
}
|
||
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"]
|
||
if status_filter:
|
||
items = [item for item in items if item.get("status") == status_filter]
|
||
if task_type:
|
||
items = [item for item in items if item.get("type") == task_type]
|
||
return {
|
||
"version": tasks["version"],
|
||
"next_id": tasks["next_id"],
|
||
"count": len(items),
|
||
"items": [task_to_dict(item) for item in sorted(items, key=lambda item: int(item.get("id", 0)), reverse=True)],
|
||
}
|
||
|
||
|
||
def replace_class_record_line(original_text: str, original_line: str, corrected_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("原上课记录在正式文件中不存在，可能已被修改")
|
||
if original_line != corrected_line and any(line.strip() == corrected_line for line in lines):
|
||
raise ValueError("修改后的上课记录已存在，不能重复写入")
|
||
lines[matched[0]] = corrected_line
|
||
trailing_newline = "\n" if original_text.endswith("\n") 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)
|
||
if task.get("status") not in {"pending", "conflict"}:
|
||
raise ValueError("该任务已处理，不能重复操作")
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
task["status"] = status
|
||
task["updated_at"] = now
|
||
task["reviewed_at"] = now
|
||
if message:
|
||
task["message"] = message
|
||
write_admin_tasks(tasks_path, tasks)
|
||
return task_to_dict(task)
|
||
|
||
|
||
def reject_admin_task(tasks_path: Path, task_id: int) -> dict:
|
||
return mark_admin_task(tasks_path, task_id, "rejected")
|
||
|
||
|
||
def approve_correction_task(tasks_path: Path, classnotes_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_correction":
|
||
raise ValueError("该任务不是上课记录纠错")
|
||
if task.get("status") not in {"pending", "conflict"}:
|
||
raise ValueError("该任务已处理，不能重复批准")
|
||
|
||
original_line = str(task.get("original_line", "")).strip()
|
||
corrected_line = str(task.get("corrected_line", "")).strip()
|
||
parse_class_record_line(original_line)
|
||
corrected = parse_class_record_line(corrected_line)
|
||
corrected_line = class_record_to_line(corrected)
|
||
|
||
original_classnotes = classnotes_path.read_text(encoding="utf-8")
|
||
try:
|
||
new_classnotes = replace_class_record_line(original_classnotes, original_line, corrected_line)
|
||
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_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["corrected_line"] = corrected_line
|
||
task["corrected"] = record_to_dict(corrected)
|
||
new_tasks = json.dumps(tasks, ensure_ascii=False, indent=2) + "\n"
|
||
|
||
backup_dir = create_data_backup(
|
||
"admin-approve-correction",
|
||
{
|
||
classnotes_path: original_classnotes,
|
||
tasks_path: original_tasks,
|
||
},
|
||
[original_line, corrected_line],
|
||
)
|
||
try:
|
||
atomic_write_text(classnotes_path, new_classnotes)
|
||
atomic_write_text(tasks_path, new_tasks)
|
||
except Exception:
|
||
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)
|
||
return text.strip("._") or "未命名"
|
||
|
||
|
||
def sha1_text(value: str, length: int = 16) -> str:
|
||
return hashlib.sha1(value.encode("utf-8")).hexdigest()[:length]
|
||
|
||
|
||
def payload_bool(value: object) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, (int, float)):
|
||
return value != 0
|
||
text = str(value or "").strip().lower()
|
||
return text in {"1", "true", "yes", "y", "高", "高置信", "可信"}
|
||
|
||
|
||
def normalize_summary_date(value: object) -> str:
|
||
text = str(value or "").strip().replace(".", "-")
|
||
if not text:
|
||
raise ValueError("课程小结缺少日期")
|
||
parsed = datetime.strptime(text, "%Y-%m-%d").date()
|
||
return parsed.isoformat()
|
||
|
||
|
||
def normalize_time_range_text(value: object) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return ""
|
||
match = TIME_RANGE_RE.fullmatch(text)
|
||
if not match:
|
||
raise ValueError(f"课程小结时间段格式错误: {text}")
|
||
start_hour = int(match.group("sh"))
|
||
start_minute = int(match.group("sm"))
|
||
end_hour = int(match.group("eh"))
|
||
end_minute = int(match.group("em"))
|
||
if start_hour > 23 or end_hour > 23 or start_minute > 59 or end_minute > 59:
|
||
raise ValueError(f"课程小结时间段超出范围: {text}")
|
||
if end_hour * 60 + end_minute <= start_hour * 60 + start_minute:
|
||
raise ValueError(f"课程小结结束时间必须晚于开始时间: {text}")
|
||
return f"{start_hour:02d}:{start_minute:02d}-{end_hour:02d}:{end_minute:02d}"
|
||
|
||
|
||
def duration_text_from_minutes(minutes: int) -> str:
|
||
return f"{minutes // 60}小时{minutes % 60}分"
|
||
|
||
|
||
def duration_minutes_from_time_range(time_range: str) -> int | None:
|
||
if not time_range:
|
||
return None
|
||
match = TIME_RANGE_RE.fullmatch(time_range)
|
||
if not match:
|
||
return None
|
||
start = int(match.group("sh")) * 60 + int(match.group("sm"))
|
||
end = int(match.group("eh")) * 60 + int(match.group("em"))
|
||
return end - start if end > start else None
|
||
|
||
|
||
def duration_minutes_from_summary(summary: dict) -> int | None:
|
||
raw_duration = summary.get("duration") or summary.get("duration_text") or ""
|
||
if raw_duration:
|
||
return int(round(parse_hours_text(str(raw_duration)) * 60))
|
||
for key in ("duration_minutes", "minutes"):
|
||
value = summary.get(key)
|
||
if value not in (None, ""):
|
||
return int(round(float(value)))
|
||
for key in ("duration_hours", "hours"):
|
||
value = summary.get(key)
|
||
if value not in (None, ""):
|
||
return int(round(float(value) * 60))
|
||
return duration_minutes_from_time_range(str(summary.get("time_range") or ""))
|
||
|
||
|
||
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())
|
||
subject = parse_subject_code(str(raw.get("subject") or "").strip())
|
||
body = str(raw.get("body") or raw.get("content") or "").strip()
|
||
if not student:
|
||
raise ValueError("课程小结缺少学生")
|
||
if not body:
|
||
raise ValueError("课程小结缺少正文")
|
||
date_iso = normalize_summary_date(raw.get("date_iso") or raw.get("date") or raw.get("class_date"))
|
||
time_range = normalize_time_range_text(raw.get("time_range") or raw.get("raw_time") or raw.get("time") or "")
|
||
minutes = duration_minutes_from_summary({**raw, "time_range": time_range})
|
||
if time_range and minutes is not None:
|
||
time_minutes = duration_minutes_from_time_range(time_range)
|
||
if time_minutes is not None and abs(time_minutes - minutes) > 1:
|
||
raise ValueError(f"课程小结时间段和时长不一致: {time_range} / {duration_text_from_minutes(minutes)}")
|
||
source_id = str(raw.get("source_id") or "").strip()
|
||
if not source_id:
|
||
source_parts = [
|
||
str(raw.get("db") or ""),
|
||
str(raw.get("local_id") or ""),
|
||
student,
|
||
date_iso,
|
||
teacher,
|
||
subject,
|
||
time_range,
|
||
body[:200],
|
||
]
|
||
source_id = sha1_text("|".join(source_parts), 24)
|
||
summary = {
|
||
"source_id": source_id,
|
||
"student": student,
|
||
"date_iso": date_iso,
|
||
"time_range": time_range,
|
||
"duration_minutes": minutes,
|
||
"duration": duration_text_from_minutes(minutes) if minutes is not None else "",
|
||
"teacher": teacher,
|
||
"subject": subject,
|
||
"group": str(raw.get("group") or "").strip(),
|
||
"sender": str(raw.get("sender") or raw.get("sender_name") 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,
|
||
"recognition_source": str(raw.get("recognition_source") or raw.get("source") or "").strip(),
|
||
"confidence": str(raw.get("confidence") or "").strip(),
|
||
"teacher_trusted": payload_bool(raw.get("teacher_trusted") or raw.get("sender_teacher_trusted")),
|
||
"remark": str(raw.get("remark") or "").strip(),
|
||
}
|
||
if not summary["message_date"] and len(summary["message_time"]) >= 10:
|
||
summary["message_date"] = summary["message_time"][:10]
|
||
return summary
|
||
|
||
|
||
def course_summary_semantic_key(summary: dict) -> str:
|
||
body_digest = sha1_text(re.sub(r"\s+", "", str(summary.get("body") or "")), 12)
|
||
parts = [
|
||
summary.get("student", ""),
|
||
summary.get("date_iso", ""),
|
||
summary.get("teacher", ""),
|
||
normalize_subject(str(summary.get("subject") or "")),
|
||
summary.get("time_range", ""),
|
||
str(summary.get("duration_minutes") or ""),
|
||
body_digest,
|
||
]
|
||
return "|".join(str(part) for part in parts)
|
||
|
||
|
||
def default_course_summary_state() -> dict:
|
||
return {"version": 1, "seen_source_ids": [], "seen_semantic_keys": [], "batches": []}
|
||
|
||
|
||
def read_course_summary_state(path: Path) -> dict:
|
||
if not path.exists():
|
||
return default_course_summary_state()
|
||
try:
|
||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"课程小结状态文件 JSON 格式错误: {path}") from exc
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("课程小结状态文件必须是 JSON 对象")
|
||
payload.setdefault("version", 1)
|
||
payload.setdefault("seen_source_ids", [])
|
||
payload.setdefault("seen_semantic_keys", [])
|
||
payload.setdefault("batches", [])
|
||
return payload
|
||
|
||
|
||
def write_course_summary_state(path: Path, state: dict) -> None:
|
||
atomic_write_text(path, json.dumps(state, ensure_ascii=False, indent=2) + "\n")
|
||
|
||
|
||
def append_operation_log(path: Path, operation: str, status: str, **fields: object) -> str:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
log_id = f"{datetime.now().strftime('%Y%m%d%H%M%S')}-{sha1_text(json.dumps(fields, ensure_ascii=False, sort_keys=True), 8)}"
|
||
row = {
|
||
"id": log_id,
|
||
"created_at": now,
|
||
"operation": operation,
|
||
"status": status,
|
||
**fields,
|
||
}
|
||
with path.open("a", encoding="utf-8") as handle:
|
||
handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")
|
||
return log_id
|
||
|
||
|
||
def list_operation_logs(path: Path, limit: int = 100, operation: str = "", status_filter: str = "", student: str = "") -> dict:
|
||
rows: list[dict] = []
|
||
if path.exists():
|
||
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
||
if not raw_line.strip():
|
||
continue
|
||
try:
|
||
item = json.loads(raw_line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if operation and item.get("operation") != operation:
|
||
continue
|
||
if status_filter and item.get("status") != status_filter:
|
||
continue
|
||
if student and student not in str(item.get("student", "")):
|
||
continue
|
||
rows.append(item)
|
||
rows = rows[-limit:]
|
||
rows.reverse()
|
||
return {"count": len(rows), "items": rows}
|
||
|
||
|
||
def normalize_filter_date(value: str) -> str:
|
||
text = value.strip().replace(".", "-")
|
||
if not text:
|
||
return ""
|
||
parts = text.split("-")
|
||
if len(parts) == 3:
|
||
text = f"{int(parts[0]):04d}-{int(parts[1]):02d}-{int(parts[2]):02d}"
|
||
return date.fromisoformat(text).isoformat()
|
||
|
||
|
||
def parse_course_summary_file_identity(root: Path, path: Path) -> dict:
|
||
relative = path.relative_to(root)
|
||
student = canonical_name(relative.parts[0]) if relative.parts else ""
|
||
stem = path.stem
|
||
parts = stem.split("_")
|
||
teacher = ""
|
||
subject = ""
|
||
if len(parts) >= 3:
|
||
student = canonical_name(parts[0])
|
||
teacher = canonical_name("_".join(parts[1:-1]))
|
||
subject = parts[-1]
|
||
return {
|
||
"student": student,
|
||
"teacher": teacher,
|
||
"subject": normalize_subject(subject),
|
||
"source_path": str(path),
|
||
"relative_path": str(relative),
|
||
}
|
||
|
||
|
||
def parse_course_summary_title_date(title: str) -> str:
|
||
match = COURSE_SUMMARY_DATE_RE.search(title)
|
||
if not match:
|
||
return ""
|
||
raw = match.group("date").replace(".", "-")
|
||
parts = raw.split("-")
|
||
if len(parts) != 3:
|
||
return ""
|
||
normalized = f"{int(parts[0]):04d}-{int(parts[1]):02d}-{int(parts[2]):02d}"
|
||
try:
|
||
return date.fromisoformat(normalized).isoformat()
|
||
except ValueError:
|
||
return ""
|
||
|
||
|
||
def iter_course_summary_markdown(root: Path) -> Iterable[dict]:
|
||
if not root.exists():
|
||
return
|
||
for path in sorted(root.rglob("*.md")):
|
||
if not path.is_file():
|
||
continue
|
||
text = path.read_text(encoding="utf-8")
|
||
matches = list(COURSE_SUMMARY_HEADING_RE.finditer(text))
|
||
if not matches:
|
||
continue
|
||
identity = parse_course_summary_file_identity(root, path)
|
||
group = ""
|
||
group_match = re.search(r"^##\s+(.+)$", text[: matches[0].start()], flags=re.M)
|
||
if group_match:
|
||
group = group_match.group(1).strip()
|
||
for index, match in enumerate(matches):
|
||
title = match.group("title").strip()
|
||
start = match.end()
|
||
end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
|
||
body = text[start:end].strip()
|
||
body_without_meta = re.sub(r"^(?:>\s+.*\n)+\s*", "", body).strip()
|
||
body_text = body_without_meta or body
|
||
item_id = sha1_text(f"{identity['relative_path']}|{title}|{index}|{body_text[:200]}", 20)
|
||
yield {
|
||
**identity,
|
||
"id": item_id,
|
||
"title": title,
|
||
"date_iso": parse_course_summary_title_date(title),
|
||
"group": group,
|
||
"body": body_text,
|
||
"body_preview": body_text[:260] + ("..." if len(body_text) > 260 else ""),
|
||
}
|
||
|
||
|
||
def course_summary_matches(item: dict, q: str, student: str, teacher: str, subject: str, date_from: str, date_to: str) -> bool:
|
||
if student and student not in str(item.get("student", "")):
|
||
return False
|
||
if teacher and teacher not in str(item.get("teacher", "")):
|
||
return False
|
||
if subject and normalize_subject(subject) not in str(item.get("subject", "")):
|
||
return False
|
||
item_date = str(item.get("date_iso") or "")
|
||
if date_from and (not item_date or item_date < date_from):
|
||
return False
|
||
if date_to and (not item_date or item_date > date_to):
|
||
return False
|
||
if q:
|
||
haystack = "\n".join(
|
||
str(item.get(key, ""))
|
||
for key in ("student", "teacher", "subject", "title", "group", "body", "relative_path")
|
||
)
|
||
return q in haystack
|
||
return True
|
||
|
||
|
||
def course_summary_matched_fields(item: dict, q: str) -> list[str]:
|
||
if not q:
|
||
return []
|
||
labels = {
|
||
"student": "学生",
|
||
"teacher": "老师",
|
||
"subject": "科目",
|
||
"title": "标题",
|
||
"group": "群名",
|
||
"body": "正文",
|
||
"relative_path": "来源",
|
||
}
|
||
return [label for key, label in labels.items() if q in str(item.get(key, ""))]
|
||
|
||
|
||
def query_course_summaries(
|
||
root: Path,
|
||
q: str = "",
|
||
student: str = "",
|
||
teacher: str = "",
|
||
subject: str = "",
|
||
date_from: str = "",
|
||
date_to: str = "",
|
||
limit: int = 200,
|
||
) -> dict:
|
||
normalized_from = normalize_filter_date(date_from)
|
||
normalized_to = normalize_filter_date(date_to)
|
||
if normalized_from and normalized_to and normalized_from > normalized_to:
|
||
raise ValueError("开始日期不能晚于结束日期")
|
||
keyword = q.strip()
|
||
matched = [
|
||
item
|
||
for item in iter_course_summary_markdown(root)
|
||
if course_summary_matches(
|
||
item,
|
||
keyword,
|
||
canonical_name(student.strip()),
|
||
canonical_name(teacher.strip()),
|
||
subject.strip(),
|
||
normalized_from,
|
||
normalized_to,
|
||
)
|
||
]
|
||
for item in matched:
|
||
item["matched_fields"] = course_summary_matched_fields(item, keyword)
|
||
matched.sort(
|
||
key=lambda item: (
|
||
str(item.get("date_iso") or "0000-00-00"),
|
||
str(item.get("student") or ""),
|
||
str(item.get("teacher") or ""),
|
||
str(item.get("subject") or ""),
|
||
str(item.get("title") or ""),
|
||
),
|
||
reverse=True,
|
||
)
|
||
limited = matched[:limit]
|
||
return {
|
||
"count": len(matched),
|
||
"returned": len(limited),
|
||
"items": limited,
|
||
}
|
||
|
||
|
||
def course_summary_path(root: Path, summary: dict) -> Path:
|
||
student = safe_filename_part(summary["student"])
|
||
teacher = safe_filename_part(summary["teacher"] or "待核对老师")
|
||
subject = safe_filename_part(summary["subject"] or "待核对科目")
|
||
return root / student / f"{student}_{teacher}_{subject}.md"
|
||
|
||
|
||
def course_summary_heading(summary: dict, existing_headings: set[str]) -> str:
|
||
subject = normalize_subject(str(summary.get("subject") or "待核对科目"))
|
||
time_range = str(summary.get("time_range") or "").strip()
|
||
base = f"{summary['date_iso']} {time_range + ' ' if time_range else ''}{subject}课堂小结"
|
||
if base not in existing_headings:
|
||
return base
|
||
return f"{base}（{sha1_text(str(summary.get('source_id') or '') + str(summary.get('body') or ''), 8)}）"
|
||
|
||
|
||
def save_course_summary_markdown(root: Path, summary: dict) -> dict:
|
||
path = course_summary_path(root, summary)
|
||
existing = path.read_text(encoding="utf-8") if path.exists() else ""
|
||
existing_headings = set(re.findall(r"^###\s+(.+)$", existing, flags=re.M))
|
||
existing_compact = re.sub(r"\s+", "", existing)
|
||
body = str(summary.get("body") or "").rstrip()
|
||
body_compact = re.sub(r"\s+", "", body)
|
||
if body_compact and body_compact in existing_compact:
|
||
return {"path": str(path), "added": False}
|
||
|
||
group_name = str(summary.get("group") or summary["student"])
|
||
heading = course_summary_heading(summary, existing_headings)
|
||
lines: list[str] = []
|
||
if not existing.strip():
|
||
lines.extend([f"# {path.stem}", "", f"## {group_name}", ""])
|
||
elif f"## {group_name}" not in existing:
|
||
lines.extend(["", f"## {group_name}", ""])
|
||
lines.extend(
|
||
[
|
||
f"### {heading}",
|
||
"",
|
||
f"> 来源ID：`{summary['source_id']}`",
|
||
f"> 发送时间：`{summary.get('message_time') or ''}`",
|
||
f"> 发送者：`{summary.get('sender') or ''}`",
|
||
"",
|
||
body,
|
||
"",
|
||
]
|
||
)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("a", encoding="utf-8") as handle:
|
||
if existing and not existing.endswith("\n"):
|
||
handle.write("\n")
|
||
handle.write("\n".join(lines).rstrip() + "\n")
|
||
return {"path": str(path), "added": True, "heading": heading}
|
||
|
||
|
||
def message_date_after_class_date(summary: dict) -> bool:
|
||
message_date = str(summary.get("message_date") or "")
|
||
if len(message_date) != 10:
|
||
return False
|
||
try:
|
||
return date.fromisoformat(str(summary["date_iso"])) > date.fromisoformat(message_date)
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def course_summary_to_class_record_line(summary: dict) -> str:
|
||
date_iso = normalize_summary_date(summary.get("date_iso"))
|
||
record_date = date_iso.replace("-", ".")
|
||
weekday = WEEKDAYS[date.fromisoformat(date_iso).weekday()]
|
||
time_range = normalize_time_range_text(summary.get("time_range"))
|
||
minutes = summary.get("duration_minutes")
|
||
if minutes is None:
|
||
raise ValueError("课程小结缺少时长")
|
||
duration = duration_text_from_minutes(int(minutes))
|
||
return (
|
||
f"{record_date}-{weekday}-{time_range}-{summary['student']}-"
|
||
f"{duration}-{summary['teacher']}-{normalize_subject(str(summary['subject']))}"
|
||
)
|
||
|
||
|
||
def auto_register_reasons(summary: dict, classnotes_path: Path, accounts_path: Path) -> tuple[list[str], str]:
|
||
reasons: list[str] = []
|
||
confidence = str(summary.get("confidence") or "").lower()
|
||
recognition_source = str(summary.get("recognition_source") or "")
|
||
if confidence not in HIGH_CONFIDENCE_VALUES and recognition_source not in AUTO_RECOGNITION_SOURCES:
|
||
reasons.append("识别置信度不足")
|
||
if not summary.get("teacher_trusted"):
|
||
reasons.append("发送者老师映射未确认")
|
||
if summary.get("teacher") in UNKNOWN_TEACHERS:
|
||
reasons.append("老师待核对")
|
||
if normalize_subject(str(summary.get("subject") or "")) in UNKNOWN_SUBJECTS:
|
||
reasons.append("科目待核对")
|
||
if not summary.get("time_range"):
|
||
reasons.append("时间段缺失")
|
||
if summary.get("duration_minutes") is None:
|
||
reasons.append("时长缺失")
|
||
if message_date_after_class_date(summary):
|
||
reasons.append("课程日期晚于消息发送日期")
|
||
|
||
line = ""
|
||
try:
|
||
line = course_summary_to_class_record_line(summary)
|
||
parse_class_record_line(line)
|
||
except ValueError as exc:
|
||
reasons.append(str(exc))
|
||
|
||
try:
|
||
find_account_index(read_accounts(accounts_path), str(summary["student"]))
|
||
except ValueError as exc:
|
||
reasons.append(str(exc))
|
||
|
||
if line:
|
||
existing_lines = {raw.strip() for raw in classnotes_path.read_text(encoding="utf-8").splitlines()}
|
||
if line in existing_lines:
|
||
reasons.append("classnotes 已存在同一条上课记录")
|
||
return reasons, line
|
||
|
||
|
||
def create_course_summary_review_task(
|
||
tasks_path: Path,
|
||
summary: dict,
|
||
proposed_line: str,
|
||
reasons: list[str],
|
||
saved_path: str = "",
|
||
) -> dict:
|
||
tasks = read_admin_tasks(tasks_path)
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
task = {
|
||
"id": int(tasks["next_id"]),
|
||
"type": "course_summary_review",
|
||
"status": "pending",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"source_id": summary["source_id"],
|
||
"student": summary["student"],
|
||
"summary": summary,
|
||
"proposed_line": proposed_line,
|
||
"reasons": reasons,
|
||
"saved_path": saved_path,
|
||
}
|
||
tasks["next_id"] = int(tasks["next_id"]) + 1
|
||
tasks["items"].append(task)
|
||
write_admin_tasks(tasks_path, tasks)
|
||
return task_to_dict(task)
|
||
|
||
|
||
def approve_course_summary_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") != "course_summary_review":
|
||
raise ValueError("该任务不是课程小结审核")
|
||
if task.get("status") not in {"pending", "conflict"}:
|
||
raise ValueError("该任务已处理，不能重复批准")
|
||
|
||
proposed_line = str(task.get("proposed_line") or "").strip()
|
||
if not proposed_line:
|
||
summary = dict(task.get("summary") or {})
|
||
proposed_line = course_summary_to_class_record_line(summary)
|
||
|
||
try:
|
||
result = register_class_record_lines(classnotes_path, accounts_path, line=proposed_line)
|
||
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
|
||
|
||
task["status"] = "approved"
|
||
task["updated_at"] = datetime.now().isoformat(timespec="seconds")
|
||
task["reviewed_at"] = task["updated_at"]
|
||
task["registered_line"] = proposed_line
|
||
task["backup_id"] = result.get("backup_id", "")
|
||
write_admin_tasks(tasks_path, tasks)
|
||
return {"task": task_to_dict(task), "backup_id": result.get("backup_id", "")}
|
||
|
||
|
||
def approve_admin_task(
|
||
tasks_path: Path,
|
||
classnotes_path: Path,
|
||
accounts_path: Path,
|
||
task_id: int,
|
||
) -> dict:
|
||
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") == "course_summary_review":
|
||
return approve_course_summary_task(tasks_path, classnotes_path, accounts_path, task_id)
|
||
raise ValueError("不支持的审核任务类型")
|
||
|
||
|
||
def ingest_course_summaries(
|
||
*,
|
||
classnotes_path: Path,
|
||
accounts_path: Path,
|
||
tasks_path: Path,
|
||
summaries_root: Path,
|
||
state_path: Path,
|
||
operation_logs_path: Path,
|
||
batch_id: str,
|
||
window: dict,
|
||
students: list[str],
|
||
summaries: list[dict],
|
||
) -> dict:
|
||
state = read_course_summary_state(state_path)
|
||
seen_source_ids = set(str(item) for item in state.get("seen_source_ids", []))
|
||
seen_semantic_keys = set(str(item) for item in state.get("seen_semantic_keys", []))
|
||
result = {
|
||
"received": len(summaries),
|
||
"saved": 0,
|
||
"auto_registered": 0,
|
||
"review_pending": 0,
|
||
"duplicates": 0,
|
||
"rejected": 0,
|
||
"operation_log_ids": [],
|
||
"items": [],
|
||
}
|
||
|
||
for raw in summaries:
|
||
normalized: dict | None = None
|
||
try:
|
||
normalized = normalize_course_summary(raw)
|
||
source_id = normalized["source_id"]
|
||
semantic_key = course_summary_semantic_key(normalized)
|
||
if source_id in seen_source_ids or semantic_key in seen_semantic_keys:
|
||
result["duplicates"] += 1
|
||
log_id = append_operation_log(
|
||
operation_logs_path,
|
||
"course_summary_ingest",
|
||
"duplicate",
|
||
batch_id=batch_id,
|
||
source_id=source_id,
|
||
student=normalized["student"],
|
||
)
|
||
result["operation_log_ids"].append(log_id)
|
||
result["items"].append({"source_id": source_id, "status": "duplicate"})
|
||
continue
|
||
|
||
saved = save_course_summary_markdown(summaries_root, normalized)
|
||
result["saved"] += 1 if saved.get("added") else 0
|
||
|
||
reasons, proposed_line = auto_register_reasons(normalized, classnotes_path, accounts_path)
|
||
if reasons:
|
||
task = create_course_summary_review_task(
|
||
tasks_path,
|
||
normalized,
|
||
proposed_line,
|
||
reasons,
|
||
saved_path=str(saved.get("path") or ""),
|
||
)
|
||
result["review_pending"] += 1
|
||
status_value = "review"
|
||
task_id = task.get("id")
|
||
backup_id = ""
|
||
else:
|
||
register_result = register_class_record_lines(classnotes_path, accounts_path, line=proposed_line)
|
||
result["auto_registered"] += 1
|
||
status_value = "auto_registered"
|
||
task_id = None
|
||
backup_id = str(register_result.get("backup_id") or "")
|
||
|
||
seen_source_ids.add(source_id)
|
||
seen_semantic_keys.add(semantic_key)
|
||
log_id = append_operation_log(
|
||
operation_logs_path,
|
||
"course_summary_ingest",
|
||
status_value,
|
||
batch_id=batch_id,
|
||
source_id=source_id,
|
||
student=normalized["student"],
|
||
teacher=normalized.get("teacher", ""),
|
||
subject=normalized.get("subject", ""),
|
||
proposed_line=proposed_line,
|
||
reasons=reasons,
|
||
task_id=task_id,
|
||
backup_id=backup_id,
|
||
saved_path=str(saved.get("path") or ""),
|
||
)
|
||
result["operation_log_ids"].append(log_id)
|
||
result["items"].append(
|
||
{
|
||
"source_id": source_id,
|
||
"status": status_value,
|
||
"task_id": task_id,
|
||
"backup_id": backup_id,
|
||
"reasons": reasons,
|
||
}
|
||
)
|
||
except Exception as exc:
|
||
result["rejected"] += 1
|
||
source_id = str((normalized or raw).get("source_id") or "")
|
||
log_id = append_operation_log(
|
||
operation_logs_path,
|
||
"course_summary_ingest",
|
||
"rejected",
|
||
batch_id=batch_id,
|
||
source_id=source_id,
|
||
student=str((normalized or raw).get("student") or ""),
|
||
error=str(exc),
|
||
)
|
||
result["operation_log_ids"].append(log_id)
|
||
result["items"].append({"source_id": source_id, "status": "rejected", "error": str(exc)})
|
||
|
||
state["seen_source_ids"] = sorted(seen_source_ids)
|
||
state["seen_semantic_keys"] = sorted(seen_semantic_keys)
|
||
state.setdefault("batches", []).append(
|
||
{
|
||
"batch_id": batch_id,
|
||
"received_at": datetime.now().isoformat(timespec="seconds"),
|
||
"window": window,
|
||
"students": students,
|
||
"result": {key: result[key] for key in ("received", "saved", "auto_registered", "review_pending", "duplicates", "rejected")},
|
||
}
|
||
)
|
||
state["batches"] = state["batches"][-200:]
|
||
write_course_summary_state(state_path, state)
|
||
return result
|
||
|
||
|
||
def parse_record_date(text: str) -> date:
|
||
return datetime.strptime(text, "%Y.%m.%d").date()
|
||
|
||
|
||
def month_range(year: int, month: int) -> tuple[date, date]:
|
||
start = date(year, month, 1)
|
||
if month == 12:
|
||
return start, date(year, 12, 31)
|
||
return start, date(year, month + 1, 1) - timedelta(days=1)
|
||
|
||
|
||
def safe_date(year: int, month: int, day: int) -> date | None:
|
||
try:
|
||
return date(year, month, day)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def safe_month_range(year: int, month: int) -> tuple[date, date] | None:
|
||
try:
|
||
return month_range(year, month)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def previous_weekend_range(today: date) -> tuple[date, date]:
|
||
days_since_monday = today.weekday()
|
||
current_week_monday = today - timedelta(days=days_since_monday)
|
||
previous_saturday = current_week_monday - timedelta(days=2)
|
||
previous_sunday = current_week_monday - timedelta(days=1)
|
||
return previous_saturday, previous_sunday
|
||
|
||
|
||
def invalid_date_range() -> tuple[date, date]:
|
||
return date.max, date.min
|
||
|
||
|
||
def merge_optional_range(
|
||
current: tuple[date | None, date | None],
|
||
new_range: tuple[date, date] | None,
|
||
) -> tuple[date, date]:
|
||
if new_range is None:
|
||
return merge_range(current, invalid_date_range())
|
||
return merge_range(current, new_range)
|
||
|
||
|
||
def merge_range(
|
||
current: tuple[date | None, date | None],
|
||
new_range: tuple[date, date],
|
||
) -> tuple[date, date]:
|
||
current_start, current_end = current
|
||
new_start, new_end = new_range
|
||
if current_start is None or current_end is None:
|
||
return new_start, new_end
|
||
return max(current_start, new_start), min(current_end, new_end)
|
||
|
||
|
||
def normalize_range(
|
||
start: date,
|
||
end: date,
|
||
end_year_was_explicit: bool,
|
||
) -> tuple[date, date]:
|
||
if start <= end:
|
||
return start, end
|
||
if not end_year_was_explicit and start.month > end.month:
|
||
adjusted_end = safe_date(start.year + 1, end.month, end.day)
|
||
if adjusted_end:
|
||
return start, adjusted_end
|
||
return end, start
|
||
|
||
|
||
def parse_explicit_date_range(match: re.Match[str], today: date) -> tuple[date, date] | None:
|
||
start_year_text = match.group("sy")
|
||
end_year_text = match.group("ey")
|
||
start_month = int(match.group("sm"))
|
||
start_day = int(match.group("sd"))
|
||
end_month = int(match.group("em")) if match.group("em") else start_month
|
||
end_day = int(match.group("ed"))
|
||
start_year = int(start_year_text) if start_year_text else today.year
|
||
end_year = int(end_year_text) if end_year_text else start_year
|
||
|
||
start = safe_date(start_year, start_month, start_day)
|
||
end = safe_date(end_year, end_month, end_day)
|
||
if start is None or end is None:
|
||
return None
|
||
return normalize_range(start, end, end_year_was_explicit=bool(end_year_text))
|
||
|
||
|
||
def blank_spans(text: str, spans: list[tuple[int, int]]) -> str:
|
||
chars = list(text)
|
||
for start, end in spans:
|
||
for index in range(start, end):
|
||
chars[index] = " "
|
||
return "".join(chars)
|
||
|
||
|
||
def parse_date_filters(query: str, today: date) -> tuple[date | None, date | None]:
|
||
result: tuple[date | None, date | None] = (None, None)
|
||
range_spans: list[tuple[int, int]] = []
|
||
|
||
for pattern in (CHINESE_DATE_RANGE_RE, NUMERIC_DATE_RANGE_RE):
|
||
for match in pattern.finditer(query):
|
||
range_spans.append(match.span())
|
||
result = merge_optional_range(result, parse_explicit_date_range(match, today))
|
||
|
||
single_date_query = blank_spans(query, range_spans)
|
||
|
||
if "今天" in query:
|
||
result = merge_range(result, (today, today))
|
||
if "昨天" in query:
|
||
yesterday = today - timedelta(days=1)
|
||
result = merge_range(result, (yesterday, yesterday))
|
||
if "前天" in query:
|
||
day_before_yesterday = today - timedelta(days=2)
|
||
result = merge_range(result, (day_before_yesterday, day_before_yesterday))
|
||
if "本月" in query:
|
||
result = merge_range(result, month_range(today.year, today.month))
|
||
if "上月" in query:
|
||
last_month_year = today.year if today.month > 1 else today.year - 1
|
||
last_month = today.month - 1 if today.month > 1 else 12
|
||
result = merge_range(result, month_range(last_month_year, last_month))
|
||
if "上周末" in query:
|
||
result = merge_range(result, previous_weekend_range(today))
|
||
|
||
for match in re.finditer(r"最近\s*(\d+)\s*天", query):
|
||
days = int(match.group(1))
|
||
if days < 1:
|
||
result = merge_range(result, invalid_date_range())
|
||
else:
|
||
result = merge_range(result, (today - timedelta(days=days - 1), today))
|
||
|
||
for match in re.finditer(r"(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*[日号]?", single_date_query):
|
||
year, month, day = map(int, match.groups())
|
||
day_date = safe_date(year, month, day)
|
||
result = merge_optional_range(result, (day_date, day_date) if day_date else None)
|
||
|
||
for match in re.finditer(r"(?<!\d)(\d{4})[./-](\d{1,2})[./-](\d{1,2})(?!\d)", single_date_query):
|
||
year, month, day = map(int, match.groups())
|
||
day_date = safe_date(year, month, day)
|
||
result = merge_optional_range(result, (day_date, day_date) if day_date else None)
|
||
|
||
for match in re.finditer(r"(\d{4})\s*年\s*(\d{1,2})\s*月(?!\s*\d)", single_date_query):
|
||
year, month = map(int, match.groups())
|
||
result = merge_optional_range(result, safe_month_range(year, month))
|
||
|
||
for match in re.finditer(r"(?<!\d)(\d{1,2})\s*月\s*(\d{1,2})\s*[日号]?", single_date_query):
|
||
month, day = map(int, match.groups())
|
||
day_date = safe_date(today.year, month, day)
|
||
result = merge_optional_range(result, (day_date, day_date) if day_date else None)
|
||
|
||
for match in re.finditer(r"(?<!\d)(\d{1,2})[./](\d{1,2})(?!\d)", single_date_query):
|
||
month, day = map(int, match.groups())
|
||
day_date = safe_date(today.year, month, day)
|
||
result = merge_optional_range(result, (day_date, day_date) if day_date else None)
|
||
|
||
for match in re.finditer(r"(?<!\d)(\d{1,2})\s*月(?!\s*\d)", single_date_query):
|
||
month = int(match.group(1))
|
||
result = merge_optional_range(result, safe_month_range(today.year, month))
|
||
|
||
return result
|
||
|
||
|
||
def unique_sorted(items: Iterable[str]) -> list[str]:
|
||
return sorted(set(items))
|
||
|
||
|
||
def role_hint_before(query: str, name: str, role: str) -> bool:
|
||
for match in re.finditer(re.escape(name), query):
|
||
prefix = query[max(0, match.start() - 4) : match.start()]
|
||
suffix = query[match.end() : match.end() + 4]
|
||
if any(word in prefix or word in suffix for word in ROLE_WORDS[role]):
|
||
return True
|
||
return False
|
||
|
||
|
||
def detect_people(query: str, records: list[ClassRecord]) -> tuple[list[str], list[str]]:
|
||
students = set()
|
||
teachers = set()
|
||
canonical_query = query
|
||
for alias, canonical in FALLBACK_ALIASES.items():
|
||
if alias in query and canonical != alias:
|
||
canonical_query += f" {canonical}"
|
||
|
||
for name in unique_sorted(record.student for record in records):
|
||
if name in canonical_query:
|
||
students.add(name)
|
||
for name in unique_sorted(record.teacher for record in records):
|
||
if name in canonical_query:
|
||
teachers.add(name)
|
||
|
||
for name in sorted(students & teachers):
|
||
student_hint = role_hint_before(canonical_query, name, "student")
|
||
teacher_hint = role_hint_before(canonical_query, name, "teacher")
|
||
if student_hint and not teacher_hint:
|
||
teachers.discard(name)
|
||
elif teacher_hint and not student_hint:
|
||
students.discard(name)
|
||
|
||
return sorted(students), sorted(teachers)
|
||
|
||
|
||
def parse_subject_code(text: str) -> str:
|
||
value = text.strip()
|
||
if value in SUBJECTS:
|
||
return value
|
||
result: list[str] = []
|
||
index = 0
|
||
while index < len(value):
|
||
if value.startswith("道法", index):
|
||
result.append("道法")
|
||
index += 2
|
||
continue
|
||
char = value[index]
|
||
result.append(SUBJECT_ALIASES.get(char, char))
|
||
index += 1
|
||
return "".join(result)
|
||
|
||
|
||
def normalize_subject(text: str) -> str:
|
||
return parse_subject_code(text).strip()
|
||
|
||
|
||
def detect_subjects(query: str) -> list[str]:
|
||
subjects = {subject for subject in SUBJECTS if subject in query}
|
||
for alias, subject in SUBJECT_ALIASES.items():
|
||
if re.search(rf"(?<![\u4e00-\u9fff]){re.escape(alias)}(?![\u4e00-\u9fff])", query):
|
||
subjects.add(subject)
|
||
for match in re.finditer(r"[一二三四五六七八九高初]*([数语英物化生史地政]|道法)+课", query):
|
||
parsed = parse_subject_code(match.group(0).replace("课", ""))
|
||
for subject in SUBJECTS:
|
||
if subject in parsed:
|
||
subjects.add(subject)
|
||
return sorted(subjects)
|
||
|
||
|
||
def build_query_spec(query: str, records: list[ClassRecord], today: date | None = None) -> QuerySpec:
|
||
effective_today = today or date.today()
|
||
start_date, end_date = parse_date_filters(query, effective_today)
|
||
students, teachers = detect_people(query, records)
|
||
subjects = detect_subjects(query)
|
||
return QuerySpec(
|
||
raw_query=query,
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
students=students,
|
||
teachers=teachers,
|
||
subjects=subjects,
|
||
)
|
||
|
||
|
||
def filter_records(records: list[ClassRecord], spec: QuerySpec) -> list[ClassRecord]:
|
||
matched: list[ClassRecord] = []
|
||
for record in records:
|
||
record_date = parse_record_date(record.date)
|
||
if spec.start_date and record_date < spec.start_date:
|
||
continue
|
||
if spec.end_date and record_date > spec.end_date:
|
||
continue
|
||
if spec.students and record.student not in spec.students:
|
||
continue
|
||
if spec.teachers and record.teacher not in spec.teachers:
|
||
continue
|
||
if spec.subjects and not any(subject in record.subject for subject in spec.subjects):
|
||
continue
|
||
matched.append(record)
|
||
return sorted(matched, key=lambda item: (item.date, item.time, item.student, item.teacher, item.subject))
|
||
|
||
|
||
def format_date_range(spec: QuerySpec) -> str:
|
||
if not spec.start_date and not spec.end_date:
|
||
return "未限制"
|
||
if spec.start_date and spec.end_date and spec.start_date > spec.end_date:
|
||
return "无效日期"
|
||
if spec.start_date == spec.end_date:
|
||
return spec.start_date.strftime("%Y.%m.%d")
|
||
start = spec.start_date.strftime("%Y.%m.%d") if spec.start_date else "不限"
|
||
end = spec.end_date.strftime("%Y.%m.%d") if spec.end_date else "不限"
|
||
return f"{start} 至 {end}"
|
||
|
||
|
||
def summarize_records(records: list[ClassRecord]) -> dict:
|
||
by_student: defaultdict[str, float] = defaultdict(float)
|
||
by_teacher: defaultdict[str, float] = defaultdict(float)
|
||
by_subject: defaultdict[str, float] = defaultdict(float)
|
||
for record in records:
|
||
by_student[record.student] += record.duration_hours
|
||
by_teacher[record.teacher] += record.duration_hours
|
||
by_subject[record.subject] += record.duration_hours
|
||
return {
|
||
"count": len(records),
|
||
"total_hours": round(sum(record.duration_hours for record in records), 2),
|
||
"students": dict(sorted((key, round(value, 2)) for key, value in by_student.items())),
|
||
"teachers": dict(sorted((key, round(value, 2)) for key, value in by_teacher.items())),
|
||
"subjects": dict(sorted((key, round(value, 2)) for key, value in by_subject.items())),
|
||
}
|
||
|
||
|
||
def record_to_dict(record: ClassRecord) -> dict:
|
||
return {
|
||
"date": record.date,
|
||
"weekday": record.weekday,
|
||
"time": record.time,
|
||
"student": record.student,
|
||
"duration": record.duration,
|
||
"duration_hours": record.duration_hours,
|
||
"teacher": record.teacher,
|
||
"subject": record.subject,
|
||
}
|
||
|
||
|
||
def account_to_dict(account: Account) -> dict:
|
||
return {
|
||
"student_id": account.student_id,
|
||
"student": account.student,
|
||
"payments": [{"date": payment.date, "hours": payment.hours} for payment in account.payments],
|
||
"payments_count": len(account.payments),
|
||
"remaining": account.remaining,
|
||
"account_status": account.account_status,
|
||
"note": account.note,
|
||
}
|
||
|
||
|
||
def has_filter_condition(spec: QuerySpec) -> bool:
|
||
return bool(spec.start_date or spec.end_date or spec.students or spec.teachers or spec.subjects)
|
||
|
||
|
||
def query_records(records: list[ClassRecord], query: str, limit: int = 200) -> dict:
|
||
spec = build_query_spec(query, records)
|
||
matched = filter_records(records, spec) if has_filter_condition(spec) else []
|
||
shown = matched[:limit] if limit > 0 else matched
|
||
return {
|
||
"query": {
|
||
"raw_query": spec.raw_query,
|
||
"date_range": format_date_range(spec),
|
||
"students": spec.students,
|
||
"teachers": spec.teachers,
|
||
"subjects": spec.subjects,
|
||
},
|
||
"summary": summarize_records(matched),
|
||
"records": [record_to_dict(record) for record in shown],
|
||
"total_records": len(matched),
|
||
"shown_records": len(shown),
|
||
}
|
||
|
||
|
||
def account_summary(accounts: list[Account]) -> dict:
|
||
result = {
|
||
"total": len(accounts),
|
||
"normal": 0,
|
||
"warning": 0,
|
||
"debt": 0,
|
||
"completed": 0,
|
||
"refunded": 0,
|
||
"other": 0,
|
||
}
|
||
for account in accounts:
|
||
status = account.account_status
|
||
if status == "正常":
|
||
result["normal"] += 1
|
||
elif status == "预警":
|
||
result["warning"] += 1
|
||
elif status == "欠费":
|
||
result["debt"] += 1
|
||
elif status == "结课":
|
||
result["completed"] += 1
|
||
elif status == "退费":
|
||
result["refunded"] += 1
|
||
else:
|
||
result["other"] += 1
|
||
return result
|
||
|
||
|
||
def filter_accounts(accounts: list[Account], keyword: str = "", status: str = "") -> list[Account]:
|
||
keyword = keyword.strip()
|
||
status = status.strip()
|
||
result = []
|
||
for account in accounts:
|
||
if keyword and keyword not in account.student and keyword not in account.student_id:
|
||
continue
|
||
if status and account.account_status != status:
|
||
continue
|
||
result.append(account)
|
||
return sorted(result, key=lambda item: (item.remaining, item.student))
|