bot/db_utils.py

@@ -18,6 +18,7 @@ from random import choices
 import sqlite3
 import config
 from custom_types import Card
+from datetime import datetime
 
 DB_PATH = config.DB_PATH
 CONNECTION: sqlite3.Connection
@@ -77,7 +78,7 @@ def get_random_card() -> Card | None:
 
 def get_cards(card_ids: list[int]) -> list[tuple]:
     '''
-    Retrieves stats for a list of card IDs.
+    Retrieves information about cards from the database by their IDs.
     Returns a list of tuples: (id, name, rarity, file_id, power, charm, wit, ...)
     '''
     if not card_ids:
@@ -87,7 +88,9 @@ def get_cards(card_ids: list[int]) -> list[tuple]:
     query = f'SELECT * FROM cards WHERE id IN ({placeholders})'
 
     CURSOR.execute(query, card_ids)
-    return CURSOR.fetchall()
+    
+    res = CURSOR.fetchall()
+    return res
 
 def get_player(username: str) -> int:
     '''Retrieve a player ID by username, or return None if not found.'''
@@ -100,6 +103,207 @@ def get_player(username: str) -> int:
         return int(player[0])
     return 0
 
+def insert_duel_request(username: str, target: str, duel_type: str) -> int:
+    '''Inserts a duel request into the database.
+    
+    Args:
+        username (str): The username of the player who initiated the duel.
+        target (str): The username of the player being challenged.
+        duel_type (str): The type of duel (e.g., casual, competitive).
+    
+    
+    '''
+
+    # get the player ids
+    player_1_1d = get_player(username)
+    player_2_id = get_player(target)
+    
+
+    competitive = duel_type
+
+    # insert the duel request and get the duel ID
+    CURSOR.execute(
+        'INSERT INTO duel_requests (player_1_id, player_2_id, competitive) VALUES (?, ?, ?)',
+        (player_1_1d, player_2_id, competitive)
+    )
+    duel_id = CURSOR.lastrowid
+
+    return duel_id
+
+def get_duel_request(duel_request_id: int) -> dict | None:
+    '''Retrieves a duel request by its ID.
+    
+    Args:
+        duel_id (int): The ID of the duel request.
+    
+    Returns:
+        dict: A dictionary containing the duel request details, or None if not found.
+    '''
+    CURSOR.execute(
+        'SELECT * FROM duel_requests WHERE duel_request_id = ?',
+        (duel_request_id,)
+    )
+    row = CURSOR.fetchone()
+    if row:
+        return dict(row)
+    return None
+
+def accept_duel_request(duel_request_id: int) -> bool:
+    ''' Sets up the duel request to an actual duel.
+    Args:
+        duel_request_id (int): The ID of the duel request to begin.
+    '''
+    create_duel(duel_request_id)
+
+    CURSOR.execute(
+        'UPDATE duel_requests SET accepted = 1 WHERE duel_request_id = ?',
+        (duel_request_id,)
+    )
+    
+def take_cards(player_id: int, count: int) -> list[int]:
+        '''Takes a specified number of random cards from a player.
+        
+        Args:
+            player_id (int): The ID of the player to take cards from.
+            count (int): The number of cards to take.
+        
+        Returns:
+            list[int]: A list of card IDs taken from the player.
+        '''
+        CURSOR.execute(
+            'SELECT card_id FROM pulls WHERE player_id = ? ORDER BY RANDOM() LIMIT ?',
+            (player_id, count)
+        )
+        rows = CURSOR.fetchall()
+        return [row['card_id'] for row in rows]
+
+def create_duel(duel_request_id: int) -> None:
+    '''Creates a duel from a duel request.
+    
+    Args:
+        duel_request_id (int): The ID of the duel request to create a duel from.
+    '''
+    CURSOR.execute(
+        'SELECT player_1_id, player_2_id, competitive FROM duel_requests WHERE duel_request_id = ?',
+        (duel_request_id,)
+    )
+    row = CURSOR.fetchone()
+
+    # Select a random player to set as attacker
+    attacker_id = choices([row['player_1_id'], row['player_2_id']])[0]
+
+    # Set the last round date to now
+    last_round_dt = datetime.now().isoformat()
+
+    # take 3 random cards from each player
+
+    player_1_cards = take_cards(row['player_1_id'], 3)
+    player_2_cards = take_cards(row['player_2_id'], 3)
+    
+    # Insert the duel into the database, only players ids, attacker id, cards, last round date and competitive flag
+
+    CURSOR.execute(
+        '''
+        INSERT INTO duels (player_1_id, player_2_id, attacker_id, player_1_cards, player_2_cards, last_round_dt, is_finished, competitive) 
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        ''',
+        (
+            row['player_1_id'],
+            row['player_2_id'],
+            attacker_id,
+            ','.join(map(str, player_1_cards)),
+            ','.join(map(str, player_2_cards)),
+            last_round_dt,
+            False,
+            row['competitive']
+        )
+    )
+    duel_id = CURSOR.lastrowid
+    if duel_id:
+        # Start duel
+        start_duel(duel_id, row, player_1_cards, player_2_cards)
+        
+        
+
+    
+
+def start_duel(duel_id: int, row: dict, player_1_cards: list[int], player_2_cards: list[int]) -> None:
+    '''Starts a duel by sending a notification to the players and sending the cards.
+    Args:
+        duel_id (int): The ID of the duel.
+        row (dict): The row containing duel request details.
+        player_1_cards (list[int]): List of card IDs for player 1.
+        player_2_cards (list[int]): List of card IDs for player 2.
+    '''
+    # Initialize the duel thread by sending a notification to the players
+    from fediverse_factory import get_fediverse_service
+    from fediverse_types import Visibility
+
+    fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
+    
+
+    if row['competitive']==1:
+        duel_type = 'Competitive'
+    else:
+        duel_type = 'Casual'
+
+    # Get both the player 1 name and the player 1 server_id
+    player_1 = CURSOR.execute(
+        'SELECT username, server_id FROM players WHERE id = ?',
+        (row['player_1_id'],)
+    ).fetchone()
+    player_1_name = player_1[0]
+    player_1_server_id = player_1[1]
+
+    # Get both the player 2 name and the player 2 server_id
+    player_2_name = CURSOR.execute(
+        'SELECT username, server_id FROM players WHERE id = ?',
+        (row['player_2_id'],)
+    ).fetchone()
+    player_2_name = player_2_name[0]
+    player_2_server_id = player_2_name[1]
+
+    # Create a thread for the duel
+    thread_id = fediverse_service.create_post(
+    text=f'⚔️ A {duel_type} Duel started between {player_1_name} and {player_2_name}! ROUND 1 has begun!',
+    visibility=Visibility.HOME
+    )
+    # Update the duel with the thread ID
+    """CURSOR.execute(
+        'UPDATE duels SET thread_id = ? WHERE id = ?',
+        (thread_id, duel_id)
+    )"""
+    
+    print(player_1_cards)
+    # Send the cards to the players
+    # Player 1
+    # Get the card details for player 1
+    player_1_cards_details = get_cards(player_1_cards)
+    print(player_1_cards_details)
+    # Send a notification to player 1 with their cards
+    fediverse_service.create_post(
+        text=f'{player_1_name} Your cards for the duel. \
+        Reply with select <1,2 or 3> to select a card for the duel.',
+        visibility=Visibility.SPECIFIED,
+        file_ids=[card['file_id'] for card in player_1_cards_details],
+        visible_user_ids=[player_1_server_id]
+    )
+
+    # Player 2
+    # Get the card details for player 2
+    player_2_cards_details = get_cards(player_2_cards)
+    # Send a notification to player 2 with their cards
+    fediverse_service.create_post(
+        text=f'{player_2_name} Your cards for the duel. You are defending!. \
+        Select a card with <1,2 or 3> and a stat with <1,2 or 3> for Power, Charm and Wit. Example: select 2 3 would select the second card and use its Wit stat.',
+        visibility=Visibility.SPECIFIED,
+        file_ids=[card['file_id'] for card in player_2_cards_details],
+        visible_user_ids=[player_2_server_id]
+    )
+
+    
+
+
 
 def insert_player(username: str) -> int:
     '''Insert a new player with default has_rolled = False and return their

bot/response.py

@@ -181,6 +181,129 @@ def do_help(author: str) -> BotResponse:
         'attachment_urls': None
     }
 
+# Dueling
+
+def duel_request(author: str, args: list[str]) -> BotResponse:
+    '''Sends a duel request to another user.'''
+    if len(args) == 0:
+        return {
+            'message': f'{author} Please specify a user to duel.',
+            'attachment_urls': None
+        }
+
+    target = args[0]
+    if not db.get_player(target):
+        return {
+            'message': f'{author} User {target} does not exist or \
+            has not signed up, please sign up before challenging to a duel.',
+            'attachment_urls': None
+        }
+    if target == author:
+        return {
+            'message': f'{author} You can\'t duel yourself! \
+            Try challenging someone else.',
+            'attachment_urls': None
+        }
+    
+    # Check if both the users have enough cards to duel
+    # TODO
+
+    duel_type = 'casual'
+    if len(args) == 1 and args[0] == "competitive":
+        duel_type = 'competitive'
+
+    duel_id = db.insert_duel_request(author, target, duel_type)
+
+    return {
+        'message': f'{target} You have been challenged to a {duel_type} duel by {author}! \
+        Reply with `accept_duel {duel_id}` to accept the challenge. Duel ID:{duel_id}.',
+        'attachment_urls': None
+    }
+
+def accept_duel(author: str, args: list[str]) -> BotResponse:
+    '''Accepts a duel request with an id.'''
+    if len(args) == 0:
+        return {
+            'message': f'{author} Please specify a duel ID.',
+            'attachment_urls': None
+        }
+
+    duel_request_id = args[0]
+    duel_request = db.get_duel_request(duel_request_id)
+    if not duel_request:
+        return {
+            'message': f'{author} Duel request with ID {duel_request_id} does not exist.',
+            'attachment_urls': None
+        }
+    
+    # Check if the author is the target of the duel request
+    if duel_request['player_2_id'] != db.get_player(author):
+        return {
+            'message': f'{author} You cannot accept this duel request. It is not \
+            addressed to you.',
+            'attachment_urls': None
+        }
+    # Check if the duel was already accepted
+    if duel_request['accepted']:
+        return {
+            'message': f'{author} This duel request has already been accepted.',
+            'attachment_urls': None
+        }
+    # Accept the duel request
+    db.accept_duel_request(duel_request_id)
+    return {
+        'message': f'{author} You have accepted the duel request with ID {duel_request_id}. \
+        The duel will shortly begin.',
+        'attachment_urls': None
+    }
+
+def update_duel(author: str, args: list[str]) -> BotResponse:
+    '''Updates the duel with the selected card and stat.'''
+    if len(args) != 2:
+        return {
+            'message': f'{author} Please specify a card number and a stat number.',
+            'attachment_urls': None
+        }
+
+    if not all(is_float(arg) for arg in args):
+        return {
+            'message': f'{author} Invalid arguments: both card number and stat number must be numbers.',
+            'attachment_urls': None
+        }
+
+    card_number = int(args[0]) - 1  # Convert to zero-based index
+    stat_number = int(args[1]) - 1  # Convert to zero-based index
+
+    duel = db.get_duel_by_player(author)
+    if not duel:
+        return {
+            'message': f'{author} You are not currently in a duel.',
+            'attachment_urls': None
+        }
+
+    player_id = db.get_player(author)
+    player_cards = db.get_player_cards(player_id)
+
+    if card_number < 0 or card_number >= len(player_cards):
+        return {
+            'message': f'{author} Invalid card number: {card_number + 1}.',
+            'attachment_urls': None
+        }
+
+    selected_card = player_cards[card_number]
+    if stat_number < 0 or stat_number >= len(selected_card['stats']):
+        return {
+            'message': f'{author} Invalid stat number: {stat_number + 1}.',
+            'attachment_urls': None
+        }
+
+    db.update_duel_selection(duel['id'], player_id, selected_card['id'], stat_number)
+
+    return {
+        'message': f'{author} You have selected {selected_card["name"]} with \
+{selected_card["stats"][stat_number]} in {["Power", "Charm", "Wit"][stat_number]} for the duel.',
+        'attachment_urls': None]
+    }
 
 def delete_account(author: str) -> BotResponse:
     return {
@@ -325,6 +448,21 @@ def generate_response(notification: ParsedNotification) -> BotResponse | None:
             res = delete_account(author)
         case 'confirm_delete_account':
             res = confirm_delete(author)
+        case 'duel_request':
+            res = duel_request(
+                author,
+                notification['arguments']
+            )
+        case 'accept_duel':
+            res = accept_duel(
+                author,
+                notification['arguments']
+            )
+        case 'select':
+            res = update_duel(
+                author,
+                notification['arguments']
+            )
         case _:
             pass
 

migrations/0007_duelv1.sql

@@ -0,0 +1,45 @@
+/*
+Kemoverse - a gacha-style bot for the Fediverse.
+Copyright © 2025 Waifu
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as
+published by the Free Software Foundation, either version 3 of the
+License, or (at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see https://www.gnu.org/licenses/. 
+*/
+CREATE TABLE duels (
+    duel_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    player_1_id INTEGER NOT NULL,
+    player_2_id INTEGER NOT NULL,
+    attacker_id INTEGER,
+    player_1_cards TEXT NOT NULL DEFAULT '[]',
+    player_2_cards TEXT NOT NULL DEFAULT '[]',
+    graveyard_p1 TEXT NOT NULL DEFAULT '[]',
+    graveyard_p2 TEXT NOT NULL DEFAULT '[]',
+    round INTEGER NOT NULL DEFAULT 1,
+    competitive BOOLEAN NOT NULL DEFAULT 0,
+    last_round_dt TEXT,
+    is_finished BOOLEAN NOT NULL DEFAULT 0,
+    points INTEGER NOT NULL DEFAULT 0,
+    thread_id TEXT DEFAULT NULL,
+    winner_id INTEGER DEFAULT NULL
+);
+
+CREATE TABLE duel_requests (
+    duel_request_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    player_1_id INTEGER NOT NULL,
+    player_2_id INTEGER NOT NULL,
+    created_at TEXT NOT NULL DEFAULT (datetime('now')),
+    accepted BOOLEAN NOT NULL DEFAULT 0,
+    competitive BOOLEAN NOT NULL DEFAULT 0
+);
+
+ALTER TABLE players ADD COLUMN server_id text NOT NULL DEFAULT '0';