Compare commits

...

74 commits

Author SHA1 Message Date
c59edbcf4f Merge pull request 'Added table and general functions for stats system' (#21) from stats_system into dev
Some notes:
- Set the default value for card stats as 0, as we don't know the max we want
- Added a get_cards function that takes a list of ids and returns all the card information from the database, there's probably better ways to implement it though
Tested and works correctly
Reviewed-on: #21
2025-06-25 20:45:12 -07:00
w
91dfe77ffa Enhance add_card function to include power, charm, and wit parameters with validation checks 2025-06-26 00:37:47 -03:00
w
7c7d299fc4 Fix SQL table name in insert_card function and update argument validation for optional attributes 2025-06-25 01:11:37 -03:00
w
23ffd55b7b typo 2025-06-25 00:59:39 -03:00
w
ad17d33660 Update card stats migration to set default values for power, charm, and wit columns to 0 2025-06-25 00:26:11 -03:00
w
8c5c860ef6 get cards function 2025-06-25 00:21:37 -03:00
w
cce941471f Enhance insert_card function to support optional power, charm, and wit parameters 2025-06-22 01:42:08 -03:00
w
7b75a7eea9 Refactor card_stats migration to add power, charm, and wit columns to the card table 2025-06-22 01:14:32 -03:00
w
e1e6808b0a Merge branch 'dev' into stats_system 2025-06-22 00:53:09 -03:00
91c4531a47 Merge pull request 'license_change' (#59) from license_change into dev
Reviewed-on: #59
2025-06-17 20:34:32 -07:00
w
7f3e0e0f8b Update copyright notices to include all contributors in various files 2025-06-18 00:21:56 -03:00
w
42180f4139 Added copyright notice 2025-06-17 23:54:55 -03:00
e3fc633fc4 Merge pull request 'pleroma-support - optionally use pleroma as a host for the bot.' (#54) from pleroma-support into dev
Reviewed-on: #54
2025-06-15 19:35:40 -07:00
Moon
c74fc17dfb Merge branch 'dev' into pleroma-support 2025-06-16 08:31:30 +09:00
w
6494ac4909 Typo that saves the west 2025-06-15 16:50:27 -03:00
6ab3e4a427 Merge pull request 'license_change' (#53) from license_change into dev
Reviewed-on: #53
2025-06-15 12:46:49 -07:00
w
b6fb6d9245 Merge branch 'license_change' of git.waifuism.life:waifu/kemoverse into license_change 2025-06-15 16:46:26 -03:00
w
5580e06028 Merge branch 'dev' into license_change 2025-06-15 16:46:03 -03:00
b08b224476 Merge branch 'dev' into license_change 2025-06-15 06:54:56 -07:00
Moon
0039717a01 Merge branch 'dev' into pleroma-support 2025-06-14 07:31:02 +09:00
Moon
bf8d9823b2 Merge branch 'dev' into pleroma-support 2025-06-14 07:18:18 +09:00
7c4fd5fc41 Merge pull request 'any-env-and-bind-address' (#58) from any-env-and-bind-address into dev
Reviewed-on: #58
Reviewed-by: VD15 <valkyriedev15@gmail.com>
2025-06-13 14:54:27 -07:00
Moon
24bfe88dc1 rm obsolete print 2025-06-14 05:57:01 +09:00
Moon
1a35750d0a restrict characters in KEMOVERSE_ENV 2025-06-14 05:51:49 +09:00
Moon
d416ae1b2d rm limitation on KEMOVERSE_ENV from another place. 2025-06-14 05:47:43 +09:00
Moon
7fd4d5db25 indicate on startup if whitelisting is enabled. 2025-06-14 05:17:39 +09:00
Moon
f4f847e577 make sure UseWhitelist is a boolean, default to True 2025-06-14 05:14:40 +09:00
Moon
2ef70801c7 fix config module ref 2025-06-14 05:08:59 +09:00
Moon
fdf21b3f5f rm check for env dev or prod so any can be used 2025-06-14 04:43:43 +09:00
Moon
337a989671 rm vestigial trusted instances config. 2025-06-14 04:34:20 +09:00
Moon
3ad4edbc45 ignore custom startup script 2025-06-14 04:10:16 +09:00
49e49bd187 Merge branch 'dev' into license_change 2025-06-13 06:56:21 -07:00
1984b67523 Merge pull request 'fix-missing-weight-column in dev' (#57) from fix-missing-weight-column into dev
Reviewed-on: #57
Reviewed-by: VD15 <valkyriedev15@gmail.com>
2025-06-13 04:14:53 -07:00
Moon
bbb64a869d Merge remote-tracking branch 'origin/master' into pleroma-support 2025-06-13 19:50:47 +09:00
Moon
fc45c688e8 Merge branch 'fix-missing-weight-column' into pleroma-support 2025-06-13 19:42:41 +09:00
Moon
77d4fa13bb rm validation of removed weight 2025-06-13 19:31:23 +09:00
Moon
bd287b096a rm reference to weight column. 2025-06-13 18:47:54 +09:00
Moon
40f018a83b make instantiation explicit so easier to test. 2025-06-13 13:53:41 +09:00
Moon
a516a9b55a minor change to make testing easier. 2025-06-13 13:36:38 +09:00
Moon
7161712f15 rm unused function that bypassed server abstraction 2025-06-13 13:26:01 +09:00
3142cf11ff Merge pull request 'Version 1.1' (#55) from dev into master
Reviewed-on: #55
2025-06-12 21:10:29 -07:00
Moon
fadd4dfe27 Merge remote-tracking branch 'origin/dev' into pleroma-support 2025-06-13 13:05:51 +09:00
d2a4e0b1fe Merge pull request '5_administrator_control' (#49) from 5_administrator_control into dev
Reviewed-on: #49
2025-06-12 20:34:08 -07:00
Moon
7b32ee7fcf had to change some things to get cursor for db working. 2025-06-13 11:23:34 +09:00
Moon
f70b2147cd Remove legacy files that bypass FediverseService abstraction
- Remove bot/client.py (replaced by fediverse factory)
- Remove bot/notification.py (replaced by bot_app.py)
- All functionality now properly uses FediverseService abstraction
2025-06-13 10:57:33 +09:00
Moon
298d7fda72 rm needless file 2025-06-13 10:52:20 +09:00
Moon
8918b5205d named some damn properties wrong 2025-06-13 10:52:20 +09:00
Moon
67b4d949fd use a library for content type detection instead of handrolled 2025-06-13 10:50:24 +09:00
Moon
89ae8a7290 back to using mastodon.py 2025-06-13 10:50:24 +09:00
Moon
91376c0eba works now but may need to revert file upload change 2025-06-13 10:50:24 +09:00
Moon
b2ca6dd59a documentation update for configurable backends 2025-06-13 10:50:03 +09:00
Moon
dac05a3ed8 completely untested refactor. bug fixes to come later. 2025-06-13 10:48:35 +09:00
Moon
fe8e7d246f actually use the abstracted type not just the id 2025-06-13 10:37:27 +09:00
Moon
a47530180d first whack at file upload 2025-06-13 10:37:27 +09:00
Moon
e0cf42f8f6 baseline fedi abstraction stuff. 2025-06-13 10:37:24 +09:00
61d8fe71bb Merge branch 'dev' into 5_administrator_control 2025-06-12 04:18:23 -07:00
w
2b194f3c9e readme changes for license 2025-06-11 22:42:57 -03:00
w
df76fb9131 adding license to main files 2025-06-11 22:29:06 -03:00
w
81c890a83d license change 2025-06-11 22:06:17 -03:00
1368c907a2 Revert "Enable WAL for DB connections"
This reverts commit 59915be661.
2025-06-08 00:09:02 +01:00
59915be661 Enable WAL for DB connections 2025-06-07 23:59:07 +01:00
4b1b8a53c7 Fix nonstandard apostrophe 2025-06-07 23:25:48 +01:00
f231b89b0a Merge branch 'dev' into 5_administrator_control 2025-06-07 15:18:48 -07:00
97b30c79f4 Add admin commands 2025-06-07 23:18:39 +01:00
8ae6e25b95 Add instance whitelist 2025-06-07 20:40:27 +01:00
fde6e1167a Add administrators 2025-06-07 19:23:17 +01:00
w
9d98299264 expanding character_stats 2025-05-28 23:52:04 -03:00
w
3b7f2006ef unifying character and stats insertion, generalize get_character 2025-05-28 23:44:59 -03:00
w
471f399176 Merge branch 'dev' into stats_system 2025-05-28 22:28:57 -03:00
w
631b66e003 Merge branch 'dev' into stats_system 2025-05-24 18:16:19 -03:00
w
3a2033a025 Add a roll timeout to the bot 2025-05-24 17:54:15 -03:00
w
74e4c86d02 Added table and general functions for stats system 2025-05-24 02:32:04 -03:00
237f17f40d Add a roll timeout to the bot 2025-05-24 02:32:04 -03:00
w
11c9cf58a3 Added table and general functions for stats system 2025-05-22 23:26:03 -03:00
44 changed files with 2386 additions and 701 deletions

17
.gitignore vendored
View file

@ -1,3 +1,19 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, VD15 and contributors.
# 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/.
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
@ -185,5 +201,6 @@ cython_debug/
gacha_game*.db
gacha_game*.db.*
config*.ini
run.sh
.idea

1
.tool-versions Normal file
View file

@ -0,0 +1 @@
nodejs 23.4.0

722
LICENSE

File diff suppressed because it is too large Load diff

82
bot/add_card.py Normal file
View file

@ -0,0 +1,82 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, Moon, VD15 and contributors.
# 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/.
import requests
import config
from fediverse_factory import get_fediverse_service
import db_utils
def add_card(name: str, rarity: int, image_url: str, power: int, charm: int, wit: int) -> tuple[int, str]:
"""
Adds a card to the database, uploading the image from a public URL to the Fediverse instance.
Args:
name (str): Card name.
rarity (int): Card rarity (e.g., 1-5).
image_url (str): Public URL of the image from the post.
power (int): Card power value.
charm (int): Card charm value.
wit (int): Card wit value.
Returns:
tuple[int, str]: Card ID and file_id.
Raises:
ValueError: If inputs are invalid.
RuntimeError: If image download/upload or database operation fails.
"""
stripped_name = name.strip()
# Validate inputs
if not stripped_name:
raise ValueError('Card name cannot be empty.')
if rarity < 1:
raise ValueError('Rarity must be a positive integer.')
if rarity not in config.RARITY_TO_WEIGHT.keys():
raise ValueError(f'Invalid rarity: {rarity}')
if not image_url:
raise ValueError('Image URL must be provided.')
if power < 0 or charm < 0 or wit < 0:
raise ValueError('Power, charm, and wit must be non-negative integers.')
try:
# Download image
response = requests.get(image_url, stream=True, timeout=30)
if response.status_code != 200:
raise RuntimeError(f"Failed to download image from {image_url}")
# Upload to Fediverse instance
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
try:
uploaded_file = fediverse_service.upload_file(response.raw)
file_id = uploaded_file.id
except RuntimeError as e:
raise RuntimeError(f"Failed to upload image: {e}") from e
# Insert into database using db_utils function
card_id = db_utils.insert_card(
stripped_name,
rarity,
file_id,
power,
charm,
wit
)
return card_id, file_id
except Exception as e:
raise

View file

@ -1,64 +0,0 @@
import requests
from misskey.exceptions import MisskeyAPIException
from client import client_connection
from db_utils import insert_character
from custom_types import Character
from config import RARITY_TO_WEIGHT
def add_character(
name: str,
rarity: int,
image_url: str) -> tuple[int, str]:
'''
Adds a character to the database, uploading the image from a public URL to
the bot's Misskey Drive.
Args:
name (str): Character name.
rarity (int): Character 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.
Raises:
ValueError: If inputs are invalid.
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.')
# Download image
response = requests.get(image_url, stream=True, timeout=30)
if response.status_code != 200:
raise RuntimeError(f'Failed to download image from {image_url}')
# Upload to bot's Drive
mk = client_connection()
try:
media = mk.drive_files_create(response.raw)
file_id = media['id']
except MisskeyAPIException as e:
raise RuntimeError(f'Failed to upload image to bot\'s Drive: {e}')\
from e
# Insert into database
character_id = insert_character(
stripped_name,
rarity,
RARITY_TO_WEIGHT[rarity],
file_id
)
return character_id, file_id

View file

@ -1,18 +1,78 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, VD15, Moon and contributors.
# 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/.
import time
import misskey as misskey
from client import client_connection
import db_utils as db
import traceback
import config
from notification import process_fediverse_notification
from db_utils import get_config, set_config, connect, setup_administrators
from fediverse_factory import get_fediverse_service
from config import NOTIFICATION_POLL_INTERVAL
from notification import process_notifications
from config import USE_WHITELIST
if __name__ == '__main__':
# Initialize the Misskey client
client = client_connection()
# Connect to DB
db.connect()
def stream_notifications():
# Initialize database connection
connect()
# Setup default administrators
setup_administrators()
# Initialize the Fediverse service
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
# Get the last seen notification ID from the database
last_seen_id = get_config("last_seen_notif_id")
# Show whitelist status
whitelist_status = "enabled" if USE_WHITELIST else "disabled"
print(f'Instance whitelisting: {whitelist_status}')
print('Listening for notifications...')
while True:
if not process_notifications(client):
time.sleep(NOTIFICATION_POLL_INTERVAL)
try:
# Get notifications from the fediverse service
notifications = fediverse_service.get_notifications(since_id=last_seen_id)
if notifications:
new_last_seen_id = last_seen_id
for notification in notifications:
notif_id = notification.id
# Skip old or same ID notifications
if last_seen_id is not None and notif_id <= last_seen_id:
continue
# Process the notification using the abstracted processor
process_fediverse_notification(notification, fediverse_service)
# Update only if this notif_id is greater
if new_last_seen_id is None or notif_id > new_last_seen_id:
new_last_seen_id = notif_id
# Save the latest seen ID
if new_last_seen_id and new_last_seen_id != last_seen_id:
set_config("last_seen_notif_id", new_last_seen_id)
last_seen_id = new_last_seen_id
time.sleep(5)
except Exception as e:
print(f"An exception has occured: {e}\n{traceback.format_exc()}")
time.sleep(5)
if __name__ == "__main__":
stream_notifications()

View file

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

View file

@ -1,5 +1,23 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, VD15, Moon and contributors.
# 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/.
'''Essentials for the bot to function'''
import configparser
import json
import re
from os import environ, path
@ -12,8 +30,10 @@ def get_config_file() -> str:
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}')
# Validate environment name contains only alphanumeric, dash, and underscore
if not re.match(r'^[a-zA-Z0-9_-]+$', env):
raise ValueError(f'KEMOVERSE_ENV "{env}" contains invalid characters. Only alphanumeric, dash (-), and underscore (_) are allowed.')
config_path: str = f'config_{env}.ini'
@ -21,7 +41,53 @@ def get_config_file() -> str:
raise ConfigError(f'Could not find {config_path}')
return config_path
def get_rarity_to_weight(config_section):
def normalize_user(user_string: str) -> str:
"""
Normalizes a user string to the format @user@domain.tld where domain is lowercase and user is case-sensitive
Args:
user_string: User string in various formats
Returns:
Normalized user string
Raises:
ValueError: If the user string is invalid or domain is malformed
"""
if not user_string or not user_string.strip():
raise ValueError("User string cannot be empty")
user_string = user_string.strip()
# Add leading @ if missing
if not user_string.startswith('@'):
user_string = '@' + user_string
# Split into user and domain parts
parts = user_string[1:].split('@', 1) # Remove leading @ and split
if len(parts) != 2:
raise ValueError(f"Invalid user format: {user_string}. Expected @user@domain.tld")
username, domain = parts
if not username:
raise ValueError("Username cannot be empty")
if not domain:
raise ValueError("Domain cannot be empty")
# Validate domain format (basic check for valid domain structure)
domain_pattern = r'^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$'
if not re.match(domain_pattern, domain):
raise ValueError(f"Invalid domain format: {domain}")
# Return normalized format: @user@domain.tld (domain lowercase, user case-sensitive)
return f"@{username}@{domain.lower()}"
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():
@ -35,18 +101,36 @@ config = configparser.ConfigParser()
config.read(get_config_file())
# Username for the bot
USER = config['credentials']['User'].lower()
if 'User' not in config['credentials'] or not config['credentials']['User'].strip():
raise ConfigError("User must be specified in config.ini under [credentials]")
USER = normalize_user(config['credentials']['User'])
# API key for the bot
KEY = config['credentials']['Token']
# Bot's Misskey instance URL
# Bot's Misskey/Pleroma instance URL
INSTANCE = config['credentials']['Instance'].lower()
# TODO: move this to db
# Instance type validation
if 'InstanceType' not in config['application']:
raise ValueError("InstanceType must be specified in config.ini")
instance_type = config['application']['InstanceType'].lower()
if instance_type not in ('misskey', 'pleroma'):
raise ValueError("InstanceType must be either 'misskey' or 'pleroma'")
INSTANCE_TYPE = instance_type
# Web server port
WEB_PORT = config['application'].getint('WebPort', 5000)
BIND_ADDRESS = config['application'].get('BindAddress', '127.0.0.1')
# 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']
DB_PATH = config['application'].get('DatabaseLocation', './gacha_game.db')
# Whether to enable the instance whitelist
USE_WHITELIST = config['application'].getboolean('UseWhitelist', True)
NOTIFICATION_POLL_INTERVAL = int(config['notification']['PollInterval'])
NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize'])

View file

@ -1,3 +1,19 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 VD15 and contributors.
# 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/.
from typing import TypedDict, List, Dict, Any
BotResponse = TypedDict('BotResponse', {
@ -5,7 +21,7 @@ BotResponse = TypedDict('BotResponse', {
'attachment_urls': List[str] | None
})
Character = TypedDict('Character', {
Card = TypedDict('Card', {
'id': int,
'name': str,
'rarity': int,

View file

@ -1,7 +1,23 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, VD15, Moon, and contributors.
# 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/.
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 +34,38 @@ def connect() -> None:
CURSOR = CONNECTION.cursor()
def get_random_character() -> Character | None:
''' Gets a random character from the database'''
CURSOR.execute('SELECT * FROM characters')
characters = CURSOR.fetchall()
def setup_administrators() -> None:
'''Creates administrator players for each handle in the config file'''
# Get default admins from config
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]
chosen = choices(characters, weights=weights, k=1)[0]
weights = [config.RARITY_TO_WEIGHT[c['rarity']] for c in cards]
chosen = choices(cards, weights=weights, k=1)[0]
return {
'id': chosen['id'],
@ -37,76 +75,181 @@ def get_random_character() -> Character | None:
'image_url': chosen['file_id']
}
def get_cards(card_ids: list[int]) -> list[tuple]:
'''
Retrieves stats for a list of card IDs.
Returns a list of tuples: (id, name, rarity, file_id, power, charm, wit, ...)
'''
if not card_ids:
return []
placeholders = ','.join('?' for _ in card_ids)
query = f'SELECT * FROM cards WHERE id IN ({placeholders})'
CURSOR.execute(query, card_ids)
return CURSOR.fetchall()
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 (?, ?)',
(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 = ?',
'SELECT id FROM players 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()
if not player:
return False
player_id = player[0]
# Delete pulls
CURSOR.execute(
'DELETE FROM pulls WHERE user_id = ?',
(user_id,)
'DELETE FROM pulls WHERE player_id = ?',
(player_id,)
)
# Delete user
# Delete player
CURSOR.execute(
'DELETE FROM users WHERE id = ?',
(user_id,)
'DELETE FROM players WHERE id = ?',
(player_id,)
)
return True
def ban_player(username: str) -> bool:
'''Adds a player to the ban list.'''
try:
CURSOR.execute(
'INSERT INTO banned_players (handle) VALUES (?)',
(username,)
)
return True
except sqlite3.IntegrityError:
return False
def insert_character(
name: str, rarity: int, weight: float, file_id: str) -> int:
'''Inserts a character'''
def unban_player(username: str) -> bool:
'''Removes a player from the ban list.'''
CURSOR.execute(
'INSERT INTO characters (name, rarity, weight, file_id) VALUES \
(?, ?, ?, ?)',
(name, rarity, weight, file_id)
'DELETE FROM banned_players WHERE handle = ?',
(username,)
)
character_id = CURSOR.lastrowid
return character_id if character_id else 0
return CURSOR.rowcount > 0
def insert_pull(user_id: int, character_id: int) -> None:
def is_player_banned(username: str) -> bool:
CURSOR.execute(
'SELECT * FROM banned_players WHERE handle = ?',
(username,)
)
row = CURSOR.fetchone()
return row is not None
def is_player_administrator(username: str) -> bool:
CURSOR.execute(
'SELECT is_administrator FROM players WHERE username = ? LIMIT 1',
(username,)
)
row = CURSOR.fetchone()
return row[0] if row else False
def insert_card(
name: str, rarity: int, file_id: str,
power: int =None, charm: int = None, wit: int = None) -> int:
'''Inserts a card'''
if power is not None and charm is not None and wit is not None:
CURSOR.execute(
'''
INSERT INTO cards (name, rarity, file_id, power, charm, wit)
VALUES (?, ?, ?, ?, ?, ?)
''',
(name, rarity, file_id, power, charm, wit)
)
else:
CURSOR.execute(
'''
INSERT INTO cards (name, rarity, file_id)
VALUES (?, ?, ?)
''',
(name, rarity, file_id)
)
card_id = CURSOR.lastrowid
return card_id if card_id else 0
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 (?, ?)',
(user_id, character_id)
'INSERT INTO pulls (player_id, card_id) VALUES (?, ?)',
(player_id, card_id)
)
def get_last_rolled_at(user_id: int) -> int:
'''Gets the timestamp when the user last rolled'''
def get_last_rolled_at(player_id: int) -> int:
'''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
# Configuration
def add_to_whitelist(instance: str) -> bool:
'''Adds an instance to the whitelist, returns false if instance was already
present'''
try:
CURSOR.execute(
'INSERT INTO instance_whitelist (tld) VALUES (?)', (instance,)
)
return True
except sqlite3.IntegrityError:
return False
def remove_from_whitelist(instance: str) -> bool:
'''Removes an instance to the whitelist, returns false if instance was not
present'''
CURSOR.execute(
'DELETE FROM instance_whitelist WHERE tld = ?', (instance,))
return CURSOR.rowcount > 0
def is_whitelisted(instance: str) -> bool:
'''Checks whether an instance is in the whitelist'''
if instance == 'local':
return True
CURSOR.execute(
'SELECT * FROM instance_whitelist WHERE tld = ?', (instance,))
row = CURSOR.fetchone()
return row is not None
def get_config(key: str) -> str:
'''Reads the value for a specified config key from the db'''

56
bot/fediverse_factory.py Normal file
View file

@ -0,0 +1,56 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Moon and contributors.
# 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/.
from fediverse_service import FediverseService
from misskey_service import MisskeyService
from pleroma_service import PleromaService
class FediverseServiceFactory:
"""Factory for creating FediverseService implementations"""
@staticmethod
def create_service(instance_type: str) -> FediverseService:
"""
Create a FediverseService implementation based on the instance type.
Args:
instance_type: The type of instance ("misskey" or "pleroma")
Returns:
FediverseService implementation (MisskeyService or PleromaService)
Raises:
ValueError: If the instance type is not supported
"""
instance_type = instance_type.lower()
if instance_type == "misskey":
return MisskeyService()
elif instance_type == "pleroma":
return PleromaService()
else:
raise ValueError(f"Unsupported instance type: {instance_type}")
def get_fediverse_service(instance_type: str) -> FediverseService:
"""
Convenience function to get a FediverseService instance
Args:
instance_type: The instance type ("misskey" or "pleroma")
"""
return FediverseServiceFactory.create_service(instance_type)

90
bot/fediverse_service.py Normal file
View file

@ -0,0 +1,90 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Moon and contributors.
# 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/.
from abc import ABC, abstractmethod
from typing import List, Optional, Union, BinaryIO
from fediverse_types import FediverseNotification, FediversePost, FediverseFile, Visibility
class FediverseService(ABC):
"""Abstract interface for Fediverse platform services (Misskey, Pleroma, etc.)"""
@abstractmethod
def get_notifications(self, since_id: Optional[str] = None) -> List[FediverseNotification]:
"""
Retrieve notifications from the Fediverse instance.
Args:
since_id: Optional ID to get notifications newer than this ID
Returns:
List of FediverseNotification objects
"""
pass
@abstractmethod
def create_post(
self,
text: str,
reply_to_id: Optional[str] = None,
visibility: Visibility = Visibility.HOME,
file_ids: Optional[List[str]] = None,
visible_user_ids: Optional[List[str]] = None
) -> str:
"""
Create a new post on the Fediverse instance.
Args:
text: The text content of the post
reply_to_id: Optional ID of post to reply to
visibility: Visibility level for the post
file_ids: Optional list of file IDs to attach
visible_user_ids: Optional list of user IDs who can see the post (for specified visibility)
Returns:
ID of the created post
"""
pass
@abstractmethod
def get_post_by_id(self, post_id: str) -> Optional[FediversePost]:
"""
Retrieve a specific post by its ID.
Args:
post_id: The ID of the post to retrieve
Returns:
FediversePost object if found, None otherwise
"""
pass
@abstractmethod
def upload_file(self, file_data: Union[BinaryIO, bytes], filename: Optional[str] = None) -> FediverseFile:
"""
Upload a file to the Fediverse instance.
Args:
file_data: File data as binary stream or bytes
filename: Optional filename for the uploaded file
Returns:
FediverseFile object with ID, URL, and other metadata
Raises:
RuntimeError: If file upload fails
"""
pass

89
bot/fediverse_types.py Normal file
View file

@ -0,0 +1,89 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Moon and contributors.
# 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/.
from dataclasses import dataclass
from typing import Optional, List, Dict, Any
from enum import Enum
class NotificationType(Enum):
MENTION = "mention"
REPLY = "reply"
FOLLOW = "follow"
FAVOURITE = "favourite"
REBLOG = "reblog"
POLL = "poll"
OTHER = "other"
class Visibility(Enum):
PUBLIC = "public"
UNLISTED = "unlisted"
HOME = "home"
FOLLOWERS = "followers"
SPECIFIED = "specified"
DIRECT = "direct"
@dataclass
class FediverseUser:
"""Common user representation across Fediverse platforms"""
id: str
username: str
host: Optional[str] = None # None for local users
display_name: Optional[str] = None
@property
def full_handle(self) -> str:
"""Returns the full fediverse handle (@user@domain or @user for local)"""
if self.host:
return f"@{self.username}@{self.host}"
return f"@{self.username}"
@dataclass
class FediverseFile:
"""Common file/attachment representation"""
id: str
url: str
type: Optional[str] = None
name: Optional[str] = None
@dataclass
class FediversePost:
"""Common post representation across Fediverse platforms"""
id: str
text: Optional[str]
user: FediverseUser
visibility: Visibility
created_at: Optional[str] = None
files: List[FediverseFile] = None
reply_to_id: Optional[str] = None
def __post_init__(self):
if self.files is None:
self.files = []
@dataclass
class FediverseNotification:
"""Common notification representation across Fediverse platforms"""
id: str
type: NotificationType
user: FediverseUser
post: Optional[FediversePost] = None
created_at: Optional[str] = None

172
bot/misskey_service.py Normal file
View file

@ -0,0 +1,172 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Moon and contributors.
# 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/.
import misskey
from typing import List, Optional, Dict, Any, Union, BinaryIO
from fediverse_service import FediverseService
from fediverse_types import (
FediverseNotification, FediversePost, FediverseUser, FediverseFile,
NotificationType, Visibility
)
import config
class MisskeyService(FediverseService):
"""Misskey implementation of FediverseService"""
def __init__(self):
self.client = misskey.Misskey(address=config.INSTANCE, i=config.KEY)
def _convert_misskey_user(self, user_data: Dict[str, Any]) -> FediverseUser:
"""Convert Misskey user data to FediverseUser"""
return FediverseUser(
id=user_data.get("id", ""),
username=user_data.get("username", "unknown"),
host=user_data.get("host"),
display_name=user_data.get("name")
)
def _convert_misskey_file(self, file_data: Dict[str, Any]) -> FediverseFile:
"""Convert Misskey file data to FediverseFile"""
return FediverseFile(
id=file_data.get("id", ""),
url=file_data.get("url", ""),
type=file_data.get("type"),
name=file_data.get("name")
)
def _convert_misskey_visibility(self, visibility: str) -> Visibility:
"""Convert Misskey visibility to our enum"""
visibility_map = {
"public": Visibility.PUBLIC,
"unlisted": Visibility.UNLISTED,
"home": Visibility.HOME,
"followers": Visibility.FOLLOWERS,
"specified": Visibility.SPECIFIED
}
return visibility_map.get(visibility, Visibility.HOME)
def _convert_to_misskey_visibility(self, visibility: Visibility) -> str:
"""Convert our visibility enum to Misskey visibility"""
visibility_map = {
Visibility.PUBLIC: "public",
Visibility.UNLISTED: "unlisted",
Visibility.HOME: "home",
Visibility.FOLLOWERS: "followers",
Visibility.SPECIFIED: "specified",
Visibility.DIRECT: "specified" # Map direct to specified for Misskey
}
return visibility_map.get(visibility, "home")
def _convert_misskey_notification_type(self, notif_type: str) -> NotificationType:
"""Convert Misskey notification type to our enum"""
type_map = {
"mention": NotificationType.MENTION,
"reply": NotificationType.REPLY,
"follow": NotificationType.FOLLOW,
"favourite": NotificationType.FAVOURITE,
"reblog": NotificationType.REBLOG,
"poll": NotificationType.POLL
}
return type_map.get(notif_type, NotificationType.OTHER)
def _convert_misskey_post(self, note_data: Dict[str, Any]) -> FediversePost:
"""Convert Misskey note data to FediversePost"""
files = []
if note_data.get("files"):
files = [self._convert_misskey_file(f) for f in note_data["files"]]
return FediversePost(
id=note_data.get("id", ""),
text=note_data.get("text"),
user=self._convert_misskey_user(note_data.get("user", {})),
visibility=self._convert_misskey_visibility(note_data.get("visibility", "home")),
created_at=note_data.get("createdAt"),
files=files,
reply_to_id=note_data.get("replyId")
)
def _convert_misskey_notification(self, notification_data: Dict[str, Any]) -> FediverseNotification:
"""Convert Misskey notification data to FediverseNotification"""
post = None
if notification_data.get("note"):
post = self._convert_misskey_post(notification_data["note"])
return FediverseNotification(
id=notification_data.get("id", ""),
type=self._convert_misskey_notification_type(notification_data.get("type", "")),
user=self._convert_misskey_user(notification_data.get("user", {})),
post=post,
created_at=notification_data.get("createdAt")
)
def get_notifications(self, since_id: Optional[str] = None) -> List[FediverseNotification]:
"""Get notifications from Misskey instance"""
params = {
'include_types': ['mention', 'reply'],
'limit': 50
}
if since_id:
params["since_id"] = since_id
notifications = self.client.i_notifications(**params)
return [self._convert_misskey_notification(notif) for notif in notifications]
def create_post(
self,
text: str,
reply_to_id: Optional[str] = None,
visibility: Visibility = Visibility.HOME,
file_ids: Optional[List[str]] = None,
visible_user_ids: Optional[List[str]] = None
) -> str:
"""Create a post on Misskey instance"""
params = {
"text": text,
"visibility": self._convert_to_misskey_visibility(visibility)
}
if reply_to_id:
params["reply_id"] = reply_to_id
if file_ids:
params["file_ids"] = file_ids
if visible_user_ids and visibility == Visibility.SPECIFIED:
params["visible_user_ids"] = visible_user_ids
response = self.client.notes_create(**params)
return response.get("createdNote", {}).get("id", "")
def get_post_by_id(self, post_id: str) -> Optional[FediversePost]:
"""Get a specific post by ID from Misskey instance"""
try:
note = self.client.notes_show(noteId=post_id)
return self._convert_misskey_post(note)
except Exception:
return None
def upload_file(self, file_data: Union[BinaryIO, bytes], filename: Optional[str] = None) -> FediverseFile:
"""Upload a file to Misskey Drive"""
try:
from misskey.exceptions import MisskeyAPIException
media = self.client.drive_files_create(file_data)
return self._convert_misskey_file(media)
except MisskeyAPIException as e:
raise RuntimeError(f"Failed to upload file to Misskey Drive: {e}") from e
except Exception as e:
raise RuntimeError(f"Unexpected error during file upload: {e}") from e

View file

@ -0,0 +1,90 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Moon and contributors.
# 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/.
"""Mock FediverseService for testing purposes"""
from typing import List, Optional, Union, BinaryIO
from fediverse_service import FediverseService
from fediverse_types import FediverseNotification, FediversePost, FediverseFile, Visibility
class MockFediverseService(FediverseService):
"""Mock implementation of FediverseService for testing"""
def __init__(self):
self.notifications = []
self.created_posts = []
self.uploaded_files = []
def get_notifications(self, since_id: Optional[str] = None) -> List[FediverseNotification]:
"""Return mock notifications, optionally filtered by since_id"""
if since_id is None:
return self.notifications
# Filter notifications newer than since_id
filtered = []
for notif in self.notifications:
if notif.id > since_id:
filtered.append(notif)
return filtered
def create_post(
self,
text: str,
reply_to_id: Optional[str] = None,
visibility: Visibility = Visibility.HOME,
file_ids: Optional[List[str]] = None,
visible_user_ids: Optional[List[str]] = None
) -> str:
"""Mock post creation, returns fake post ID"""
post_id = f"mock_post_{len(self.created_posts)}"
# Store the post for assertions
self.created_posts.append({
'id': post_id,
'text': text,
'reply_to_id': reply_to_id,
'visibility': visibility,
'file_ids': file_ids,
'visible_user_ids': visible_user_ids
})
return post_id
def upload_file(self, file_data: Union[BinaryIO, bytes]) -> FediverseFile:
"""Mock file upload, returns fake file"""
file_id = f"mock_file_{len(self.uploaded_files)}"
mock_file = FediverseFile(
id=file_id,
url=f"https://example.com/files/{file_id}",
type="image/png",
name="test_file.png"
)
self.uploaded_files.append(mock_file)
return mock_file
# Helper methods for testing
def add_mock_notification(self, notification: FediverseNotification):
"""Add a mock notification for testing"""
self.notifications.append(notification)
def clear_all(self):
"""Clear all mock data"""
self.notifications.clear()
self.created_posts.clear()
self.uploaded_files.clear()

View file

@ -1,51 +1,71 @@
import traceback
from typing import Dict, Any
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, Moon, VD15, and contributors.
import misskey
from misskey.exceptions import MisskeyAPIException
# 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.
from config import NOTIFICATION_BATCH_SIZE
# 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/.
import config
from parsing import parse_notification
from db_utils import get_config, set_config
from db_utils import is_whitelisted, is_player_banned
from response import generate_response
from custom_types import BotResponse
# Define your whitelist
# TODO: move to config
WHITELISTED_INSTANCES: list[str] = []
from fediverse_factory import get_fediverse_service
from fediverse_types import FediverseNotification, NotificationType, Visibility
def process_notification(
client: misskey.Misskey,
notification: Dict[str, Any]) -> None:
'''Processes an individual notification'''
user = notification.get('user', {})
username = user.get('username', 'unknown')
host = user.get('host') # None if local user
def process_fediverse_notification(notification: FediverseNotification, fediverse_service=None) -> None:
'''Processes an individual fediverse notification using the abstraction'''
if fediverse_service is None:
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
# Get user and instance info
username = notification.user.username
host = notification.user.host
instance = host if host else 'local'
if not (instance in WHITELISTED_INSTANCES or instance == 'local'):
# Check whitelist
if config.USE_WHITELIST and not is_whitelisted(instance):
print(f'⚠️ Blocked notification from untrusted instance: {instance}')
return
# Copy visibility of the post that was received when replying (so if people
# don't want to dump a bunch of notes on home they don't have to)
visibility = notification['note']['visibility']
if visibility != 'specified':
visibility = 'home'
# Only process mentions and replies
if notification.type not in (NotificationType.MENTION, NotificationType.REPLY):
return
notif_type = notification.get('type', 'unknown')
notif_id = notification.get('id')
# Return early if no post attached
if not notification.post:
return
# Determine visibility for reply
if notification.post.visibility != Visibility.SPECIFIED:
visibility = Visibility.HOME
else:
visibility = Visibility.SPECIFIED
notif_type = notification.type.value
notif_id = notification.id
print(f'📨 <{notif_id}> [{notif_type}] from @{username}@{instance}')
# 🧠 Send to the parser
parsed_notification = parse_notification(notification, client)
parsed_notification = parse_notification(notification, fediverse_service)
if not parsed_notification:
return
# Get the note Id to reply to
note_id = notification.get('note', {}).get('id')
author = parsed_notification['author']
if is_player_banned(author):
print(f'⚠️ Blocked notification from banned player: {author}')
return
# Get the response
response: BotResponse | None = generate_response(parsed_notification)
@ -53,70 +73,14 @@ def process_notification(
if not response:
return
client.notes_create(
# Handle attachment URLs (convert to file IDs if needed)
file_ids = response['attachment_urls'] if response['attachment_urls'] else None
# Send response using fediverse service
fediverse_service.create_post(
text=response['message'],
reply_id=note_id,
reply_to_id=notification.post.id,
visibility=visibility,
file_ids=response['attachment_urls']
# TODO: write actual visible users ids so pleromers can use the bot
# privately
# visible_user_ids=[]
file_ids=file_ids
# visible_user_ids=[] # TODO: write actual visible users ids so pleromers can use the bot privately
)
def process_notifications(client: misskey.Misskey) -> bool:
'''Processes a batch of unread notifications. Returns False if there are
no more notifications to process.'''
last_seen_id = get_config('last_seen_notif_id')
# process_notification writes to last_seen_id, so make a copy
new_last_seen_id = last_seen_id
try:
notifications = client.i_notifications(
# Fetch notifications we haven't seen yet. This option is a bit
# tempermental, sometimes it'll include since_id, sometimes it
# won't. We need to keep track of what notifications we've
# already processed.
since_id=last_seen_id,
# Let misskey handle the filtering
include_types=['mention', 'reply'],
# And handle the batch size while we're at it
limit=NOTIFICATION_BATCH_SIZE
)
# No notifications. Wait the poll period.
if not notifications:
return False
# Iterate oldest to newest
for notification in notifications:
try:
# Skip if we've processed already
notif_id = notification.get('id', '')
if notif_id <= last_seen_id:
continue
# Update new_last_seen_id and process
new_last_seen_id = notif_id
process_notification(client, notification)
except Exception as e:
print(f'An exception has occured while processing a \
notification: {e}')
print(traceback.format_exc())
# If we got as many notifications as we requested, there are probably
# more in the queue
return len(notifications) == NOTIFICATION_BATCH_SIZE
except MisskeyAPIException as e:
print(f'An exception has occured while reading notifications: {e}\n')
print(traceback.format_exc())
finally:
# Quality jank right here, but finally lets us update the last_seen_id
# even if we hit an exception or return early
if new_last_seen_id > last_seen_id:
set_config('last_seen_notif_id', new_last_seen_id)
return False

View file

@ -1,29 +1,58 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu, Moon, VD15, and contributors.
# 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/.
import re
from typing import Dict, Any
import misskey
import config
from response import generate_response
from fediverse_factory import get_fediverse_service
from fediverse_types import FediverseNotification, NotificationType, Visibility
from custom_types import ParsedNotification
def parse_notification(notification: FediverseNotification, fediverse_service=None):
'''Parses any notifications received by the bot and sends any commands to
generate_response()'''
def parse_notification(
notification: Dict[str, Any],
client: misskey.Misskey) -> ParsedNotification | None:
'''Parses any notifications received by the bot'''
if fediverse_service is None:
fediverse_service = get_fediverse_service(config.INSTANCE_TYPE)
# We get the type of notification to filter the ones that we actually want
# to parse
if notification.type not in (NotificationType.MENTION, NotificationType.REPLY):
return # Ignore anything that isn't a mention
# Return early if no post attached
if not notification.post:
return
# We want the visibility to be related to the type that was received (so if
# people don't want to dump a bunch of notes on home they don't have to)
if notification.post.visibility != Visibility.SPECIFIED:
visibility = Visibility.HOME
else:
visibility = Visibility.SPECIFIED
# Get the full Activitypub ID of the user
user = notification.get("user", {})
username = user.get("username", "unknown")
host = user.get("host")
# Local users may not have a hostname attached
full_user = f"@{username}" if not host else f"@{username}@{host}"
full_user = notification.user.full_handle
note_obj = notification.get("note", {})
note_text = note_obj.get("text")
note_id = note_obj.get("id")
note_text = notification.post.text
note_id = notification.post.id
note = note_text.strip().lower() if note_text else ""
# Split words into tokens
parts = note.split()
# Check for both short and fully-qualified name mentions
username_variants = [
@ -31,22 +60,41 @@ def parse_notification(
f'@{config.USER.split("@")[1]}'
]
# Make sure the notification text explicitly mentions the bot
if not any(variant in note for variant in username_variants):
# Notifs must consist of the initial mention and at least one other token
if len(parts) <= 1:
return None
# Find command and arguments after the mention
# Removes all mentions
# regex = mentions that start with @ and may contain @domain
cleaned_text = re.sub(r"@\w+(?:@\S+)?", "", note).strip()
parts = cleaned_text.split()
# Make sure the first token is a mention to the bot
if not parts[0] in username_variants:
return None
command = parts[0].lower() if parts else None
arguments = parts[1:] if len(parts) > 1 else []
command = parts[1].lower()
arguments = parts[2:] if len(parts) > 2 else []
return {
# Create ParsedNotification object for the new response system
parsed_notification: ParsedNotification = {
'author': full_user,
'command': command,
'arguments': arguments,
'note_obj': note_obj
'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:
return
# Handle attachment URLs (convert to file IDs if needed)
file_ids = response['attachment_urls'] if response['attachment_urls'] else None
fediverse_service.create_post(
text=response['message'],
reply_to_id=note_id,
visibility=visibility,
file_ids=file_ids
#visible_user_ids=[] #todo: write actual visible users ids so pleromers can use the bot privately
)

204
bot/pleroma_service.py Normal file
View file

@ -0,0 +1,204 @@
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Moon and contributors.
# 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/.
from mastodon import Mastodon
from typing import List, Optional, Dict, Any, Union, BinaryIO
import io
import filetype
from fediverse_service import FediverseService
from fediverse_types import (
FediverseNotification, FediversePost, FediverseUser, FediverseFile,
NotificationType, Visibility
)
import config
class PleromaService(FediverseService):
"""Pleroma implementation of FediverseService using Mastodon.py"""
def __init__(self):
self.client = Mastodon(
access_token=config.KEY,
api_base_url=config.INSTANCE
)
def _convert_mastodon_user(self, user_data: Dict[str, Any]) -> FediverseUser:
"""Convert Mastodon/Pleroma user data to FediverseUser"""
acct = user_data.get("acct", "")
if "@" in acct:
username, host = acct.split("@", 1)
else:
username = acct
host = None
return FediverseUser(
id=str(user_data.get("id", "")),
username=username,
host=host,
display_name=user_data.get("display_name")
)
def _convert_mastodon_file(self, file_data: Dict[str, Any]) -> FediverseFile:
"""Convert Mastodon/Pleroma media attachment to FediverseFile"""
return FediverseFile(
id=str(file_data.get("id", "")),
url=file_data.get("url", ""),
type=file_data.get("type"),
name=file_data.get("description")
)
def _convert_mastodon_visibility(self, visibility: str) -> Visibility:
"""Convert Mastodon/Pleroma visibility to our enum"""
visibility_map = {
"public": Visibility.PUBLIC,
"unlisted": Visibility.UNLISTED,
"private": Visibility.FOLLOWERS,
"direct": Visibility.DIRECT
}
return visibility_map.get(visibility, Visibility.PUBLIC)
def _convert_to_mastodon_visibility(self, visibility: Visibility) -> str:
"""Convert our visibility enum to Mastodon/Pleroma visibility"""
visibility_map = {
Visibility.PUBLIC: "public",
Visibility.UNLISTED: "unlisted",
Visibility.HOME: "unlisted", # Map home to unlisted for Pleroma
Visibility.FOLLOWERS: "private",
Visibility.SPECIFIED: "direct", # Map specified to direct for Pleroma
Visibility.DIRECT: "direct"
}
return visibility_map.get(visibility, "public")
def _convert_mastodon_notification_type(self, notif_type: str) -> NotificationType:
"""Convert Mastodon/Pleroma notification type to our enum"""
type_map = {
"mention": NotificationType.MENTION,
"follow": NotificationType.FOLLOW,
"favourite": NotificationType.FAVOURITE,
"reblog": NotificationType.REBLOG,
"poll": NotificationType.POLL
}
return type_map.get(notif_type, NotificationType.OTHER)
def _convert_mastodon_status(self, status_data: Dict[str, Any]) -> FediversePost:
"""Convert Mastodon/Pleroma status data to FediversePost"""
files = []
if status_data.get("media_attachments"):
files = [self._convert_mastodon_file(f) for f in status_data["media_attachments"]]
# Extract plain text from HTML content
content = status_data.get("content", "")
# Basic HTML stripping - in production you might want to use a proper HTML parser
import re
plain_text = re.sub(r'<[^>]+>', '', content) if content else None
return FediversePost(
id=str(status_data.get("id", "")),
text=plain_text,
user=self._convert_mastodon_user(status_data.get("account", {})),
visibility=self._convert_mastodon_visibility(status_data.get("visibility", "public")),
created_at=status_data.get("created_at"),
files=files,
reply_to_id=str(status_data["in_reply_to_id"]) if status_data.get("in_reply_to_id") else None
)
def _convert_mastodon_notification(self, notification_data: Dict[str, Any]) -> FediverseNotification:
"""Convert Mastodon/Pleroma notification data to FediverseNotification"""
post = None
if notification_data.get("status"):
post = self._convert_mastodon_status(notification_data["status"])
return FediverseNotification(
id=str(notification_data.get("id", "")),
type=self._convert_mastodon_notification_type(notification_data.get("type", "")),
user=self._convert_mastodon_user(notification_data.get("account", {})),
post=post,
created_at=notification_data.get("created_at")
)
def get_notifications(self, since_id: Optional[str] = None) -> List[FediverseNotification]:
"""Get notifications from Pleroma instance"""
params = {}
if since_id:
params["since_id"] = since_id
notifications = self.client.notifications(**params)
return [self._convert_mastodon_notification(notif) for notif in notifications]
def create_post(
self,
text: str,
reply_to_id: Optional[str] = None,
visibility: Visibility = Visibility.HOME,
file_ids: Optional[List[str]] = None,
visible_user_ids: Optional[List[str]] = None
) -> str:
"""Create a post on Pleroma instance"""
params = {
"status": text,
"visibility": self._convert_to_mastodon_visibility(visibility)
}
if reply_to_id:
params["in_reply_to_id"] = reply_to_id
if file_ids:
params["media_ids"] = file_ids
# Note: Pleroma/Mastodon doesn't have direct equivalent to visible_user_ids
# For direct messages, you typically mention users in the status text
response = self.client.status_post(**params)
return str(response.get("id", ""))
def get_post_by_id(self, post_id: str) -> Optional[FediversePost]:
"""Get a specific post by ID from Pleroma instance"""
try:
status = self.client.status(post_id)
return self._convert_mastodon_status(status)
except Exception:
return None
def upload_file(self, file_data: Union[BinaryIO, bytes], filename: Optional[str] = None) -> FediverseFile:
"""Upload a file to Pleroma instance"""
try:
# Convert file_data to bytes for MIME detection
if hasattr(file_data, 'read'):
# Check if we can seek back
try:
current_pos = file_data.tell()
file_bytes = file_data.read()
file_data.seek(current_pos)
file_data = io.BytesIO(file_bytes)
except (io.UnsupportedOperation, OSError):
# Non-seekable stream, already read all data
file_data = io.BytesIO(file_bytes)
else:
file_bytes = file_data
file_data = io.BytesIO(file_bytes)
# Use filetype library for robust MIME detection
kind = filetype.guess(file_bytes)
if kind is not None:
mime_type = kind.mime
else:
# Fallback to image/jpeg if detection fails
mime_type = 'image/jpeg'
media = self.client.media_post(file_data, mime_type=mime_type, description=filename)
return self._convert_mastodon_file(media)
except Exception as e:
raise RuntimeError(f"Failed to upload file to Pleroma: {e}") from e

View file

@ -1,22 +1,38 @@
#Kemoverse - a gacha-style bot for the Fediverse.
#Copyright © 2025 Waifu, VD15 and contributors.
#
#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/.
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
import db_utils as db
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'''
user_id = get_player(author)
'''Determines whether the user can roll, then pulls a random card'''
user_id = db.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
'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)
date = db.get_last_rolled_at(user_id)
# No date means it's users first roll
if date:
@ -45,39 +61,43 @@ def do_roll(author: str) -> BotResponse:
'attachment_urls': None
}
character = get_random_character()
card = db.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'])
stars = '⭐️' * character['rarity']
db.insert_pull(user_id, card['id'])
stars = '⭐️' * card['rarity']
return {
'message': f'{author} 🎲 Congrats! You rolled {stars} \
**{character['name']}**\nShe\'s all yours now~ 💖✨',
'attachment_urls': [character['image_url']]
**{card['name']}**\nShe\'s all yours now~ 💖✨',
'attachment_urls': [card['image_url']]
}
def do_signup(author: str) -> BotResponse:
'''Registers a new user if they havent signed up yet.'''
user_id = get_player(author)
user_id = db.get_player(author)
if user_id:
return {
'message':f'{author} 👀 Youre already signed up! Let the rolling begin~ 🎲',
'message': f'{author} 👀 Youre already signed up! Let the rolling \
begin~ 🎲',
'attachment_urls': None
}
new_user_id = insert_player(author)
new_user_id = db.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,22 +111,22 @@ 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
}
if len(arguments) != 2:
if not(len(arguments) in (2,5)):
return {
'message': f'{author} Please specify the following attributes \
in order: name, rarity',
in order: name, rarity. Optionally add [power, charm, wit].',
'attachment_urls': None
}
@ -116,20 +136,35 @@ in order: name, rarity',
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(
if len(arguments) == 2:
pass
else:
if not all(arg.isnumeric() for arg in arguments[2:]):
return {
'message': f'{author} Invalid attributes: power, charm and \
wit must be numbers.',
'attachment_urls': None
}
card_id, file_id = add_card(
name=arguments[0],
rarity=int(arguments[1]),
image_url=image_url,
power=int(arguments[2]),
charm=int(arguments[3]),
wit=int(arguments[4])
)
return {
'message': f'{author} Added {arguments[0]}, ID {card_id}.',
'attachment_urls': [file_id]
}
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,59 +172,145 @@ 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 \
- `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
'message': f'{author} Here\'s what I can do:\n\
- `roll` Pulls a random card.\n\
- `create <name> <rarity>` Creates a card using a given image.\n\
- `signup` Registers your account.\n\
- `delete_account` Deletes your account.\n\
- `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'
'message': f'{author} ⚠️ This will permanently delete your account \
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)
def confirm_delete(author: str) -> BotResponse:
db.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_whitelist(author: str, args: list[str]) -> BotResponse:
if len(args) == 0:
return {
'message': f'{author} Please specify an instance to whitelist',
'attachment_urls': None
}
if db.add_to_whitelist(args[0]):
return {
'message': f'{author} Whitelisted instance: {args[0]}',
'attachment_urls': None
}
else:
return {
'message': f'{author} Instance already whitelisted: {args[0]}',
'attachment_urls': None
}
def do_unwhitelist(author: str, args: list[str]) -> BotResponse:
if len(args) == 0:
return {
'message': f'{author} Please specify an instance to remove from \
the whitelist',
'attachment_urls': None
}
if db.remove_from_whitelist(args[0]):
return {
'message': f'{author} Unwhitelisted instance: {args[0]}',
'attachment_urls': None
}
else:
return {
'message': f'{author} Instance not whitelisted: {args[0]}',
'attachment_urls': None
}
def do_ban(author: str, args: list[str]) -> BotResponse:
if len(args) == 0:
return {
'message': f'{author} Please specify a user to ban',
'attachment_urls': None
}
if db.is_player_administrator(args[0]):
return {
'message': f'{author} Cannot ban other administrators.',
'attachment_urls': None
}
if db.ban_player(args[0]):
# Delete banned player's account
db.delete_player(args[0])
return {
'message': f'{author} 🔨 **BONK!** Get banned, {args[0]}!',
'attachment_urls': None
}
else:
return {
'message': f'{author} Player is already banned: {args[0]}',
'attachment_urls': None
}
def do_unban(author: str, args: list[str]) -> BotResponse:
if len(args) == 0:
return {
'message': f'{author} Please specify a user to unban',
'attachment_urls': None
}
if db.unban_player(args[0]):
return {
'message': f'{author} Player unbanned: {args[0]}!',
'attachment_urls': None
}
else:
return {
'message': f'{author} Player was not banned: {args[0]}',
'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)
player_id = db.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 +321,29 @@ 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 _:
pass
# if not user_is_administrator:
return res
# Commands beyond this point require the user to be an administrator
if not db.is_player_administrator(author):
return res
# Admin commands
match command:
case 'whitelist':
res = do_whitelist(author, notification['arguments'])
case 'unwhitelist':
res = do_unwhitelist(author, notification['arguments'])
case 'ban':
res = do_ban(author, notification['arguments'])
case 'unban':
res = do_unban(author, notification['arguments'])
case _:
pass
# Administrator commands go here
return res

49
contributing.md Normal file
View file

@ -0,0 +1,49 @@
<!--
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/.
-->
# Contributing to Kemoverse
Contributions are welcome with the following requirements:
## Licensing
- All contributions must be licensed under the **AGPLv3-or-later** or a compatible license.
- If you include code from another project (e.g., MIT Expat), please **list the license and copyright holders** clearly.
- If your contribution introduces code under a different license, you **must clarify this**, so the project can remain license-compliant.
## Attribution
- Please **add your name** to the license header of any file where youve made a **nontrivial change**.
- Nontrivial changes include:
- New features
- Logic changes
- Major refactoring or structure changes
- Not: typo fixes or simple reformatting
## Commit Messages (optional but appreciated)
- Try to write clear, descriptive commit messages.
## Communication
- If you're planning a major change or feature, please open an issue or contact the maintainers first.
- This helps avoid duplicated work and makes collaboration easier.
---
Thank you for helping grow the Kemoverse 💫

View file

@ -1,3 +1,19 @@
# 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/.
import subprocess
import os
from watchdog.observers import Observer

View file

@ -1,3 +1,20 @@
<!--
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 Waifu and contributors.
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/.
-->
# 🎲 Kemoverse Documentation
Welcome to the developer documentation for **Kemoverse**, a gacha trading card game in the Fediverse!

View file

@ -1,3 +1,20 @@
<!--
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 Waifu, VD15 and contributors.
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/.
-->
## 🧪 Installation

View file

@ -1,3 +1,20 @@
<!--
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 Waifu and contributors.
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/.
-->
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.
---

View file

@ -1,10 +1,37 @@
; Kemoverse - a gacha-style bot for the Fediverse.
; Copyright © 2025 Waifu and VD15
; 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/.
; Rename me to config.ini and put your values in here
[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
; Instance type - either "misskey" or "pleroma"
InstanceType = misskey
; Web server port (default: 5000)
WebPort = 5000
; Web server bind address (default: 127.0.0.1, set to 0.0.0.0 to listen on all interfaces)
BindAddress = 127.0.0.1
; Whether to lmit access to the bot via an instance whitelist
; The whitelist can be adjusted via the application
UseWhitelist = False
[gacha]
; Number of seconds players have to wait between rolls
@ -32,4 +59,3 @@ User = @bot@example.tld
; API key for the bot
; Generate one by going to Settings > API > Generate access token
Token = abcdefghijklmnopqrstuvwxyz012345

View file

@ -1,3 +1,20 @@
/*
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 VD15 and contributors.
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 IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,

View file

@ -1 +1,18 @@
/*
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 VD15 and contributors.
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/.
*/
INSERT OR IGNORE INTO config VALUES ("last_seen_notif_id", 0);

View file

@ -1 +1,19 @@
/*
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 Waifu and contributors.
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/.
*/
ALTER TABLE characters DROP COLUMN weight;

View file

@ -0,0 +1,22 @@
/*
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 VD15 and contributors.
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/.
*/
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;

View file

@ -0,0 +1,19 @@
/*
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 VD15
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/.
*/
ALTER TABLE players ADD COLUMN is_administrator BOOLEAN NOT NULL DEFAULT 0;

View file

@ -0,0 +1,25 @@
/*
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 VD15
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 IF NOT EXISTS instance_whitelist (
tld TEXT UNIQUE PRIMARY KEY
);
CREATE TABLE IF NOT EXISTS banned_players (
handle TEXT UNIQUE PRIMARY KEY
);

View file

@ -0,0 +1,26 @@
/*
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/.
*/
ALTER TABLE cards
ADD COLUMN power INTEGER NOT NULL DEFAULT 0;
ALTER TABLE cards
ADD COLUMN charm INTEGER NOT NULL DEFAULT 0;
ALTER TABLE cards
ADD COLUMN wit INTEGER NOT NULL DEFAULT 0;

View file

@ -1,6 +1,24 @@
<!--
Kemoverse - a gacha-style bot for the Fediverse.
Copyright © 2025 Waifu and contributors.
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/.
-->
# 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. Currently designed for use with Misskey. 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>
@ -47,13 +65,13 @@ A gacha-style bot for the Fediverse built with Python. Users can roll for charac
- 🌐 Web app to generate cards from images
### 🌍 Fediverse Support
✅ Anyone from the fediverse can play, but the server only works using a Misskey instance. Want to rewrite the program in Elixir for Pleroma? Let us know!
✅ Anyone from the fediverse can play! The bot supports both Misskey and Pleroma instances through configurable backends.
## 🗃️ Tech Stack
- Python (3.12+)
- SQLite
- Fediverse API integration (via Misskey endpoints)
- Fediverse API integration (Misskey and Pleroma support)
- Flask
- Modular DB design for extensibility
@ -61,16 +79,42 @@ 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.
## 📝 License
Unless stated otherwise, this repository is:
**Copyright © 2025 Waifu and contributors**
**Licensed under the GNU Affero General Public License v3.0 or later (AGPL-3.0-or-later)**
---
### 🛠️ What this means for you:
- You are free to **use**, **modify**, and **redistribute** the code, as long as you preserve the same license.
- If you run a modified version of this software as part of a service (e.g., a website or bot), you must also **share the source code** of your modifications with users.
A copy of the license should also be included in this repository.
If not, you can always find it at [gnu.org/licenses](https://www.gnu.org/licenses/).
---
The AGPL exists to **protect user freedom**, especially in networked and server-side software. If you enhance or build upon this project, please help the community by sharing your changes too.
Unless explicitly stated otherwise, **all files in this repository are covered by the AGPL v3.0 or any later version**.
```mermaid
flowchart TD
subgraph Player Interaction
A1[Misskey bot]
A1[Fediverse bot]
A2[Web]
end
subgraph Misskey
B1[Misskey instance]
subgraph Fediverse
B1[Fediverse instance]
end
subgraph Bot
@ -78,7 +122,7 @@ flowchart TD
C2[Notification parser]
C3[Gacha roll logic]
C4[Database interface]
C5[Misskey API poster]
C5[Fediverse API poster]
end
subgraph Website

View file

@ -1,3 +1,19 @@
# 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/.
blinker==1.9.0
click==8.1.8
Flask==3.1.0
@ -6,3 +22,5 @@ Jinja2==3.1.6
MarkupSafe==3.0.2
Werkzeug==3.1.3
Misskey.py==4.1.0
Mastodon.py==1.8.1
filetype==1.2.0

View file

@ -1,3 +1,19 @@
#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/.
import sqlite3
import traceback
import os
@ -57,16 +73,14 @@ def perform_migration(cursor: sqlite3.Cursor, migration: tuple[int, str]) -> Non
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}')
print(f'Running in "{env}" mode')
config = ConfigParser()
config.read(config_path)
db_path = config['application']['DatabaseLocation']
@ -96,7 +110,6 @@ def main():
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

View file

@ -1,5 +1,21 @@
#!/bin/bash
# Kemoverse - a gacha-style bot for the Fediverse.
# Copyright © 2025 Waifu and VD15
# 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/.
# Navigate to the project directory (optional)
cd "$(dirname "$0")"

View file

@ -1,13 +1,34 @@
#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/.
import sqlite3
import sys
import os
# Add bot directory to path to import config
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'bot'))
import config
from flask import Flask, render_template, abort
from werkzeug.exceptions import HTTPException
app = Flask(__name__)
DB_PATH = "./gacha_game.db" # Adjust path if needed
def get_db_connection():
conn = sqlite3.connect(DB_PATH)
conn = sqlite3.connect(config.DB_PATH)
conn.row_factory = sqlite3.Row
return conn
@ -68,4 +89,4 @@ def submit_character():
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
app.run(host=config.BIND_ADDRESS, port=config.WEB_PORT, debug=True)

View file

@ -1,13 +1,30 @@
/*
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/.
*/
body {
font-family: "Segoe UI", Tahoma, Geneva, Verdana, sans-serif;
background-color: #f4f6fa;
background-color: #FAFAFA;
color: #333;
margin: 0;
padding: 0;
}
header {
background-color: #7289da;
background-color: #5aa02c;
color: white;
padding: 20px;
text-align: center;

View file

@ -1,3 +1,20 @@
<!--
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/.
-->
<!DOCTYPE html>
<html>
<head>

View file

@ -1,3 +1,20 @@
<!--
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/.
-->
{% extends "_base.html" %}
{% block title %}
{{ error.code }}

View file

@ -1,3 +1,20 @@
<!--
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/.
-->
{% extends "_base.html" %}
{% block content %}

View file

@ -1,3 +1,21 @@
<!--
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/.
-->
{% extends "_base.html" %}
{% block header %}

View file

@ -1,3 +1,20 @@
<!--
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/.
-->
{% extends "_base.html" %}
{% block content %}

View file

@ -1,3 +1,20 @@
<!--
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/.
-->
{% extends "_base.html" %}
{% block content %}
<div class="profile">