Codex
324 lines
12 KiB
Python
324 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import re
|
||
import shutil
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from app.data import ( # noqa: E402
|
||
append_operation_log,
|
||
course_summary_to_class_record_line,
|
||
course_summary_semantic_key,
|
||
create_course_summary_review_task,
|
||
normalize_course_summary,
|
||
read_admin_tasks,
|
||
read_course_summary_state,
|
||
safe_filename_part,
|
||
sha1_text,
|
||
write_course_summary_state,
|
||
)
|
||
|
||
|
||
EXCLUDE_MARKDOWN = {
|
||
"微信聊天记录ID映射表.md",
|
||
"采集记录表.md",
|
||
"课程记录核对表.md",
|
||
}
|
||
EXCLUDE_PREFIXES = (
|
||
"已采集小结记录",
|
||
"classnotes缺失",
|
||
"课程小结缺失",
|
||
"疑似 classnotes 出错",
|
||
"疑似课程小结待复核",
|
||
"正文日期晚于发送日期待复核",
|
||
)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="导入历史课程小结 Markdown 到 VPS 数据目录")
|
||
parser.add_argument("--source", required=True, type=Path, help="本机历史课程小结采集目录")
|
||
parser.add_argument("--target", default=Path("/data/course_summaries"), type=Path, help="VPS 课程小结正式目录")
|
||
parser.add_argument("--state", default=Path("/data/course_summary_state.json"), type=Path, help="课程小结状态文件")
|
||
parser.add_argument("--tasks", default=Path("/data/admin_tasks.json"), type=Path, help="管理任务文件")
|
||
parser.add_argument("--operation-logs", default=Path("/data/operation_logs.jsonl"), type=Path, help="操作日志文件")
|
||
parser.add_argument("--missing-table", type=Path, help="历史 classnotes缺失.txt；不传则尝试 source/classnotes缺失.txt")
|
||
parser.add_argument("--dry-run", action="store_true", help="只统计，不写入")
|
||
return parser.parse_args()
|
||
|
||
|
||
def should_import_markdown(path: Path, source: Path) -> bool:
|
||
if path.name in EXCLUDE_MARKDOWN or any(path.name.startswith(prefix) for prefix in EXCLUDE_PREFIXES):
|
||
return False
|
||
if any(part in {"__pycache__", "临时解析归档", "推送归档", "推送失败队列"} for part in path.relative_to(source).parts):
|
||
return False
|
||
return path.suffix.lower() == ".md"
|
||
|
||
|
||
def split_name_from_path(path: Path) -> tuple[str, str, str]:
|
||
parts = path.stem.split("_")
|
||
student = path.parent.name if path.parent.name else (parts[0] if parts else "")
|
||
teacher = parts[1] if len(parts) >= 2 else ""
|
||
subject = parts[2] if len(parts) >= 3 else ""
|
||
return student, teacher, subject
|
||
|
||
|
||
def iter_markdown_entries(path: Path) -> list[dict]:
|
||
text = path.read_text(encoding="utf-8", errors="ignore")
|
||
student, teacher, subject = split_name_from_path(path)
|
||
group = student
|
||
current_title = ""
|
||
current_body: list[str] = []
|
||
entries: list[dict] = []
|
||
|
||
def flush() -> None:
|
||
if not current_title:
|
||
return
|
||
body = "\n".join(current_body).strip()
|
||
date_match = re.search(r"(\d{4}[.-]\d{2}[.-]\d{2})", current_title)
|
||
if not date_match or not body:
|
||
return
|
||
time_match = re.search(r"(\d{1,2}:\d{2}-\d{1,2}:\d{2})", current_title)
|
||
source_seed = f"{path}|{current_title}|{body[:120]}"
|
||
entries.append(
|
||
{
|
||
"source_id": f"history:{sha1_text(source_seed, 24)}",
|
||
"student": student,
|
||
"date_iso": date_match.group(1).replace(".", "-"),
|
||
"time_range": time_match.group(1) if time_match else "",
|
||
"teacher": teacher,
|
||
"subject": subject,
|
||
"group": group,
|
||
"title": current_title,
|
||
"body": body,
|
||
"recognition_source": "history_import",
|
||
"confidence": "history",
|
||
"teacher_trusted": False,
|
||
"remark": f"历史导入：{path}",
|
||
}
|
||
)
|
||
|
||
for line in text.splitlines():
|
||
if line.startswith("## ") and not line.startswith("### "):
|
||
group = line.removeprefix("## ").strip() or group
|
||
continue
|
||
if line.startswith("### "):
|
||
flush()
|
||
current_title = line.removeprefix("### ").strip()
|
||
current_body = []
|
||
continue
|
||
if current_title:
|
||
if line.startswith("> 来源ID") or line.startswith("> 发送时间") or line.startswith("> 发送者"):
|
||
continue
|
||
current_body.append(line)
|
||
flush()
|
||
return entries
|
||
|
||
|
||
def copy_markdown_files(source: Path, target: Path, dry_run: bool) -> tuple[int, int]:
|
||
copied = 0
|
||
scanned = 0
|
||
for path in sorted(source.rglob("*.md")):
|
||
if not should_import_markdown(path, source):
|
||
continue
|
||
scanned += 1
|
||
relative = path.relative_to(source)
|
||
dest = target / relative
|
||
if dest.exists() and dest.read_text(encoding="utf-8", errors="ignore") == path.read_text(encoding="utf-8", errors="ignore"):
|
||
continue
|
||
copied += 1
|
||
if not dry_run:
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(path, dest)
|
||
return scanned, copied
|
||
|
||
|
||
def rebuild_state_from_markdown(source: Path, state_path: Path, dry_run: bool) -> tuple[int, int]:
|
||
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", []))
|
||
imported = 0
|
||
skipped = 0
|
||
for path in sorted(source.rglob("*.md")):
|
||
if not should_import_markdown(path, source):
|
||
continue
|
||
for entry in iter_markdown_entries(path):
|
||
try:
|
||
summary = normalize_course_summary(entry)
|
||
seen_source_ids.add(summary["source_id"])
|
||
seen_semantic_keys.add(course_summary_semantic_key(summary))
|
||
imported += 1
|
||
except ValueError:
|
||
skipped += 1
|
||
state["seen_source_ids"] = sorted(seen_source_ids)
|
||
state["seen_semantic_keys"] = sorted(seen_semantic_keys)
|
||
state.setdefault("batches", []).append(
|
||
{
|
||
"batch_id": f"history-import-{datetime.now().strftime('%Y%m%d%H%M%S')}",
|
||
"received_at": datetime.now().isoformat(timespec="seconds"),
|
||
"window": {"mode": "历史导入"},
|
||
"students": [],
|
||
"result": {"received": imported, "saved": 0, "auto_registered": 0, "review_pending": 0, "duplicates": 0, "rejected": skipped},
|
||
}
|
||
)
|
||
state["batches"] = state["batches"][-200:]
|
||
if not dry_run:
|
||
write_course_summary_state(state_path, state)
|
||
return imported, skipped
|
||
|
||
|
||
def parse_missing_table(path: Path) -> list[dict]:
|
||
if not path.exists():
|
||
return []
|
||
rows: list[dict] = []
|
||
headers: list[str] = []
|
||
for raw_line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
||
line = raw_line.strip()
|
||
if not line.startswith("| ") or "---" in line:
|
||
continue
|
||
cells = [cell.strip() for cell in line.strip("|").split("|")]
|
||
if cells and cells[0] == "日期":
|
||
headers = cells
|
||
continue
|
||
if not headers or len(cells) != len(headers):
|
||
continue
|
||
row = dict(zip(headers, cells, strict=False))
|
||
if row.get("学生"):
|
||
rows.append(row)
|
||
return rows
|
||
|
||
|
||
def time_range_from_text(text: str) -> str:
|
||
match = re.search(r"(\d{1,2})[:：](\d{2})\s*[-－–—]\s*(\d{1,2})[:：](\d{2})", text)
|
||
if not match:
|
||
return ""
|
||
start_hour, start_minute, end_hour, end_minute = map(int, match.groups())
|
||
start = start_hour * 60 + start_minute
|
||
end = end_hour * 60 + end_minute
|
||
if end <= start:
|
||
return ""
|
||
return f"{start_hour:02d}:{start_minute:02d}-{end_hour:02d}:{end_minute:02d}"
|
||
|
||
|
||
def duration_from_time_range(time_range: str) -> tuple[str, int | None]:
|
||
match = re.fullmatch(r"(\d{2}):(\d{2})-(\d{2}):(\d{2})", time_range)
|
||
if not match:
|
||
return "", None
|
||
start_hour, start_minute, end_hour, end_minute = map(int, match.groups())
|
||
minutes = end_hour * 60 + end_minute - (start_hour * 60 + start_minute)
|
||
if minutes <= 0:
|
||
return "", None
|
||
return f"{minutes // 60}小时{minutes % 60}分", minutes
|
||
|
||
|
||
def source_summary_for_missing_row(source: Path, row: dict) -> dict:
|
||
relative_path = str(row.get("来源文件") or "").strip()
|
||
if not relative_path:
|
||
return {}
|
||
source_path = source / relative_path
|
||
if not source_path.exists():
|
||
return {}
|
||
|
||
target_date = str(row.get("日期") or "").replace(".", "-")
|
||
best: dict = {}
|
||
for entry in iter_markdown_entries(source_path):
|
||
if entry.get("date_iso") != target_date:
|
||
continue
|
||
time_range = str(entry.get("time_range") or "") or time_range_from_text(str(entry.get("body") or ""))
|
||
duration, minutes = duration_from_time_range(time_range)
|
||
candidate = {
|
||
"title": entry.get("title", ""),
|
||
"body": entry.get("body", ""),
|
||
"time_range": time_range,
|
||
"duration": duration,
|
||
"duration_minutes": minutes,
|
||
}
|
||
if time_range:
|
||
return candidate
|
||
if not best:
|
||
best = candidate
|
||
return best
|
||
|
||
|
||
def import_missing_tasks(source: Path, missing_table: Path, tasks_path: Path, dry_run: bool) -> int:
|
||
rows = parse_missing_table(missing_table)
|
||
if not rows:
|
||
return 0
|
||
existing = read_admin_tasks(tasks_path)
|
||
existing_source_ids = {str(item.get("source_id") or "") for item in existing.get("items", [])}
|
||
created = 0
|
||
for row in rows:
|
||
source_seed = "|".join(str(row.get(key, "")) for key in ("日期", "学生", "老师", "科目", "来源文件", "备注"))
|
||
source_id = f"history-missing:{sha1_text(source_seed, 20)}"
|
||
if source_id in existing_source_ids:
|
||
continue
|
||
summary = {
|
||
"source_id": source_id,
|
||
"student": row.get("学生", ""),
|
||
"date_iso": str(row.get("日期", "")).replace(".", "-"),
|
||
"time_range": row.get("时间段", ""),
|
||
"duration": row.get("时长", ""),
|
||
"teacher": row.get("老师", ""),
|
||
"subject": row.get("科目", ""),
|
||
"group": row.get("群聊", ""),
|
||
"title": "历史 classnotes 缺失",
|
||
"body": f"历史 classnotes 缺失项：{source_seed}",
|
||
"recognition_source": "history_missing_table",
|
||
"confidence": "review",
|
||
"teacher_trusted": False,
|
||
"remark": row.get("备注", ""),
|
||
}
|
||
summary.update({key: value for key, value in source_summary_for_missing_row(source, row).items() if value not in ("", None)})
|
||
proposed_line = ""
|
||
try:
|
||
proposed_line = course_summary_to_class_record_line(summary)
|
||
except ValueError:
|
||
pass
|
||
created += 1
|
||
if not dry_run:
|
||
create_course_summary_review_task(
|
||
tasks_path,
|
||
summary,
|
||
proposed_line,
|
||
["历史 classnotes缺失导入，默认只进入审核，不自动扣课时"],
|
||
saved_path=row.get("来源文件", ""),
|
||
)
|
||
existing_source_ids.add(source_id)
|
||
return created
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
source = args.source.resolve()
|
||
target = args.target.resolve()
|
||
missing_table = args.missing_table or (source / "classnotes缺失.txt")
|
||
scanned, copied = copy_markdown_files(source, target, args.dry_run)
|
||
imported, skipped = rebuild_state_from_markdown(target if target.exists() else source, args.state, args.dry_run)
|
||
review_tasks = import_missing_tasks(source, missing_table, args.tasks, args.dry_run)
|
||
if not args.dry_run:
|
||
append_operation_log(
|
||
args.operation_logs,
|
||
"history_course_summary_import",
|
||
"completed",
|
||
source=str(source),
|
||
target=str(target),
|
||
scanned_files=scanned,
|
||
copied_files=copied,
|
||
indexed_summaries=imported,
|
||
skipped_summaries=skipped,
|
||
review_tasks=review_tasks,
|
||
)
|
||
print(
|
||
f"历史小结导入完成：扫描 Markdown {scanned} 个，复制 {copied} 个，"
|
||
f"索引小结 {imported} 条，跳过 {skipped} 条，生成审核任务 {review_tasks} 条。"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|