Compare commits

..

11 commits

28 changed files with 463 additions and 964 deletions

8
.gitignore vendored
View file

@ -147,7 +147,6 @@ venv.bak/
/site /site
# mypy # mypy
.mypy.ini
.mypy_cache/ .mypy_cache/
.dmypy.json .dmypy.json
dmypy.json dmypy.json
@ -182,8 +181,5 @@ cython_debug/
.cursorindexingignore .cursorindexingignore
# Custom stuff # Custom stuff
gacha_game*.db *.db
gacha_game*.db.* *.ini
config*.ini
.idea

View file

@ -1,7 +1,6 @@
import requests import requests
from fediverse_factory import get_fediverse_service from fediverse_factory import get_fediverse_service
from db_utils import get_db_connection from db_utils import get_db_connection
from config import RARITY_TO_WEIGHT
def add_character(name: str, rarity: int, weight: float, image_url: str) -> tuple[int, str]: def add_character(name: str, rarity: int, weight: float, image_url: str) -> tuple[int, str]:
""" """
@ -20,22 +19,19 @@ def add_character(name: str, rarity: int, weight: float, image_url: str) -> tupl
ValueError: If inputs are invalid. ValueError: If inputs are invalid.
RuntimeError: If image download/upload or database operation fails. RuntimeError: If image download/upload or database operation fails.
""" """
stripped_name = name.strip()
# Validate inputs
if not stripped_name:
raise ValueError('Character name cannot be empty.')
if rarity < 1:
raise ValueError('Rarity must be a positive integer.')
if rarity not in RARITY_TO_WEIGHT.keys():
raise ValueError(f'Invalid rarity: {rarity}')
if not image_url:
raise ValueError('Image URL must be provided.')
try: try:
# Validate inputs
if not name or not name.strip():
raise ValueError("Character name cannot be empty.")
if not isinstance(rarity, int) or rarity < 1:
raise ValueError("Rarity must be a positive integer.")
if not isinstance(weight, (int, float)) or weight <= 0:
raise ValueError("Weight must be a positive number.")
if not image_url:
raise ValueError("Image URL must be provided.")
# Download image # Download image
response = requests.get(image_url, stream=True, timeout=30) response = requests.get(image_url, stream=True)
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}")
@ -52,7 +48,7 @@ def add_character(name: str, rarity: int, weight: float, image_url: str) -> tupl
cur = conn.cursor() cur = conn.cursor()
cur.execute( cur.execute(
'INSERT INTO characters (name, rarity, weight, file_id) VALUES (?, ?, ?, ?)', 'INSERT INTO characters (name, rarity, weight, file_id) VALUES (?, ?, ?, ?)',
(stripped_name, rarity, float(weight), file_id) (name.strip(), rarity, float(weight), file_id)
) )
conn.commit() conn.commit()
character_id = cur.lastrowid character_id = cur.lastrowid

View file

@ -1,19 +1,21 @@
import time import time
import traceback import traceback
from parsing import parse_notification from parsing import parse_notification
from db_utils import get_config, set_config from db_utils import get_or_create_user, add_pull, get_config, set_config
from fediverse_factory import get_fediverse_service from fediverse_factory import get_fediverse_service
import config import config
def stream_notifications():
# Initialize the Fediverse service # Initialize the Fediverse service
fediverse_service = get_fediverse_service() fediverse_service = get_fediverse_service()
# Get the last seen notification ID from the database # Get trusted instances from config
last_seen_id = get_config("last_seen_notif_id") whitelisted_instances = config.TRUSTED_INSTANCES
whitelisted_instances = getattr(config, 'TRUSTED_INSTANCES', [])
def stream_notifications():
print("Starting filtered notification stream...")
last_seen_id = get_config("last_seen_notif_id")
print('Listening for notifications...')
while True: while True:
try: try:
# Get notifications from the fediverse service # Get notifications from the fediverse service
@ -63,5 +65,8 @@ def stream_notifications():
print(f"An exception has occured: {e}\n{traceback.format_exc()}") print(f"An exception has occured: {e}\n{traceback.format_exc()}")
time.sleep(5) time.sleep(5)
if __name__ == "__main__": if __name__ == "__main__":
stream_notifications() stream_notifications()

5
bot/client.py Normal file
View file

@ -0,0 +1,5 @@
import misskey
import config
def client_connection():
return misskey.Misskey(address=config.INSTANCE, i=config.KEY)

View file

