nex_math/backend/seed/__main__.py

"""初始化数据库：RBAC 账号、学习档案、题库与教材目录。

用法：在 backend/ 目录下执行 python -m seed [--force]
--force 会清空全部学习数据与题库后重新导入，保留用户与角色。
"""
from __future__ import annotations

import argparse
import json
from datetime import date, timedelta

from database import DATABASE_URL, SessionLocal, ensure_schema
from models import (
    AttemptItem,
    AttemptSession,
    DailyCompletion,
    ErrorEntry,
    Question,
    User,
    UserKnowledge,
)
from seed.catalog import ensure_catalog
from seed.data import KNOWLEDGE, QUESTIONS
from seed.rbac import DEFAULT_USERS, ensure_rbac
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 load_seed(force: bool = False) -> bool:
    ensure_schema()
    db = SessionLocal()
    created = False
    try:
        ensure_rbac(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:
            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)
        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）。")