#!/usr/bin/env python3 """Kithara Studio admin CLI — manage users, invite codes, and system settings.""" import argparse import sys import os import asyncio import uuid import secrets sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'backend', 'src')) import aiosqlite import bcrypt DEFAULT_DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data', 'kithara.db') db_path = DEFAULT_DB_PATH async def get_db(): db = await aiosqlite.connect(db_path) db.row_factory = aiosqlite.Row return db async def cmd_list_users(args): db = await get_db() try: async with db.execute("SELECT id, username, display_name, role, api_key IS NOT NULL as has_api_key, created_at FROM users ORDER BY created_at") as cursor: rows = await cursor.fetchall() if not rows: print("No users found.") return print(f"{'ID':<36} {'Username':<20} {'Display Name':<20} {'Role':<10} {'API Key':<8} {'Created'}") print("-" * 120) for row in rows: print(f"{row['id']:<36} {row['username']:<20} {row['display_name'] or '':<20} {row['role']:<10} {'Yes' if row['has_api_key'] else 'No':<8} {row['created_at']}") finally: await db.close() async def cmd_create_user(args): db = await get_db() try: async with db.execute("SELECT id FROM users WHERE username = ?", (args.username,)) as cursor: if await cursor.fetchone(): print(f"Error: Username '{args.username}' already exists.") return password_hash = bcrypt.hashpw(args.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') user_id = str(uuid.uuid4()) await db.execute( "INSERT INTO users (id, username, password_hash, display_name, role) VALUES (?, ?, ?, ?, ?)", (user_id, args.username, password_hash, args.display_name or args.username, args.role) ) await db.commit() print(f"Created user '{args.username}' (id: {user_id}, role: {args.role})") finally: await db.close() async def cmd_delete_user(args): db = await get_db() try: async with db.execute("SELECT id, username FROM users WHERE id = ? OR username = ?", (args.identifier, args.identifier)) as cursor: row = await cursor.fetchone() if not row: print(f"Error: User '{args.identifier}' not found.") return if not args.confirm: print(f"Are you sure you want to delete user '{row['username']}' (id: {row['id']})? Use --confirm to proceed.") return await db.execute("DELETE FROM users WHERE id = ?", (row['id'],)) await db.commit() print(f"Deleted user '{row['username']}' (id: {row['id']})") finally: await db.close() async def cmd_change_role(args): db = await get_db() try: result = await db.execute("UPDATE users SET role = ? WHERE id = ? OR username = ?", (args.role, args.identifier, args.identifier)) await db.commit() if result.rowcount == 0: print(f"Error: User '{args.identifier}' not found.") else: print(f"Updated role to '{args.role}' for user '{args.identifier}'") finally: await db.close() async def cmd_reset_password(args): db = await get_db() try: password_hash = bcrypt.hashpw(args.new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') result = await db.execute( "UPDATE users SET password_hash = ? WHERE id = ? OR username = ?", (password_hash, args.identifier, args.identifier) ) await db.commit() if result.rowcount == 0: print(f"Error: User '{args.identifier}' not found.") else: print(f"Reset password for user '{args.identifier}'") finally: await db.close() async def cmd_create_invite(args): db = await get_db() try: code = args.code or secrets.token_urlsafe(16) await db.execute( "INSERT INTO invite_codes (code, created_by, max_uses, current_uses) VALUES (?, ?, ?, ?)", (code, 'admin', args.max_uses, 0) ) await db.commit() print(f"Created invite code: {code}") if args.max_uses > 1: print(f" Max uses: {args.max_uses}") finally: await db.close() async def cmd_list_invites(args): db = await get_db() try: async with db.execute("SELECT code, created_by, max_uses, current_uses, created_at FROM invite_codes ORDER BY created_at DESC") as cursor: rows = await cursor.fetchall() if not rows: print("No invite codes found.") return print(f"{'Code':<25} {'Created By':<15} {'Uses':<10} {'Max Uses':<10} {'Created'}") print("-" * 100) for row in rows: print(f"{row['code']:<25} {row['created_by']:<15} {row['current_uses']:<10} {row['max_uses']:<10} {row['created_at']}") finally: await db.close() async def cmd_revoke_invite(args): db = await get_db() try: result = await db.execute("DELETE FROM invite_codes WHERE code = ?", (args.code,)) await db.commit() if result.rowcount == 0: print(f"Error: Invite code '{args.code}' not found.") else: print(f"Revoked invite code: {args.code}") finally: await db.close() def main(): global db_path parser = argparse.ArgumentParser(description='Kithara Studio Admin CLI') parser.add_argument('--db', default=DEFAULT_DB_PATH, help='Path to kithara.db') subparsers = parser.add_subparsers(dest='command', help='Available commands') subparsers.add_parser('list-users', help='List all users') create_user = subparsers.add_parser('create-user', help='Create a new user') create_user.add_argument('username', help='Username') create_user.add_argument('--password', required=True, help='Password') create_user.add_argument('--display-name', help='Display name (defaults to username)') create_user.add_argument('--role', default='user', choices=['user', 'admin'], help='User role') delete_user = subparsers.add_parser('delete-user', help='Delete a user') delete_user.add_argument('identifier', help='User ID or username') delete_user.add_argument('--confirm', action='store_true', help='Confirm deletion') change_role = subparsers.add_parser('change-role', help='Change user role') change_role.add_argument('identifier', help='User ID or username') change_role.add_argument('role', choices=['user', 'admin'], help='New role') reset_pw = subparsers.add_parser('reset-password', help='Reset a user password') reset_pw.add_argument('identifier', help='User ID or username') reset_pw.add_argument('new_password', help='New password') create_invite = subparsers.add_parser('create-invite', help='Create an invite code') create_invite.add_argument('--code', help='Custom invite code (auto-generated if not specified)') create_invite.add_argument('--max-uses', type=int, default=1, help='Max number of uses') subparsers.add_parser('list-invites', help='List all invite codes') revoke_invite = subparsers.add_parser('revoke-invite', help='Revoke an invite code') revoke_invite.add_argument('code', help='Invite code to revoke') args = parser.parse_args() if not args.command: parser.print_help() sys.exit(1) db_path = args.db cmd_map = { 'list-users': cmd_list_users, 'create-user': cmd_create_user, 'delete-user': cmd_delete_user, 'change-role': cmd_change_role, 'reset-password': cmd_reset_password, 'create-invite': cmd_create_invite, 'list-invites': cmd_list_invites, 'revoke-invite': cmd_revoke_invite, } asyncio.run(cmd_map[args.command](args)) if __name__ == '__main__': main()