Codex
712 lines
26 KiB
Python
712 lines
26 KiB
Python
from __future__ import annotations
|
||
|
||
from datetime import date, datetime, timedelta
|
||
import re
|
||
import secrets
|
||
from typing import Any
|
||
|
||
from .data import (
|
||
SUBJECTS,
|
||
canonical_teacher_name,
|
||
class_record_to_line,
|
||
course_summary_to_class_record_line,
|
||
extract_course_summary_from_text,
|
||
normalize_course_summary,
|
||
normalize_lines,
|
||
parse_class_record_line,
|
||
parse_payment_line,
|
||
)
|
||
from .repository import list_student_names
|
||
|
||
|
||
REGISTER_TYPES = {"class_record", "payment", "course_summary"}
|
||
FIELD_LABELS = {
|
||
"class_record": {
|
||
"student": "学生",
|
||
"date": "日期",
|
||
"time": "时间",
|
||
"teacher": "老师",
|
||
"subject": "科目",
|
||
"duration": "时长",
|
||
},
|
||
"payment": {
|
||
"student": "学生",
|
||
"date": "缴费日期",
|
||
"hours": "课时数",
|
||
},
|
||
"course_summary": {
|
||
"student": "学生",
|
||
"date": "日期",
|
||
"time": "时间",
|
||
"teacher": "老师",
|
||
"subject": "科目",
|
||
"body": "小结正文",
|
||
},
|
||
}
|
||
REQUIRED_FIELDS = {
|
||
"class_record": ["student", "date", "time", "teacher", "subject"],
|
||
"payment": ["student", "date", "hours"],
|
||
"course_summary": ["student", "date", "time", "teacher", "subject", "body"],
|
||
}
|
||
DATE_RE = re.compile(r"(?P<y>\d{4})[./年-]\s*(?P<m>\d{1,2})[./月-]\s*(?P<d>\d{1,2})")
|
||
MONTH_DAY_RE = re.compile(r"(?<!\d)(?P<m>\d{1,2})\s*[月./-]\s*(?P<d>\d{1,2})\s*(?:日|号)?")
|
||
TIME_RANGE_FLEX_RE = re.compile(
|
||
r"(?P<sh>\d{1,2})\s*(?:[:：点.])\s*(?P<sm>\d{1,2})?\s*"
|
||
r"(?:-|－|—|–|~|至|到)\s*"
|
||
r"(?P<eh>\d{1,2})\s*(?:[:：点.])\s*(?P<em>\d{1,2})?"
|
||
)
|
||
TIME_RANGE_HOUR_RE = re.compile(
|
||
r"(?<![\d./-])(?P<sh>\d{1,2})\s*(?:-|－|—|–|~|至|到)\s*(?P<eh>\d{1,2})(?![\d./-])"
|
||
)
|
||
TIME_RANGE_COMPACT_RE = re.compile(
|
||
r"(?<!\d)(?P<start>\d{3,4})\s*(?:-|－|—|–|~|至|到)\s*(?P<end>\d{3,4})(?!\d)"
|
||
)
|
||
TEACHER_RE = re.compile(r"(?P<teacher>[\u4e00-\u9fa5A-Za-z0-9]{1,8}老师)")
|
||
HOURS_RE = re.compile(r"(?P<hours>\d+(?:\.\d+)?)\s*(?:课时|小时)")
|
||
MAX_SESSION_AGE_SECONDS = 30 * 60
|
||
MAX_SESSIONS = 100
|
||
WAITING_PLACEHOLDER = "[待补充]"
|
||
_SESSIONS: dict[str, dict[str, Any]] = {}
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now()
|
||
|
||
|
||
def _cleanup_sessions() -> None:
|
||
cutoff = _now() - timedelta(seconds=MAX_SESSION_AGE_SECONDS)
|
||
stale = [
|
||
session_id
|
||
for session_id, session in _SESSIONS.items()
|
||
if session.get("updated_at", _now()) < cutoff
|
||
]
|
||
for session_id in stale:
|
||
_SESSIONS.pop(session_id, None)
|
||
if len(_SESSIONS) <= MAX_SESSIONS:
|
||
return
|
||
ordered = sorted(_SESSIONS.items(), key=lambda item: item[1].get("updated_at", _now()))
|
||
for session_id, _session in ordered[: len(_SESSIONS) - MAX_SESSIONS]:
|
||
_SESSIONS.pop(session_id, None)
|
||
|
||
|
||
def _session_for(conversation_id: str | None, register_type: str, text: str) -> tuple[str, dict[str, Any]]:
|
||
_cleanup_sessions()
|
||
if conversation_id and conversation_id in _SESSIONS:
|
||
session = _SESSIONS[conversation_id]
|
||
else:
|
||
conversation_id = secrets.token_urlsafe(16)
|
||
session = {"type": register_type, "text": text, "answers": {}, "created_at": _now()}
|
||
_SESSIONS[conversation_id] = session
|
||
session["type"] = register_type
|
||
if text:
|
||
session["text"] = text
|
||
session["updated_at"] = _now()
|
||
return conversation_id, session
|
||
|
||
|
||
def normalize_register_type(value: str) -> str:
|
||
text = str(value or "").strip()
|
||
aliases = {
|
||
"class": "class_record",
|
||
"class_records": "class_record",
|
||
"class-records": "class_record",
|
||
"上课记录": "class_record",
|
||
"payments": "payment",
|
||
"缴费": "payment",
|
||
"course-summaries": "course_summary",
|
||
"course_summaries": "course_summary",
|
||
"课程小结": "course_summary",
|
||
}
|
||
register_type = aliases.get(text, text)
|
||
if register_type not in REGISTER_TYPES:
|
||
raise ValueError("不支持的登记类型")
|
||
return register_type
|
||
|
||
|
||
def collect_input_lines(
|
||
text: str | None = None,
|
||
lines: list[str] | None = None,
|
||
register_type: str = "",
|
||
) -> list[str]:
|
||
if lines is not None:
|
||
return normalize_lines(lines=lines)
|
||
raw = str(text or "")
|
||
if register_type in {"class_record", "payment"}:
|
||
return normalize_lines(lines=raw.splitlines())
|
||
if "\n\n" in raw:
|
||
chunks = [item.strip() for item in re.split(r"\n\s*\n", raw) if item.strip()]
|
||
if chunks:
|
||
return chunks
|
||
return normalize_lines(line=raw)
|
||
|
||
|
||
def _field_labels(register_type: str) -> dict[str, str]:
|
||
return FIELD_LABELS.get(register_type, {})
|
||
|
||
|
||
def _questions_for_missing(register_type: str, missing_fields: list[str]) -> list[str]:
|
||
labels = _field_labels(register_type)
|
||
return [f"请补充{labels.get(field, field)}" for field in missing_fields]
|
||
|
||
|
||
def _known_students() -> list[str]:
|
||
try:
|
||
return list_student_names()
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def _normalize_date(value: object) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return ""
|
||
for match in DATE_RE.finditer(text):
|
||
year = int(match.group("y"))
|
||
month = int(match.group("m"))
|
||
day = int(match.group("d"))
|
||
try:
|
||
datetime(year, month, day)
|
||
except ValueError:
|
||
continue
|
||
return f"{year:04d}-{month:02d}-{day:02d}"
|
||
for match in MONTH_DAY_RE.finditer(text):
|
||
year = date.today().year
|
||
month = int(match.group("m"))
|
||
day = int(match.group("d"))
|
||
try:
|
||
datetime(year, month, day)
|
||
except ValueError:
|
||
continue
|
||
return f"{year:04d}-{month:02d}-{day:02d}"
|
||
return ""
|
||
|
||
|
||
def _record_date(value: str) -> str:
|
||
return value.replace("-", ".")
|
||
|
||
|
||
def _weekday(value: str) -> str:
|
||
weekdays = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
|
||
return weekdays[datetime.strptime(value, "%Y-%m-%d").date().weekday()]
|
||
|
||
|
||
def _normalize_time_range(value: object) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return ""
|
||
standard = re.search(r"\d{1,2}:\d{2}\s*(?:-|－|—|–|~|至|到)\s*\d{1,2}:\d{2}", text)
|
||
if standard:
|
||
raw = re.sub(r"\s*(?:－|—|–|~|至|到)\s*", "-", standard.group(0))
|
||
raw = re.sub(r"\s*-\s*", "-", raw)
|
||
start, end = raw.split("-", 1)
|
||
sh, sm = [int(part) for part in start.split(":", 1)]
|
||
eh, em = [int(part) for part in end.split(":", 1)]
|
||
else:
|
||
match = TIME_RANGE_FLEX_RE.search(text)
|
||
if match:
|
||
sh = int(match.group("sh"))
|
||
sm = int(match.group("sm") or 0)
|
||
eh = int(match.group("eh"))
|
||
em = int(match.group("em") or 0)
|
||
else:
|
||
compact = TIME_RANGE_COMPACT_RE.search(text)
|
||
if compact:
|
||
start = compact.group("start").zfill(4)
|
||
end = compact.group("end").zfill(4)
|
||
sh = int(start[:-2])
|
||
sm = int(start[-2:])
|
||
eh = int(end[:-2])
|
||
em = int(end[-2:])
|
||
else:
|
||
hour_only = TIME_RANGE_HOUR_RE.search(text)
|
||
if not hour_only:
|
||
return ""
|
||
sh = int(hour_only.group("sh"))
|
||
sm = 0
|
||
eh = int(hour_only.group("eh"))
|
||
em = 0
|
||
if sh > 23 or eh > 23 or sm > 59 or em > 59:
|
||
return ""
|
||
if eh * 60 + em <= sh * 60 + sm:
|
||
return ""
|
||
return f"{sh:02d}:{sm:02d}-{eh:02d}:{em:02d}"
|
||
|
||
|
||
def _duration_from_time_range(time_range: str) -> str:
|
||
start, end = time_range.split("-", 1)
|
||
sh, sm = [int(part) for part in start.split(":", 1)]
|
||
eh, em = [int(part) for part in end.split(":", 1)]
|
||
minutes = eh * 60 + em - sh * 60 - sm
|
||
return f"{minutes // 60}小时{minutes % 60}分"
|
||
|
||
|
||
def _duration_from_text(value: object) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return ""
|
||
if re.fullmatch(r"\d+小时\d+分", text):
|
||
return text
|
||
match = re.search(r"(?P<h>\d+)\s*小时\s*(?P<m>\d+)?\s*分?", text)
|
||
if match:
|
||
return f"{int(match.group('h'))}小时{int(match.group('m') or 0)}分"
|
||
match = re.search(r"(?P<hours>\d+(?:\.\d+)?)\s*(?:课时|小时)", text)
|
||
if match:
|
||
total = int(round(float(match.group("hours")) * 60))
|
||
return f"{total // 60}小时{total % 60}分"
|
||
return ""
|
||
|
||
|
||
def _extract_student(text: str) -> str:
|
||
match = re.search(r"(?:学生|学员)[:：\s]*(?P<student>[\u4e00-\u9fa5A-Za-z0-9]{2,8})", text)
|
||
if match:
|
||
return match.group("student")
|
||
for student in _known_students():
|
||
if student and student in text:
|
||
return student
|
||
match = re.search(r"^\s*(?P<student>[\u4e00-\u9fa5]{2,4})(?=\s|[,，。；;:]|\d)", text)
|
||
return match.group("student") if match else ""
|
||
|
||
|
||
def _extract_class_record_student(text: str, teacher: str, subject: str) -> str:
|
||
student = _extract_student(text)
|
||
if student:
|
||
return student
|
||
remainder = text
|
||
for pattern in (DATE_RE, MONTH_DAY_RE, TIME_RANGE_FLEX_RE, TIME_RANGE_COMPACT_RE, TIME_RANGE_HOUR_RE):
|
||
remainder = pattern.sub(" ", remainder)
|
||
remainder = re.sub(r"\d{1,2}:\d{2}\s*(?:-|－|—|–|~|至|到)\s*\d{1,2}:\d{2}", " ", remainder)
|
||
remainder = re.sub(r"星期[一二三四五六日]", " ", remainder)
|
||
remainder = re.sub(r"\d+(?:\.\d+)?\s*(?:课时|小时)", " ", remainder)
|
||
remainder = re.sub(r"\d+\s*小时\s*\d*\s*分?", " ", remainder)
|
||
if teacher:
|
||
remainder = re.sub(rf"{re.escape(teacher)}(?:老师|教师)?", " ", remainder)
|
||
if subject:
|
||
remainder = remainder.replace(subject, " ")
|
||
remainder = re.sub(r"(?:学生|学员|老师|教师|科目|课程|时间|日期|上课)", " ", remainder)
|
||
for candidate in re.findall(r"[\u4e00-\u9fa5]{2,4}", remainder):
|
||
if candidate not in SUBJECTS and candidate not in {"老师", "教师", "学生", "学员"}:
|
||
return candidate
|
||
return ""
|
||
|
||
|
||
def _extract_teacher(text: str) -> str:
|
||
match = TEACHER_RE.search(text)
|
||
return canonical_teacher_name(match.group("teacher")) if match else ""
|
||
|
||
|
||
def _normalize_teacher_hint(value: object) -> str:
|
||
text = canonical_teacher_name(str(value or "").strip())
|
||
return re.sub(r"^以下都是", "", text).strip()
|
||
|
||
|
||
def _extract_subject(text: str) -> str:
|
||
for subject in SUBJECTS:
|
||
if subject in text:
|
||
return subject
|
||
match = re.search(r"(?:科目|课程)[:：\s]*(?P<subject>[\u4e00-\u9fa5A-Za-z0-9]{1,12})", text)
|
||
return match.group("subject") if match else ""
|
||
|
||
|
||
def _merge_answers(fields: dict[str, str], answers: dict[str, str]) -> dict[str, str]:
|
||
merged = dict(fields)
|
||
for key, value in answers.items():
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
continue
|
||
if key == "followup" or key.startswith("items."):
|
||
continue
|
||
merged[key] = text
|
||
return merged
|
||
|
||
|
||
def _answers_for_item(answers: dict[str, str], index: int, *, include_legacy: bool = False) -> dict[str, str]:
|
||
prefix = f"items.{index}."
|
||
item_answers = {
|
||
key.removeprefix(prefix): value
|
||
for key, value in answers.items()
|
||
if key.startswith(prefix)
|
||
}
|
||
if include_legacy:
|
||
for key, value in answers.items():
|
||
if key == "followup" or key.startswith("items."):
|
||
continue
|
||
item_answers.setdefault(key, value)
|
||
return item_answers
|
||
|
||
|
||
def _field_is_missing(register_type: str, field: str, fields: dict[str, str]) -> bool:
|
||
value = str(fields.get(field) or "").strip()
|
||
if not value:
|
||
return True
|
||
if field == "date":
|
||
return not _normalize_date(value)
|
||
if register_type == "class_record" and field == "time":
|
||
return not _normalize_time_range(value)
|
||
if register_type == "payment" and field == "hours":
|
||
hours = re.sub(r"\s*(?:课时|小时)$", "", value)
|
||
try:
|
||
return float(hours) <= 0
|
||
except ValueError:
|
||
return True
|
||
return False
|
||
|
||
|
||
def _missing_fields(register_type: str, fields: dict[str, str]) -> list[str]:
|
||
return [field for field in REQUIRED_FIELDS[register_type] if _field_is_missing(register_type, field, fields)]
|
||
|
||
|
||
def _needs_info_response(
|
||
register_type: str,
|
||
fields: dict[str, str],
|
||
missing_fields: list[str],
|
||
*,
|
||
summary: str = "",
|
||
timed_out: bool = False,
|
||
error: str = "",
|
||
draft_line: str = "",
|
||
draft_items: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"status": "needs_info",
|
||
"standard_lines": [],
|
||
"questions": _questions_for_missing(register_type, missing_fields) or [_question_for_error(register_type, error or "信息不完整")],
|
||
"summary": summary or "请补齐缺失字段后再次生成预览",
|
||
"ai_used": False,
|
||
"fields": fields,
|
||
"missing_fields": missing_fields,
|
||
"field_labels": _field_labels(register_type),
|
||
"timed_out": timed_out,
|
||
"recognition_source": "partial_local",
|
||
"error": error,
|
||
"draft_line": draft_line,
|
||
"draft_items": draft_items or [],
|
||
}
|
||
|
||
|
||
def _standard_from_fields(register_type: str, fields: dict[str, str]) -> str:
|
||
if register_type == "class_record":
|
||
date_value = _normalize_date(fields.get("date"))
|
||
time_range = _normalize_time_range(fields.get("time"))
|
||
if not date_value or not time_range:
|
||
raise ValueError("上课记录日期或时间缺失")
|
||
duration = _duration_from_text(fields.get("duration")) or _duration_from_time_range(time_range)
|
||
return (
|
||
f"{_record_date(date_value)}-{_weekday(date_value)}-{time_range}-"
|
||
f"{fields['student']}-{duration}-{fields['teacher']}-{fields['subject']}"
|
||
)
|
||
if register_type == "payment":
|
||
date_value = _normalize_date(fields.get("date"))
|
||
if not date_value:
|
||
raise ValueError("缴费日期缺失")
|
||
hours = str(fields.get("hours") or "").strip()
|
||
hours = re.sub(r"\s*(?:课时|小时)$", "", hours)
|
||
return f"{fields['student']}-{date_value}:{hours}"
|
||
date_value = _normalize_date(fields.get("date"))
|
||
time_range = _normalize_time_range(fields.get("time"))
|
||
return "\n".join(
|
||
[
|
||
f"学生：{fields['student']}",
|
||
f"日期：{date_value}",
|
||
f"时间：{time_range}",
|
||
f"老师：{fields['teacher']}",
|
||
f"科目：{fields['subject']}",
|
||
"小结：",
|
||
str(fields.get("body") or "").strip(),
|
||
]
|
||
)
|
||
|
||
|
||
def _draft_from_fields(register_type: str, fields: dict[str, str]) -> str:
|
||
if register_type == "class_record":
|
||
date_iso = _normalize_date(fields.get("date"))
|
||
time_range = _normalize_time_range(fields.get("time"))
|
||
student = fields.get("student") or WAITING_PLACEHOLDER
|
||
date_value = _record_date(date_iso) if date_iso else WAITING_PLACEHOLDER
|
||
weekday = _weekday(date_iso) if date_iso else WAITING_PLACEHOLDER
|
||
teacher = _normalize_teacher_hint(fields.get("teacher")) or WAITING_PLACEHOLDER
|
||
duration = _duration_from_text(fields.get("duration")) or (
|
||
_duration_from_time_range(time_range) if time_range else WAITING_PLACEHOLDER
|
||
)
|
||
subject = fields.get("subject") or WAITING_PLACEHOLDER
|
||
return f"{date_value}-{weekday}-{time_range or WAITING_PLACEHOLDER}-{student}-{duration}-{teacher}-{subject}"
|
||
if register_type == "payment":
|
||
student = fields.get("student") or WAITING_PLACEHOLDER
|
||
date_value = _normalize_date(fields.get("date")) or WAITING_PLACEHOLDER
|
||
hours = str(fields.get("hours") or "").strip() or WAITING_PLACEHOLDER
|
||
return f"{student}-{date_value}:{hours}"
|
||
student = fields.get("student") or "【学生】"
|
||
date_value = _normalize_date(fields.get("date")) or "【日期】"
|
||
time_range = _normalize_time_range(fields.get("time")) or "【时间】"
|
||
teacher = _normalize_teacher_hint(fields.get("teacher")) or "【老师】"
|
||
subject = fields.get("subject") or "【科目】"
|
||
body = str(fields.get("body") or "").strip() or "【小结正文】"
|
||
return "\n".join(
|
||
[
|
||
f"学生：{student}",
|
||
f"日期：{date_value}",
|
||
f"时间：{time_range}",
|
||
f"老师：{teacher}",
|
||
f"科目：{subject}",
|
||
"小结：",
|
||
body,
|
||
]
|
||
)
|
||
|
||
|
||
def _question_for_error(register_type: str, message: str) -> str:
|
||
if register_type == "class_record":
|
||
return f"请按标准格式补齐或修改上课记录：{message}"
|
||
if register_type == "payment":
|
||
return f"请按“学生-YYYY-MM-DD:课时”补齐或修改缴费记录：{message}"
|
||
return f"请补齐或修改课程小结信息：{message}"
|
||
|
||
|
||
def _extract_class_record_fields(text: str) -> dict[str, str]:
|
||
teacher = _extract_teacher(text)
|
||
subject = _extract_subject(text)
|
||
return {
|
||
"student": _extract_class_record_student(text, teacher, subject),
|
||
"date": _normalize_date(text),
|
||
"time": _normalize_time_range(text),
|
||
"teacher": teacher,
|
||
"subject": subject,
|
||
"duration": _duration_from_text(text),
|
||
}
|
||
|
||
|
||
def _extract_payment_fields(text: str) -> dict[str, str]:
|
||
hours_match = HOURS_RE.search(text)
|
||
return {
|
||
"student": _extract_student(text),
|
||
"date": _normalize_date(text),
|
||
"hours": hours_match.group("hours") if hours_match else "",
|
||
}
|
||
|
||
|
||
def _extract_course_summary_fields(text: str) -> dict[str, str]:
|
||
try:
|
||
raw = extract_course_summary_from_text(text, known_students=_known_students())
|
||
except ValueError:
|
||
raw = {}
|
||
body = str(raw.get("body") or "").strip()
|
||
if body == text.strip() and any(label in text for label in ("小结", "正文", "内容")):
|
||
body = re.split(r"(?:小结|正文|内容)[:：]?", text, maxsplit=1)[-1].strip()
|
||
return {
|
||
"student": str(raw.get("student") or _extract_student(text)).strip(),
|
||
"date": _normalize_date(raw.get("date_iso") or text),
|
||
"time": _normalize_time_range(raw.get("time_range") or text),
|
||
"teacher": _normalize_teacher_hint(raw.get("teacher") or _extract_teacher(text)),
|
||
"subject": str(raw.get("subject") or _extract_subject(text)).strip(),
|
||
"body": body or text.strip(),
|
||
}
|
||
|
||
|
||
def extract_local_fields(register_type: str, lines: list[str]) -> dict[str, str]:
|
||
text = "\n".join(lines).strip()
|
||
if register_type == "class_record":
|
||
return _extract_class_record_fields(text)
|
||
if register_type == "payment":
|
||
return _extract_payment_fields(text)
|
||
return _extract_course_summary_fields(text)
|
||
|
||
|
||
def extract_line_fields(register_type: str, line: str) -> dict[str, str]:
|
||
if register_type == "class_record":
|
||
return _extract_class_record_fields(line)
|
||
if register_type == "payment":
|
||
return _extract_payment_fields(line)
|
||
return _extract_course_summary_fields(line)
|
||
|
||
|
||
def multi_line_fields_preview(register_type: str, lines: list[str], answers: dict[str, str]) -> dict[str, Any]:
|
||
draft_items: list[dict[str, Any]] = []
|
||
standard_lines: list[str] = []
|
||
all_ready = True
|
||
questions: list[str] = []
|
||
union_missing: list[str] = []
|
||
|
||
for index, line in enumerate(lines):
|
||
fields = extract_line_fields(register_type, line)
|
||
fields = _merge_answers(fields, _answers_for_item(answers, index, include_legacy=len(lines) == 1))
|
||
missing = _missing_fields(register_type, fields)
|
||
error = ""
|
||
standard_line = ""
|
||
if missing:
|
||
all_ready = False
|
||
else:
|
||
try:
|
||
standard_line = validate_standard_lines(register_type, [_standard_from_fields(register_type, fields)])[0]
|
||
standard_lines.append(standard_line)
|
||
except (ValueError, KeyError) as exc:
|
||
all_ready = False
|
||
error = str(exc)
|
||
|
||
for field in missing:
|
||
if field not in union_missing:
|
||
union_missing.append(field)
|
||
labels = _field_labels(register_type)
|
||
questions.extend(f"第 {index + 1} 条请补充{labels.get(field, field)}" for field in missing)
|
||
if error and not missing:
|
||
questions.append(f"第 {index + 1} 条{_question_for_error(register_type, error)}")
|
||
|
||
draft_items.append(
|
||
{
|
||
"index": index,
|
||
"source_line": line,
|
||
"fields": fields,
|
||
"missing_fields": missing,
|
||
"questions": _questions_for_missing(register_type, missing),
|
||
"draft_line": standard_line or _draft_from_fields(register_type, fields),
|
||
"error": error,
|
||
}
|
||
)
|
||
|
||
if all_ready:
|
||
return {
|
||
"status": "ready",
|
||
"standard_lines": standard_lines,
|
||
"questions": [],
|
||
"summary": "已通过本地规则生成预览",
|
||
"ai_used": False,
|
||
"fields": {},
|
||
"missing_fields": [],
|
||
"field_labels": _field_labels(register_type),
|
||
"timed_out": False,
|
||
"recognition_source": "local",
|
||
"draft_items": [],
|
||
}
|
||
|
||
return _needs_info_response(
|
||
register_type,
|
||
draft_items[0]["fields"] if draft_items else {},
|
||
union_missing,
|
||
summary="请逐条补齐缺失字段后再次生成预览",
|
||
error="",
|
||
draft_line="\n".join(str(item.get("draft_line") or "") for item in draft_items),
|
||
draft_items=draft_items,
|
||
)
|
||
|
||
|
||
def fields_preview(register_type: str, fields: dict[str, str]) -> dict[str, Any] | None:
|
||
missing = _missing_fields(register_type, fields)
|
||
if missing:
|
||
return _needs_info_response(register_type, fields, missing, draft_line=_draft_from_fields(register_type, fields))
|
||
try:
|
||
standard_lines = validate_standard_lines(register_type, [_standard_from_fields(register_type, fields)])
|
||
except (ValueError, KeyError) as exc:
|
||
return _needs_info_response(
|
||
register_type,
|
||
fields,
|
||
[],
|
||
summary="脚本无法生成可校验预览，请修改字段后再试",
|
||
error=str(exc),
|
||
draft_line=_draft_from_fields(register_type, fields),
|
||
)
|
||
return {
|
||
"status": "ready",
|
||
"standard_lines": standard_lines,
|
||
"questions": [],
|
||
"summary": "已通过本地规则生成预览",
|
||
"ai_used": False,
|
||
"fields": fields,
|
||
"missing_fields": [],
|
||
"field_labels": _field_labels(register_type),
|
||
"timed_out": False,
|
||
"recognition_source": "local",
|
||
}
|
||
|
||
|
||
def standard_lines_preview(register_type: str, standard_lines: list[str], summary: str, *, ai_used: bool, source: str) -> dict[str, Any]:
|
||
return {
|
||
"status": "ready",
|
||
"standard_lines": standard_lines,
|
||
"questions": [],
|
||
"summary": summary,
|
||
"ai_used": ai_used,
|
||
"fields": {},
|
||
"missing_fields": [],
|
||
"field_labels": _field_labels(register_type),
|
||
"timed_out": False,
|
||
"recognition_source": source,
|
||
}
|
||
|
||
|
||
def validate_standard_lines(register_type: str, lines: list[str]) -> list[str]:
|
||
if not lines:
|
||
raise ValueError("标准行不能为空")
|
||
standard_lines: list[str] = []
|
||
for index, line in enumerate(lines):
|
||
text = str(line or "").strip()
|
||
if not text:
|
||
continue
|
||
try:
|
||
if register_type == "class_record":
|
||
standard_lines.append(class_record_to_line(parse_class_record_line(text)))
|
||
elif register_type == "payment":
|
||
text = re.sub(r":\s*(\d+(?:\.\d+)?)\s*(?:课时|小时)\s*$", r":\1", text)
|
||
student, payment = parse_payment_line(text)
|
||
hours = int(payment.hours) if float(payment.hours).is_integer() else payment.hours
|
||
standard_lines.append(f"{student}-{payment.date}:{hours}")
|
||
else:
|
||
raw = extract_course_summary_from_text(text, index)
|
||
normalized = normalize_course_summary(raw)
|
||
course_summary_to_class_record_line(normalized)
|
||
standard_lines.append(text)
|
||
except ValueError as exc:
|
||
raise ValueError(f"第 {index + 1} 条校验失败：{exc}") from exc
|
||
if not standard_lines:
|
||
raise ValueError("标准行不能为空")
|
||
return standard_lines
|
||
|
||
|
||
def local_standard_preview(register_type: str, lines: list[str]) -> dict[str, Any] | None:
|
||
try:
|
||
standard_lines = validate_standard_lines(register_type, lines)
|
||
except ValueError:
|
||
return None
|
||
return standard_lines_preview(register_type, standard_lines, "已按标准格式通过本地校验", ai_used=False, source="local")
|
||
|
||
|
||
def preview_register(
|
||
*,
|
||
register_type: str,
|
||
text: str | None = None,
|
||
lines: list[str] | None = None,
|
||
answers: dict[str, str] | None = None,
|
||
conversation_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
normalized_type = normalize_register_type(register_type)
|
||
source_lines = collect_input_lines(text=text, lines=lines, register_type=normalized_type)
|
||
joined_text = "\n\n".join(source_lines)
|
||
conversation_id, session = _session_for(conversation_id, normalized_type, joined_text)
|
||
merged_answers = {**dict(session.get("answers") or {}), **(answers or {})}
|
||
session["answers"] = merged_answers
|
||
|
||
local = local_standard_preview(normalized_type, source_lines)
|
||
if local is not None:
|
||
return {"conversation_id": conversation_id, "type": normalized_type, **local}
|
||
|
||
if normalized_type in {"class_record", "payment"}:
|
||
return {
|
||
"conversation_id": conversation_id,
|
||
"type": normalized_type,
|
||
**multi_line_fields_preview(normalized_type, source_lines, merged_answers),
|
||
}
|
||
|
||
local_fields = _merge_answers(extract_local_fields(normalized_type, source_lines), merged_answers)
|
||
local_ready = fields_preview(normalized_type, local_fields)
|
||
if local_ready is not None:
|
||
return {"conversation_id": conversation_id, "type": normalized_type, **local_ready}
|
||
missing = _missing_fields(normalized_type, local_fields)
|
||
return {
|
||
"conversation_id": conversation_id,
|
||
"type": normalized_type,
|
||
**_needs_info_response(
|
||
normalized_type,
|
||
local_fields,
|
||
missing,
|
||
summary="脚本无法补齐全部字段，请补充信息后再次生成预览",
|
||
draft_line=_draft_from_fields(normalized_type, local_fields),
|
||
),
|
||
}
|