"""初始化数据库：RBAC 账号、学习档案、题库与教材目录。 用法：在 backend/ 目录下执行 python -m seed [--force] --force 会清空全部学习数据与题库后重新导入，保留用户与角色。 这里是**唯一**会补种数据的地方：后端每次启动只对齐表结构 （``database.ensure_schema``），不再补章节 / 知识点 / 教材等业务数据， 因此自己准备的 math.db 覆盖回来后不会被种子改动。 """ from __future__ import annotations import argparse import json from datetime import date, timedelta from sqlalchemy import text from database import ( DATABASE_URL, SessionLocal, ensure_schema, migrate_legacy_courses, ) from models import ( AttemptItem, AttemptSession, DailyCompletion, ErrorEntry, Question, User, UserKnowledge, ) from seed.catalog import ensure_catalog from seed.data import KNOWLEDGE, QUESTIONS from seed.ebooks import ensure_ebooks from seed.knowledge_graph import ensure_knowledge_graph from seed.rbac import DEFAULT_USERS, ensure_rbac from services import ebooks as ebook_store from services.learning_profile import ( ensure_default_profiles, remove_unused_admin_profile, user_has_admin_role, ) from services.llm_config import ensure_llm_channels from services.knowledge_service import set_question_knowledge_names def _insert_questions(db) -> None: for item in QUESTIONS: db.add( Question( stem=item["stem"], options=json.dumps(item["options"], ensure_ascii=False), correct_index=item["correct_index"], knowledge_name=item["knowledge_name"], explanation=item["explanation"], ) ) def _demo_history(db, student: User) -> None: questions = db.query(Question).order_by(Question.id.asc()).all() configs = [ # (标题, 题目, 答错题号集合, 几天前) ("第一轮基础题", questions[:10], {2, 4, 7}, 3), ("第二轮综合题", questions[10:18], {1, 3, 5}, 1), ] for title, questions, wrong_indexes, days_ago in configs: if not questions: continue correct_count = len(questions) - len(wrong_indexes) session = AttemptSession( user_id=student.id, mode="legacy", title=title, total=len(questions), correct=correct_count, happened_on=date.today() - timedelta(days=days_ago), ) db.add(session) db.flush() for index, question in enumerate(questions, start=1): is_correct = index not in wrong_indexes selected = ( question.correct_index if is_correct else (question.correct_index + 1) % 4 ) teacher_text = question.explanation or "结合定义与图像性质重新判断。" question.used_count = (question.used_count or 0) + 1 db.add( AttemptItem( attempt_id=session.id, user_id=student.id, question_index=index, question_id=question.id, stem=question.stem, options=question.options, correct_index=question.correct_index, selected=selected, correct=is_correct, knowledge_name=question.knowledge_name, knowledge_names=json.dumps( [question.knowledge_name], ensure_ascii=False ), explanation=teacher_text, ) ) if not is_correct: db.add( ErrorEntry( user_id=student.id, attempt_id=session.id, source_label=f"{title} · {index}", knowledge_name=question.knowledge_name, knowledge_names=json.dumps( [question.knowledge_name], ensure_ascii=False ), diagnosis="答题错误，需要复习", question_id=question.id, stem=question.stem, options=question.options, selected=selected, correct_index=question.correct_index, explanation=teacher_text, ) ) # 保持每日任务演示数据（昨天已完成） db.add(DailyCompletion(user_id=student.id, done_on=date.today() - timedelta(days=1))) def _backfill_derived_fields(db) -> None: """给存量数据补派生字段（题目使用次数、知识点名快照）。 以前这些 UPDATE 挂在后端启动流程里，现在只属于初始化： 自己准备好的库覆盖回来后不会被重新计算 / 覆盖。 """ db.execute( text( "UPDATE questions SET used_count = (" "SELECT COUNT(*) FROM attempt_items " "WHERE attempt_items.question_id = questions.id" ") WHERE id IN (SELECT DISTINCT question_id FROM attempt_items " "WHERE question_id IS NOT NULL)" ) ) for question in db.query(Question).all(): has_links = db.execute( text( "SELECT 1 FROM question_knowledge " "WHERE question_id = :qid LIMIT 1" ), {"qid": question.id}, ).first() if has_links is None: set_question_knowledge_names( db, question, [question.knowledge_name], attach_to_chapter=True, ) for row in db.query(ErrorEntry).all(): if not row.knowledge_names or row.knowledge_names == "[]": row.knowledge_names = json.dumps( [row.knowledge_name], ensure_ascii=False ) for row in db.query(AttemptItem).all(): if not row.knowledge_names or row.knowledge_names == "[]": row.knowledge_names = json.dumps( [row.knowledge_name], ensure_ascii=False ) db.commit() def load_seed(force: bool = False) -> bool: ensure_schema() # 补列之后再搬数据：迁移要写 knowledge_resources.course_id migrate_legacy_courses() db = SessionLocal() created = False try: ensure_rbac(db) ensure_knowledge_graph(db) users = { username: db.query(User).filter(User.username == username).first() for username, _, _, _ in DEFAULT_USERS } ensure_default_profiles(db) ensure_llm_channels(db) if force: # --force 会清掉题库与学习数据：先留一份可回退的备份 from dbtool import backup as backup_database saved = backup_database(label="pre-force") print(f"[seed] --force 将清空题库与学习数据，已先备份：{saved}") db.query(AttemptItem).delete() db.query(DailyCompletion).delete() db.query(ErrorEntry).delete() db.query(AttemptSession).delete() db.query(UserKnowledge).delete() db.query(Question).delete() ensure_default_profiles(db) _insert_questions(db) db.flush() for question in db.query(Question).all(): set_question_knowledge_names( db, question, [question.knowledge_name] ) _demo_history(db, users["student"]) created = True elif db.query(Question).count() == 0: _insert_questions(db) db.flush() for question in db.query(Question).all(): set_question_knowledge_names( db, question, [question.knowledge_name] ) _demo_history(db, users["student"]) created = True db.commit() ensure_catalog(db, force_sync=force) # 电子书与教材目录同源：种子跑完就把 res/ 的书和磁盘上的存量电子书挂好 ensure_ebooks(db) ebook_store.backfill_pages(db) _backfill_derived_fields(db) for user in users.values(): if user and user_has_admin_role(db, user.id): remove_unused_admin_profile(db, user.id) return created finally: db.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description="初始化数学学习系统数据库") parser.add_argument("--force", action="store_true", help="清空学习数据后重新导入种子") args = parser.parse_args() created = load_seed(force=args.force) print(f"数据库：{DATABASE_URL}") if created: total = len(QUESTIONS) print( "已完成初始化：默认账号 admin/admin123、student/student123，" f"{total} 道题目。" ) else: print("已存在数据，跳过导入（重置可运行：python -m seed --force）。")