forked from waifu/kemoverse
Add a roll timeout to the bot
This commit is contained in:
parent
11c9cf58a3
commit
237f17f40d
5 changed files with 133 additions and 74 deletions
|
@ -32,7 +32,7 @@ def add_character(name: str, rarity: int, weight: float, image_url: str) -> tupl
|
||||||
raise ValueError("Image URL must be provided.")
|
raise ValueError("Image URL must be provided.")
|
||||||
|
|
||||||
# Download image
|
# Download image
|
||||||
response = requests.get(image_url, stream=True)
|
response = requests.get(image_url, stream=True, timeout=30)
|
||||||
if response.status_code != 200:
|
if response.status_code != 200:
|
||||||
raise RuntimeError(f"Failed to download image from {image_url}")
|
raise RuntimeError(f"Failed to download image from {image_url}")
|
||||||
|
|
||||||
|
@ -55,9 +55,6 @@ def add_character(name: str, rarity: int, weight: float, image_url: str) -> tupl
|
||||||
character_id = cur.lastrowid
|
character_id = cur.lastrowid
|
||||||
|
|
||||||
return character_id, file_id
|
return character_id, file_id
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
raise
|
|
||||||
finally:
|
finally:
|
||||||
if 'conn' in locals():
|
if 'conn' in locals():
|
||||||
conn.close()
|
conn.close()
|
|
@ -4,22 +4,20 @@ config = configparser.ConfigParser()
|
||||||
config.read('config.ini')
|
config.read('config.ini')
|
||||||
|
|
||||||
# Username for the bot
|
# Username for the bot
|
||||||
USER = config['application']['BotUser']
|
USER = config['credentials']['User']
|
||||||
|
|
||||||
# API key for the bot
|
# API key for the bot
|
||||||
KEY = config['application']['ApiKey']
|
KEY = config['credentials']['Token']
|
||||||
# Bot's Misskey instance URL
|
# Bot's Misskey instance URL
|
||||||
INSTANCE = config['application']['InstanceUrl']
|
INSTANCE = config['credentials']['Instance']
|
||||||
|
|
||||||
# SQLite Database location
|
|
||||||
DB_PATH = config['application']['DatabaseLocation']
|
|
||||||
|
|
||||||
# Extra stuff for control of the bot
|
|
||||||
|
|
||||||
# TODO: move this to db
|
# TODO: move this to db
|
||||||
# Fedi handles in the traditional 'user@domain.tld' style, allows these users
|
# Fedi handles in the traditional 'user@domain.tld' style, allows these users
|
||||||
# to use extra admin exclusive commands with the bot'''
|
# to use extra admin exclusive commands with the bot
|
||||||
ADMINS = config['application']['DefaultAdmins']
|
ADMINS = config['application']['DefaultAdmins']
|
||||||
|
# SQLite Database location
|
||||||
|
DB_PATH = config['application']['DatabaseLocation']
|
||||||
|
|
||||||
NOTIFICATION_POLL_INTERVAL = int(config['application']['NotificationPollInterval'])
|
NOTIFICATION_POLL_INTERVAL = int(config['notification']['PollInterval'])
|
||||||
NOTIFICATION_BATCH_SIZE = int(config['application']['NotificationBatchSize'])
|
NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize'])
|
||||||
|
|
||||||
|
GACHA_ROLL_INTERVAL = int(config['gacha']['RollInterval'])
|
||||||
|
|
|
@ -1,5 +1,4 @@
|
||||||
import sqlite3
|
import sqlite3
|
||||||
import random
|
|
||||||
import config
|
import config
|
||||||
|
|
||||||
DB_PATH = config.DB_PATH
|
DB_PATH = config.DB_PATH
|
||||||
|
@ -48,6 +47,15 @@ def add_pull(user_id, character_id):
|
||||||
conn.commit()
|
conn.commit()
|
||||||
conn.close()
|
conn.close()
|
||||||
|
|
||||||
|
def get_last_rolled_at(user_id):
|
||||||
|
'''Gets the timestamp when the user last rolled'''
|
||||||
|
conn = get_db_connection()
|
||||||
|
cur = conn.cursor()
|
||||||
|
cur.execute("SELECT timestamp FROM pulls WHERE user_id = ? ORDER BY timestamp DESC", \
|
||||||
|
(user_id,))
|
||||||
|
row = cur.fetchone()
|
||||||
|
conn.close()
|
||||||
|
return row[0] if row else None
|
||||||
|
|
||||||
# Configuration
|
# Configuration
|
||||||
|
|
||||||
|
|
112
bot/response.py
112
bot/response.py
|
@ -1,6 +1,8 @@
|
||||||
import random
|
import random
|
||||||
from db_utils import get_or_create_user, add_pull, get_db_connection
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from db_utils import get_or_create_user, add_pull, get_db_connection, get_last_rolled_at
|
||||||
from add_character import add_character
|
from add_character import add_character
|
||||||
|
from config import GACHA_ROLL_INTERVAL
|
||||||
|
|
||||||
def get_character():
|
def get_character():
|
||||||
''' Gets a random character from the database'''
|
''' Gets a random character from the database'''
|
||||||
|
@ -18,48 +20,71 @@ def get_character():
|
||||||
|
|
||||||
return chosen['id'], chosen['name'], chosen['file_id'], chosen['rarity']
|
return chosen['id'], chosen['name'], chosen['file_id'], chosen['rarity']
|
||||||
|
|
||||||
|
def do_roll(full_user):
|
||||||
|
'''Determines whether the user can roll, then pulls a random character'''
|
||||||
|
user_id = get_or_create_user(full_user)
|
||||||
|
|
||||||
|
# Get date of user's last roll
|
||||||
|
date = get_last_rolled_at(user_id)
|
||||||
|
|
||||||
|
# No date means it's users first roll
|
||||||
|
if date:
|
||||||
|
# SQLite timestamps returned by the DB are always in UTC
|
||||||
|
# Below timestamps are to be converted to UTC
|
||||||
|
prev = datetime.strptime(date + '+0000', '%Y-%m-%d %H:%M:%S%z')
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
|
||||||
|
time_since_last_roll = now - prev
|
||||||
|
roll_interval = timedelta(seconds=GACHA_ROLL_INTERVAL)
|
||||||
|
duration = roll_interval - time_since_last_roll
|
||||||
|
|
||||||
|
# User needs to wait before they can roll again
|
||||||
|
if time_since_last_roll < roll_interval:
|
||||||
|
remaining_duration = None
|
||||||
|
if duration.seconds > 3600:
|
||||||
|
remaining_duration = f'{-(duration.seconds // -3600)} hours'
|
||||||
|
elif duration.seconds > 60:
|
||||||
|
remaining_duration = f'{-(duration.seconds // -60)} minutes'
|
||||||
|
else:
|
||||||
|
remaining_duration = f'{duration.seconds} seconds'
|
||||||
|
|
||||||
|
return f'{full_user} ⏱️ Please wait another {remaining_duration} before rolling again.'
|
||||||
|
|
||||||
|
character_id, character_name, file_id, rarity = get_character()
|
||||||
|
|
||||||
|
if not character_id:
|
||||||
|
return f'{full_user} Uwaaa... something went wrong! No characters found. 😿'
|
||||||
|
|
||||||
|
add_pull(user_id,character_id)
|
||||||
|
stars = '⭐️' * rarity
|
||||||
|
return([f"@{full_user} 🎲 Congrats! You rolled {stars} **{character_name}**\n\
|
||||||
|
She's all yours now~ 💖✨",[file_id]])
|
||||||
|
|
||||||
def is_float(val):
|
def is_float(val):
|
||||||
|
'''Returns true if `val` can be converted to a float'''
|
||||||
try:
|
try:
|
||||||
float(val)
|
float(val)
|
||||||
return True
|
return True
|
||||||
except ValueError:
|
except ValueError:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
def do_create(full_user, arguments, note_obj):
|
||||||
def generate_response(parsed_command):
|
'''Creates a character'''
|
||||||
|
|
||||||
'''Given a command with arguments, processes the game state and
|
|
||||||
returns a response'''
|
|
||||||
|
|
||||||
command, full_user, arguments, note_obj = parsed_command
|
|
||||||
|
|
||||||
if command == "roll":
|
|
||||||
user_id = get_or_create_user(full_user)
|
|
||||||
character_id, character_name, file_id, rarity = get_character()
|
|
||||||
|
|
||||||
if not character_id:
|
|
||||||
#TODO: Can't have tuples of a single element
|
|
||||||
# Return these as a dict or object instead.
|
|
||||||
return(f"@{full_user} Uwaaa... something went wrong! No characters found. 😿")
|
|
||||||
|
|
||||||
add_pull(user_id,character_id)
|
|
||||||
stars = '⭐️' * rarity
|
|
||||||
return([f"@{full_user} 🎲 Congrats! You rolled {stars} **{character_name}**\nShe's all yours now~ 💖✨",[file_id]])
|
|
||||||
|
|
||||||
if command == "create":
|
|
||||||
# Example call from bot logic
|
# Example call from bot logic
|
||||||
image_url = note_obj.get("files", [{}])[0].get("url") if note_obj.get("files") else None
|
image_url = note_obj.get('files', [{}])[0].get('url') if note_obj.get('files') else None
|
||||||
if not image_url:
|
if not image_url:
|
||||||
return "You need an image to create a character, dumbass."
|
return f'{full_user}{full_user} You need an image to create a character, dumbass.'
|
||||||
|
|
||||||
if len(arguments) != 3:
|
if len(arguments) != 3:
|
||||||
return "Please specify the following attributes in order: name, rarity, drop weighting"
|
return '{full_user}Please specify the following attributes in order: \
|
||||||
|
name, rarity, drop weighting'
|
||||||
|
|
||||||
if not (arguments[1].isnumeric() and 1 <= int(arguments[1]) <= 5):
|
if not (arguments[1].isnumeric() and 1 <= int(arguments[1]) <= 5):
|
||||||
return f"Invalid rarity: '{arguments[1]}' must be a number between 1 and 5"
|
return f'{full_user}Invalid rarity: \'{arguments[1]}\' must be a number between 1 and 5'
|
||||||
|
|
||||||
if not (is_float(arguments[2]) and 0.0 < float(arguments[2]) <= 1.0):
|
if not (is_float(arguments[2]) and 0.0 < float(arguments[2]) <= 1.0):
|
||||||
return f"Invalid drop weight: '{arguments[2]}' must be a decimal value between 0.0 and 1.0"
|
return f'{full_user}Invalid drop weight: \'{arguments[2]}\' \
|
||||||
|
must be a decimal value between 0.0 and 1.0'
|
||||||
|
|
||||||
character_id, file_id = add_character(
|
character_id, file_id = add_character(
|
||||||
name=arguments[0],
|
name=arguments[0],
|
||||||
|
@ -67,5 +92,32 @@ def generate_response(parsed_command):
|
||||||
weight=float(arguments[2]),
|
weight=float(arguments[2]),
|
||||||
image_url=image_url
|
image_url=image_url
|
||||||
)
|
)
|
||||||
return([f"Added {arguments[0]}, ID {character_id}.",[file_id]])
|
return([f'{full_user}Added {arguments[0]}, ID {character_id}.',[file_id]])
|
||||||
return None
|
|
||||||
|
|
||||||
|
def do_help(full_user):
|
||||||
|
'''Provides a list of commands that the bot can do.'''
|
||||||
|
return f'{full_user} Here\'s what I can do:\n \
|
||||||
|
- `roll` Pulls a random character.\
|
||||||
|
- `create <name> <rarity> <weight>` Creates a character using a given image.\
|
||||||
|
- `help` Shows this message'
|
||||||
|
|
||||||
|
def do_invalid_command(command, full_user):
|
||||||
|
'''Generic response when an unknown or invalid command is sent'''
|
||||||
|
return f'{full_user} Unrecognised command: {command}\n\
|
||||||
|
Message \'help\' to get a list of valid commands'
|
||||||
|
|
||||||
|
def generate_response(parsed_command):
|
||||||
|
'''Given a command with arguments, processes the game state and
|
||||||
|
returns a response'''
|
||||||
|
|
||||||
|
command, full_user, arguments, note_obj = parsed_command
|
||||||
|
match command:
|
||||||
|
case 'roll':
|
||||||
|
return do_roll(full_user)
|
||||||
|
case 'create':
|
||||||
|
return do_create(full_user, arguments, note_obj)
|
||||||
|
case 'help':
|
||||||
|
return do_help(command)
|
||||||
|
case _:
|
||||||
|
return do_invalid_command(command, full_user)
|
||||||
|
|
|
@ -1,23 +1,27 @@
|
||||||
; Rename me to config.ini and put your values in here
|
; Rename me to config.ini and put your values in here
|
||||||
[application]
|
[application]
|
||||||
; Full fedi handle of the bot user
|
|
||||||
BotUser = @bot@example.tld
|
|
||||||
|
|
||||||
; API key for the bot
|
|
||||||
; Generate one by going to Settings > API > Generate access token
|
|
||||||
ApiKey = abcdefghijklmnopqrstuvwxyz012345
|
|
||||||
|
|
||||||
; Fully qualified URL of the instance hosting the bot
|
|
||||||
InstanceUrl = http://example.tld
|
|
||||||
|
|
||||||
; Comma separated list of fedi handles for any administrator users
|
; Comma separated list of fedi handles for any administrator users
|
||||||
|
; More can be added through the application
|
||||||
DefaultAdmins = ['admin@example.tld']
|
DefaultAdmins = ['admin@example.tld']
|
||||||
|
|
||||||
; SQLite Database location
|
; SQLite Database location
|
||||||
DatabaseLocation = ./gacha_game.db
|
DatabaseLocation = ./gacha_game.db
|
||||||
|
|
||||||
; Number of seconds to sleep while awaiting new notifications
|
[gacha]
|
||||||
NotificationPollInterval = 5
|
; Number of seconds players have to wait between rolls
|
||||||
|
RollInterval = 72000
|
||||||
|
|
||||||
|
[notification]
|
||||||
|
; Number of seconds to sleep while awaiting new notifications
|
||||||
|
PollInterval = 5
|
||||||
|
; Number of notifications to process at once (max 100)
|
||||||
|
BatchSize = 10
|
||||||
|
|
||||||
|
[credentials]
|
||||||
|
; Fully qualified URL of the instance hosting the bot
|
||||||
|
Instance = http://example.tld
|
||||||
|
; Full fedi handle of the bot user
|
||||||
|
User = @bot@example.tld
|
||||||
|
; API key for the bot
|
||||||
|
; Generate one by going to Settings > API > Generate access token
|
||||||
|
Token = abcdefghijklmnopqrstuvwxyz012345
|
||||||
|
|
||||||
; Number of notifications to process at once (limit 100)
|
|
||||||
NotificationBatchSize = 10
|
|
||||||
|
|
Loading…
Add table
Reference in a new issue