baseline fedi abstraction stuff.
This commit is contained in:
parent
a693a2189d
commit
bf5bc2212c
9 changed files with 500 additions and 0 deletions
1
.tool-versions
Normal file
1
.tool-versions
Normal file
|
@ -0,0 +1 @@
|
|||
nodejs 23.4.0
|
31
CLAUDE.md
Normal file
31
CLAUDE.md
Normal file
|
@ -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).
|
|
@ -14,6 +14,16 @@ INSTANCE = config['application']['InstanceUrl']
|
|||
# SQLite Database location
|
||||
DB_PATH = config['application']['DatabaseLocation']
|
||||
|
||||
# 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
|
||||
|
|
31
bot/fediverse_factory.py
Normal file
31
bot/fediverse_factory.py
Normal file
|
@ -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()
|
57
bot/fediverse_service.py
Normal file
57
bot/fediverse_service.py
Normal file
|
@ -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
|
73
bot/fediverse_types.py
Normal file
73
bot/fediverse_types.py
Normal file
|
@ -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
|
141
bot/misskey_service.py
Normal file
141
bot/misskey_service.py
Normal file
|
@ -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
|
155
bot/pleroma_service.py
Normal file
155
bot/pleroma_service.py
Normal file
|
@ -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
|
|
@ -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
|
||||
|
|
Loading…
Add table
Reference in a new issue