Compare commits
10 commits
master
...
any-env-an
Author | SHA1 | Date | |
---|---|---|---|
|
24bfe88dc1 | ||
|
1a35750d0a | ||
|
d416ae1b2d | ||
|
7fd4d5db25 | ||
|
f4f847e577 | ||
|
2ef70801c7 | ||
|
fdf21b3f5f | ||
1984b67523 | |||
|
77d4fa13bb | ||
|
bd287b096a |
8 changed files with 81 additions and 24 deletions
|
@ -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
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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'])
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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],
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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
|
||||
|
||||
|
|
|
@ -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)
|
||||
|
|
Loading…
Add table
Reference in a new issue