kemoverse/bot/misskey_service.py

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=str(note_data.get("id", "")),  # Misskey IDs (aid/meid/ulid/objectid) are lexicographically sortable as strings
            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=str(note_data.get("replyId", "")) if note_data.get("replyId") else None
        )
    
    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=str(notification_data.get("id", "")),  # Misskey IDs are lexicographically sortable as strings
            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