@ -1,45 +1,18 @@
'''Essentials for the bot to function''' '''Essentials for the bot to function'''
import configparser import configparser
from os import environ, path
class ConfigError(Exception):
'''Could not find config file'''
def get_config_file() -> str:
'''Gets the path to the config file in the current environment'''
env: str | None = environ.get('KEMOVERSE_ENV')
if not env:
raise ConfigError('Error: KEMOVERSE_ENV is unset')
if not (env in ['prod', 'dev']):
raise ConfigError(f'Error: Invalid environment: {env}')
config_path: str = f'config_{env}.ini'
if not path.isfile(config_path):
raise ConfigError(f'Could not find {config_path}')
return config_path
def get_rarity_to_weight(config_section):
"""Parses Rarity_X keys from config and returns a {rarity: weight} dict."""
rarity_weights = {}
for key, value in config_section.items():
if key.startswith("rarity_"):
rarity = int(key.removeprefix("rarity_"))
rarity_weights[rarity] = float(value)
return rarity_weights
config = configparser.ConfigParser() config = configparser.ConfigParser()
config.read(get_config_file()) config.read('config.ini')
# Username for the bot # Username for the bot
USER = config['credentials']['User'].lower() USER = config['application']['BotUser']
# API key for the bot # API key for the bot
KEY = config['credentials']['Token'] KEY = config['application']['ApiKey']
# Bot's Misskey/Pleroma instance URL # Bot's Misskey instance URL
INSTANCE = config['credentials']['Instance'].lower() INSTANCE = config['application']['InstanceUrl']
# SQLite Database location
DB_PATH = config['application']['DatabaseLocation']
# Instance type validation # Instance type validation
if 'InstanceType' not in config['application']: if 'InstanceType' not in config['application']:
@ -62,14 +35,5 @@ TRUSTED_INSTANCES = [instance.strip() for instance in trusted_instances_str.spli
# 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['notification']['PollInterval'])
NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize'])
GACHA_ROLL_INTERVAL = int(config['gacha']['RollInterval'])
RARITY_TO_WEIGHT = get_rarity_to_weight(config['gacha'])

View file

@ -1,21 +0,0 @@
from typing import TypedDict, List, Dict, Any
BotResponse = TypedDict('BotResponse', {
'message': str,
'attachment_urls': List[str] | None
})
Character = TypedDict('Character', {
'id': int,
'name': str,
'rarity': int,
'weight': float,
'image_url': str
})
ParsedNotification = TypedDict('ParsedNotification', {
'author': str,
'command': str | None,
'arguments': List[str],
'note_obj': Dict[str, Any]
})

View file

@ -1,121 +1,58 @@
from random import choices
import sqlite3 import sqlite3
import random
import config import config
from custom_types import Character
DB_PATH = config.DB_PATH DB_PATH = config.DB_PATH
CONNECTION: sqlite3.Connection
CURSOR: sqlite3.Cursor
def get_db_connection():
def connect() -> None:
'''Creates a connection to the database''' '''Creates a connection to the database'''
print('Connecting to the database...') conn = sqlite3.connect(DB_PATH)
global CONNECTION conn.row_factory = sqlite3.Row
global CURSOR return conn
CONNECTION = sqlite3.connect(DB_PATH, autocommit=True)
CONNECTION.row_factory = sqlite3.Row
CURSOR = CONNECTION.cursor()
def get_or_create_user(username):
def get_random_character() -> Character | None: '''Retrieves an ID for a given user, if the user does not exist, it will be
''' Gets a random character from the database''' created.'''
CURSOR.execute('SELECT * FROM characters') conn = get_db_connection()
characters = CURSOR.fetchall() conn.row_factory = sqlite3.Row
cur = conn.cursor()
if not characters: cur.execute('SELECT id FROM users WHERE username = ?', (username,))
return None user = cur.fetchone()
weights = [config.RARITY_TO_WEIGHT[c['rarity']] for c in characters]
chosen = choices(characters, weights=weights, k=1)[0]
return {
'id': chosen['id'],
'name': chosen['name'],
'rarity': chosen['rarity'],
'weight': config.RARITY_TO_WEIGHT[chosen['rarity']],
'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: if user:
return int(user[0]) conn.close()
return user[0]
def insert_player(username: str) -> int: # New user starts with has_rolled = False
'''Insert a new player with default has_rolled = False and return their user ID.''' cur.execute(
CURSOR.execute(
'INSERT INTO users (username, has_rolled) VALUES (?, ?)', 'INSERT INTO users (username, has_rolled) VALUES (?, ?)',
(username, False) (username, False)
) )
return CURSOR.lastrowid conn.commit()
user_id = cur.lastrowid
conn.close()
return user_id
def delete_player(username: str) -> bool: def add_pull(user_id, character_id):
'''Permanently deletes a user and all their pulls.'''
CURSOR.execute(
'SELECT id FROM users WHERE username = ?',
(username,)
)
user = CURSOR.fetchone()
user_id = user[0]
# Delete pulls
CURSOR.execute(
'DELETE FROM pulls WHERE user_id = ?',
(user_id,)
)
# Delete user
CURSOR.execute(
'DELETE FROM users WHERE id = ?',
(user_id,)
)
return True
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 \
(?, ?, ?, ?)',
(name, rarity, weight, file_id)
)
character_id = CURSOR.lastrowid
return character_id if character_id else 0
def insert_pull(user_id: int, character_id: int) -> None:
'''Creates a pull in the database''' '''Creates a pull in the database'''
CURSOR.execute( conn = get_db_connection()
'INSERT INTO pulls (user_id, character_id) VALUES (?, ?)', cur = conn.cursor()
(user_id, character_id) cur.execute('INSERT INTO pulls (user_id, character_id) VALUES (?, ?)', (user_id, character_id))
) conn.commit()
conn.close()
def get_config(key):
def get_last_rolled_at(user_id: int) -> int:
'''Gets the timestamp when the user last rolled'''
CURSOR.execute(
"SELECT timestamp FROM pulls WHERE user_id = ? ORDER BY timestamp \
DESC",
(user_id,))
row = CURSOR.fetchone()
return row[0] if row else 0
def get_config(key: str) -> str:
'''Reads the value for a specified config key from the db''' '''Reads the value for a specified config key from the db'''
CURSOR.execute("SELECT value FROM config WHERE key = ?", (key,)) conn = get_db_connection()
row = CURSOR.fetchone() cur = conn.cursor()
return row[0] if row else '' cur.execute("SELECT value FROM config WHERE key = ?", (key,))
row = cur.fetchone()
conn.close()
return row[0] if row else None
def set_config(key, value):
def set_config(key: str, value: str) -> None:
'''Writes the value for a specified config key to the db''' '''Writes the value for a specified config key to the db'''
CURSOR.execute("INSERT OR REPLACE INTO config (key, value) VALUES (?, ?)", conn = get_db_connection()
(key, value)) cur = conn.cursor()
cur.execute("INSERT OR REPLACE INTO config (key, value) VALUES (?, ?)", (key, value))
conn.commit()
conn.close()

70
bot/gacha_response.py Normal file
View file

@ -0,0 +1,70 @@
import random
from db_utils import get_or_create_user, add_pull, get_db_connection
from add_character import add_character
from fediverse_types import FediversePost
def get_character():
''' Gets a random character from the database'''
conn = get_db_connection()
cur = conn.cursor()
cur.execute('SELECT * FROM characters')
characters = cur.fetchall()
conn.close()
if not characters:
return None, None, None, None
weights = [c['weight'] for c in characters]
chosen = random.choices(characters, weights=weights, k=1)[0]
return chosen['id'], chosen['name'], chosen['file_id'], chosen['rarity']
def is_float(val):
try:
float(val)
return True
except ValueError:
return False
# TODO: See issue #3, separate command parsing from game logic.
def gacha_response(command: str, full_user: str, arguments: list, post: FediversePost):
'''Parses a given command with arguments, processes the game state and
returns a response'''
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
image_url = post.files[0].url if post.files else None
if not image_url:
return "You need an image to create a character, dumbass."
if len(arguments) != 3:
return "Please specify the following attributes in order: name, rarity, drop weighting"
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"
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"
character_id, file_id = add_character(
name=arguments[0],
rarity=int(arguments[1]),
weight=float(arguments[2]),
image_url=image_url
)
return([f"Added {arguments[0]}, ID {character_id}.",[file_id]])
return None

View file

@ -1,13 +1,12 @@
import re import random, re
import config import config
from response import generate_response from gacha_response import gacha_response
from fediverse_factory import get_fediverse_service from fediverse_factory import get_fediverse_service
from fediverse_types import FediverseNotification, NotificationType, Visibility from fediverse_types import FediverseNotification, NotificationType, Visibility
from custom_types import ParsedNotification
def parse_notification(notification: FediverseNotification, fediverse_service=None): def parse_notification(notification: FediverseNotification, fediverse_service=None):
'''Parses any notifications received by the bot and sends any commands to '''Parses any notifications received by the bot and sends any commands to
generate_response()''' gacha_response()'''
if fediverse_service is None: if fediverse_service is None:
fediverse_service = get_fediverse_service() fediverse_service = get_fediverse_service()
@ -44,41 +43,32 @@ def parse_notification(notification: FediverseNotification, fediverse_service=No
# Make sure the notification text explicitly mentions the bot # Make sure the notification text explicitly mentions the bot
if not any(variant in note for variant in username_variants): if not any(variant in note for variant in username_variants):
return None return
# Find command and arguments after the mention # Find command and arguments after the mention
# Removes all mentions # Removes all mentions (regex = mentions that start with @ and may contain @domain)
# regex = mentions that start with @ and may contain @domain
cleaned_text = re.sub(r"@\w+(?:@\S+)?", "", note).strip() cleaned_text = re.sub(r"@\w+(?:@\S+)?", "", note).strip()
parts = cleaned_text.split() parts = cleaned_text.split()
command = parts[0].lower() if parts else None command = parts[0].lower() if parts else None
arguments = parts[1:] if len(parts) > 1 else [] arguments = parts[1:] if len(parts) > 1 else []
# Create ParsedNotification object for the new response system # TODO: move response generation to a different function
parsed_notification: ParsedNotification = { response = gacha_response(command.lower(), full_user, arguments, notification.post)
'author': full_user,
'command': command,
'arguments': arguments,
'note_obj': {
'id': note_id,
'text': note_text,
'files': [{'url': f.url} for f in notification.post.files] if notification.post.files else []
}
}
# Generate response using the new system
response = generate_response(parsed_notification)
if not response: if not response:
return return
# Handle attachment URLs (convert to file IDs if needed) if isinstance(response, str):
file_ids = response['attachment_urls'] if response['attachment_urls'] else None
fediverse_service.create_post( fediverse_service.create_post(
text=response['message'], text=response,
reply_to_id=note_id,
visibility=visibility
)
else:
fediverse_service.create_post(
text=response[0],
reply_to_id=note_id, reply_to_id=note_id,
visibility=visibility, visibility=visibility,
file_ids=file_ids file_ids=response[1]
#visible_user_ids=[] #todo: write actual visible users ids so pleromers can use the bot privately #visible_user_ids=[] #todo: write actual visible users ids so pleromers can use the bot privately
) )

View file

@ -1,215 +0,0 @@
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, \
get_random_character
from add_character import add_character
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'''
user_id = get_player(author)
if not user_id:
return {
'message':f'{author} 🛑 You havent signed up yet! Use the `signup` command to start playing.',
'attachment_urls': None
}
# 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(str(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 {
'message': f'{author} ⏱️ Please wait another \
{remaining_duration} before rolling again.',
'attachment_urls': None
}
character = get_random_character()
if not character:
return {
'message': f'{author} Uwaaa... something went wrong! No \
characters found. 😿',
'attachment_urls': None
}
insert_pull(user_id, character['id'])
stars = '⭐️' * character['rarity']
return {
'message': f'{author} 🎲 Congrats! You rolled {stars} \
**{character['name']}**\nShe\'s all yours now~ 💖✨',
'attachment_urls': [character['image_url']]
}
def do_signup(author: str) -> BotResponse:
'''Registers a new user if they havent signed up yet.'''
user_id = get_player(author)
if user_id:
return {
'message':f'{author} 👀 Youre 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!',
'attachment_urls': None
}
def is_float(val: Any) -> bool:
'''Returns true if `val` can be converted to a float'''
try:
float(val)
return True
except ValueError:
return False
def do_create(
author: str,
arguments: List[str],
note_obj: Dict[str, Any]) -> BotResponse:
'''Creates a character'''
# 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, \
dumbass.',
'attachment_urls': None
}
if len(arguments) != 3:
return {
'message': f'{author} Please specify the following attributes \
in order: name, rarity, weight',
'attachment_urls': None
}
if not (arguments[1].isnumeric() and 1 <= int(arguments[1]) <= 5):
return {
'message': f'{author} Invalid rarity: \'{arguments[1]}\' must \
be a number between 1 and 5',
'attachment_urls': None
}
if not (is_float(arguments[2]) and 0.0 < float(arguments[2]) <= 1.0):
return {
'message': f'{author} Invalid drop weight: \'{arguments[2]}\' \
must be a decimal value between 0.0 and 1.0',
'attachment_urls': None
}
character_id, file_id = add_character(
name=arguments[0],
rarity=int(arguments[1]),
weight=float(arguments[2]),
image_url=image_url
)
return {
'message': f'{author} Added {arguments[0]}, ID {character_id}.',
'attachment_urls': [file_id]
}
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 \
- `roll` Pulls a random character.\
- `create <name> <rarity>` Creates a character using a given image.\
- `signup` Registers your account.\
- `delete_account` Deletes your account.\
- `help` Shows this message',
'attachment_urls': None
}
def delete_account(author: str) -> BotResponse:
return {
'message':f'{author} ⚠️ This will permanently delete your account and all your cards.\n'
'If youre sure, reply with `confirm_delete` 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 🕊️✨',
'attachment_urls': None
}
def generate_response(notification: ParsedNotification) -> BotResponse | None:
'''Given a command with arguments, processes the game state and
returns a response'''
# Temporary response variable
res: BotResponse | None = None
# TODO: Check if the user has an account
author = notification['author']
user_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 'signup':
res = do_signup(author)
case 'help':
res = do_help(author)
case 'roll':
res = do_roll(author)
case _:
pass
if not user_id:
return res
# User commands
match command:
case 'create':
res = do_create(
author,
notification['arguments'],
notification['note_obj']
)
case 'signup':
res = do_signup(author)
case 'delete_account':
res = delete_account(author)
case 'confirm_delete':
res = confirm_delete(author)
case _:
pass
# if not user_is_administrator:
return res
# Administrator commands go here

67
db.py Normal file
View file

@ -0,0 +1,67 @@
import sqlite3
import configparser
# Read config to get database location
config = configparser.ConfigParser()
config.read('config.ini')
# Connect to SQLite database (or create it if it doesn't exist)
db_path = config['application']['DatabaseLocation']
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create tables
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
has_rolled BOOLEAN NOT NULL DEFAULT 0
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS characters (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
rarity INTEGER NOT NULL,
weight REAL NOT NULL,
file_id TEXT NOT NULL
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS pulls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
character_id INTEGER,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id),
FOREIGN KEY (character_id) REFERENCES characters(id)
)
''')
cursor.execute("""
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT
)
""")
""" # Insert example characters into the database if they don't already exist
characters = [
('Murakami-san', 1, 0.35),
('Mastodon-kun', 2, 0.25),
('Pleroma-tan', 3, 0.2),
('Misskey-tan', 4, 0.15),
('Syuilo-mama', 5, 0.05)
]
cursor.executemany('''
INSERT OR IGNORE INTO characters (name, rarity, weight) VALUES (?, ?, ?)
''', characters)
"""
# Commit changes and close
conn.commit()
conn.close()

View file

@ -1,24 +0,0 @@
# 🎲 Kemoverse Documentation
Welcome to the developer documentation for **Kemoverse**, a gacha trading card game in the Fediverse!
Features collectible cards, rarity-based pulls, and integration with Misskey.
Name comes from Kemonomimi and Fediverse.
---
## 📁 Table of Contents
- [Installation](./install.md)
- [Game Design](./design.md)
- [Bot Architecture](./bot.md)
- [Database Structure](./database.md)
- [Card System](./cards.md)
- [Web UI](./web.md)
- [Theming and Assets](./theme.md)
- [Contributing](./contributing.md)
- [FAQ](./faq.md)
---

View file

@ -1,83 +0,0 @@
## 🧪 Installation
### Download and install dependencies
Clone the repo
```sh
git clone https://git.waifuism.life/waifu/kemoverse.git
cd kemoverse
```
Setup a virtual environment (Optional, recommended)
```sh
python3 -m venv venv
source venv/bin/activate
```
Install project dependencies via pip
```sh
python3 -m pip install -r requirements.txt
```
### Setup config file
A sample config file is included with the project as a template: `example_config.ini`
Create a copy of this file and replace its' values with your own. Consult the
template for more information about individual config values and their meaning.
Config files are environment-specific. Use `config_dev.ini` for development and
`config_prod.ini` for production. Switch between environments using the
`KEMOVERSE_ENV` environment variable.
```sh
cp example_config.ini config_dev.ini
# Edit config_dev.ini
```
### Setup database
To set up the database, run:
```sh
KEMOVERSE_ENV=dev python3 setup_db.py
```
### Run the bot
```sh
KEMOVERSE_ENV=dev ./startup.sh
```
If all goes well, you should now be able to interact with the bot.
### Running in production
To run the the in a production environment, use `KEMOVERSE_ENV=prod`. You will
also need to create a `config_prod.ini` file and run the database setup step
again if pointing prod to a different database. (you are pointing dev and prod
to different databases, right? 🤨)
### Updating
To update the bot, first pull new changes from upstream:
```sh
git pull
```
Then run any database migrations. We recommend testing in dev beforehand to
make sure nothing breaks in the update process.
**Always backup your prod database before running any migrations!**
```sh
# Backup database file
cp gacha_game_dev.db gacha_game_dev.db.bak
# Run migrations
KEMOVERSE_ENV=dev python3 setup_db.py
```

View file

@ -1,33 +0,0 @@
Welcome to the **Visual Identity** guide for the Kemoverse. This page contains the standard colors, logos, and graphic elements used across the game (cards, UI, web presence, bots, etc). Please follow these guidelines to ensure consistency.
---
## 🟢 Primary Color Palette
| Color Name | Hex Code | Usage |
|----------------|------------|--------------------------------------|
| Green | `#5aa02c` | Main buttons, links, headers |
| Midnight Black | `#1A1A1A` | Backgrounds, dark mode |
| Misty White | `#FAFAFA` | Default backgrounds, light text bg |
| Soft Gray | `#CCCCCC` | Borders, placeholders, separators |
| Highlight Green | `#8dd35f` | Alerts, emphasis, icons |
| Rarity Gold | `#FFD700` | Special rare cards, SSR outlines |
| Rarity Silver | `#C0C0C0` | Rare card text, stat glow effects |
> ✅ Use `Green` and `Misty White` for the standard UI. Avoid mixing in extra palettes unless explicitly needed.
---
## 🖼 Logos
### Main Logo
<p align="center">
<img src="../web/static/logo.png" alt="Fediverse Gacha Bot Logo" width="300" height="auto">
</p>
- File: `web/static/logo.png`
- Usage: Website header, favicon, bot avatar, watermark
---

View file

@ -1,41 +1,24 @@
; Rename me to config.ini and put your values in here ; Rename me to config.ini and put your values in here
[application] [application]
; Comma separated list of fedi handles for any administrator users
; More can be added through the application
DefaultAdmins = ['admin@example.tld']
; SQLite Database location
DatabaseLocation = ./gacha_game.db
[gacha]
; Number of seconds players have to wait between rolls
RollInterval = 72000
; Rarity drop weights (1 to 5 stars)
; Format: rarity=weight per line
; In order: common, uncommon, rare, epic and legendary (Example values below)
Rarity_1 = 0.7
Rarity_2 = 0.2
Rarity_3 = 0.08
Rarity_4 = 0.015
Rarity_5 = 0.005
[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 ; Full fedi handle of the bot user
User = @bot@example.tld BotUser = @bot@example.tld
; API key for the bot ; API key for the bot
; Generate one by going to Settings > API > Generate access token ; Generate one by going to Settings > API > Generate access token
Token = abcdefghijklmnopqrstuvwxyz012345 ApiKey = abcdefghijklmnopqrstuvwxyz012345
; Fully qualified URL of the instance hosting the bot
InstanceUrl = http://example.tld
; Instance type - either "misskey" or "pleroma" ; Instance type - either "misskey" or "pleroma"
InstanceType = misskey InstanceType = misskey
; Comma separated list of fedi handles for any administrator users
DefaultAdmins = ['admin@example.tld']
; SQLite Database location
DatabaseLocation = ./gacha_game.db
; Web server port (default: 5000) ; Web server port (default: 5000)
WebPort = 5000 WebPort = 5000

View file

@ -1,28 +0,0 @@
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
has_rolled BOOLEAN NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS characters (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
rarity INTEGER NOT NULL,
weight REAL NOT NULL,
file_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS pulls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
character_id INTEGER,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id),
FOREIGN KEY (character_id) REFERENCES characters(id)
);
CREATE TABLE IF NOT EXISTS config (
key TEXT PRIMARY KEY,
value TEXT
);
INSERT OR IGNORE INTO config VALUES ("schema_version", 0);

View file

@ -1 +0,0 @@
INSERT OR IGNORE INTO config VALUES ("last_seen_notif_id", 0);

View file

@ -1 +0,0 @@
ALTER TABLE characters DROP COLUMN weight;

View file

@ -1,46 +1,34 @@
# Kemoverse # Kemoverse
A gacha-style bot for the Fediverse built with Python. Users can roll for characters, trade, duel, and perhaps engage with popularity-based mechanics. Supports both Misskey and Pleroma instances. Name comes from Kemonomimi and Fediverse. A gacha-style bot for the Fediverse built with Python. Users can roll for characters, trade, duel, and perhaps engage with popularity-based mechanics. Supports both Misskey and Pleroma instances. Name comes from Kemonomimi and Fediverse.
<p align="center">
<img src="./web/static/logo.png" alt="Fediverse Gacha Bot Logo" width="300" height="auto">
</p>
## 📝 Docs ![Fediverse Gacha Bot Logo](./web/static/logo.png)
👉 [**Start reading the docs**](./docs/index.md)
🤌 [**Install instructions for those in a rush**](docs/install.md)
## 🔧 Features ## 🔧 Features
### ✅ Implemented ### ✅ Implemented
- 🎲 Character roll system - 🎲 Character roll system
- 🧠 Core database structure for cards - 🎴 Cards stats system
- 📦 Basic support for storing pulls per player - 🧠 Core database structure for characters and stats
- ⏱️ Time-based limitations on rolls - 📦 Basic support for storing pulls per user
- ⚠️ Explicit account creation/deletion
### 🧩 In Progress ### 🧩 In Progress
- 📝 Whitelist system to limit access - 📝 Whitelist system to limit access
- ⏱️ Time-based limitations on rolls
- ⚔️ Dueling system
## 🧠 Planned Features (Long Term)
## 🧠 Roadmap
[See our v2.0 board for more details](https://git.waifuism.life/waifu/kemoverse/projects/3)
### 🛒 Gameplay & Collection ### 🛒 Gameplay & Collection
- 🔁 **Trading system** between players - 🔁 **Trading system** between users
- ⭐ **Favorite characters** (pin them or set profiles) - ⭐ **Favorite characters** (pin them or set profiles)
- 📢 **Public post announcements** for rare card pulls - 📢 **Public post announcements** for rare card pulls
- 📊 **Stats** for cards
- 🎮 **Games** to play
- ⚔️ Dueling
- 🧮 **Leaderboards** - 🧮 **Leaderboards**
- Most traded cards - Most traded Characters
- Most owned cards - Most owned Characters
- Most voted cards - Most voted Characters
- Most popular cards (via usage-based popularity metrics) - Most popular Characters (via usage-based popularity metrics)
- Users with the rarest cards - Users with the rarest Characters
### 🎨 Card Aesthetics ### 🎨 Card Aesthetics
- 🖼️ Simple card template for character rendering - 🖼️ Simple card template for character rendering
@ -51,7 +39,7 @@ A gacha-style bot for the Fediverse built with Python. Users can roll for charac
## 🗃️ Tech Stack ## 🗃️ Tech Stack
- Python (3.12+) - Python (3.11+)
- SQLite - SQLite
- Fediverse API integration (Misskey and Pleroma support) - Fediverse API integration (Misskey and Pleroma support)
- Flask - Flask
@ -61,6 +49,11 @@ A gacha-style bot for the Fediverse built with Python. Users can roll for charac
The bot is meant to feel *light, fun, and competitive*. Mixing social, gacha and duel tactics. The bot is meant to feel *light, fun, and competitive*. Mixing social, gacha and duel tactics.
## 🧪 Getting Started (coming soon)
Instructions on installing dependencies, initializing the database, and running the bot locally will go here.
```mermaid ```mermaid
flowchart TD flowchart TD

View file

@ -1,129 +0,0 @@
import sqlite3
import traceback
import os
import argparse
from configparser import ConfigParser
from typing import List, Tuple
class DBNotFoundError(Exception):
'''Could not find the database location'''
class InvalidMigrationError(Exception):
'''Migration file has an invalid name'''
class KemoverseEnvUnset(Exception):
'''KEMOVERSE_ENV is not set or has an invalid value'''
class ConfigError(Exception):
'''Could not find the config file for the current environment'''
def get_migrations() -> List[Tuple[int, str]] | InvalidMigrationError:
'''Returns a list of migration files in numeric order.'''
# Store transaction id and filename separately
sql_files: List[Tuple[int, str]] = []
migrations_dir = 'migrations'
for filename in os.listdir(migrations_dir):
joined_path = os.path.join(migrations_dir, filename)
# Ignore anything that isn't a .sql file
if not (os.path.isfile(joined_path) and filename.endswith('.sql')):
print(f'{filename} is not a .sql file, ignoring...')
continue
parts = filename.split('_', 1)
# Invalid filename format
if len(parts) < 2 or not parts[0].isdigit():
raise InvalidMigrationError(f'Invalid migration file: {filename}')
sql_files.append((int(parts[0]), joined_path))
# Get sorted list of files by migration number
sql_files.sort(key=lambda x: x[0])
return sql_files
def perform_migration(cursor: sqlite3.Cursor, migration: tuple[int, str]) -> None:
'''Performs a migration on the DB'''
print(f'Performing migration {migration[1]}...')
# Open and execute the sql script
with open(migration[1], encoding='utf-8') as file:
script = file.read()
cursor.executescript(script)
# Update the schema version
cursor.execute('UPDATE config SET value = ? WHERE key = "schema_version"', (migration[0],))
def get_db_path() -> str | DBNotFoundError:
'''Gets the DB path from config.ini'''
env = os.environ.get('KEMOVERSE_ENV')
if not (env and env in ['prod', 'dev']):
raise KemoverseEnvUnset
print(f'Running in "{env}" mode')
config_path = f'config_{env}.ini'
if not os.path.isfile(config_path):
raise ConfigError(f'Could not find {config_path}')
config = ConfigParser()
config.read(config_path)
db_path = config['application']['DatabaseLocation']
if not db_path:
raise DBNotFoundError()
return db_path
def get_current_migration(cursor: sqlite3.Cursor) -> int:
'''Gets the current schema version of the database'''
try:
cursor.execute('SELECT value FROM config WHERE key = ?', ('schema_version',))
version = cursor.fetchone()
return -1 if not version else int(version[0])
except sqlite3.Error:
print('Error getting schema version')
# Database has not been initialized yet
return -1
def main():
'''Does the thing'''
# Connect to the DB
db_path = ''
try:
db_path = get_db_path()
except ConfigError as ex:
print(ex)
return
except KemoverseEnvUnset:
print('Error: KEMOVERSE_ENV is either not set or has an invalid value.')
print('Please set KEMOVERSE_ENV to either "dev" or "prod" before running.')
print(traceback.format_exc())
return
conn = sqlite3.connect(db_path, autocommit=False)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Obtain list of migrations to run
migrations = get_migrations()
# Determine schema version
current_migration = get_current_migration(cursor)
print(f'Current schema version: {current_migration}')
# Run any migrations newer than current schema
for migration in migrations:
if migration[0] <= current_migration:
print(f'Migration already up: {migration[1]}')
continue
try:
perform_migration(cursor, migration)
conn.commit()
except Exception as ex:
print(f'An error occurred while applying {migration[1]}: {ex}, aborting...')
print(traceback.format_exc())
conn.rollback()
break
conn.close()
if __name__ == '__main__':
main()

View file

@ -1,3 +1,4 @@
from flask import Flask, render_template
import sqlite3 import sqlite3
import sys import sys
import os import os
@ -6,9 +7,6 @@ import os
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'bot')) sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'bot'))
import config import config
from flask import Flask, render_template, abort
from werkzeug.exceptions import HTTPException
app = Flask(__name__) app = Flask(__name__)
DB_PATH = config.DB_PATH DB_PATH = config.DB_PATH
@ -17,14 +15,6 @@ def get_db_connection():
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
return conn return conn
@app.errorhandler(HTTPException)
def handle_exception(error):
return render_template("_error.html", error=error), error.code
@app.route("/i404")
def i404():
return abort(404)
@app.route('/') @app.route('/')
def index(): def index():
conn = get_db_connection() conn = get_db_connection()
@ -49,8 +39,6 @@ def user_profile(user_id):
cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,)) cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
user = cursor.fetchone() user = cursor.fetchone()
if user is None:
abort(404)
cursor.execute(''' cursor.execute('''
SELECT pulls.timestamp, characters.name as character_name, characters.rarity SELECT pulls.timestamp, characters.name as character_name, characters.rarity

View file

@ -1,92 +0,0 @@
body {
font-family: "Segoe UI", Tahoma, Geneva, Verdana, sans-serif;
background-color: #f4f6fa;
color: #333;
margin: 0;
padding: 0;
}
header {
background-color: #7289da;
color: white;
padding: 20px;
text-align: center;
}
header h1 {
margin: 0;
font-size: 2.5em;
}
header p {
margin-top: 5px;
font-size: 1.1em;
}
.container {
max-width: 800px;
margin: 30px auto;
padding: 20px;
background-color: #ffffff;
border-radius: 10px;
box-shadow: 0 3px 10px rgba(0, 0, 0, 0.07);
}
h2 {
border-bottom: 1px solid #ccc;
padding-bottom: 8px;
margin-top: 30px;
}
ul {
list-style-type: none;
padding-left: 0;
}
li {
margin: 10px 0;
}
a {
text-decoration: none;
color: #2c3e50;
font-weight: bold;
background-color: #e3eaf3;
padding: 8px 12px;
border-radius: 6px;
display: inline-block;
transition: background-color 0.2s;
}
a:hover {
background-color: #cdd8e6;
}
.leaderboard-entry {
margin-bottom: 8px;
padding: 6px 10px;
background: #f9fafc;
border-left: 4px solid #7289da;
border-radius: 5px;
}
footer {
text-align: center;
margin-top: 40px;
font-size: 0.9em;
color: #888;
}
.note {
background: #fcfcf0;
border: 1px dashed #bbb;
padding: 10px;
border-radius: 8px;
margin-top: 30px;
font-size: 0.95em;
color: #666;
}
.footer-link {
margin: 0 10px;
}

View file

@ -1,30 +0,0 @@
<!DOCTYPE html>
<html>
<head>
<link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='style.css') }}">
<link rel="shortcut icon" href="{{ url_for('static', filename='logo.png') }}">
<title>
{% if title %}
{{ title }}
{% else %}
{% block title %}{% endblock %}
{% endif %}
&nbsp;|&nbsp;Kemoverse
</title>
</head>
<body>
<header>
{% block header %}{% endblock %}
</header>
<div class="container">
{% block content %}{% endblock %}
</div>
<footer>
<a class="footer-link" href="{{ url_for('about') }}">About</a>
<a class="footer-link" href="{{ url_for('submit_character') }}">Submit a Character</a>
{% block footer_extra %}{% endblock %}
</footer>
</body>
</html>

