schedule-management-system/backend/app/services/seed.py

from datetime import time
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.security import hash_password
from backend.app.core.config import settings
from backend.app.models import Role, Permission, RolePermission, User, Campus, Department, Shift


PERMISSIONS = [
    ("shifts.view", "查看班次", "shifts"),
    ("shifts.create", "新增班次", "shifts"),
    ("shifts.edit", "编辑班次", "shifts"),
    ("shifts.delete", "删除班次", "shifts"),
    ("schedule.view", "查看排班", "schedule"),
    ("schedule.create", "新增排班", "schedule"),
    ("schedule.edit", "编辑排班", "schedule"),
    ("schedule.delete", "删除排班", "schedule"),
    ("schedule.swap", "调班", "schedule"),
    ("schedule.substitute", "替班", "schedule"),
    ("schedule.stopClinic", "停诊", "schedule"),
    ("schedule.export", "导出排班", "schedule"),
    ("duty.view", "查看值班", "duty"),
    ("duty.create", "新增值班", "duty"),
    ("duty.edit", "编辑值班", "duty"),
    ("duty.delete", "删除值班", "duty"),
    ("duty.export", "导出值班", "duty"),
    ("statistics.view", "查看统计", "statistics"),
    ("statistics.export", "导出统计", "statistics"),
    ("users.view", "查看人员", "users"),
    ("users.create", "新增人员", "users"),
    ("users.edit", "编辑人员", "users"),
    ("users.delete", "删除人员", "users"),
    ("users.assignRole", "分配角色", "users"),
]

ROLE_TEMPLATES = {
    "管理员": [
        "shifts.view", "shifts.create", "shifts.edit", "shifts.delete",
        "schedule.view", "schedule.create", "schedule.edit", "schedule.delete", "schedule.swap", "schedule.substitute", "schedule.stopClinic", "schedule.export",
        "duty.view", "duty.create", "duty.edit", "duty.delete", "duty.export",
        "statistics.view", "statistics.export",
        "users.view", "users.create", "users.edit", "users.delete", "users.assignRole"
    ],
    "护士长": [
        "schedule.view", "schedule.create", "schedule.edit", "schedule.swap", "schedule.substitute", "schedule.stopClinic",
        "duty.view", "duty.create", "duty.edit",
        "statistics.view"
    ],
    "医生": [
        "schedule.view",
        "duty.view",
        "statistics.view"
    ],
    "护士": [
        "schedule.view",
        "duty.view"
    ],
    "排班员": [
        "shifts.view", "shifts.create", "shifts.edit", "shifts.delete",
        "schedule.view", "schedule.create", "schedule.edit", "schedule.delete", "schedule.swap", "schedule.substitute", "schedule.stopClinic",
        "duty.view", "duty.create", "duty.edit", "duty.delete",
        "statistics.view", "statistics.export"
    ]
}


async def seed_data(db: AsyncSession) -> None:
    permission_exists = await db.execute(select(Permission.code).limit(1))
    if not permission_exists.first():
        for code, name, module in PERMISSIONS:
            db.add(Permission(code=code, name=name, module=module))

    role_exists = await db.execute(select(Role.id).limit(1))
    if not role_exists.first():
        roles = {}
        for name, permissions in ROLE_TEMPLATES.items():
            role = Role(name=name, description=f"{name}角色模板", is_system=True)
            db.add(role)
            roles[name] = (role, permissions)
        await db.flush()
        for role, permissions in roles.values():
            for code in permissions:
                db.add(RolePermission(role_id=role.id, permission_code=code))

    campus_exists = await db.execute(select(Campus.id).limit(1))
    if not campus_exists.first():
        campuses = [
            Campus(name="主院区"),
            Campus(name="分院区"),
            Campus(name="东院区")
        ]
        db.add_all(campuses)
        await db.flush()
        dept_seed = [
            (campuses[0].id, "心内科"),
            (campuses[0].id, "呼吸科"),
            (campuses[0].id, "神经内科"),
            (campuses[0].id, "消化内科"),
            (campuses[1].id, "骨科"),
            (campuses[1].id, "普外科"),
            (campuses[1].id, "泌尿外科"),
            (campuses[2].id, "儿科"),
            (campuses[2].id, "妇产科")
        ]
        for campus_id, name in dept_seed:
            db.add(Department(campus_id=campus_id, name=name))

    shift_exists = await db.execute(select(Shift.id).limit(1))
    if not shift_exists.first():
        db.add_all([
            Shift(name="白班", start_time=time.fromisoformat("08:00"), end_time=time.fromisoformat("16:00"), enabled=True),
            Shift(name="夜班", start_time=time.fromisoformat("16:00"), end_time=time.fromisoformat("08:00"), enabled=True, remark="跨天班次"),
            Shift(name="门诊", start_time=time.fromisoformat("09:00"), end_time=time.fromisoformat("12:00"), enabled=True),
            Shift(name="急诊", start_time=time.fromisoformat("00:00"), end_time=time.fromisoformat("23:59"), enabled=True),
        ])

    if settings.seed_admin_account and settings.seed_admin_password:
        admin_exists = await db.execute(select(User).where(User.account == settings.seed_admin_account))
        if not admin_exists.scalar_one_or_none():
            role_result = await db.execute(select(Role).where(Role.name == "管理员"))
            admin_role = role_result.scalar_one_or_none()
            if admin_role:
                db.add(User(
                    name="系统管理员",
                    account=settings.seed_admin_account,
                    phone="13800000000",
                    title="系统管理员",
                    role_id=admin_role.id,
                    status="active",
                    password_hash=hash_password(settings.seed_admin_password),
                    token_version=1
                ))

    await db.commit()