baseline fedi abstraction stuff.

.tool-versions

@@ -0,0 +1 @@
+nodejs 23.4.0

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).

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

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()

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

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

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

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

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