View file

@ -1,8 +0,0 @@
{% extends "_base.html" %}
{% block title %}
{{ error.code }}
{% endblock %}
{% block content %}
<h2>{{ error.code }}&nbsp;-&nbsp;{{ error.name }} </h2>
<p>{{ error.description }}</p>
{% endblock %}

View file

@ -1,9 +1,13 @@
{% extends "_base.html" %} <!DOCTYPE html>
<html>
{% block content %} <head>
<title>About - Misskey Gacha Center</title>
</head>
<body>
<h1>About This Gacha</h1> <h1>About This Gacha</h1>
<p>This is a playful Misskey-themed gacha tracker made with Flask and SQLite.</p> <p>This is a playful Misskey-themed gacha tracker made with Flask and SQLite.</p>
<p>All rolls are stored, stats are tracked, and characters are added manually for now.</p> <p>All rolls are stored, stats are tracked, and characters are added manually for now.</p>
<p>Built with love, chaos, and way too much caffeine ☕.</p> <p>Built with love, chaos, and way too much caffeine ☕.</p>
<a href="{{ url_for('index') }}">← Back to Home</a> <a href="{{ url_for('index') }}">← Back to Home</a>
{% endblock %} </body>
</html>

View file

@ -1,11 +1,110 @@
{% extends "_base.html" %} <!DOCTYPE html>
<html>
<head>
<title>Misskey Gacha Center</title>
<style>
body {
font-family: "Segoe UI", Tahoma, Geneva, Verdana, sans-serif;
background-color: #f4f6fa;
color: #333;
margin: 0;
padding: 0;
}
{% block header %} header {
background-color: #7289da;
color: white;
padding: 20px;
text-align: center;
}
header h1 {
margin: 0;
font-size: 2.5em;
}
header p {
margin-top: 5px;
font-size: 1.1em;
}
.container {
max-width: 800px;
margin: 30px auto;
padding: 20px;
background-color: #ffffff;
border-radius: 10px;
box-shadow: 0 3px 10px rgba(0, 0, 0, 0.07);
}
h2 {
border-bottom: 1px solid #ccc;
padding-bottom: 8px;
margin-top: 30px;
}
ul {
list-style-type: none;
padding-left: 0;
}
li {
margin: 10px 0;
}
a {
text-decoration: none;
color: #2c3e50;
font-weight: bold;
background-color: #e3eaf3;
padding: 8px 12px;
border-radius: 6px;
display: inline-block;
transition: background-color 0.2s;
}
a:hover {
background-color: #cdd8e6;
}
.leaderboard-entry {
margin-bottom: 8px;
padding: 6px 10px;
background: #f9fafc;
border-left: 4px solid #7289da;
border-radius: 5px;
}
footer {
text-align: center;
margin-top: 40px;
font-size: 0.9em;
color: #888;
}
.note {
background: #fcfcf0;
border: 1px dashed #bbb;
padding: 10px;
border-radius: 8px;
margin-top: 30px;
font-size: 0.95em;
color: #666;
}
.footer-link {
margin: 0 10px;
}
</style>
</head>
<body>
<header>
<h1>Misskey Gacha Center</h1> <h1>Misskey Gacha Center</h1>
<p>Track your luck. Compare your pulls. Compete with friends.</p> <p>Track your luck. Compare your pulls. Compete with friends.</p>
{% endblock %} </header>
{% block content %} <div class="container">
<h2>🎖️ Leaderboard: Most Rolls</h2> <h2>🎖️ Leaderboard: Most Rolls</h2>
{% for user in top_users %} {% for user in top_users %}
@ -26,4 +125,13 @@
<div class="note"> <div class="note">
🚀 This is a fun little gacha tracker! More features coming soon. Want to help shape it? 🚀 This is a fun little gacha tracker! More features coming soon. Want to help shape it?
</div> </div>
{% endblock %}
</div>
<footer>
<a class="footer-link" href="{{ url_for('about') }}">About</a>
<a class="footer-link" href="{{ url_for('submit_character') }}">Submit a Character</a>
</footer>
</body>
</html>

