Codex
957 lines
34 KiB
Python
957 lines
34 KiB
Python
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
from dataclasses import dataclass, replace
|
||
from datetime import date, datetime, timedelta
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
from pathlib import Path
|
||
import shutil
|
||
import tempfile
|
||
from typing import Iterable
|
||
|
||
|
||
SUBJECTS = ["数学", "语文", "英语", "物理", "化学", "生物", "历史", "地理", "政治", "道法"]
|
||
SUBJECT_ALIASES = {
|
||
"数": "数学",
|
||
"语": "语文",
|
||
"英": "英语",
|
||
"物": "物理",
|
||
"化": "化学",
|
||
"生": "生物",
|
||
"史": "历史",
|
||
"地": "地理",
|
||
"政": "政治",
|
||
}
|
||
FALLBACK_ALIASES = {
|
||
"施亿涵": "施忆涵",
|
||
"施凯其": "施凯萁",
|
||
"宇浩": "黄宇澔",
|
||
"肖恒罄": "肖恒馨",
|
||
}
|
||
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})$")
|
||
BACKUP_DIR_RE = re.compile(r"^\d{8}-\d{6}-\d{6}-")
|
||
BACKUP_KEEP_COUNT = 50
|
||
WEEKDAYS = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
|
||
ROLE_WORDS = {
|
||
"student": ("学生", "学员", "孩子", "同学"),
|
||
"teacher": ("老师", "教师"),
|
||
}
|
||
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)"
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Payment:
|
||
date: str
|
||
hours: float
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Account:
|
||
student_id: str
|
||
student: str
|
||
payments: list[Payment]
|
||
remaining: float
|
||
account_status: str
|
||
note: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ClassRecord:
|
||
date: str
|
||
weekday: str
|
||
time: str
|
||
student: str
|
||
duration: str
|
||
duration_hours: float
|
||
teacher: str
|
||
subject: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class QuerySpec:
|
||
raw_query: str
|
||
start_date: date | None
|
||
end_date: date | None
|
||
students: list[str]
|
||
teachers: list[str]
|
||
subjects: list[str]
|
||
|
||
|
||
class DuplicateRecordError(ValueError):
|
||
pass
|
||
|
||
|
||
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 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 atomic_write_text(path: Path, text: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
original_stat = path.stat() if path.exists() else None
|
||
temp_path: Path | None = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
|
||
temp_path = Path(handle.name)
|
||
handle.write(text)
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
if original_stat is not None:
|
||
os.chmod(temp_path, original_stat.st_mode & 0o7777)
|
||
try:
|
||
os.chown(temp_path, original_stat.st_uid, original_stat.st_gid)
|
||
except OSError:
|
||
pass
|
||
os.replace(temp_path, path)
|
||
temp_path = None
|
||
try:
|
||
dir_fd = os.open(path.parent, os.O_DIRECTORY)
|
||
try:
|
||
os.fsync(dir_fd)
|
||
finally:
|
||
os.close(dir_fd)
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
if temp_path is not None and temp_path.exists():
|
||
temp_path.unlink()
|
||
|
||
|
||
def apply_source_permissions(target: Path, source: Path) -> None:
|
||
if not source.exists():
|
||
return
|
||
source_stat = source.stat()
|
||
os.chmod(target, source_stat.st_mode & 0o7777)
|
||
try:
|
||
os.chown(target, source_stat.st_uid, source_stat.st_gid)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def data_backup_root(paths: Iterable[Path]) -> Path:
|
||
path_list = list(paths)
|
||
if not path_list:
|
||
raise ValueError("备份文件不能为空")
|
||
return path_list[0].parent / "backups"
|
||
|
||
|
||
def create_backup_directory(backup_root: Path, operation: str) -> Path:
|
||
backup_root.mkdir(parents=True, exist_ok=True)
|
||
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
||
base_name = f"{timestamp}-{operation}"
|
||
for index in range(1000):
|
||
suffix = "" if index == 0 else f"-{index}"
|
||
backup_dir = backup_root / f"{base_name}{suffix}"
|
||
try:
|
||
backup_dir.mkdir()
|
||
return backup_dir
|
||
except FileExistsError:
|
||
continue
|
||
raise RuntimeError("无法创建唯一备份目录")
|
||
|
||
|
||
def create_data_backup(
|
||
operation: str,
|
||
file_contents: dict[Path, str],
|
||
submitted_lines: list[str],
|
||
) -> Path:
|
||
backup_root = data_backup_root(file_contents.keys())
|
||
backup_dir = create_backup_directory(backup_root, operation)
|
||
try:
|
||
submitted_text = "\n".join(submitted_lines)
|
||
metadata = {
|
||
"backup_id": backup_dir.name,
|
||
"created_at": datetime.now().isoformat(timespec="microseconds"),
|
||
"operation": operation,
|
||
"submitted_lines_count": len(submitted_lines),
|
||
"submitted_lines_sha256": hashlib.sha256(submitted_text.encode("utf-8")).hexdigest(),
|
||
"files": [],
|
||
}
|
||
for source_path, content in file_contents.items():
|
||
target_path = backup_dir / source_path.name
|
||
atomic_write_text(target_path, content)
|
||
apply_source_permissions(target_path, source_path)
|
||
source_stat = source_path.stat()
|
||
metadata["files"].append(
|
||
{
|
||
"name": source_path.name,
|
||
"source_path": str(source_path),
|
||
"size": len(content.encode("utf-8")),
|
||
"mtime": source_stat.st_mtime,
|
||
}
|
||
)
|
||
metadata_path = backup_dir / "metadata.json"
|
||
atomic_write_text(
|
||
metadata_path,
|
||
json.dumps(metadata, ensure_ascii=False, indent=2) + "\n",
|
||
)
|
||
return backup_dir
|
||
except Exception:
|
||
shutil.rmtree(backup_dir, ignore_errors=True)
|
||
raise
|
||
|
||
|
||
def prune_data_backups(backup_root: Path, keep: int = BACKUP_KEEP_COUNT) -> None:
|
||
if keep < 1 or not backup_root.exists():
|
||
return
|
||
backup_dirs = sorted(
|
||
path
|
||
for path in backup_root.iterdir()
|
||
if path.is_dir() and BACKUP_DIR_RE.match(path.name)
|
||
)
|
||
for backup_dir in backup_dirs[:-keep]:
|
||
shutil.rmtree(backup_dir)
|
||
|
||
|
||
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 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 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 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))
|