修复课程小结失败回滚和日志迁移副作用

app/app/data.py

@@ -1398,6 +1398,24 @@ def sha1_text(value: str, length: int = 16) -> str:
     return hashlib.sha1(value.encode("utf-8")).hexdigest()[:length]
 
 
+def snapshot_text_files(paths: Iterable[Path]) -> dict[Path, str | None]:
+    snapshots: dict[Path, str | None] = {}
+    for path in paths:
+        if path in snapshots:
+            continue
+        snapshots[path] = path.read_text(encoding="utf-8") if path.exists() else None
+    return snapshots
+
+
+def restore_text_file_snapshots(snapshots: dict[Path, str | None]) -> None:
+    for path, original_text in snapshots.items():
+        if original_text is None:
+            if path.exists():
+                path.unlink()
+            continue
+        atomic_write_text(path, original_text)
+
+
 def payload_bool(value: object) -> bool:
     if isinstance(value, bool):
         return value
@@ -3739,6 +3757,24 @@ def approve_admin_task(
     raise ValueError("不支持的审核任务类型")
 
 
+COURSE_SUMMARY_RESULT_COUNTER_KEYS = ("saved", "auto_registered", "review_pending", "duplicates", "rejected")
+
+
+def snapshot_course_summary_result(result: dict) -> dict:
+    return {
+        **{key: result[key] for key in COURSE_SUMMARY_RESULT_COUNTER_KEYS},
+        "operation_log_ids": list(result["operation_log_ids"]),
+        "items": list(result["items"]),
+    }
+
+
+def restore_course_summary_result(result: dict, snapshot: dict) -> None:
+    for key in COURSE_SUMMARY_RESULT_COUNTER_KEYS:
+        result[key] = snapshot[key]
+    result["operation_log_ids"] = list(snapshot["operation_log_ids"])
+    result["items"] = list(snapshot["items"])
+
+
 def register_course_summary_texts(
     *,
     classnotes_path: Path,
@@ -3767,14 +3803,25 @@ def register_course_summary_texts(
     for index, text in enumerate(texts):
         raw: dict | None = None
         normalized: dict | None = None
+        file_snapshots: dict[Path, str | None] | None = None
+        result_snapshot = snapshot_course_summary_result(result)
         try:
             raw = extract_course_summary_from_text(text, index, known_students=known_students)
             normalized = normalize_course_summary(raw)
             source_id = normalized["source_id"]
             semantic_key = course_summary_semantic_key(normalized)
+            summary_path = course_summary_path(summaries_root, normalized)
             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", []))
+            file_snapshots = snapshot_text_files([
+                classnotes_path,
+                accounts_path,
+                tasks_path,
+                state_path,
+                operation_logs_path,
+                summary_path,
+            ])
             if source_id in seen_source_ids or semantic_key in seen_semantic_keys:
                 duplicate_reasons, duplicate_conflicts, source_id_duplicate, semantic_duplicate = course_summary_duplicate_review_context(
                     summaries_root,
@@ -3794,7 +3841,7 @@ def register_course_summary_texts(
                     normalized,
                     proposed_line,
                     review_reasons,
-                    str(course_summary_path(summaries_root, normalized)),
+                    str(summary_path),
                     extra_fields={
                         "duplicate_source": summary_as_duplicate_candidate(normalized),
                         "duplicate_reasons": duplicate_reasons,
@@ -3838,7 +3885,7 @@ def register_course_summary_texts(
                     proposed_line=proposed_line,
                     reasons=review_reasons,
                     task_id=str(task.get("id") or ""),
-                    saved_path=str(course_summary_path(summaries_root, normalized)),
+                    saved_path=str(summary_path),
                     duplicate_source_id=source_id_duplicate,
                     duplicate_semantic_key=semantic_duplicate,
                 )
@@ -4027,7 +4074,12 @@ def register_course_summary_texts(
                     "backup_id": str(register_result.get("backup_id") or ""),
                 }
             )
-        except ValueError as exc:
+        except Exception as exc:
+            if file_snapshots is not None:
+                restore_text_file_snapshots(file_snapshots)
+            restore_course_summary_result(result, result_snapshot)
+            if not isinstance(exc, ValueError):
+                raise
             source_id = str((normalized or raw or {}).get("source_id") or f"manual:{sha1_text(text, 24)}")
             student = str((normalized or raw or {}).get("student") or "")
             log_id = append_operation_log(
@@ -4080,6 +4132,10 @@ def ingest_course_summaries(
     for raw in summaries:
         normalized: dict | None = None
         backup_id = ""
+        file_snapshots: dict[Path, str | None] | None = None
+        result_snapshot = snapshot_course_summary_result(result)
+        seen_source_ids_before = set(seen_source_ids)
+        seen_semantic_keys_before = set(seen_semantic_keys)
         try:
             normalized = normalize_course_summary(raw)
             ai_questions = [
@@ -4089,6 +4145,15 @@ def ingest_course_summaries(
             ]
             source_id = normalized["source_id"]
             semantic_key = course_summary_semantic_key(normalized)
+            summary_path = course_summary_path(summaries_root, normalized)
+            file_snapshots = snapshot_text_files([
+                classnotes_path,
+                accounts_path,
+                tasks_path,
+                state_path,
+                operation_logs_path,
+                summary_path,
+            ])
             if source_id in seen_source_ids or semantic_key in seen_semantic_keys:
                 duplicate_reasons, duplicate_conflicts, source_id_duplicate, semantic_duplicate = course_summary_duplicate_review_context(
                     summaries_root,
@@ -4108,7 +4173,7 @@ def ingest_course_summaries(
                     normalized,
                     proposed_line,
                     review_reasons,
-                    str(course_summary_path(summaries_root, normalized)),
+                    str(summary_path),
                     extra_fields={
                         "duplicate_source": summary_as_duplicate_candidate(normalized),
                         "duplicate_reasons": duplicate_reasons,
@@ -4135,7 +4200,7 @@ def ingest_course_summaries(
                     proposed_line=proposed_line,
                     reasons=review_reasons,
                     task_id=task.get("id"),
-                    saved_path=str(course_summary_path(summaries_root, normalized)),
+                    saved_path=str(summary_path),
                     duplicate_source_id=source_id_duplicate,
                     duplicate_semantic_key=semantic_duplicate,
                 )
@@ -4242,6 +4307,11 @@ def ingest_course_summaries(
                 }
             )
         except Exception as exc:
+            if file_snapshots is not None:
+                restore_text_file_snapshots(file_snapshots)
+            seen_source_ids = seen_source_ids_before
+            seen_semantic_keys = seen_semantic_keys_before
+            restore_course_summary_result(result, result_snapshot)
             ai_questions = [
                 str(item).strip()
                 for item in ((raw if isinstance(raw, dict) else {}).get("ai_questions") or [])

app/app/main.py

@@ -3,7 +3,8 @@ from __future__ import annotations
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
 
-from .config import USE_SQLITE_SOURCE
+from .config import OPERATION_LOGS_PATH, USE_SQLITE_SOURCE, write_lock
+from .data import migrate_operation_log_labels
 from .repository import ensure_runtime_cache
 from .routers import accounts, admin, ai_register, health, ingest, pages, records
 
@@ -15,6 +16,8 @@ app = FastAPI(title="新时空教务管理系统", version="1.0.0")
 def prepare_sqlite_runtime_cache() -> None:
     if USE_SQLITE_SOURCE:
         ensure_runtime_cache()
+    with write_lock:
+        migrate_operation_log_labels(OPERATION_LOGS_PATH)
 
 
 @app.exception_handler(ValueError)

app/app/routers/admin.py

@@ -24,7 +24,6 @@ from ..data import (
     link_existing_course_summary_task,
     list_admin_tasks,
     list_operation_logs,
-    migrate_operation_log_labels,
     query_course_summaries,
     reject_admin_task,
     resolve_duplicate_course_summary_task,
@@ -95,7 +94,6 @@ def admin_operation_logs(
     student: str = Query(""),
     _user: str = Depends(verify_admin_auth),
 ):
-    migrate_operation_log_labels(OPERATION_LOGS_PATH)
     return list_operation_logs(
         OPERATION_LOGS_PATH,
         limit=limit,