fitguide/scripts/seed.py

"""Bootstrap default permissions, roles, and the first superuser.

Idempotent: safe to run repeatedly. Reads the superuser credentials from
settings (env vars or .env).

Usage:
    uv run python -m scripts.seed
"""

import asyncio

from app.core.config import settings
from app.core.database import AsyncSessionLocal, init_db
from app.crud import attachment_point_name as apn_crud
from app.crud import part_type as part_type_crud
from app.crud import role as role_crud
from app.crud import sub_part_type as sub_part_type_crud
from app.crud import user as user_crud

ATTACHMENT_POINT_NAMES = ["Objective", "Ocular", "Rail"]

PART_TYPES = ["Flip Cover", "ARD", "Polarizer", "Adapter Ring", "Scope"]

SUB_PART_TYPES = {
    "Scope": ["Riflescope", "Binoculars", "Combat Optic", "Red Dot", "Spotter"],
}

# Default permission catalogue: "<resource>:<action>".
PERMISSIONS = [
    ("users:read", "View users"),
    ("users:write", "Create and update users"),
    ("users:delete", "Delete users"),
    ("roles:read", "View roles and permissions"),
    ("roles:write", "Create, update and delete roles"),
    ("iso_threads:read", "View ISO threads"),
    ("iso_threads:write", "Create and update ISO threads"),
    ("iso_threads:delete", "Delete ISO threads"),
    ("unc_threads:read", "View UNC threads"),
    ("unc_threads:write", "Create and update UNC threads"),
    ("unc_threads:delete", "Delete UNC threads"),
    ("manufacturers:read", "View manufacturers"),
    ("manufacturers:write", "Create and update manufacturers"),
    ("manufacturers:delete", "Delete manufacturers"),
    ("scope_models:read", "View scope models"),
    ("scope_models:write", "Create and update scope models"),
    ("scope_models:delete", "Delete scope models"),
    ("part_colours:read", "View part colours"),
    ("part_colours:write", "Create and update part colours"),
    ("part_colours:delete", "Delete part colours"),
    ("part_types:read", "View part types"),
    ("part_types:write", "Create and update part types"),
    ("part_types:delete", "Delete part types"),
    ("sub_part_types:read", "View sub part types"),
    ("sub_part_types:write", "Create and update sub part types"),
    ("sub_part_types:delete", "Delete sub part types"),
    ("imported_parts:read", "View imported parts"),
    ("imported_parts:write", "Sync and configure imported parts"),
    ("imported_parts:delete", "Delete imported parts"),
    ("static_connections:read", "View static connections"),
    ("static_connections:write", "Create and update static connections"),
    ("static_connections:delete", "Delete static connections"),
    ("clamp_connections:read", "View clamp connections"),
    ("clamp_connections:write", "Create and update clamp connections"),
    ("clamp_connections:delete", "Delete clamp connections"),
    ("attachment_point_names:read", "View attachment point names"),
    ("attachment_point_names:write", "Create and update attachment point names"),
    ("attachment_point_names:delete", "Delete attachment point names"),
    ("attachment_points:read", "View attachment points"),
    ("attachment_points:write", "Create and update attachment points"),
    ("attachment_points:delete", "Delete attachment points"),
    ("augmented_parts:read", "View augmented parts"),
    ("augmented_parts:write", "Create and update augmented parts"),
    ("augmented_parts:delete", "Delete augmented parts"),
    ("fits:read", "View part fits"),
    ("fits:write", "Create, promote and recompute part fits"),
    ("fits:delete", "Delete part fits"),
    ("audit_log:read", "View audit log"),
]

# Role auto-assigned to new (registered/LDAP-provisioned) users unless an admin
# marks a different one as default in the UI.
DEFAULT_ROLE = "user"

# Default roles -> permission names. "admin" gets everything.
ROLES = {
    "admin": [name for name, _ in PERMISSIONS],  # includes audit_log:read
    "user": [
        "users:read",
        "iso_threads:read",
        "unc_threads:read",
        "manufacturers:read",
        "scope_models:read",
        "part_colours:read",
        "part_types:read",
        "sub_part_types:read",
        "imported_parts:read",
        "static_connections:read",
        "clamp_connections:read",
        "attachment_point_names:read",
        "attachment_points:read",
        "augmented_parts:read",
        "fits:read",
    ],
}


async def seed() -> None:
    await init_db()

    async with AsyncSessionLocal() as db:
        # Permissions
        perms = {}
        for name, description in PERMISSIONS:
            perm = await role_crud.get_permission_by_name(db, name)
            if perm is None:
                perm = await role_crud.create_permission(
                    db, name=name, description=description
                )
            perms[name] = perm

        # Roles
        roles = {}
        for role_name, perm_names in ROLES.items():
            role = await role_crud.get_by_name(db, role_name)
            if role is None:
                role = await role_crud.create(
                    db,
                    name=role_name,
                    description=f"{role_name.title()} role",
                    permission_ids=[perms[p].id for p in perm_names],
                )
            roles[role_name] = role

        # Ensure a default role is marked (idempotent).
        if await role_crud.get_default(db) is None:
            default_role = roles.get(DEFAULT_ROLE)
            if default_role is not None:
                await role_crud.set_default(db, default_role)

        # Attachment point names
        for apn_name in ATTACHMENT_POINT_NAMES:
            if await apn_crud.get_by_name(db, apn_name) is None:
                await apn_crud.create(db, name=apn_name)

        # Part types
        for pt_name in PART_TYPES:
            if await part_type_crud.get_by_name(db, pt_name) is None:
                await part_type_crud.create(db, name=pt_name)

        # Sub part types
        for pt_name, spt_names in SUB_PART_TYPES.items():
            pt = await part_type_crud.get_by_name(db, pt_name)
            if pt is not None:
                for spt_name in spt_names:
                    if (
                        await sub_part_type_crud.get_by_name_and_part_type(
                            db, spt_name, pt.id
                        )
                        is None
                    ):
                        await sub_part_type_crud.create(
                            db, name=spt_name, part_type_id=pt.id
                        )

        # Superuser
        existing = await user_crud.get_by_email(db, settings.first_superuser_email)
        if existing is None:
            await user_crud.create(
                db,
                email=settings.first_superuser_email,
                password=settings.first_superuser_password,
                full_name="Administrator",
                is_superuser=True,
                role_ids=[roles["admin"].id],
            )
            print(f"Created superuser: {settings.first_superuser_email}")
        else:
            print(f"Superuser already exists: {settings.first_superuser_email}")

    print("Seed complete.")


if __name__ == "__main__":
    asyncio.run(seed())