Compare commits

...
Sign in to create a new pull request.

10 commits

Author SHA1 Message Date
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
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
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
8 changed files with 81 additions and 24 deletions

View file

@ -58,7 +58,6 @@ def add_card(
card_id = insert_card(
stripped_name,
rarity,
RARITY_TO_WEIGHT[rarity],
file_id
)
return card_id, file_id

View file

@ -3,7 +3,7 @@ import misskey as misskey
from client import client_connection
import db_utils as db
from config import NOTIFICATION_POLL_INTERVAL
from config import NOTIFICATION_POLL_INTERVAL, USE_WHITELIST
from notification import process_notifications
if __name__ == '__main__':
@ -15,6 +15,10 @@ if __name__ == '__main__':
# Setup default administrators
db.setup_administrators()
# 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):

View file

@ -1,6 +1,7 @@
'''Essentials for the bot to function'''
import configparser
import json
import re
from os import environ, path
@ -13,8 +14,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'
@ -23,6 +26,50 @@ def get_config_file() -> str:
return config_path
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."""
@ -38,19 +85,26 @@ 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
INSTANCE = config['credentials']['Instance'].lower()
# 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 = 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']['UseWhitelist']
USE_WHITELIST = config['application'].getboolean('UseWhitelist', True)
NOTIFICATION_POLL_INTERVAL = int(config['notification']['PollInterval'])
NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize'])

View file

@ -150,12 +150,11 @@ def is_player_administrator(username: str) -> bool:
def insert_card(
name: str, rarity: int, weight: float, file_id: str) -> int:
name: str, rarity: int, file_id: str) -> int:
'''Inserts a card'''
CURSOR.execute(
'INSERT INTO cards (name, rarity, weight, file_id) VALUES \
(?, ?, ?, ?)',
(name, rarity, weight, file_id)
'INSERT INTO cards (name, rarity, file_id) VALUES (?, ?, ?)',
(name, rarity, file_id)
)
card_id = CURSOR.lastrowid
return card_id if card_id else 0

View file

@ -120,12 +120,6 @@ 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
}
card_id, file_id = add_card(
name=arguments[0],

View file

@ -5,6 +5,11 @@
DefaultAdmins = ["@localadmin", "@remoteadmin@example.tld"]
; SQLite Database location
DatabaseLocation = ./gacha_game.db
; 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

View file

@ -57,16 +57,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 +94,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,10 +1,15 @@
import sqlite3
import sys
from pathlib import Path
# Add parent directory to Python path so we can import from bot/
sys.path.append(str(Path(__file__).parent.parent))
from bot.config import WEB_PORT, BIND_ADDRESS, DB_PATH
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)
@ -68,4 +73,4 @@ def submit_character():
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
app.run(host=BIND_ADDRESS, port=WEB_PORT, debug=True)