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/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..2e60baf --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,31 @@ +# Kemoverse - Fediverse Gacha Bot + +A Python-based gacha-style bot for the Fediverse that interfaces with either Misskey or Pleroma instances. + +## Project Overview +- **Language**: Python 3.11+ +- **Framework**: Flask for web interface +- **Database**: SQLite +- **Fediverse Support**: Configurable for Misskey or Pleroma instances +- **API Integration**: Uses Misskey.py library for Fediverse API communication + +## Key Components +- `bot/` - Core bot functionality including gacha mechanics, character management, and API interactions +- `web/` - Flask web application for user interface and card management +- `db.py` - Database utilities and schema management +- `config.ini` - Instance configuration (API keys, instance URL, admin users) + +## Configuration +The bot is configured via `config.ini` to connect to either a Misskey or Pleroma Fediverse instance. Key settings include: +- Instance URL and API credentials +- Bot user account details +- Administrator user permissions +- Database location + +## Development Commands +- Install dependencies: `pip install -r requirements.txt` +- Run bot: `python dev_runner.py` +- Start web interface: `python web/app.py` + +## Testing +Run tests with appropriate Python testing framework (check for pytest, unittest, or similar in project). \ No newline at end of file diff --git a/bot/config.py b/bot/config.py index 227f949..5b5848f 100644 --- a/bot/config.py +++ b/bot/config.py @@ -38,9 +38,21 @@ 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 + +# Extra stuff for control of the bot + # TODO: move this to db # Fedi handles in the traditional 'user@domain.tld' style, allows these users # to use extra admin exclusive commands with the bot 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..2d12d89 --- /dev/null +++ b/bot/fediverse_service.py @@ -0,0 +1,57 @@ +from abc import ABC, abstractmethod +from typing import List, Optional +from fediverse_types import FediverseNotification, FediversePost, 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 \ 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..845ed27 --- /dev/null +++ b/bot/misskey_service.py @@ -0,0 +1,141 @@ +import misskey +from typing import List, Optional, Dict, Any +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 = {} + if since_id: + params["sinceId"] = 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["replyId"] = reply_to_id + + if file_ids: + params["fileIds"] = file_ids + + if visible_user_ids and visibility == Visibility.SPECIFIED: + params["visibleUserIds"] = 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 \ No newline at end of file diff --git a/bot/pleroma_service.py b/bot/pleroma_service.py new file mode 100644 index 0000000..1e53cd7 --- /dev/null +++ b/bot/pleroma_service.py @@ -0,0 +1,155 @@ +from mastodon import Mastodon +from typing import List, Optional, Dict, Any +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 \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 1db6f9b..3d237b4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ Jinja2==3.1.6 MarkupSafe==3.0.2 Werkzeug==3.1.3 Misskey.py==4.1.0 +Mastodon.py==1.8.1