diff --git a/.gitignore b/.gitignore index e5543ec..a05fd62 100644 --- a/.gitignore +++ b/.gitignore @@ -186,4 +186,4 @@ gacha_game*.db gacha_game*.db.* config*.ini -.idea \ No newline at end of file +.idea diff --git a/.tool-versions b/.tool-versions new file mode 100644 index 0000000..45e4b9f --- /dev/null +++ b/.tool-versions @@ -0,0 +1 @@ +nodejs 23.4.0 diff --git a/bot/add_card.py b/bot/add_card.py index fcaab43..ffa1120 100644 --- a/bot/add_card.py +++ b/bot/add_card.py @@ -1,32 +1,25 @@ import requests -from misskey.exceptions import MisskeyAPIException -from client import client_connection -from db_utils import insert_card -from custom_types import Card +from fediverse_factory import get_fediverse_service +import db_utils from config import RARITY_TO_WEIGHT - -def add_card( - name: str, - rarity: int, - image_url: str) -> tuple[int, str]: - ''' - Adds a card to the database, uploading the image from a public URL to - the bot's Misskey Drive. +def add_card(name: str, rarity: int, weight: float, image_url: str) -> 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 (e.g., from - note['files'][i]['url']). + weight (float): Pull weight (e.g., 0.02). + image_url (str): Public URL of the image from the post. Returns: - tuple[int, str]: Card ID and bot's Drive file_id. + 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() @@ -40,25 +33,29 @@ def add_card( 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 + # 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}") - # Insert into database - card_id = insert_card( + # Upload to Fediverse instance + fediverse_service = get_fediverse_service() + 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, - RARITY_TO_WEIGHT[rarity], + float(weight), file_id - ) - return card_id, file_id + ) + + return card_id, file_id + + except Exception as e: + raise diff --git a/bot/bot_app.py b/bot/bot_app.py index ed2772b..dabc219 100644 --- a/bot/bot_app.py +++ b/bot/bot_app.py @@ -1,21 +1,55 @@ import time -import misskey as misskey -from client import client_connection -import db_utils as db - -from config import NOTIFICATION_POLL_INTERVAL -from notification import process_notifications - -if __name__ == '__main__': - # Initialize the Misskey client - client = client_connection() - # Connect to DB - db.connect() +import traceback +from notification import process_fediverse_notification +from db_utils import get_config, set_config, connect, setup_administrators +from fediverse_factory import get_fediverse_service +def stream_notifications(): + # Initialize database connection + connect() + # Setup default administrators - db.setup_administrators() + setup_administrators() + + # Initialize the Fediverse service + fediverse_service = get_fediverse_service() + + # Get the last seen notification ID from the database + last_seen_id = get_config("last_seen_notif_id") 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() diff --git a/bot/client.py b/bot/client.py deleted file mode 100644 index 57f9f4e..0000000 --- a/bot/client.py +++ /dev/null @@ -1,6 +0,0 @@ -import misskey -import config - - -def client_connection() -> misskey.Misskey: - return misskey.Misskey(address=config.INSTANCE, i=config.KEY) diff --git a/bot/config.py b/bot/config.py index 9737608..11541cd 100644 --- a/bot/config.py +++ b/bot/config.py @@ -41,16 +41,33 @@ config.read(get_config_file()) USER = config['credentials']['User'].lower() # 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() +# 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) + +# Trusted instances +trusted_instances_str = config['application'].get('TrustedInstances', '') +TRUSTED_INSTANCES = [instance.strip() for instance in trusted_instances_str.split(',') if instance.strip()] + # 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'] # Whether to enable the instance whitelist -USE_WHITELIST = config['application']['UseWhitelist'] +USE_WHITELIST = config['application'].getboolean('UseWhitelist', False) NOTIFICATION_POLL_INTERVAL = int(config['notification']['PollInterval']) NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize']) diff --git a/bot/fediverse_factory.py b/bot/fediverse_factory.py new file mode 100644 index 0000000..f2a9017 --- /dev/null +++ b/bot/fediverse_factory.py @@ -0,0 +1,31 @@ +from fediverse_service import FediverseService +from misskey_service import MisskeyService +from pleroma_service import PleromaService +import config + + +class FediverseServiceFactory: + """Factory for creating FediverseService implementations based on configuration""" + + @staticmethod + def create_service() -> FediverseService: + """ + Create a FediverseService implementation based on the configured instance type. + + Returns: + FediverseService implementation (MisskeyService or PleromaService) + + Raises: + ValueError: If the instance type is not supported + """ + if config.INSTANCE_TYPE == "misskey": + return MisskeyService() + elif config.INSTANCE_TYPE == "pleroma": + return PleromaService() + else: + raise ValueError(f"Unsupported instance type: {config.INSTANCE_TYPE}") + + +def get_fediverse_service() -> FediverseService: + """Convenience function to get a FediverseService instance""" + return FediverseServiceFactory.create_service() \ No newline at end of file diff --git a/bot/fediverse_service.py b/bot/fediverse_service.py new file mode 100644 index 0000000..1292d63 --- /dev/null +++ b/bot/fediverse_service.py @@ -0,0 +1,74 @@ +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 \ No newline at end of file diff --git a/bot/fediverse_types.py b/bot/fediverse_types.py new file mode 100644 index 0000000..c1580b6 --- /dev/null +++ b/bot/fediverse_types.py @@ -0,0 +1,73 @@ +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 \ No newline at end of file diff --git a/bot/misskey_service.py b/bot/misskey_service.py new file mode 100644 index 0000000..510b32b --- /dev/null +++ b/bot/misskey_service.py @@ -0,0 +1,156 @@ +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 \ No newline at end of file diff --git a/bot/notification.py b/bot/notification.py index deb8ec6..b69a496 100644 --- a/bot/notification.py +++ b/bot/notification.py @@ -1,45 +1,50 @@ import traceback from typing import Dict, Any -import misskey -from misskey.exceptions import MisskeyAPIException - from config import NOTIFICATION_BATCH_SIZE, USE_WHITELIST from parsing import parse_notification from db_utils import get_config, set_config, 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() + + # Get user and instance info + username = notification.user.username + host = notification.user.host instance = host if host else 'local' + # Check whitelist if 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 @@ -49,23 +54,22 @@ def process_notification( print(f'⚠️ Blocked notification from banned player: {author}') return - # Get the note Id to reply to - note_id = notification.get('note', {}).get('id') - # Get the response response: BotResponse | None = generate_response(parsed_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 ) diff --git a/bot/parsing.py b/bot/parsing.py index e1e8583..476e774 100644 --- a/bot/parsing.py +++ b/bot/parsing.py @@ -1,27 +1,38 @@ 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()''' + + if fediverse_service is None: + fediverse_service = get_fediverse_service() -def parse_notification( - notification: Dict[str, Any], - client: misskey.Misskey) -> ParsedNotification | None: - '''Parses any notifications received by the bot''' + # 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 @@ -44,9 +55,30 @@ def parse_notification( 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 + ) diff --git a/bot/pleroma_service.py b/bot/pleroma_service.py new file mode 100644 index 0000000..21320d1 --- /dev/null +++ b/bot/pleroma_service.py @@ -0,0 +1,188 @@ +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 \ No newline at end of file diff --git a/bot/response.py b/bot/response.py index b49de1e..ce7d937 100644 --- a/bot/response.py +++ b/bot/response.py @@ -107,10 +107,10 @@ dumbass.', 'attachment_urls': None } - if len(arguments) != 2: + if len(arguments) != 3: return { 'message': f'{author} Please specify the following attributes \ -in order: name, rarity', +in order: name, rarity, weight', 'attachment_urls': None } @@ -130,6 +130,7 @@ must be a decimal value between 0.0 and 1.0', card_id, file_id = add_card( name=arguments[0], rarity=int(arguments[1]), + weight=float(arguments[2]), image_url=image_url ) return { diff --git a/example_config.ini b/example_config.ini index 0ea2422..516d552 100644 --- a/example_config.ini +++ b/example_config.ini @@ -5,7 +5,14 @@ DefaultAdmins = ["@localadmin", "@remoteadmin@example.tld"] ; SQLite Database location DatabaseLocation = ./gacha_game.db -; Whether to lmit access to the bot via an instance whitelist +; Instance type - either "misskey" or "pleroma" +InstanceType = misskey +; Web server port (default: 5000) +WebPort = 5000 +; Comma-separated list of trusted fediverse instances (leave empty to allow only local users) +; Example: TrustedInstances = mastodon.social,misskey.io,pleroma.example.com +TrustedInstances = +; Whether to limit access to the bot via an instance whitelist ; The whitelist can be adjusted via the application UseWhitelist = False @@ -35,4 +42,3 @@ User = @bot@example.tld ; API key for the bot ; Generate one by going to Settings > API > Generate access token Token = abcdefghijklmnopqrstuvwxyz012345 - diff --git a/readme.md b/readme.md index 437b09f..6ae910c 100644 --- a/readme.md +++ b/readme.md @@ -1,6 +1,6 @@ # 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.

Fediverse Gacha Bot Logo

@@ -47,13 +47,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 @@ -65,12 +65,12 @@ The bot is meant to feel *light, fun, and competitive*. Mixing social, gacha and 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 +78,7 @@ flowchart TD C2[Notification parser] C3[Gacha roll logic] C4[Database interface] - C5[Misskey API poster] + C5[Fediverse API poster] end subgraph Website diff --git a/requirements.txt b/requirements.txt index 1db6f9b..4a749b2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,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 diff --git a/web/app.py b/web/app.py index 61ed38f..9837741 100644 --- a/web/app.py +++ b/web/app.py @@ -1,10 +1,16 @@ 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 +DB_PATH = config.DB_PATH def get_db_connection(): conn = sqlite3.connect(DB_PATH) @@ -68,4 +74,4 @@ def submit_character(): if __name__ == '__main__': - app.run(host='0.0.0.0', port=5000, debug=True) + app.run(host='0.0.0.0', port=config.WEB_PORT, debug=True)