Codex
233 lines
9.0 KiB
Python
233 lines
9.0 KiB
Python
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())
|