#!/usr/bin/env python3 """ Seed initial admin user, roles, and menus Creates: 1. Two roles: admin and user 2. Admin user: cosmo / cosmo 3. Admin menu structure Usage: python scripts/seed_admin.py """ import asyncio import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from sqlalchemy import select from app.database import AsyncSessionLocal from app.models.db import User, Role, Menu, RoleMenu import bcrypt import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def hash_password(password: str) -> str: """Hash password using bcrypt""" return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') async def main(): """Seed admin data""" async with AsyncSessionLocal() as session: try: # 1. Create roles logger.info("Creating roles...") # Check if roles already exist result = await session.execute(select(Role)) existing_roles = result.scalars().all() if existing_roles: logger.info(f"Roles already exist: {[r.name for r in existing_roles]}") admin_role = next((r for r in existing_roles if r.name == 'admin'), None) user_role = next((r for r in existing_roles if r.name == 'user'), None) else: admin_role = Role( name='admin', display_name='管理员', description='系统管理员,拥有所有权限' ) user_role = Role( name='user', display_name='普通用户', description='普通用户,仅有基本访问权限' ) session.add(admin_role) session.add(user_role) await session.flush() logger.info(f"✓ Created roles: admin, user") # 2. Create admin user logger.info("Creating admin user...") # Check if admin user already exists result = await session.execute( select(User).where(User.username == 'cosmo') ) existing_user = result.scalar_one_or_none() if existing_user: logger.info(f"Admin user 'cosmo' already exists (id={existing_user.id})") admin_user = existing_user else: admin_user = User( username='cosmo', password_hash=hash_password('cosmo'), email='admin@cosmo.com', full_name='Cosmo Administrator', is_active=True ) session.add(admin_user) await session.flush() # Assign admin role to user using direct insert to avoid lazy loading from app.models.db.user import user_roles await session.execute( user_roles.insert().values( user_id=admin_user.id, role_id=admin_role.id ) ) await session.flush() logger.info(f"✓ Created admin user: cosmo / cosmo") # 3. Create admin menus logger.info("Creating admin menus...") # Check if menus already exist result = await session.execute(select(Menu)) existing_menus = result.scalars().all() if existing_menus: logger.info(f"Menus already exist ({len(existing_menus)} menus)") else: # Root menu items dashboard_menu = Menu( name='dashboard', title='控制台', icon='dashboard', path='/admin/dashboard', component='admin/Dashboard', sort_order=1, is_active=True, description='系统控制台' ) data_management_menu = Menu( name='data_management', title='数据管理', icon='database', path=None, # Parent menu, no direct path component=None, sort_order=2, is_active=True, description='数据管理模块' ) session.add(dashboard_menu) session.add(data_management_menu) await session.flush() # Sub-menu items under data_management celestial_bodies_menu = Menu( parent_id=data_management_menu.id, name='celestial_bodies', title='天体数据列表', icon='planet', path='/admin/celestial-bodies', component='admin/CelestialBodies', sort_order=1, is_active=True, description='查看和管理天体数据' ) static_data_menu = Menu( parent_id=data_management_menu.id, name='static_data', title='静态数据列表', icon='data', path='/admin/static-data', component='admin/StaticData', sort_order=2, is_active=True, description='查看和管理静态数据(星座、星系等)' ) nasa_data_menu = Menu( parent_id=data_management_menu.id, name='nasa_data', title='NASA数据下载管理', icon='download', path='/admin/nasa-data', component='admin/NasaData', sort_order=3, is_active=True, description='管理NASA Horizons数据下载' ) session.add(celestial_bodies_menu) session.add(static_data_menu) session.add(nasa_data_menu) await session.flush() logger.info(f"✓ Created {5} menu items") # 4. Assign all menus to admin role logger.info("Assigning menus to admin role...") all_menus = [ dashboard_menu, data_management_menu, celestial_bodies_menu, static_data_menu, nasa_data_menu ] for menu in all_menus: role_menu = RoleMenu(role_id=admin_role.id, menu_id=menu.id) session.add(role_menu) await session.flush() logger.info(f"✓ Assigned {len(all_menus)} menus to admin role") await session.commit() logger.info("\n" + "=" * 60) logger.info("Admin data seeded successfully!") logger.info("=" * 60) logger.info("Admin credentials:") logger.info(" Username: cosmo") logger.info(" Password: cosmo") logger.info("=" * 60) except Exception as e: await session.rollback() logger.error(f"Error seeding admin data: {e}") import traceback traceback.print_exc() sys.exit(1) if __name__ == "__main__": asyncio.run(main())