Add administrators

bot/add_card.py

@@ -1,27 +1,27 @@
 import requests
 from misskey.exceptions import MisskeyAPIException
 from client import client_connection
-from db_utils import insert_character
+from db_utils import insert_card
-from custom_types import Character
+from custom_types import Card
 from config import RARITY_TO_WEIGHT
 
 
-def add_character(
+def add_card(
         name: str,
         rarity: int,
         image_url: str) -> tuple[int, str]:
     '''
-    Adds a character to the database, uploading the image from a public URL to
+    Adds a card to the database, uploading the image from a public URL to
     the bot's Misskey Drive.
 
     Args:
-        name (str): Character name.
+        name (str): Card name.
-        rarity (int): Character rarity (e.g., 1-5).
+        rarity (int): Card rarity (e.g., 1-5).
         image_url (str): Public URL of the image from the post (e.g., from
             note['files'][i]['url']).
 
     Returns:
-        tuple[int, str]: Character ID and bot's Drive file_id.
+        tuple[int, str]: Card ID and bot's Drive file_id.
 
     Raises:
         ValueError: If inputs are invalid.
@@ -32,7 +32,7 @@ def add_character(
 
     # Validate inputs
     if not stripped_name:
-        raise ValueError('Character name cannot be empty.')
+        raise ValueError('Card name cannot be empty.')
     if rarity < 1:
         raise ValueError('Rarity must be a positive integer.')
     if rarity not in RARITY_TO_WEIGHT.keys():
@@ -55,10 +55,10 @@ def add_character(
                 from e
 
     # Insert into database
-    character_id = insert_character(
+    card_id = insert_card(
             stripped_name,
             rarity,
             RARITY_TO_WEIGHT[rarity],
             file_id
     )
-    return character_id, file_id
+    return card_id, file_id

bot/bot_app.py

@@ -12,6 +12,9 @@ if __name__ == '__main__':
     # Connect to DB
     db.connect()
 
+    # Setup default administrators
+    db.setup_administrators()
+
     print('Listening for notifications...')
     while True:
         if not process_notifications(client):

bot/config.py

@@ -1,5 +1,6 @@
 '''Essentials for the bot to function'''
 import configparser
+import json
 from os import environ, path
 
 
@@ -21,7 +22,9 @@ def get_config_file() -> str:
         raise ConfigError(f'Could not find {config_path}')
     return config_path
 
-def get_rarity_to_weight(config_section):
+
+def get_rarity_to_weight(
+        config_section: configparser.SectionProxy) -> dict[int, float]:
     """Parses Rarity_X keys from config and returns a {rarity: weight} dict."""
     rarity_weights = {}
     for key, value in config_section.items():
@@ -41,10 +44,9 @@ KEY = config['credentials']['Token']
 # Bot's Misskey instance URL
 INSTANCE = config['credentials']['Instance'].lower()
 
-# TODO: move this to db
 # Fedi handles in the traditional 'user@domain.tld' style, allows these users
 # to use extra admin exclusive commands with the bot
-ADMINS = config['application']['DefaultAdmins']
+ADMINS = json.loads(config['application']['DefaultAdmins'])
 # SQLite Database location
 DB_PATH = config['application']['DatabaseLocation']
 
@@ -53,4 +55,4 @@ NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize'])
 
 GACHA_ROLL_INTERVAL = int(config['gacha']['RollInterval'])
 
-RARITY_TO_WEIGHT = get_rarity_to_weight(config['gacha'])
+RARITY_TO_WEIGHT = get_rarity_to_weight(config['gacha'])

bot/custom_types.py

@@ -5,7 +5,7 @@ BotResponse = TypedDict('BotResponse', {
     'attachment_urls': List[str] | None
 })
 
-Character = TypedDict('Character', {
+Card = TypedDict('Card', {
     'id': int,
     'name': str,
     'rarity': int,

bot/db_utils.py

@@ -1,7 +1,7 @@
 from random import choices
 import sqlite3
 import config
-from custom_types import Character
+from custom_types import Card
 
 DB_PATH = config.DB_PATH
 CONNECTION: sqlite3.Connection
@@ -18,16 +18,38 @@ def connect() -> None:
     CURSOR = CONNECTION.cursor()
 
 
-def get_random_character() -> Character | None:
+def setup_administrators() -> None:
-    ''' Gets a random character from the database'''
+    '''Creates administrator players for each handle in the config file'''
-    CURSOR.execute('SELECT * FROM characters')
+    # Get default admins from config
-    characters = CURSOR.fetchall()
+    for username in config.ADMINS:
+        player_id = get_player(username)
+        if player_id == 0:
+            # Create player if not exists
+            print(f'Creating administrator player: {username}')
+            CURSOR.execute(
+                'INSERT INTO players (username, has_rolled, is_administrator) \
+                    VALUES (?, ?, ?)',
+                (username, False, True)
+            )
+        else:
+            # Update is_administrator if exists
+            print(f'Granting administrator to player: {username}')
+            CURSOR.execute(
+                'UPDATE players SET is_administrator = 1 WHERE id = ?',
+                (player_id,)
+            )
 
-    if not characters:
+
+def get_random_card() -> Card | None:
+    ''' Gets a random card from the database'''
+    CURSOR.execute('SELECT * FROM cards')
+    cards = CURSOR.fetchall()
+
+    if not cards:
         return None
 
-    weights = [config.RARITY_TO_WEIGHT[c['rarity']] for c in characters]
+    weights = [config.RARITY_TO_WEIGHT[c['rarity']] for c in cards]
-    chosen = choices(characters, weights=weights, k=1)[0]
+    chosen = choices(cards, weights=weights, k=1)[0]
 
     return {
         'id': chosen['id'],
@@ -37,73 +59,89 @@ def get_random_character() -> Character | None:
         'image_url': chosen['file_id']
     }
 
+
 def get_player(username: str) -> int:
     '''Retrieve a player ID by username, or return None if not found.'''
-    CURSOR.execute('SELECT id FROM users WHERE username = ?', (username,))
-    user = CURSOR.fetchone()
-    if user:
-        return int(user[0])
-
-def insert_player(username: str) -> int:
-    '''Insert a new player with default has_rolled = False and return their user ID.'''
     CURSOR.execute(
-        'INSERT INTO users (username, has_rolled) VALUES (?, ?)',
+        'SELECT id FROM players WHERE username = ?',
-        (username, False)
-    )
-    return CURSOR.lastrowid
-
-def delete_player(username: str) -> bool:
-    '''Permanently deletes a user and all their pulls.'''
-    CURSOR.execute(
-        'SELECT id FROM users WHERE username = ?',
         (username,)
     )
-    user = CURSOR.fetchone()
+    player = CURSOR.fetchone()
+    if player:
+        return int(player[0])
+    return 0
 
-    user_id = user[0]
+
+def insert_player(username: str) -> int:
+    '''Insert a new player with default has_rolled = False and return their
+    player ID.'''
+    CURSOR.execute(
+        'INSERT INTO players (username, has_rolled) VALUES (?, ?)',
+        (username, False)
+    )
+    return CURSOR.lastrowid if CURSOR.lastrowid else 0
+
+
+def delete_player(username: str) -> bool:
+    '''Permanently deletes a player and all their pulls.'''
+    CURSOR.execute(
+        'SELECT id FROM players WHERE username = ?',
+        (username,)
+    )
+    player = CURSOR.fetchone()
+
+    player_id = player[0]
 
     # Delete pulls
     CURSOR.execute(
-        'DELETE FROM pulls WHERE user_id = ?',
+        'DELETE FROM pulls WHERE player_id = ?',
-        (user_id,)
+        (player_id,)
     )
 
-    # Delete user
+    # Delete player
     CURSOR.execute(
-        'DELETE FROM users WHERE id = ?',
+        'DELETE FROM players WHERE id = ?',
-        (user_id,)
+        (player_id,)
     )
 
     return True
 
 
-
+def is_player_administrator(player_id: int) -> bool:
-def insert_character(
-        name: str, rarity: int, weight: float, file_id: str) -> int:
-    '''Inserts a character'''
     CURSOR.execute(
-        'INSERT INTO characters (name, rarity, weight, file_id) VALUES \
+        'SELECT is_administrator FROM PLAYERS WHERE id = ? LIMIT 1',
+        (player_id,)
+    )
+    row = CURSOR.fetchone()
+    return row[0] if row else False
+
+
+def insert_card(
+        name: str, rarity: int, weight: float, file_id: str) -> int:
+    '''Inserts a card'''
+    CURSOR.execute(
+        'INSERT INTO cards (name, rarity, weight, file_id) VALUES \
 (?, ?, ?, ?)',
         (name, rarity, weight, file_id)
     )
-    character_id = CURSOR.lastrowid
+    card_id = CURSOR.lastrowid
-    return character_id if character_id else 0
+    return card_id if card_id else 0
 
 
-def insert_pull(user_id: int, character_id: int) -> None:
+def insert_pull(player_id: int, card_id: int) -> None:
     '''Creates a pull in the database'''
     CURSOR.execute(
-        'INSERT INTO pulls (user_id, character_id) VALUES (?, ?)',
+        'INSERT INTO pulls (player_id, card_id) VALUES (?, ?)',
-        (user_id, character_id)
+        (player_id, card_id)
     )
 
 
-def get_last_rolled_at(user_id: int) -> int:
+def get_last_rolled_at(player_id: int) -> int:
-    '''Gets the timestamp when the user last rolled'''
+    '''Gets the timestamp when the player last rolled'''
     CURSOR.execute(
-        "SELECT timestamp FROM pulls WHERE user_id = ? ORDER BY timestamp \
+        "SELECT timestamp FROM pulls WHERE player_id = ? ORDER BY timestamp \
 DESC",
-        (user_id,))
+        (player_id,))
     row = CURSOR.fetchone()
     return row[0] if row else 0
 

bot/response.py

@@ -1,19 +1,20 @@
 from datetime import datetime, timedelta, timezone
 from typing import TypedDict, Any, List, Dict
-from db_utils import get_player, insert_player, delete_player, insert_pull, get_last_rolled_at, \
+from db_utils import get_player, insert_player, delete_player, insert_pull, \
-        get_random_character
+        get_last_rolled_at, get_random_card, is_player_administrator
-from add_character import add_character
+from add_card import add_card
 from config import GACHA_ROLL_INTERVAL
 from custom_types import BotResponse, ParsedNotification
 
 
 def do_roll(author: str) -> BotResponse:
-    '''Determines whether the user can roll, then pulls a random character'''
+    '''Determines whether the user can roll, then pulls a random card'''
     user_id = get_player(author)
     if not user_id:
         return {
-        'message':f'{author} 🛑 You haven’t signed up yet! Use the `signup` command to start playing.',
+            'message': f'{author} 🛑 You haven’t signed up yet! Use the \
-        'attachment_urls': None
+`signup` command to start playing.',
+            'attachment_urls': None
         }
     # Get date of user's last roll
     date = get_last_rolled_at(user_id)
@@ -45,39 +46,43 @@ def do_roll(author: str) -> BotResponse:
                 'attachment_urls': None
             }
 
-    character = get_random_character()
+    card = get_random_card()
 
-    if not character:
+    if not card:
         return {
             'message': f'{author} Uwaaa... something went wrong! No \
-characters found. 😿',
+cards found. 😿',
             'attachment_urls': None
         }
 
-    insert_pull(user_id, character['id'])
+    insert_pull(user_id, card['id'])
-    stars = '⭐️' * character['rarity']
+    stars = '⭐️' * card['rarity']
     return {
         'message': f'{author} 🎲 Congrats! You rolled {stars} \
-**{character['name']}**\nShe\'s all yours now~ 💖✨',
+**{card['name']}**\nShe\'s all yours now~ 💖✨',
-        'attachment_urls': [character['image_url']]
+        'attachment_urls': [card['image_url']]
     }
 
+
 def do_signup(author: str) -> BotResponse:
     '''Registers a new user if they haven’t signed up yet.'''
     user_id = get_player(author)
 
     if user_id:
         return {
-            'message':f'{author} 👀 You’re already signed up! Let the rolling begin~ 🎲',
+            'message': f'{author} 👀 You’re already signed up! Let the rolling \
+begin~ 🎲',
             'attachment_urls': None
         }
 
     new_user_id = insert_player(author)
     return {
-            'message': f'{author} ✅ Signed up successfully! Your gacha destiny begins now... ✨ Use the roll command to start!',
+            'message': f'{author} ✅ Signed up successfully! Your gacha \
+destiny begins now... ✨ Use the roll command to start!',
             'attachment_urls': None
         }
 
+
 def is_float(val: Any) -> bool:
     '''Returns true if `val` can be converted to a float'''
     try:
@@ -91,14 +96,14 @@ def do_create(
         author: str,
         arguments: List[str],
         note_obj: Dict[str, Any]) -> BotResponse:
-    '''Creates a character'''
+    '''Creates a card'''
     # Example call from bot logic
     image_url = note_obj.get('files', [{}])[0].get('url') \
         if note_obj.get('files') else None
 
     if not image_url:
         return {
-            'message': f'{author} You need an image to create a character, \
+            'message': f'{author} You need an image to create a card, \
 dumbass.',
             'attachment_urls': None
         }
@@ -123,13 +128,13 @@ must be a decimal value between 0.0 and 1.0',
             'attachment_urls': None
         }
 
-    character_id, file_id = add_character(
+    card_id, file_id = add_card(
         name=arguments[0],
         rarity=int(arguments[1]),
         image_url=image_url
     )
     return {
-        'message': f'{author} Added {arguments[0]}, ID {character_id}.',
+        'message': f'{author} Added {arguments[0]}, ID {card_id}.',
         'attachment_urls': [file_id]
     }
 
@@ -137,30 +142,43 @@ must be a decimal value between 0.0 and 1.0',
 def do_help(author: str) -> BotResponse:
     '''Provides a list of commands that the bot can do.'''
     return {
-        'message':f'{author} Here\'s what I can do:\n \
+        'message': f'{author} Here\'s what I can do:\n\
-            - `roll` Pulls a random character.\
+- `roll` Pulls a random card.\n\
-            - `create <name> <rarity>` Creates a character using a given image.\
+- `create <name> <rarity>` Creates a card using a given image.\n\
-            - `signup` Registers your account.\
+- `signup` Registers your account.\n\
-            - `delete_account` Deletes your account.\
+- `delete_account` Deletes your account.\n\
-            - `help` Shows this message',
+- `help` Shows this message',
-            'attachment_urls': None
+        'attachment_urls': None
     }
-            
+
+
 def delete_account(author: str) -> BotResponse:
     return {
-        'message':f'{author} ⚠️ This will permanently delete your account and all your cards.\n'
+        'message': f'{author} ⚠️ This will permanently delete your account \
-        'If you’re sure, reply with `confirm_delete` to proceed.\n\n'
+and all your cards.\n'
+        'If you’re sure, reply with `confirm_delete_account` to proceed.\n\n'
         '**There is no undo.** Your gacha luck will be lost to the void... 💀✨',
         'attachment_urls': None
 
     }
 
+
 def confirm_delete(author: str) -> BotResponse:
-    
     delete_player(author)
 
     return {
-        'message':f'{author} 🧼 Your account and all your cards have been deleted. RIP your gacha history 🕊️✨',
+        'message': f'{author} 🧼 Your account and all your cards have been \
+deleted. RIP your gacha history 🕊️✨',
+        'attachment_urls': None
+    }
+
+
+def do_admin_test(author: str) -> BotResponse:
+    player_id = get_player(author)
+    is_admin = is_player_administrator(player_id)
+    return {
+        'message': f'{author} You are {"not " if not is_admin else ""}an \
+admin.',
         'attachment_urls': None
     }
 
@@ -171,25 +189,23 @@ def generate_response(notification: ParsedNotification) -> BotResponse | None:
 
     # Temporary response variable
     res: BotResponse | None = None
-    # TODO: Check if the user has an account
     author = notification['author']
-    user_id = get_player(author)
+    player_id = get_player(author)
     command = notification['command']
-    # Check if the user is an administrator
-    # user_is_administrator = user_is_administrator()
 
     # Unrestricted commands
     match command:
+        case 'roll':
+            res = do_roll(author)
         case 'signup':
             res = do_signup(author)
         case 'help':
             res = do_help(author)
-        case 'roll':
-            res = do_roll(author)
         case _:
             pass
 
-    if not user_id:
+    # Commands beyond this point require the user to have an account
+    if not player_id:
         return res
 
     # User commands
@@ -200,15 +216,18 @@ def generate_response(notification: ParsedNotification) -> BotResponse | None:
                 notification['arguments'],
                 notification['note_obj']
             )
-        case 'signup':
-            res = do_signup(author)
         case 'delete_account':
             res = delete_account(author)
-        case 'confirm_delete':
+        case 'confirm_delete_account':
             res = confirm_delete(author)
+        case 'admin_test':
+            res = do_admin_test(author)
         case _:
             pass
-    # if not user_is_administrator:
+
-    return res
+    # Commands beyond this point require the user to be an administrator
+    if not is_player_administrator(player_id):
+        return res
 
     # Administrator commands go here
+    return res

example_config.ini

@@ -2,7 +2,7 @@
 [application]
 ; Comma separated list of fedi handles for any administrator users
 ; More can be added through the application
-DefaultAdmins    = ['admin@example.tld']
+DefaultAdmins    = ["@localadmin", "remoteadmin@example.tld"]
 ; SQLite Database location
 DatabaseLocation = ./gacha_game.db
 

migrations/0003_rename_tables.sql

@@ -0,0 +1,4 @@
+ALTER TABLE users RENAME TO players;
+ALTER TABLE characters RENAME TO cards;
+ALTER TABLE pulls RENAME user_id TO player_id;
+ALTER TABLE pulls RENAME character_id TO card_id;

migrations/0004_add_administrators.sql

@@ -0,0 +1 @@
+ALTER TABLE players ADD COLUMN is_administrator BOOLEAN NOT NULL DEFAULT 0;