waifu/kemoverse
waifu/kemoverse
3
2
Fork 3
Code Issues 13 Pull requests 4 Projects 1 Releases Packages Wiki Activity

completely untested refactor. bug fixes to come later.

Browse source
This commit is contained in:
Moon 2025-06-12 11:22:20 +09:00
parent f4499f7afb
commit fc3d638bf4
4 changed files with 48 additions and 52 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
bot/add_character.py
View file

@ -1,20 +1,19 @@
import requests
from misskey.exceptions import MisskeyAPIException
from client import client_connection
from fediverse_factory import get_fediverse_service
from db_utils import get_db_connection
def add_character(name: str, rarity: int, weight: float, 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.
Adds a character to the database, uploading the image from a public URL to the Fediverse instance.
Args:
name (str): Character name.
rarity (int): Character rarity (e.g., 1-5).
weight (float): Pull weight (e.g., 0.02).
image_url (str): Public URL of the image from the post (e.g., from note['files'][i]['url']).
image_url (str): Public URL of the image from the post.
Returns:
tuple[int, str]: Character ID and bot's Drive file_id.
tuple[int, str]: Character ID and file_id.
Raises:
ValueError: If inputs are invalid.
@ -36,13 +35,13 @@ def add_character(name: str, rarity: int, weight: float, image_url: str) -> tupl
if response.status_code != 200:
raise RuntimeError(f"Failed to download image from {image_url}")
# Upload to bot's Drive
mk = client_connection()
# Upload to Fediverse instance
fediverse_service = get_fediverse_service()
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
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
conn = get_db_connection()

29
bot/bot_app.py
View file

@ -1,12 +1,11 @@
import time
import traceback
import misskey
from parsing import parse_notification
from db_utils import get_or_create_user, add_pull, get_config, set_config
from client import client_connection
from fediverse_factory import get_fediverse_service
# Initialize the Misskey client
client = client_connection()
# Initialize the Fediverse service
fediverse_service = get_fediverse_service()
# Define your whitelist
# TODO: move to config
@ -19,40 +18,34 @@ def stream_notifications():
while True:
try:
# May be able to mark notifications as read using misskey.py and
# filter them out here. This function also takes a since_id we
# could use as well
notifications = client.i_notifications()
# Get notifications from the fediverse service
notifications = fediverse_service.get_notifications(since_id=last_seen_id)
if notifications:
# Oldest to newest
notifications.reverse()
new_last_seen_id = last_seen_id
for notification in notifications:
notif_id = notification.get("id")
notif_id = notification.id
# Skip old or same ID notifications
if last_seen_id is not None and notif_id <= last_seen_id:
continue
user = notification.get("user", {})
username = user.get("username", "unknown")
host = user.get("host") # None if local user
username = notification.user.username
host = notification.user.host
instance = host if host else "local"
if instance in whitelisted_instances or instance == "local":
note = notification.get("note", {}).get("text", "")
notif_type = notification.get("type", "unknown")
note = notification.post.text if notification.post else ""
notif_type = notification.type.value
print(f"📨 [{notif_type}] from @{username}@{instance}")
print(f"💬 {note}")
print("-" * 30)
# 🧠 Send to the parser
parse_notification(notification,client)
parse_notification(notification, fediverse_service)
else:
print(f"⚠️ Blocked notification from untrusted instance: {host}")

5
bot/gacha_response.py
View file

@ -1,6 +1,7 @@
import random
from db_utils import get_or_create_user, add_pull, get_db_connection
from add_character import add_character
from fediverse_types import FediversePost
def get_character():
''' Gets a random character from the database'''
@ -27,7 +28,7 @@ def is_float(val):
# TODO: See issue #3, separate command parsing from game logic.
def gacha_response(command,full_user, arguments,note_obj):
def gacha_response(command: str, full_user: str, arguments: list, post: FediversePost):
'''Parses a given command with arguments, processes the game state and
returns a response'''
@ -46,7 +47,7 @@ def gacha_response(command,full_user, arguments,note_obj):
if command == "create":
# Example call from bot logic
image_url = note_obj.get("files", [{}])[0].get("url") if note_obj.get("files") else None
image_url = post.files[0].url if post.files else None
if not image_url:
return "You need an image to create a character, dumbass."

45
bot/parsing.py
View file

@ -1,34 +1,37 @@
import random, re
import config
from gacha_response import gacha_response
from fediverse_factory import get_fediverse_service
from fediverse_types import FediverseNotification, NotificationType, Visibility
def parse_notification(notification,client):
'''Oarses any notifications received by the bot and sends any commands to
def parse_notification(notification: FediverseNotification, fediverse_service=None):
'''Parses any notifications received by the bot and sends any commands to
gacha_response()'''
if fediverse_service is None:
fediverse_service = get_fediverse_service()
# We get the type of notification to filter the ones that we actually want
# to parse
notif_type = notification.get("type")
if not notif_type in ('mention', 'reply'):
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)
visibility = notification["note"]["visibility"]
if visibility != "specified":
visibility = "home"
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 ""
@ -51,20 +54,20 @@ def parse_notification(notification,client):
arguments = parts[1:] if len(parts) > 1 else []
# TODO: move response generation to a different function
response = gacha_response(command.lower(),full_user, arguments, note_obj)
response = gacha_response(command.lower(), full_user, arguments, notification.post)
if not response:
return
if isinstance(response, str):
client.notes_create(
fediverse_service.create_post(
text=response,
reply_id=note_id,
reply_to_id=note_id,
visibility=visibility
)
else:
client.notes_create(
fediverse_service.create_post(
text=response[0],
reply_id=note_id,
reply_to_id=note_id,
visibility=visibility,
file_ids=response[1]
#visible_user_ids=[] #todo: write actual visible users ids so pleromers can use the bot privately