ajyy/backend/app/services/seed.py

"""幂等种子数据：角色、菜单、管理员、热线/二维码配置"""
import uuid
from typing import Dict, List

import bcrypt
from sqlalchemy.orm import Session

from app.core.config import get_settings
from app.models.business import HotlineConfig, QrcodeConfig
from app.models.system import SysMenu, SysRole, SysRoleMenu, SysUser


def _uuid() -> str:
    return str(uuid.uuid4())


def _hash_password(plain: str) -> str:
    return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


# 菜单定义：(menu_code, menu_name, path, menu_type, sort, parent_code|None)
MENU_DEFS = [
    ("dashboard", "工作台", "/dashboard", "C", 10, None),
    ("feedback", "在线反馈", "/feedback", "C", 20, None),
    ("suggestion", "建言献策", "/suggestion", "C", 30, None),
    ("hotline", "监督热线", "/hotline", "C", 40, None),
    ("qrcode", "二维码配置", "/qrcode", "C", 50, None),
    ("user", "用户管理", "/user", "C", 60, None),
    ("role", "角色授权", "/role", "C", 70, None),
    ("audit", "操作审计", "/audit", "C", 80, None),
    ("feedback:handle", "反馈办理", "", "F", 21, "feedback"),
    ("suggestion:handle", "建言办理", "", "F", 31, "suggestion"),
    ("hotline:write", "热线编辑", "", "F", 41, "hotline"),
    ("qrcode:write", "二维码编辑", "", "F", 51, "qrcode"),
    ("user:manage", "用户管理操作", "", "F", 61, "user"),
    ("role:grant", "角色授权操作", "", "F", 71, "role"),
]

ROLE_DEFS = [
    ("admin", "系统管理员", "全部菜单与按钮权限"),
    ("handler", "办理员", "工作台、反馈、建言及办理权限"),
    ("viewer", "查阅员", "工作台、反馈、建言查阅（无办理）"),
]

# role_code -> menu_code 列表；admin 用特殊标记表示全部
ROLE_MENU_CODES: Dict[str, List[str] | str] = {
    "admin": "*",
    "handler": [
        "dashboard",
        "feedback",
        "suggestion",
        "feedback:handle",
        "suggestion:handle",
    ],
    "viewer": ["dashboard", "feedback", "suggestion"],
}


def seed_if_empty(db: Session) -> None:
    """若库中尚无角色则写入种子；已有则直接返回（幂等）。"""
    if db.query(SysRole).count() > 0:
        return

    settings = get_settings()
    now_by = "system"

    # --- roles ---
    role_by_code: Dict[str, SysRole] = {}
    for code, name, remark in ROLE_DEFS:
        role = SysRole(
            id=_uuid(),
            role_code=code,
            role_name=name,
            remark=remark,
            del_flag="0",
            create_by=now_by,
            update_by=now_by,
        )
        db.add(role)
        role_by_code[code] = role
    db.flush()

    # --- menus ---
    menu_by_code: Dict[str, SysMenu] = {}
    for code, name, path, mtype, sort, parent_code in MENU_DEFS:
        parent_id = "0"
        if parent_code:
            parent_id = menu_by_code[parent_code].id
        menu = SysMenu(
            id=_uuid(),
            parent_id=parent_id,
            menu_code=code,
            menu_name=name,
            path=path,
            menu_type=mtype,
            sort=sort,
            visible="1",
            del_flag="0",
            create_by=now_by,
            update_by=now_by,
        )
        db.add(menu)
        menu_by_code[code] = menu
    db.flush()

    # --- role_menu ---
    for role_code, codes in ROLE_MENU_CODES.items():
        role = role_by_code[role_code]
        if codes == "*":
            target_codes = list(menu_by_code.keys())
        else:
            target_codes = codes  # type: ignore[assignment]
        for menu_code in target_codes:
            db.add(
                SysRoleMenu(
                    id=_uuid(),
                    role_id=role.id,
                    menu_id=menu_by_code[menu_code].id,
                )
            )

    # --- admin user ---
    admin_role = role_by_code["admin"]
    db.add(
        SysUser(
            id=_uuid(),
            username=settings.INIT_ADMIN_USERNAME,
            password_hash=_hash_password(settings.INIT_ADMIN_PASSWORD),
            display_name="管理员",
            role_id=admin_role.id,
            status="0",
            del_flag="0",
            create_by=now_by,
            update_by=now_by,
        )
    )

    # --- hotline / qrcode defaults ---
    if db.query(HotlineConfig).count() == 0:
        db.add(
            HotlineConfig(
                id=_uuid(),
                phone="0551-00000000",
                email="jj@example.com",
                address="安徽建工集团医院纪检室",
                work_time="工作日 9:00-17:00",
                del_flag="0",
                create_by=now_by,
                update_by=now_by,
            )
        )
    if db.query(QrcodeConfig).count() == 0:
        db.add(
            QrcodeConfig(
                id=_uuid(),
                land_url="https://example.com/h5/",
                title="安建医院纪检扫码反馈",
                image_url="",
                del_flag="0",
                create_by=now_by,
                update_by=now_by,
            )
        )

    db.commit()