View file

@ -1,8 +1,12 @@
{% extends "_base.html" %} <!DOCTYPE html>
<html>
{% block content %} <head>
<title>Submit a Character - Misskey Gacha Center</title>
</head>
<body>
<h1>Submit a Character</h1> <h1>Submit a Character</h1>
<p>Want to add a new character to the gacha pool?</p> <p>Want to add a new character to the gacha pool?</p>
<p>This feature will be available soon. Stay tuned!</p> <p>This feature will be available soon. Stay tuned!</p>
<a href="{{ url_for('index') }}">← Back to Home</a> <a href="{{ url_for('index') }}">← Back to Home</a>
{% endblock %} </body>
</html>

View file

@ -1,5 +1,57 @@
{% extends "_base.html" %} <!DOCTYPE html>
{% block content %} <html>
<head>
<title>{{ user['username'] }}'s Rolls</title>
<style>
body {
font-family: Arial, sans-serif;
background-color: #f4f4f8;
margin: 0;
padding: 20px;
}
.profile, .pulls {
background-color: white;
padding: 15px;
border-radius: 10px;
box-shadow: 0 0 10px rgba(0,0,0,0.1);
margin-bottom: 20px;
}
h1, h2 {
margin-top: 0;
}
ul {
list-style-type: none;
padding: 0;
}
li {
padding: 10px 0;
border-bottom: 1px solid #eee;
}
.rarity {
color: gold;
font-weight: bold;
margin-left: 8px;
}
.timestamp {
color: #888;
font-size: 0.9em;
}
a {
display: inline-block;
margin-top: 20px;
color: #333;
text-decoration: none;
background-color: #ddd;
padding: 8px 12px;
border-radius: 5px;
}
a:hover {
background-color: #bbb;
}
</style>
</head>
<body>
<div class="profile"> <div class="profile">
<h1>{{ user['username'] }}'s Gacha Rolls</h1> <h1>{{ user['username'] }}'s Gacha Rolls</h1>
<p>User ID: {{ user['id'] }}</p> <p>User ID: {{ user['id'] }}</p>
@ -20,4 +72,6 @@
</div> </div>
<a href="{{ url_for('index') }}">← Back to Users</a> <a href="{{ url_for('index') }}">← Back to Users</a>
{% endblock %}
</body>
</html>