From 70a5af3020d2221025452250e5918988f11fe083 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 24 Jun 2026 17:27:50 +0800 Subject: [PATCH] Migrate education data to SQLite --- Makefile | 8 +- app/.env.example | 5 + app/Makefile | 10 +- app/README.md | 40 +- app/app/api_utils.py | 9 +- app/app/config.py | 64 ++- app/app/db.py | 195 ++++++++ app/app/main.py | 8 + app/app/repository.py | 633 ++++++++++++++++++++++++++ app/app/routers/accounts.py | 10 +- app/app/routers/health.py | 8 + app/docker-compose.yml | 4 + app/scripts/migrate_text_to_sqlite.py | 232 ++++++++++ 13 files changed, 1203 insertions(+), 23 deletions(-) create mode 100644 app/app/db.py create mode 100644 app/app/repository.py create mode 100644 app/scripts/migrate_text_to_sqlite.py diff --git a/Makefile b/Makefile index 88d0994..cc53e5b 100644 --- a/Makefile +++ b/Makefile @@ -4,7 +4,7 @@ APP_DIR := app APP_MAKE := $(MAKE) -C $(APP_DIR) --no-print-directory COMPOSE := docker compose -f $(APP_DIR)/docker-compose.yml -.PHONY: check smoke data-hash build up ps health deploy logs install-gitea-backup compose-config +.PHONY: check smoke data-hash migrate-sqlite-dry-run migrate-sqlite build up ps health deploy logs install-gitea-backup compose-config check: $(APP_MAKE) check @@ -15,6 +15,12 @@ smoke: data-hash: $(APP_MAKE) data-hash +migrate-sqlite-dry-run: + $(APP_MAKE) migrate-sqlite-dry-run + +migrate-sqlite: + $(APP_MAKE) migrate-sqlite + build: $(APP_MAKE) build diff --git a/app/.env.example b/app/.env.example index 46fa570..fd76662 100644 --- a/app/.env.example +++ b/app/.env.example @@ -8,6 +8,11 @@ BASIC_AUTH_PASSWORD=change-me ADMIN_AUTH_PASSWORD=change-me INGEST_AUTH_TOKEN=change-this-ingest-token +USE_SQLITE_SOURCE=1 +SQLITE_DB_PATH=/data/xsk_education.db +RUNTIME_DATA_ROOT=/data/runtime_text_cache +LEGACY_TEXT_ROOT=/data + CLASSNOTES_PATH=/data/classnotes.txt ACCOUNTS_PATH=/data/学生课时账户.md TEACHERS_PATH=/data/教师档案.md diff --git a/app/Makefile b/app/Makefile index f8ac2e9..df59718 100644 --- a/app/Makefile +++ b/app/Makefile @@ -1,9 +1,9 @@ SHELL := /bin/bash APP_PORT ?= 18080 -DATA_FILES := ../data/classnotes.txt ../data/学生课时账户.md $(wildcard ../data/admin_tasks.json) $(wildcard ../data/course_summary_state.json) +DATA_FILES := $(wildcard ../data/xsk_education.db) $(wildcard ../data/sqlite_migration_report_*.json) $(wildcard ../archives/text-source-before-sqlite-*.tar.gz.sha256) -.PHONY: check smoke data-hash build up ps health deploy logs install-gitea-backup +.PHONY: check smoke data-hash migrate-sqlite-dry-run migrate-sqlite build up ps health deploy logs install-gitea-backup check: node --check app/static/app.js @@ -16,6 +16,12 @@ smoke: data-hash: sha256sum $(DATA_FILES) +migrate-sqlite-dry-run: + python3 scripts/migrate_text_to_sqlite.py --data-root ../data --db-path ../data/xsk_education.db --archives-root ../archives --dry-run + +migrate-sqlite: + python3 scripts/migrate_text_to_sqlite.py --data-root ../data --db-path ../data/xsk_education.db --archives-root ../archives + build: docker compose build diff --git a/app/README.md b/app/README.md index bc49045..6a8a394 100644 --- a/app/README.md +++ b/app/README.md @@ -1,6 +1,6 @@ # 新时空教务管理系统 -这是一个面向新时空教务业务的综合管理系统。迁移后 VPS 是正式业务数据主机，负责保存 `classnotes.txt`、`学生课时账户.md`、课程小结库、审核任务和操作记录；本机只保留微信群聊天记录采集/识别，并把课程小结批量推送到 VPS。 +这是一个面向新时空教务业务的综合管理系统。迁移后 VPS 是正式业务数据主机，负责保存 SQLite 数据库 `/data/xsk_education.db`；本机只保留微信群聊天记录采集/识别，并把课程小结批量推送到 VPS。 ## 目录 @@ -8,6 +8,7 @@ - `MAINTENANCE.md`：日常检查、部署、数据保护和回滚流程。 - `Makefile`：常用维护命令入口。 - `scripts/deploy_to_vps.py`：部署新时空教务管理系统到 VPS。 +- `scripts/migrate_text_to_sqlite.py`：一次性把旧纯文本、JSON、JSONL 和课程小结 Markdown 迁移进 SQLite。 - `scripts/import_course_summaries.py`：一次性导入历史课程小结 Markdown。 - `scripts/sync_to_vps.py`：旧版正式数据同步脚本，迁移后不要继续常驻运行。 - `scripts/smoke_test.js`：轻量前端行为烟测。 @@ -71,7 +72,7 @@ XSK_PYTHON_IMAGE='python:3.12-slim' ## 手动同步数据 -迁移完成后不要再用本机 `classnotes.txt` 和 `学生课时账户.md` 覆盖 VPS。下面命令只保留给迁移前或灾难恢复时使用，日常新增课程小结应走 `POST /api/ingest/course-summaries`。 +迁移完成后不要再用本机 `classnotes.txt` 和 `学生课时账户.md` 覆盖 VPS。SQLite 数据库 `/data/xsk_education.db` 是唯一事实源；运行时生成的 `/data/runtime_text_cache/` 只是兼容旧业务逻辑的缓存。下面命令只保留给迁移前或灾难恢复时使用，日常新增课程小结应走 `POST /api/ingest/course-summaries`。 ```bash XSK_USE_SSHPASS=1 \ @@ -87,7 +88,31 @@ python3 scripts/sync_to_vps.py --once --use-sshpass 远端数据目录： ```text -/root/新时空教务管理系统/data/ +/root/新时空教务管理系统/data/xsk_education.db +``` + +## SQLite 迁移 + +首次迁移先 dry-run： + +```bash +cd /root/新时空教务管理系统 +make migrate-sqlite-dry-run +``` + +确认严格校验通过后执行正式迁移： + +```bash +make migrate-sqlite +``` + +正式迁移会生成： + +```text +/root/新时空教务管理系统/data/xsk_education.db +/root/新时空教务管理系统/data/sqlite_migration_report_<时间>.json +/root/新时空教务管理系统/archives/text-source-before-sqlite-<时间>.tar.gz +/root/新时空教务管理系统/archives/text-source-before-sqlite-<时间>.tar.gz.sha256 ``` ## 课程小结推送 @@ -131,15 +156,10 @@ docker compose exec xsk-education-management python scripts/import_course_summar ## 数据备份 -通过登记 API、课程小结自动入账或审核批准修改正式课时数据时，服务会在写入前自动备份本次会改动的业务文件。正式数据和辅助状态位于： +通过登记 API、课程小结自动入账或审核批准修改正式课时数据时，服务会以 SQLite 事务写入 `/data/xsk_education.db`。旧纯文本事实源已封存为归档包；运行时文本缓存位于： ```text -/root/新时空教务管理系统/data/classnotes.txt -/root/新时空教务管理系统/data/学生课时账户.md -/root/新时空教务管理系统/data/教师档案.md -/root/新时空教务管理系统/data/course_summaries/ -/root/新时空教务管理系统/data/course_summary_state.json -/root/新时空教务管理系统/data/operation_logs.jsonl +/root/新时空教务管理系统/data/runtime_text_cache/ ``` 备份目录位于： diff --git a/app/app/api_utils.py b/app/app/api_utils.py index daa1c82..574be15 100644 --- a/app/app/api_utils.py +++ b/app/app/api_utils.py @@ -6,8 +6,9 @@ from pathlib import Path from fastapi import HTTPException, Request from pydantic import ValidationError -from .config import ACCOUNTS_PATH, CLASSNOTES_PATH, TEACHERS_PATH +from .config import ACCOUNTS_PATH, CLASSNOTES_PATH, TEACHERS_PATH, USE_SQLITE_SOURCE from .data import Account, Payment, Teacher, read_accounts, read_classnotes, read_teachers +from .repository import ensure_runtime_cache from .schemas import AccountPayload, RegisterLinesPayload, TeacherPayload @@ -38,18 +39,24 @@ async def read_register_payload(request: Request) -> RegisterLinesPayload: def load_records(): + if USE_SQLITE_SOURCE: + ensure_runtime_cache() if not CLASSNOTES_PATH.exists(): raise HTTPException(status_code=503, detail=f"课程记录文件不存在: {CLASSNOTES_PATH}") return read_classnotes(CLASSNOTES_PATH) def load_accounts(): + if USE_SQLITE_SOURCE: + ensure_runtime_cache() if not ACCOUNTS_PATH.exists(): raise HTTPException(status_code=503, detail=f"课时账户文件不存在: {ACCOUNTS_PATH}") return read_accounts(ACCOUNTS_PATH) def load_teachers(): + if USE_SQLITE_SOURCE: + ensure_runtime_cache() if not TEACHERS_PATH.exists(): return [] return read_teachers(TEACHERS_PATH) diff --git a/app/app/config.py b/app/app/config.py index ddd266f..34c0b7f 100644 --- a/app/app/config.py +++ b/app/app/config.py @@ -8,13 +8,35 @@ import threading APP_DIR = Path(__file__).resolve().parent STATIC_DIR = APP_DIR / "static" -CLASSNOTES_PATH = Path(os.getenv("CLASSNOTES_PATH", "/data/classnotes.txt")) -ACCOUNTS_PATH = Path(os.getenv("ACCOUNTS_PATH", "/data/学生课时账户.md")) -TEACHERS_PATH = Path(os.getenv("TEACHERS_PATH", "/data/教师档案.md")) -ADMIN_TASKS_PATH = Path(os.getenv("ADMIN_TASKS_PATH", "/data/admin_tasks.json")) -COURSE_SUMMARIES_ROOT = Path(os.getenv("COURSE_SUMMARIES_ROOT", "/data/course_summaries")) -COURSE_SUMMARY_STATE_PATH = Path(os.getenv("COURSE_SUMMARY_STATE_PATH", "/data/course_summary_state.json")) -OPERATION_LOGS_PATH = Path(os.getenv("OPERATION_LOGS_PATH", "/data/operation_logs.jsonl")) +SQLITE_DB_PATH = Path(os.getenv("SQLITE_DB_PATH", "/data/xsk_education.db")) +RUNTIME_DATA_ROOT = Path(os.getenv("RUNTIME_DATA_ROOT", "/data/runtime_text_cache")) +LEGACY_TEXT_ROOT = Path(os.getenv("LEGACY_TEXT_ROOT", "/data")) +USE_SQLITE_SOURCE = os.getenv("USE_SQLITE_SOURCE", "1").strip().lower() not in {"0", "false", "no"} + +if USE_SQLITE_SOURCE: + CLASSNOTES_PATH = RUNTIME_DATA_ROOT / "classnotes.txt" + ACCOUNTS_PATH = RUNTIME_DATA_ROOT / "学生课时账户.md" + TEACHERS_PATH = RUNTIME_DATA_ROOT / "教师档案.md" + ADMIN_TASKS_PATH = RUNTIME_DATA_ROOT / "admin_tasks.json" + COURSE_SUMMARIES_ROOT = RUNTIME_DATA_ROOT / "course_summaries" + COURSE_SUMMARY_STATE_PATH = RUNTIME_DATA_ROOT / "course_summary_state.json" + OPERATION_LOGS_PATH = RUNTIME_DATA_ROOT / "operation_logs.jsonl" +else: + CLASSNOTES_PATH = Path(os.getenv("CLASSNOTES_PATH", "/data/classnotes.txt")) + ACCOUNTS_PATH = Path(os.getenv("ACCOUNTS_PATH", "/data/学生课时账户.md")) + TEACHERS_PATH = Path(os.getenv("TEACHERS_PATH", "/data/教师档案.md")) + ADMIN_TASKS_PATH = Path(os.getenv("ADMIN_TASKS_PATH", "/data/admin_tasks.json")) + COURSE_SUMMARIES_ROOT = Path(os.getenv("COURSE_SUMMARIES_ROOT", "/data/course_summaries")) + COURSE_SUMMARY_STATE_PATH = Path(os.getenv("COURSE_SUMMARY_STATE_PATH", "/data/course_summary_state.json")) + OPERATION_LOGS_PATH = Path(os.getenv("OPERATION_LOGS_PATH", "/data/operation_logs.jsonl")) + +LEGACY_CLASSNOTES_PATH = Path(os.getenv("LEGACY_CLASSNOTES_PATH", str(LEGACY_TEXT_ROOT / "classnotes.txt"))) +LEGACY_ACCOUNTS_PATH = Path(os.getenv("LEGACY_ACCOUNTS_PATH", str(LEGACY_TEXT_ROOT / "学生课时账户.md"))) +LEGACY_TEACHERS_PATH = Path(os.getenv("LEGACY_TEACHERS_PATH", str(LEGACY_TEXT_ROOT / "教师档案.md"))) +LEGACY_ADMIN_TASKS_PATH = Path(os.getenv("LEGACY_ADMIN_TASKS_PATH", str(LEGACY_TEXT_ROOT / "admin_tasks.json"))) +LEGACY_COURSE_SUMMARIES_ROOT = Path(os.getenv("LEGACY_COURSE_SUMMARIES_ROOT", str(LEGACY_TEXT_ROOT / "course_summaries"))) +LEGACY_COURSE_SUMMARY_STATE_PATH = Path(os.getenv("LEGACY_COURSE_SUMMARY_STATE_PATH", str(LEGACY_TEXT_ROOT / "course_summary_state.json"))) +LEGACY_OPERATION_LOGS_PATH = Path(os.getenv("LEGACY_OPERATION_LOGS_PATH", str(LEGACY_TEXT_ROOT / "operation_logs.jsonl"))) BASIC_AUTH_PASSWORD = os.getenv("BASIC_AUTH_PASSWORD", "") ACCOUNTS_AUTH_PASSWORD = os.getenv("ACCOUNTS_AUTH_PASSWORD") or os.getenv("ACCOUNT_AUTH_PASSWORD", "") @@ -26,4 +48,30 @@ ACCOUNTS_SESSION_COOKIE = "xsk_accounts_session" ADMIN_SESSION_COOKIE = "xsk_admin_session" SESSION_MAX_AGE = 60 * 60 * 24 * 30 -write_lock = threading.Lock() +class SQLiteBackedWriteLock: + def __init__(self) -> None: + self._lock = threading.Lock() + + def __enter__(self): + self._lock.acquire() + if USE_SQLITE_SOURCE: + from .repository import export_database_to_runtime_cache + + export_database_to_runtime_cache() + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + try: + if USE_SQLITE_SOURCE: + from .repository import export_database_to_runtime_cache, sync_runtime_cache_to_database + + if exc_type is None: + sync_runtime_cache_to_database() + else: + export_database_to_runtime_cache() + finally: + self._lock.release() + return False + + +write_lock = SQLiteBackedWriteLock() if USE_SQLITE_SOURCE else threading.Lock() diff --git a/app/app/db.py b/app/app/db.py new file mode 100644 index 0000000..3d3528c --- /dev/null +++ b/app/app/db.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +from contextlib import contextmanager +import sqlite3 +from pathlib import Path +from typing import Iterator + +from .config import SQLITE_DB_PATH + + +SCHEMA_VERSION = 1 + + +def connect(db_path: Path | None = None) -> sqlite3.Connection: + path = db_path or SQLITE_DB_PATH + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("PRAGMA busy_timeout = 5000") + conn.execute("PRAGMA journal_mode = WAL") + return conn + + +@contextmanager +def transaction(db_path: Path | None = None) -> Iterator[sqlite3.Connection]: + conn = connect(db_path) + try: + conn.execute("BEGIN IMMEDIATE") + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +def initialize_schema(conn: sqlite3.Connection) -> None: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS metadata ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS students ( + student_id TEXT PRIMARY KEY, + student TEXT NOT NULL UNIQUE, + account_status TEXT NOT NULL, + note TEXT NOT NULL DEFAULT '', + source_remaining REAL NOT NULL DEFAULT 0, + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS account_transactions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + student_id TEXT NOT NULL REFERENCES students(student_id) ON DELETE CASCADE, + tx_date TEXT NOT NULL, + hours REAL NOT NULL, + tx_type TEXT NOT NULL, + source TEXT NOT NULL DEFAULT '', + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS class_records ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + record_key TEXT NOT NULL UNIQUE, + record_date TEXT NOT NULL, + weekday TEXT NOT NULL, + time_range TEXT NOT NULL, + duration_minutes INTEGER NOT NULL, + student_id TEXT REFERENCES students(student_id) ON DELETE SET NULL, + student TEXT NOT NULL, + teacher TEXT NOT NULL, + subject TEXT NOT NULL, + raw_line TEXT NOT NULL, + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS teachers ( + teacher_id TEXT PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + alias TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + note TEXT NOT NULL DEFAULT '', + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS teacher_subjects ( + teacher_id TEXT NOT NULL REFERENCES teachers(teacher_id) ON DELETE CASCADE, + subject TEXT NOT NULL, + sort_order INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (teacher_id, subject) + ); + + CREATE TABLE IF NOT EXISTS admin_tasks ( + id INTEGER PRIMARY KEY, + task_type TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT '', + updated_at TEXT NOT NULL DEFAULT '', + student TEXT NOT NULL DEFAULT '', + source_id TEXT NOT NULL DEFAULT '', + payload_json TEXT NOT NULL, + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS admin_task_state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS operation_logs ( + log_id TEXT PRIMARY KEY, + created_at TEXT NOT NULL DEFAULT '', + operation TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT '', + student TEXT NOT NULL DEFAULT '', + payload_json TEXT NOT NULL, + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS course_summaries ( + id TEXT PRIMARY KEY, + relative_path TEXT NOT NULL, + title TEXT NOT NULL, + body TEXT NOT NULL, + raw_body TEXT NOT NULL DEFAULT '', + group_name TEXT NOT NULL DEFAULT '', + student TEXT NOT NULL DEFAULT '', + teacher TEXT NOT NULL DEFAULT '', + subject TEXT NOT NULL DEFAULT '', + date_iso TEXT NOT NULL DEFAULT '', + time_range TEXT NOT NULL DEFAULT '', + message_time TEXT NOT NULL DEFAULT '', + sender TEXT NOT NULL DEFAULT '', + source_id TEXT NOT NULL DEFAULT '', + semantic_key TEXT NOT NULL DEFAULT '', + content_hash TEXT NOT NULL, + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS course_summary_seen_keys ( + kind TEXT NOT NULL, + value TEXT NOT NULL, + PRIMARY KEY (kind, value) + ); + + CREATE TABLE IF NOT EXISTS ingest_batches ( + batch_id TEXT PRIMARY KEY, + received_at TEXT NOT NULL DEFAULT '', + payload_json TEXT NOT NULL, + sort_order INTEGER NOT NULL DEFAULT 0 + ); + + CREATE TABLE IF NOT EXISTS migration_audit ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL, + kind TEXT NOT NULL, + payload_json TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_class_records_student_date ON class_records(student, record_date); + CREATE INDEX IF NOT EXISTS idx_class_records_teacher ON class_records(teacher); + CREATE INDEX IF NOT EXISTS idx_course_summaries_identity ON course_summaries(student, teacher, subject, date_iso, time_range); + CREATE INDEX IF NOT EXISTS idx_operation_logs_created_at ON operation_logs(created_at); + """ + ) + conn.execute( + "INSERT OR REPLACE INTO metadata(key, value) VALUES('schema_version', ?)", + (str(SCHEMA_VERSION),), + ) + + +def database_meta(db_path: Path | None = None) -> dict: + path = db_path or SQLITE_DB_PATH + if not path.exists(): + return {"exists": False, "path": str(path)} + stat = path.stat() + version = "" + try: + with connect(path) as conn: + initialize_schema(conn) + row = conn.execute("SELECT value FROM metadata WHERE key = 'schema_version'").fetchone() + version = str(row["value"]) if row else "" + except sqlite3.DatabaseError: + version = "unreadable" + return { + "exists": True, + "path": str(path), + "size": stat.st_size, + "mtime": stat.st_mtime, + "schema_version": version, + } diff --git a/app/app/main.py b/app/app/main.py index d1c1604..1978404 100644 --- a/app/app/main.py +++ b/app/app/main.py @@ -3,12 +3,20 @@ from __future__ import annotations from fastapi import FastAPI, Request from fastapi.responses import JSONResponse +from .config import USE_SQLITE_SOURCE +from .repository import ensure_runtime_cache from .routers import accounts, admin, ai_register, health, ingest, pages, records app = FastAPI(title="新时空教务管理系统", version="1.0.0") +@app.on_event("startup") +def prepare_sqlite_runtime_cache() -> None: + if USE_SQLITE_SOURCE: + ensure_runtime_cache() + + @app.exception_handler(ValueError) async def value_error_handler(_request: Request, exc: ValueError): return JSONResponse(status_code=500, content={"detail": str(exc)}) diff --git a/app/app/repository.py b/app/app/repository.py new file mode 100644 index 0000000..20d5db4 --- /dev/null +++ b/app/app/repository.py @@ -0,0 +1,633 @@ +from __future__ import annotations + +from collections import defaultdict +from datetime import date, datetime +import hashlib +import json +import os +from pathlib import Path +import shutil +import sqlite3 +from typing import Iterable + +from .config import ( + ACCOUNTS_PATH, + ADMIN_TASKS_PATH, + CLASSNOTES_PATH, + COURSE_SUMMARIES_ROOT, + COURSE_SUMMARY_STATE_PATH, + OPERATION_LOGS_PATH, + RUNTIME_DATA_ROOT, + SQLITE_DB_PATH, + TEACHERS_PATH, +) +from .db import connect, initialize_schema, transaction +from .domain import Account, ClassRecord, Payment, Teacher +from .storage import atomic_write_text + + +CLASSNOTES_HEADER = """# 课程记录 +# 格式：日期-星期-时间段-学生姓名-时长-老师姓名-科目 +# 示例：2025.09.14-星期日-16:00-18:00-余峻珲-2小时0分-杨达伟-数学 + +""" + +ACCOUNTS_HEADER = """# 学生课时账户 + +说明： + +- 本表由 SQLite 运行时缓存生成，正式事实源为 `/data/xsk_education.db`。 +- `剩余课时` 由缴费/退费流水与上课记录实时重算。 + +| 学生ID | 学生姓名 | 缴费记录 | 剩余课时 | 账户状态 | 备注 | +|---|---|---|---|---|---| +""" + +TEACHERS_HEADER = """# 教师档案 + +| 教师ID | 教师姓名 | 别名 | 任教学科 | 状态 | 备注 | +| ------ | -------- | ---- | -------- | ---- | ---- | +""" + + +def sha1_text(text: str, length: int = 40) -> str: + return hashlib.sha1(text.encode("utf-8")).hexdigest()[:length] + + +def _json_dumps(value: object) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True) + + +def _record_key(record: ClassRecord) -> str: + return "|".join( + [ + record.student, + record.date, + record.time, + record.duration, + record.teacher, + record.subject, + ] + ) + + +def _class_record_to_line(record: ClassRecord) -> str: + return f"{record.date}-{record.weekday}-{record.time}-{record.student}-{record.duration}-{record.teacher}-{record.subject}" + + +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 _account_status(stored_status: str, remaining: float) -> str: + if stored_status in {"结课", "退费"}: + return stored_status + if remaining < 0: + return "欠费" + if remaining < 10: + return "预警" + return "正常" + + +def _duration_text_from_minutes(minutes: int) -> str: + return f"{minutes // 60}小时{minutes % 60}分" + + +def _parse_course_summary_blocks(root: Path) -> list[dict]: + from .data import iter_course_summary_markdown + + if not root.exists(): + return [] + return list(iter_course_summary_markdown(root)) + + +def _read_sources( + *, + classnotes_path: Path, + accounts_path: Path, + teachers_path: Path, + tasks_path: Path, + summaries_root: Path, + state_path: Path, + operation_logs_path: Path, +) -> dict: + from .data import ( + read_accounts, + read_admin_tasks, + read_classnotes, + read_course_summary_state, + read_operation_log_rows, + read_teachers, + ) + + return { + "records": read_classnotes(classnotes_path) if classnotes_path.exists() else [], + "accounts": read_accounts(accounts_path) if accounts_path.exists() else [], + "teachers": read_teachers(teachers_path) if teachers_path.exists() else [], + "tasks": read_admin_tasks(tasks_path) if tasks_path.exists() else {"version": 1, "next_id": 1, "items": []}, + "summaries": _parse_course_summary_blocks(summaries_root), + "state": read_course_summary_state(state_path) if state_path.exists() else {"version": 1, "seen_source_ids": [], "seen_semantic_keys": [], "batches": []}, + "logs": read_operation_log_rows(operation_logs_path) if operation_logs_path.exists() else [], + } + + +def _clear_business_tables(conn: sqlite3.Connection) -> None: + for table in [ + "teacher_subjects", + "account_transactions", + "class_records", + "students", + "teachers", + "admin_tasks", + "admin_task_state", + "operation_logs", + "course_summaries", + "course_summary_seen_keys", + "ingest_batches", + ]: + conn.execute(f"DELETE FROM {table}") + + +def _insert_sources( + conn: sqlite3.Connection, + sources: dict, + *, + allow_balance_adjustments: bool, +) -> dict: + accounts: list[Account] = sources["accounts"] + records: list[ClassRecord] = sources["records"] + teachers: list[Teacher] = sources["teachers"] + tasks: dict = sources["tasks"] + summaries: list[dict] = sources["summaries"] + state: dict = sources["state"] + logs: list[dict] = sources["logs"] + + account_by_student = {account.student: account for account in accounts} + for index, account in enumerate(accounts): + conn.execute( + """ + INSERT INTO students(student_id, student, account_status, note, source_remaining, sort_order) + VALUES(?, ?, ?, ?, ?, ?) + """, + (account.student_id, account.student, account.account_status, account.note, account.remaining, index), + ) + for payment_index, payment in enumerate(account.payments): + conn.execute( + """ + INSERT INTO account_transactions(student_id, tx_date, hours, tx_type, source, sort_order) + VALUES(?, ?, ?, ?, ?, ?) + """, + ( + account.student_id, + payment.date, + float(payment.hours), + "payment" if payment.hours >= 0 else "refund", + "account_text", + payment_index, + ), + ) + + seen_record_keys: set[str] = set() + used_hours_by_student: defaultdict[str, float] = defaultdict(float) + for index, record in enumerate(records): + key = _record_key(record) + if key in seen_record_keys: + raise ValueError(f"课程记录重复: {_class_record_to_line(record)}") + seen_record_keys.add(key) + account = account_by_student.get(record.student) + if account is None: + raise ValueError(f"上课学生没有课时账户: {record.student}") + minutes = int(round(record.duration_hours * 60)) + used_hours_by_student[record.student] = round(used_hours_by_student[record.student] + record.duration_hours, 2) + conn.execute( + """ + INSERT INTO class_records(record_key, record_date, weekday, time_range, duration_minutes, + student_id, student, teacher, subject, raw_line, sort_order) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + key, + record.date, + record.weekday, + record.time, + minutes, + account.student_id, + record.student, + record.teacher, + record.subject, + _class_record_to_line(record), + index, + ), + ) + + balance_mismatches: list[dict] = [] + if allow_balance_adjustments: + for account in accounts: + paid = round(sum(payment.hours for payment in account.payments), 2) + used = round(used_hours_by_student.get(account.student, 0.0), 2) + derived = round(paid - used, 2) + delta = round(account.remaining - derived, 2) + if abs(delta) <= 0.011: + continue + conn.execute( + """ + INSERT INTO account_transactions(student_id, tx_date, hours, tx_type, source, sort_order) + VALUES(?, ?, ?, ?, ?, ?) + """, + (account.student_id, date.today().isoformat(), delta, "manual_adjustment", "runtime_account_remaining", 100000), + ) + else: + for account in accounts: + paid = round(sum(payment.hours for payment in account.payments), 2) + used = round(used_hours_by_student.get(account.student, 0.0), 2) + derived = round(paid - used, 2) + delta = round(derived - account.remaining, 2) + if abs(delta) > 0.011: + balance_mismatches.append( + { + "student": account.student, + "student_id": account.student_id, + "old_remaining": account.remaining, + "recomputed_remaining": derived, + "difference": delta, + "paid_hours": paid, + "used_hours": used, + "status": "resolved_by_recompute", + } + ) + + for index, teacher in enumerate(teachers): + conn.execute( + """ + INSERT INTO teachers(teacher_id, name, alias, status, note, sort_order) + VALUES(?, ?, ?, ?, ?, ?) + """, + (teacher.teacher_id, teacher.name, teacher.alias, teacher.status, teacher.note, index), + ) + for subject_index, subject in enumerate(teacher.subjects): + conn.execute( + "INSERT INTO teacher_subjects(teacher_id, subject, sort_order) VALUES(?, ?, ?)", + (teacher.teacher_id, subject, subject_index), + ) + + conn.execute("INSERT INTO admin_task_state(key, value) VALUES('version', ?)", (str(tasks.get("version", 1)),)) + conn.execute("INSERT INTO admin_task_state(key, value) VALUES('next_id', ?)", (str(tasks.get("next_id", 1)),)) + for index, task in enumerate(tasks.get("items") or []): + conn.execute( + """ + INSERT INTO admin_tasks(id, task_type, status, created_at, updated_at, student, source_id, payload_json, sort_order) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + int(task.get("id") or index + 1), + str(task.get("type") or ""), + str(task.get("status") or ""), + str(task.get("created_at") or ""), + str(task.get("updated_at") or ""), + str(task.get("student") or task.get("corrected", {}).get("student") or ""), + str(task.get("source_id") or ""), + json.dumps(task, ensure_ascii=False, sort_keys=True), + index, + ), + ) + + for index, item in enumerate(logs): + log_id = str(item.get("id") or f"legacy-log-{index + 1}") + conn.execute( + """ + INSERT OR REPLACE INTO operation_logs(log_id, created_at, operation, status, student, payload_json, sort_order) + VALUES(?, ?, ?, ?, ?, ?, ?) + """, + ( + log_id, + str(item.get("created_at") or ""), + str(item.get("operation") or ""), + str(item.get("status") or ""), + str(item.get("student") or ""), + json.dumps({**item, "id": log_id}, ensure_ascii=False, sort_keys=True), + index, + ), + ) + + for index, item in enumerate(summaries): + item_id = str(item.get("id") or sha1_text(f"{item.get('relative_path')}|{item.get('title')}|{index}", 20)) + body = str(item.get("body") or "") + raw_body = str(item.get("raw_body") or body) + content_hash = sha1_text(f"{item.get('relative_path')}|{item.get('title')}|{body}") + conn.execute( + """ + INSERT OR REPLACE INTO course_summaries(id, relative_path, title, body, raw_body, group_name, + student, teacher, subject, date_iso, time_range, + message_time, sender, source_id, semantic_key, content_hash, sort_order) + VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + item_id, + str(item.get("relative_path") or ""), + str(item.get("title") or ""), + body, + raw_body, + str(item.get("group") or ""), + str(item.get("student") or ""), + str(item.get("teacher") or ""), + str(item.get("subject") or ""), + str(item.get("date_iso") or ""), + str(item.get("time_range") or ""), + str(item.get("message_time") or ""), + str(item.get("sender") or ""), + str(item.get("source_id") or ""), + str(item.get("semantic_key") or ""), + content_hash, + index, + ), + ) + + for value in state.get("seen_source_ids") or []: + conn.execute("INSERT OR IGNORE INTO course_summary_seen_keys(kind, value) VALUES('source_id', ?)", (str(value),)) + for value in state.get("seen_semantic_keys") or []: + conn.execute("INSERT OR IGNORE INTO course_summary_seen_keys(kind, value) VALUES('semantic_key', ?)", (str(value),)) + for index, batch in enumerate(state.get("batches") or []): + batch_id = str(batch.get("batch_id") or f"legacy-batch-{index + 1}") + conn.execute( + """ + INSERT OR REPLACE INTO ingest_batches(batch_id, received_at, payload_json, sort_order) + VALUES(?, ?, ?, ?) + """, + (batch_id, str(batch.get("received_at") or ""), json.dumps({**batch, "batch_id": batch_id}, ensure_ascii=False, sort_keys=True), index), + ) + + return { + "records": len(records), + "accounts": len(accounts), + "teachers": len(teachers), + "admin_tasks": len(tasks.get("items") or []), + "operation_logs": len(logs), + "course_summaries": len(summaries), + "balance_mismatches": balance_mismatches, + } + + +def replace_database_from_paths( + conn: sqlite3.Connection, + *, + classnotes_path: Path, + accounts_path: Path, + teachers_path: Path, + tasks_path: Path, + summaries_root: Path, + state_path: Path, + operation_logs_path: Path, + allow_balance_adjustments: bool, +) -> dict: + initialize_schema(conn) + sources = _read_sources( + classnotes_path=classnotes_path, + accounts_path=accounts_path, + teachers_path=teachers_path, + tasks_path=tasks_path, + summaries_root=summaries_root, + state_path=state_path, + operation_logs_path=operation_logs_path, + ) + _clear_business_tables(conn) + return _insert_sources(conn, sources, allow_balance_adjustments=allow_balance_adjustments) + + +def sync_runtime_cache_to_database() -> dict: + with transaction(SQLITE_DB_PATH) as conn: + return replace_database_from_paths( + conn, + classnotes_path=CLASSNOTES_PATH, + accounts_path=ACCOUNTS_PATH, + teachers_path=TEACHERS_PATH, + tasks_path=ADMIN_TASKS_PATH, + summaries_root=COURSE_SUMMARIES_ROOT, + state_path=COURSE_SUMMARY_STATE_PATH, + operation_logs_path=OPERATION_LOGS_PATH, + allow_balance_adjustments=True, + ) + + +def _class_records_from_db(conn: sqlite3.Connection) -> list[str]: + rows = conn.execute( + "SELECT raw_line FROM class_records ORDER BY sort_order, id" + ).fetchall() + return [str(row["raw_line"]) for row in rows] + + +def _payments_by_student(conn: sqlite3.Connection) -> dict[str, list[Payment]]: + rows = conn.execute( + """ + SELECT student_id, tx_date, hours + FROM account_transactions + ORDER BY student_id, sort_order, id + """ + ).fetchall() + result: dict[str, list[Payment]] = defaultdict(list) + for row in rows: + result[str(row["student_id"])].append(Payment(date=str(row["tx_date"]), hours=float(row["hours"]))) + return result + + +def _used_hours_by_student(conn: sqlite3.Connection) -> dict[str, float]: + rows = conn.execute( + """ + SELECT student_id, duration_minutes + FROM class_records + WHERE student_id IS NOT NULL + ORDER BY sort_order, id + """ + ).fetchall() + result: dict[str, float] = defaultdict(float) + for row in rows: + student_id = str(row["student_id"]) + duration_hours = round(int(row["duration_minutes"] or 0) / 60.0, 2) + result[student_id] = round(result[student_id] + duration_hours, 2) + return dict(result) + + +def _accounts_from_db(conn: sqlite3.Connection) -> list[Account]: + payment_map = _payments_by_student(conn) + used_map = _used_hours_by_student(conn) + accounts: list[Account] = [] + rows = conn.execute( + "SELECT student_id, student, account_status, note FROM students ORDER BY sort_order, student_id" + ).fetchall() + for row in rows: + student_id = str(row["student_id"]) + payments = payment_map.get(student_id, []) + paid = round(sum(payment.hours for payment in payments), 2) + remaining = round(paid - used_map.get(student_id, 0.0), 2) + status = _account_status(str(row["account_status"]), remaining) + accounts.append( + Account( + student_id=student_id, + student=str(row["student"]), + payments=payments, + remaining=remaining, + account_status=status, + note=str(row["note"] or ""), + ) + ) + return accounts + + +def _teachers_from_db(conn: sqlite3.Connection) -> list[Teacher]: + subject_rows = conn.execute( + "SELECT teacher_id, subject FROM teacher_subjects ORDER BY teacher_id, sort_order" + ).fetchall() + subjects: dict[str, list[str]] = defaultdict(list) + for row in subject_rows: + subjects[str(row["teacher_id"])].append(str(row["subject"])) + rows = conn.execute( + "SELECT teacher_id, name, alias, status, note FROM teachers ORDER BY sort_order, teacher_id" + ).fetchall() + return [ + Teacher( + teacher_id=str(row["teacher_id"]), + name=str(row["name"]), + alias=str(row["alias"] or ""), + subjects=subjects.get(str(row["teacher_id"]), []), + status=str(row["status"]), + note=str(row["note"] or ""), + ) + for row in rows + ] + + +def _tasks_from_db(conn: sqlite3.Connection) -> dict: + state_rows = conn.execute("SELECT key, value FROM admin_task_state").fetchall() + state = {str(row["key"]): str(row["value"]) for row in state_rows} + task_rows = conn.execute("SELECT payload_json FROM admin_tasks ORDER BY sort_order, id").fetchall() + return { + "version": int(state.get("version") or 1), + "next_id": int(state.get("next_id") or 1), + "items": [json.loads(str(row["payload_json"])) for row in task_rows], + } + + +def _operation_logs_from_db(conn: sqlite3.Connection) -> list[dict]: + rows = conn.execute("SELECT payload_json FROM operation_logs ORDER BY sort_order, created_at, log_id").fetchall() + return [json.loads(str(row["payload_json"])) for row in rows] + + +def _summary_state_from_db(conn: sqlite3.Connection) -> dict: + source_ids = [ + str(row["value"]) + for row in conn.execute("SELECT value FROM course_summary_seen_keys WHERE kind = 'source_id' ORDER BY value") + ] + semantic_keys = [ + str(row["value"]) + for row in conn.execute("SELECT value FROM course_summary_seen_keys WHERE kind = 'semantic_key' ORDER BY value") + ] + batches = [ + json.loads(str(row["payload_json"])) + for row in conn.execute("SELECT payload_json FROM ingest_batches ORDER BY sort_order, received_at, batch_id") + ] + return {"version": 1, "seen_source_ids": source_ids, "seen_semantic_keys": semantic_keys, "batches": batches[-200:]} + + +def _write_accounts(path: Path, accounts: list[Account]) -> None: + lines = [ACCOUNTS_HEADER.rstrip("\n")] + for account in accounts: + payments = ",".join(_format_payment(payment) for payment in account.payments) + lines.append( + f"| {account.student_id} | {account.student} | {payments} | " + f"{_format_number(account.remaining)} | {account.account_status} | {account.note} |" + ) + atomic_write_text(path, "\n".join(lines).rstrip() + "\n") + + +def _write_teachers(path: Path, teachers: list[Teacher]) -> None: + lines = [TEACHERS_HEADER.rstrip("\n")] + for teacher in teachers: + lines.append( + f"| {teacher.teacher_id} | {teacher.name} | {teacher.alias} | " + f"{'、'.join(teacher.subjects)} | {teacher.status} | {teacher.note} |" + ) + atomic_write_text(path, "\n".join(lines).rstrip() + "\n") + + +def _write_course_summaries(conn: sqlite3.Connection, root: Path) -> None: + if root.exists(): + shutil.rmtree(root) + root.mkdir(parents=True, exist_ok=True) + grouped: dict[str, list[sqlite3.Row]] = defaultdict(list) + for row in conn.execute("SELECT * FROM course_summaries ORDER BY relative_path, sort_order"): + relative_path = str(row["relative_path"] or "未归档/未命名.md") + grouped[relative_path].append(row) + for relative_path, rows in grouped.items(): + target = root / relative_path + target.parent.mkdir(parents=True, exist_ok=True) + first = rows[0] + heading = f"# {first['student']} {first['teacher']}".strip() + group = str(first["group_name"] or "") + blocks = [heading, ""] + if group: + blocks.extend([f"## {group}", ""]) + for row in rows: + blocks.append(f"### {row['title']}") + blocks.append("") + raw_body = str(row["raw_body"] or "").strip() + body = str(row["body"] or "").strip() + if raw_body: + blocks.append(raw_body) + else: + meta_lines = [] + if row["source_id"]: + meta_lines.append(f"> 来源ID：`{row['source_id']}`") + if row["message_time"]: + meta_lines.append(f"> 发送时间：`{row['message_time']}`") + if row["sender"]: + meta_lines.append(f"> 发送者：`{row['sender']}`") + if meta_lines: + blocks.extend(meta_lines) + blocks.append("") + blocks.append(body) + blocks.append("") + atomic_write_text(target, "\n".join(blocks).rstrip() + "\n") + + +def export_database_to_runtime_cache() -> dict: + if not SQLITE_DB_PATH.exists(): + return {"ok": False, "reason": "database_missing", "path": str(SQLITE_DB_PATH)} + RUNTIME_DATA_ROOT.mkdir(parents=True, exist_ok=True) + with connect(SQLITE_DB_PATH) as conn: + initialize_schema(conn) + class_lines = _class_records_from_db(conn) + atomic_write_text(CLASSNOTES_PATH, CLASSNOTES_HEADER + "\n".join(class_lines).rstrip() + ("\n" if class_lines else "")) + _write_accounts(ACCOUNTS_PATH, _accounts_from_db(conn)) + _write_teachers(TEACHERS_PATH, _teachers_from_db(conn)) + atomic_write_text(ADMIN_TASKS_PATH, json.dumps(_tasks_from_db(conn), ensure_ascii=False, indent=2) + "\n") + logs = _operation_logs_from_db(conn) + atomic_write_text( + OPERATION_LOGS_PATH, + "".join(json.dumps(item, ensure_ascii=False, sort_keys=True) + "\n" for item in logs), + ) + atomic_write_text(COURSE_SUMMARY_STATE_PATH, json.dumps(_summary_state_from_db(conn), ensure_ascii=False, indent=2) + "\n") + _write_course_summaries(conn, COURSE_SUMMARIES_ROOT) + return { + "ok": True, + "records": len(class_lines), + "accounts": len(_accounts_from_db(conn)), + "operation_logs": len(logs), + } + + +def ensure_runtime_cache() -> dict: + return export_database_to_runtime_cache() + + +def source_file_hashes(paths: Iterable[Path]) -> dict[str, str]: + hashes: dict[str, str] = {} + for path in paths: + if not path.exists() or not path.is_file(): + continue + hashes[str(path)] = hashlib.sha256(path.read_bytes()).hexdigest() + return hashes diff --git a/app/app/routers/accounts.py b/app/app/routers/accounts.py index 3eb8bb3..66fcf9a 100644 --- a/app/app/routers/accounts.py +++ b/app/app/routers/accounts.py @@ -12,10 +12,13 @@ from ..config import ( CLASSNOTES_PATH, COURSE_SUMMARIES_ROOT, COURSE_SUMMARY_STATE_PATH, + LEGACY_TEXT_ROOT, OPERATION_LOGS_PATH, TEACHERS_PATH, + USE_SQLITE_SOURCE, write_lock, ) +from ..db import database_meta from ..data import ( ACCOUNT_STATUSES, TEACHER_STATUSES, @@ -45,7 +48,7 @@ router = APIRouter() def compact_duration_text(hours: float) -> str: text = duration_text_from_hours(hours) - return text.removesuffix("0分") + return text[:-2] if text.endswith("小时0分") else text @router.post("/api/register/class-records") @@ -134,6 +137,11 @@ def account_health(_user: str = Depends(verify_accounts_auth)): "teachers": file_meta(TEACHERS_PATH), "classnotes": file_meta(CLASSNOTES_PATH), "course_summaries": file_meta(COURSE_SUMMARIES_ROOT), + "database": database_meta(), + "legacy_archive": { + "source_mode": "sqlite" if USE_SQLITE_SOURCE else "text", + "text_root": str(LEGACY_TEXT_ROOT), + }, "accounts_count": len(accounts), "teachers_count": len(teachers), "active_teachers_count": sum(1 for teacher in teachers if teacher.status == "在岗"), diff --git a/app/app/routers/health.py b/app/app/routers/health.py index 6c887dc..0e48e4d 100644 --- a/app/app/routers/health.py +++ b/app/app/routers/health.py @@ -9,9 +9,12 @@ from ..config import ( CLASSNOTES_PATH, COURSE_SUMMARIES_ROOT, COURSE_SUMMARY_STATE_PATH, + LEGACY_TEXT_ROOT, OPERATION_LOGS_PATH, TEACHERS_PATH, + USE_SQLITE_SOURCE, ) +from ..db import database_meta from ..data import account_summary @@ -31,6 +34,11 @@ def health(_user: str = Depends(verify_records_auth)): "course_summaries": file_meta(COURSE_SUMMARIES_ROOT), "course_summary_state": file_meta(COURSE_SUMMARY_STATE_PATH), "operation_logs": file_meta(OPERATION_LOGS_PATH), + "database": database_meta(), + "legacy_archive": { + "source_mode": "sqlite" if USE_SQLITE_SOURCE else "text", + "text_root": str(LEGACY_TEXT_ROOT), + }, "records_count": len(records), "accounts_count": len(accounts), "teachers_count": len(teachers), diff --git a/app/docker-compose.yml b/app/docker-compose.yml index c913c12..156265b 100644 --- a/app/docker-compose.yml +++ b/app/docker-compose.yml @@ -10,6 +10,10 @@ services: - .env environment: TZ: ${TZ:-Asia/Shanghai} + USE_SQLITE_SOURCE: ${USE_SQLITE_SOURCE:-1} + SQLITE_DB_PATH: ${SQLITE_DB_PATH:-/data/xsk_education.db} + RUNTIME_DATA_ROOT: ${RUNTIME_DATA_ROOT:-/data/runtime_text_cache} + LEGACY_TEXT_ROOT: ${LEGACY_TEXT_ROOT:-/data} CLASSNOTES_PATH: ${CLASSNOTES_PATH:-/data/classnotes.txt} ACCOUNTS_PATH: ${ACCOUNTS_PATH:-/data/学生课时账户.md} TEACHERS_PATH: ${TEACHERS_PATH:-/data/教师档案.md} diff --git a/app/scripts/migrate_text_to_sqlite.py b/app/scripts/migrate_text_to_sqlite.py new file mode 100644 index 0000000..1d48497 --- /dev/null +++ b/app/scripts/migrate_text_to_sqlite.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +import argparse +from datetime import datetime +import hashlib +import json +import os +from pathlib import Path +import shutil +import sqlite3 +import sys +import tarfile + + +APP_ROOT = Path(__file__).resolve().parents[1] +REPO_ROOT = APP_ROOT.parent +if str(APP_ROOT) not in sys.path: + sys.path.insert(0, str(APP_ROOT)) + +from app.db import SCHEMA_VERSION, connect, initialize_schema # noqa: E402 +from app.repository import replace_database_from_paths, source_file_hashes # noqa: E402 + + +EXPECTED_COUNTS = { + "records": 1548, + "accounts": 45, + "teachers": 13, +} + + +def sha256_path(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def data_paths(data_root: Path) -> dict[str, Path]: + return { + "classnotes_path": data_root / "classnotes.txt", + "accounts_path": data_root / "学生课时账户.md", + "teachers_path": data_root / "教师档案.md", + "tasks_path": data_root / "admin_tasks.json", + "summaries_root": data_root / "course_summaries", + "state_path": data_root / "course_summary_state.json", + "operation_logs_path": data_root / "operation_logs.jsonl", + } + + +def validate_counts(summary: dict, *, strict_expected: bool) -> list[str]: + errors: list[str] = [] + for key in ["records", "accounts", "teachers", "admin_tasks", "operation_logs", "course_summaries"]: + if int(summary.get(key) or 0) < 0: + errors.append(f"{key} 数量异常") + if strict_expected: + for key, expected in EXPECTED_COUNTS.items(): + actual = int(summary.get(key) or 0) + if actual != expected: + errors.append(f"{key} 数量应为 {expected}，实际为 {actual}") + mismatch_count = len(summary.get("balance_mismatches") or []) + if mismatch_count != 4: + errors.append(f"余额差异应为 4 个学生，实际为 {mismatch_count}") + return errors + + +def validate_database(db_path: Path, summary: dict, *, strict_expected: bool) -> dict: + errors = validate_counts(summary, strict_expected=strict_expected) + with connect(db_path) as conn: + table_counts = { + "records": conn.execute("SELECT COUNT(*) FROM class_records").fetchone()[0], + "accounts": conn.execute("SELECT COUNT(*) FROM students").fetchone()[0], + "teachers": conn.execute("SELECT COUNT(*) FROM teachers").fetchone()[0], + "admin_tasks": conn.execute("SELECT COUNT(*) FROM admin_tasks").fetchone()[0], + "operation_logs": conn.execute("SELECT COUNT(*) FROM operation_logs").fetchone()[0], + "course_summaries": conn.execute("SELECT COUNT(*) FROM course_summaries").fetchone()[0], + } + for key, value in table_counts.items(): + if int(summary.get(key) or 0) != int(value): + errors.append(f"SQLite {key} 数量不一致: source={summary.get(key)} sqlite={value}") + duplicate_records = conn.execute( + """ + SELECT record_key, COUNT(*) AS c + FROM class_records + GROUP BY record_key + HAVING c > 1 + LIMIT 1 + """ + ).fetchone() + if duplicate_records: + errors.append(f"SQLite 中存在重复课程记录: {duplicate_records['record_key']}") + missing_account = conn.execute( + """ + SELECT student + FROM class_records + WHERE student_id IS NULL + LIMIT 1 + """ + ).fetchone() + if missing_account: + errors.append(f"SQLite 中存在没有账户的上课学生: {missing_account['student']}") + return {"ok": not errors, "errors": errors, "table_counts": table_counts} + + +def write_audit(conn: sqlite3.Connection, payload: dict) -> None: + conn.execute( + "INSERT INTO migration_audit(created_at, kind, payload_json) VALUES(?, ?, ?)", + ( + datetime.now().isoformat(timespec="seconds"), + "text_to_sqlite", + json.dumps(payload, ensure_ascii=False, sort_keys=True), + ), + ) + + +def create_archive(data_root: Path, archives_root: Path, report_path: Path, timestamp: str) -> tuple[Path, Path]: + archives_root.mkdir(parents=True, exist_ok=True) + archive_path = archives_root / f"text-source-before-sqlite-{timestamp}.tar.gz" + with tarfile.open(archive_path, "w:gz") as archive: + for name in [ + "classnotes.txt", + "学生课时账户.md", + "教师档案.md", + "admin_tasks.json", + "operation_logs.jsonl", + "course_summary_state.json", + ]: + path = data_root / name + if path.exists(): + archive.add(path, arcname=name) + summaries_root = data_root / "course_summaries" + if summaries_root.exists(): + archive.add(summaries_root, arcname="course_summaries") + archive.add(report_path, arcname=report_path.name) + sha_path = archive_path.with_suffix(archive_path.suffix + ".sha256") + sha_path.write_text(f"{sha256_path(archive_path)} {archive_path.name}\n", encoding="utf-8") + return archive_path, sha_path + + +def migrate(args: argparse.Namespace) -> dict: + data_root = args.data_root.resolve() + db_path = args.db_path.resolve() + tmp_path = Path(str(db_path) + ".tmp") + timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") + report_path = data_root / f"sqlite_migration_report_{timestamp}.json" + + if tmp_path.exists(): + tmp_path.unlink() + if tmp_path.with_suffix(tmp_path.suffix + "-wal").exists(): + tmp_path.with_suffix(tmp_path.suffix + "-wal").unlink() + if tmp_path.with_suffix(tmp_path.suffix + "-shm").exists(): + tmp_path.with_suffix(tmp_path.suffix + "-shm").unlink() + + with connect(tmp_path) as conn: + initialize_schema(conn) + conn.commit() + conn.execute("BEGIN IMMEDIATE") + try: + summary = replace_database_from_paths( + conn, + **data_paths(data_root), + allow_balance_adjustments=False, + ) + conn.commit() + except Exception: + conn.rollback() + raise + + validation = validate_database(tmp_path, summary, strict_expected=not args.no_strict_expected) + source_hashes = source_file_hashes( + [ + data_root / "classnotes.txt", + data_root / "学生课时账户.md", + data_root / "教师档案.md", + data_root / "admin_tasks.json", + data_root / "operation_logs.jsonl", + data_root / "course_summary_state.json", + ] + ) + report = { + "created_at": datetime.now().isoformat(timespec="seconds"), + "schema_version": SCHEMA_VERSION, + "data_root": str(data_root), + "db_path": str(db_path), + "summary": summary, + "validation": validation, + "source_hashes": source_hashes, + "dry_run": bool(args.dry_run), + } + report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + if not validation["ok"]: + raise SystemExit("迁移校验失败：\n" + "\n".join(validation["errors"])) + + with connect(tmp_path) as conn: + write_audit(conn, {**report, "report_path": str(report_path)}) + + if args.dry_run: + print(json.dumps({**report, "tmp_db_path": str(tmp_path), "report_path": str(report_path)}, ensure_ascii=False, indent=2)) + return report + + db_path.parent.mkdir(parents=True, exist_ok=True) + if db_path.exists(): + backup_db = db_path.with_name(f"{db_path.name}.before-sqlite-migration-{timestamp}") + shutil.copy2(db_path, backup_db) + report["previous_db_backup"] = str(backup_db) + os.replace(tmp_path, db_path) + for suffix in ["-wal", "-shm"]: + sidecar = Path(str(tmp_path) + suffix) + if sidecar.exists(): + os.replace(sidecar, Path(str(db_path) + suffix)) + + archive_path, sha_path = create_archive(data_root, args.archives_root.resolve(), report_path, timestamp) + report["archive_path"] = str(archive_path) + report["archive_sha256_path"] = str(sha_path) + report["dry_run"] = False + report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print(json.dumps({**report, "report_path": str(report_path)}, ensure_ascii=False, indent=2)) + return report + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="将新时空纯文本教务数据迁移到 SQLite") + parser.add_argument("--data-root", type=Path, default=REPO_ROOT / "data") + parser.add_argument("--db-path", type=Path, default=Path("/data/xsk_education.db")) + parser.add_argument("--archives-root", type=Path, default=REPO_ROOT / "archives") + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--no-strict-expected", action="store_true", help="不校验当前生产数据的固定计数") + return parser.parse_args() + + +if __name__ == "__main__": + migrate(parse_args())