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