Compare commits
No commits in common. "34df5fc6eea2dddfb835200c2ff6cd8a2d01d181" and "a693a2189da6bafebab8f77328d08ef3d9f03398" have entirely different histories.
34df5fc6ee
...
a693a2189d
17 changed files with 63 additions and 643 deletions
|
@ -1 +0,0 @@
|
||||||
nodejs 23.4.0
|
|
31
CLAUDE.md
31
CLAUDE.md
|
@ -1,31 +0,0 @@
|
||||||
# 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).
|
|
|
@ -1,19 +1,20 @@
|
||||||
import requests
|
import requests
|
||||||
from fediverse_factory import get_fediverse_service
|
from misskey.exceptions import MisskeyAPIException
|
||||||
|
from client import client_connection
|
||||||
from db_utils import get_db_connection
|
from db_utils import get_db_connection
|
||||||
|
|
||||||
def add_character(name: str, rarity: int, weight: float, image_url: str) -> tuple[int, str]:
|
def add_character(name: str, rarity: int, weight: float, image_url: str) -> tuple[int, str]:
|
||||||
"""
|
"""
|
||||||
Adds a character to the database, uploading the image from a public URL to the Fediverse instance.
|
Adds a character to the database, uploading the image from a public URL to the bot's Misskey Drive.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
name (str): Character name.
|
name (str): Character name.
|
||||||
rarity (int): Character rarity (e.g., 1-5).
|
rarity (int): Character rarity (e.g., 1-5).
|
||||||
weight (float): Pull weight (e.g., 0.02).
|
weight (float): Pull weight (e.g., 0.02).
|
||||||
image_url (str): Public URL of the image from the post.
|
image_url (str): Public URL of the image from the post (e.g., from note['files'][i]['url']).
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
tuple[int, str]: Character ID and file_id.
|
tuple[int, str]: Character ID and bot's Drive file_id.
|
||||||
|
|
||||||
Raises:
|
Raises:
|
||||||
ValueError: If inputs are invalid.
|
ValueError: If inputs are invalid.
|
||||||
|
@ -35,13 +36,13 @@ def add_character(name: str, rarity: int, weight: float, image_url: str) -> tupl
|
||||||
if response.status_code != 200:
|
if response.status_code != 200:
|
||||||
raise RuntimeError(f"Failed to download image from {image_url}")
|
raise RuntimeError(f"Failed to download image from {image_url}")
|
||||||
|
|
||||||
# Upload to Fediverse instance
|
# Upload to bot's Drive
|
||||||
fediverse_service = get_fediverse_service()
|
mk = client_connection()
|
||||||
try:
|
try:
|
||||||
uploaded_file = fediverse_service.upload_file(response.raw)
|
media = mk.drive_files_create(response.raw)
|
||||||
file_id = uploaded_file.id
|
file_id = media["id"]
|
||||||
except RuntimeError as e:
|
except MisskeyAPIException as e:
|
||||||
raise RuntimeError(f"Failed to upload image: {e}") from e
|
raise RuntimeError(f"Failed to upload image to bot's Drive: {e}") from e
|
||||||
|
|
||||||
# Insert into database
|
# Insert into database
|
||||||
conn = get_db_connection()
|
conn = get_db_connection()
|
||||||
|
|
|
@ -1,11 +1,12 @@
|
||||||
import time
|
import time
|
||||||
import traceback
|
import traceback
|
||||||
|
import misskey
|
||||||
from parsing import parse_notification
|
from parsing import parse_notification
|
||||||
from db_utils import get_or_create_user, add_pull, get_config, set_config
|
from db_utils import get_or_create_user, add_pull, get_config, set_config
|
||||||
from fediverse_factory import get_fediverse_service
|
from client import client_connection
|
||||||
|
|
||||||
# Initialize the Fediverse service
|
# Initialize the Misskey client
|
||||||
fediverse_service = get_fediverse_service()
|
client = client_connection()
|
||||||
|
|
||||||
# Define your whitelist
|
# Define your whitelist
|
||||||
# TODO: move to config
|
# TODO: move to config
|
||||||
|
@ -18,34 +19,40 @@ def stream_notifications():
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
# Get notifications from the fediverse service
|
# May be able to mark notifications as read using misskey.py and
|
||||||
notifications = fediverse_service.get_notifications(since_id=last_seen_id)
|
# filter them out here. This function also takes a since_id we
|
||||||
|
# could use as well
|
||||||
|
notifications = client.i_notifications()
|
||||||
|
|
||||||
if notifications:
|
if notifications:
|
||||||
|
# Oldest to newest
|
||||||
|
notifications.reverse()
|
||||||
|
|
||||||
new_last_seen_id = last_seen_id
|
new_last_seen_id = last_seen_id
|
||||||
|
|
||||||
for notification in notifications:
|
for notification in notifications:
|
||||||
notif_id = notification.id
|
notif_id = notification.get("id")
|
||||||
|
|
||||||
# Skip old or same ID notifications
|
# Skip old or same ID notifications
|
||||||
if last_seen_id is not None and notif_id <= last_seen_id:
|
if last_seen_id is not None and notif_id <= last_seen_id:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
username = notification.user.username
|
user = notification.get("user", {})
|
||||||
host = notification.user.host
|
username = user.get("username", "unknown")
|
||||||
|
host = user.get("host") # None if local user
|
||||||
|
|
||||||
instance = host if host else "local"
|
instance = host if host else "local"
|
||||||
|
|
||||||
if instance in whitelisted_instances or instance == "local":
|
if instance in whitelisted_instances or instance == "local":
|
||||||
note = notification.post.text if notification.post else ""
|
note = notification.get("note", {}).get("text", "")
|
||||||
notif_type = notification.type.value
|
notif_type = notification.get("type", "unknown")
|
||||||
|
|
||||||
print(f"📨 [{notif_type}] from @{username}@{instance}")
|
print(f"📨 [{notif_type}] from @{username}@{instance}")
|
||||||
print(f"💬 {note}")
|
print(f"💬 {note}")
|
||||||
print("-" * 30)
|
print("-" * 30)
|
||||||
|
|
||||||
# 🧠 Send to the parser
|
# 🧠 Send to the parser
|
||||||
parse_notification(notification, fediverse_service)
|
parse_notification(notification,client)
|
||||||
|
|
||||||
else:
|
else:
|
||||||
print(f"⚠️ Blocked notification from untrusted instance: {host}")
|
print(f"⚠️ Blocked notification from untrusted instance: {host}")
|
||||||
|
|
|
@ -14,19 +14,6 @@ INSTANCE = config['application']['InstanceUrl']
|
||||||
# SQLite Database location
|
# SQLite Database location
|
||||||
DB_PATH = config['application']['DatabaseLocation']
|
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
|
|
||||||
|
|
||||||
# Web server port
|
|
||||||
WEB_PORT = config['application'].getint('WebPort', 5000)
|
|
||||||
|
|
||||||
# Extra stuff for control of the bot
|
# Extra stuff for control of the bot
|
||||||
|
|
||||||
# TODO: move this to db
|
# TODO: move this to db
|
||||||
|
|
|
@ -1,31 +0,0 @@
|
||||||
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()
|
|
|
@ -1,74 +0,0 @@
|
||||||
from abc import ABC, abstractmethod
|
|
||||||
from typing import List, Optional, Union, BinaryIO
|
|
||||||
from fediverse_types import FediverseNotification, FediversePost, FediverseFile, 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
|
|
||||||
|
|
||||||
@abstractmethod
|
|
||||||
def upload_file(self, file_data: Union[BinaryIO, bytes], filename: Optional[str] = None) -> FediverseFile:
|
|
||||||
"""
|
|
||||||
Upload a file to the Fediverse instance.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
file_data: File data as binary stream or bytes
|
|
||||||
filename: Optional filename for the uploaded file
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
FediverseFile object with ID, URL, and other metadata
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
RuntimeError: If file upload fails
|
|
||||||
"""
|
|
||||||
pass
|
|
|
@ -1,73 +0,0 @@
|
||||||
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
|
|
|
@ -1,7 +1,6 @@
|
||||||
import random
|
import random
|
||||||
from db_utils import get_or_create_user, add_pull, get_db_connection
|
from db_utils import get_or_create_user, add_pull, get_db_connection
|
||||||
from add_character import add_character
|
from add_character import add_character
|
||||||
from fediverse_types import FediversePost
|
|
||||||
|
|
||||||
def get_character():
|
def get_character():
|
||||||
''' Gets a random character from the database'''
|
''' Gets a random character from the database'''
|
||||||
|
@ -28,7 +27,7 @@ def is_float(val):
|
||||||
|
|
||||||
|
|
||||||
# TODO: See issue #3, separate command parsing from game logic.
|
# TODO: See issue #3, separate command parsing from game logic.
|
||||||
def gacha_response(command: str, full_user: str, arguments: list, post: FediversePost):
|
def gacha_response(command,full_user, arguments,note_obj):
|
||||||
'''Parses a given command with arguments, processes the game state and
|
'''Parses a given command with arguments, processes the game state and
|
||||||
returns a response'''
|
returns a response'''
|
||||||
|
|
||||||
|
@ -47,7 +46,7 @@ def gacha_response(command: str, full_user: str, arguments: list, post: Fedivers
|
||||||
|
|
||||||
if command == "create":
|
if command == "create":
|
||||||
# Example call from bot logic
|
# Example call from bot logic
|
||||||
image_url = post.files[0].url if post.files else None
|
image_url = note_obj.get("files", [{}])[0].get("url") if note_obj.get("files") else None
|
||||||
if not image_url:
|
if not image_url:
|
||||||
return "You need an image to create a character, dumbass."
|
return "You need an image to create a character, dumbass."
|
||||||
|
|
||||||
|
|
|
@ -1,153 +0,0 @@
|
||||||
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 = {}
|
|
||||||
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
|
|
||||||
|
|
||||||
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
|
|
|
@ -1,37 +1,34 @@
|
||||||
import random, re
|
import random, re
|
||||||
import config
|
import config
|
||||||
from gacha_response import gacha_response
|
from gacha_response import gacha_response
|
||||||
from fediverse_factory import get_fediverse_service
|
|
||||||
from fediverse_types import FediverseNotification, NotificationType, Visibility
|
|
||||||
|
|
||||||
def parse_notification(notification: FediverseNotification, fediverse_service=None):
|
def parse_notification(notification,client):
|
||||||
'''Parses any notifications received by the bot and sends any commands to
|
'''Oarses any notifications received by the bot and sends any commands to
|
||||||
gacha_response()'''
|
gacha_response()'''
|
||||||
|
|
||||||
if fediverse_service is None:
|
|
||||||
fediverse_service = get_fediverse_service()
|
|
||||||
|
|
||||||
# We get the type of notification to filter the ones that we actually want
|
# We get the type of notification to filter the ones that we actually want
|
||||||
# to parse
|
# to parse
|
||||||
if notification.type not in (NotificationType.MENTION, NotificationType.REPLY):
|
|
||||||
return # Ignore anything that isn't a mention
|
|
||||||
|
|
||||||
# Return early if no post attached
|
notif_type = notification.get("type")
|
||||||
if not notification.post:
|
if not notif_type in ('mention', 'reply'):
|
||||||
return
|
return # Ignore anything that isn't a mention
|
||||||
|
|
||||||
# We want the visibility to be related to the type that was received (so if
|
# We want the visibility to be related to the type that was received (so if
|
||||||
# people don't want to dump a bunch of notes on home they don't have to)
|
# people don't want to dump a bunch of notes on home they don't have to)
|
||||||
if notification.post.visibility != Visibility.SPECIFIED:
|
visibility = notification["note"]["visibility"]
|
||||||
visibility = Visibility.HOME
|
if visibility != "specified":
|
||||||
else:
|
visibility = "home"
|
||||||
visibility = Visibility.SPECIFIED
|
|
||||||
|
|
||||||
# Get the full Activitypub ID of the user
|
# Get the full Activitypub ID of the user
|
||||||
full_user = notification.user.full_handle
|
user = notification.get("user", {})
|
||||||
|
username = user.get("username", "unknown")
|
||||||
|
host = user.get("host")
|
||||||
|
# Local users may not have a hostname attached
|
||||||
|
full_user = f"@{username}" if not host else f"@{username}@{host}"
|
||||||
|
|
||||||
note_text = notification.post.text
|
note_obj = notification.get("note", {})
|
||||||
note_id = notification.post.id
|
note_text = note_obj.get("text")
|
||||||
|
note_id = note_obj.get("id")
|
||||||
|
|
||||||
note = note_text.strip().lower() if note_text else ""
|
note = note_text.strip().lower() if note_text else ""
|
||||||
|
|
||||||
|
@ -54,20 +51,20 @@ def parse_notification(notification: FediverseNotification, fediverse_service=No
|
||||||
arguments = parts[1:] if len(parts) > 1 else []
|
arguments = parts[1:] if len(parts) > 1 else []
|
||||||
|
|
||||||
# TODO: move response generation to a different function
|
# TODO: move response generation to a different function
|
||||||
response = gacha_response(command.lower(), full_user, arguments, notification.post)
|
response = gacha_response(command.lower(),full_user, arguments, note_obj)
|
||||||
if not response:
|
if not response:
|
||||||
return
|
return
|
||||||
|
|
||||||
if isinstance(response, str):
|
if isinstance(response, str):
|
||||||
fediverse_service.create_post(
|
client.notes_create(
|
||||||
text=response,
|
text=response,
|
||||||
reply_to_id=note_id,
|
reply_id=note_id,
|
||||||
visibility=visibility
|
visibility=visibility
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
fediverse_service.create_post(
|
client.notes_create(
|
||||||
text=response[0],
|
text=response[0],
|
||||||
reply_to_id=note_id,
|
reply_id=note_id,
|
||||||
visibility=visibility,
|
visibility=visibility,
|
||||||
file_ids=response[1]
|
file_ids=response[1]
|
||||||
#visible_user_ids=[] #todo: write actual visible users ids so pleromers can use the bot privately
|
#visible_user_ids=[] #todo: write actual visible users ids so pleromers can use the bot privately
|
||||||
|
|
|
@ -1,188 +0,0 @@
|
||||||
from mastodon import Mastodon
|
|
||||||
from typing import List, Optional, Dict, Any, Union, BinaryIO
|
|
||||||
import io
|
|
||||||
import filetype
|
|
||||||
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
|
|
||||||
|
|
||||||
def upload_file(self, file_data: Union[BinaryIO, bytes], filename: Optional[str] = None) -> FediverseFile:
|
|
||||||
"""Upload a file to Pleroma instance"""
|
|
||||||
try:
|
|
||||||
# Convert file_data to bytes for MIME detection
|
|
||||||
if hasattr(file_data, 'read'):
|
|
||||||
# Check if we can seek back
|
|
||||||
try:
|
|
||||||
current_pos = file_data.tell()
|
|
||||||
file_bytes = file_data.read()
|
|
||||||
file_data.seek(current_pos)
|
|
||||||
file_data = io.BytesIO(file_bytes)
|
|
||||||
except (io.UnsupportedOperation, OSError):
|
|
||||||
# Non-seekable stream, already read all data
|
|
||||||
file_data = io.BytesIO(file_bytes)
|
|
||||||
else:
|
|
||||||
file_bytes = file_data
|
|
||||||
file_data = io.BytesIO(file_bytes)
|
|
||||||
|
|
||||||
# Use filetype library for robust MIME detection
|
|
||||||
kind = filetype.guess(file_bytes)
|
|
||||||
if kind is not None:
|
|
||||||
mime_type = kind.mime
|
|
||||||
else:
|
|
||||||
# Fallback to image/jpeg if detection fails
|
|
||||||
mime_type = 'image/jpeg'
|
|
||||||
|
|
||||||
media = self.client.media_post(file_data, mime_type=mime_type, description=filename)
|
|
||||||
return self._convert_mastodon_file(media)
|
|
||||||
except Exception as e:
|
|
||||||
raise RuntimeError(f"Failed to upload file to Pleroma: {e}") from e
|
|
8
db.py
8
db.py
|
@ -1,13 +1,7 @@
|
||||||
import sqlite3
|
import sqlite3
|
||||||
import configparser
|
|
||||||
|
|
||||||
# Read config to get database location
|
|
||||||
config = configparser.ConfigParser()
|
|
||||||
config.read('config.ini')
|
|
||||||
|
|
||||||
# Connect to SQLite database (or create it if it doesn't exist)
|
# Connect to SQLite database (or create it if it doesn't exist)
|
||||||
db_path = config['application']['DatabaseLocation']
|
conn = sqlite3.connect('gacha_game.db')
|
||||||
conn = sqlite3.connect(db_path)
|
|
||||||
cursor = conn.cursor()
|
cursor = conn.cursor()
|
||||||
|
|
||||||
# Create tables
|
# Create tables
|
||||||
|
|
|
@ -10,14 +10,8 @@ ApiKey = abcdefghijklmnopqrstuvwxyz012345
|
||||||
; Fully qualified URL of the instance hosting the bot
|
; Fully qualified URL of the instance hosting the bot
|
||||||
InstanceUrl = http://example.tld
|
InstanceUrl = http://example.tld
|
||||||
|
|
||||||
; Instance type - either "misskey" or "pleroma"
|
|
||||||
InstanceType = misskey
|
|
||||||
|
|
||||||
; Comma separated list of fedi handles for any administrator users
|
; Comma separated list of fedi handles for any administrator users
|
||||||
DefaultAdmins = ['admin@example.tld']
|
DefaultAdmins = ['admin@example.tld']
|
||||||
|
|
||||||
; SQLite Database location
|
; SQLite Database location
|
||||||
DatabaseLocation = ./gacha_game.db
|
DatabaseLocation = ./gacha_game.db
|
||||||
|
|
||||||
; Web server port (default: 5000)
|
|
||||||
WebPort = 5000
|
|
14
readme.md
14
readme.md
|
@ -1,6 +1,6 @@
|
||||||
# Kemoverse
|
# Kemoverse
|
||||||
|
|
||||||
A gacha-style bot for the Fediverse built with Python. Users can roll for characters, trade, duel, and perhaps engage with popularity-based mechanics. Supports both Misskey and Pleroma instances. Name comes from Kemonomimi and Fediverse.
|
A gacha-style bot for the Fediverse built with Python. Users can roll for characters, trade, duel, and perhaps engage with popularity-based mechanics. Currently designed for use with Misskey. Name comes from Kemonomimi and Fediverse.
|
||||||
|
|
||||||

|

|
||||||
|
|
||||||
|
@ -35,13 +35,13 @@ A gacha-style bot for the Fediverse built with Python. Users can roll for charac
|
||||||
- 🌐 Web app to generate cards from images
|
- 🌐 Web app to generate cards from images
|
||||||
|
|
||||||
### 🌍 Fediverse Support
|
### 🌍 Fediverse Support
|
||||||
✅ Anyone from the fediverse can play! The bot supports both Misskey and Pleroma instances through configurable backends.
|
✅ Anyone from the fediverse can play, but the server only works using a Misskey instance. Want to rewrite the program in Elixir for Pleroma? Let us know!
|
||||||
|
|
||||||
## 🗃️ Tech Stack
|
## 🗃️ Tech Stack
|
||||||
|
|
||||||
- Python (3.11+)
|
- Python (3.11+)
|
||||||
- SQLite
|
- SQLite
|
||||||
- Fediverse API integration (Misskey and Pleroma support)
|
- Fediverse API integration (via Misskey endpoints)
|
||||||
- Flask
|
- Flask
|
||||||
- Modular DB design for extensibility
|
- Modular DB design for extensibility
|
||||||
|
|
||||||
|
@ -58,12 +58,12 @@ Instructions on installing dependencies, initializing the database, and running
|
||||||
flowchart TD
|
flowchart TD
|
||||||
|
|
||||||
subgraph Player Interaction
|
subgraph Player Interaction
|
||||||
A1[Fediverse bot]
|
A1[Misskey bot]
|
||||||
A2[Web]
|
A2[Web]
|
||||||
end
|
end
|
||||||
|
|
||||||
subgraph Fediverse
|
subgraph Misskey
|
||||||
B1[Fediverse instance]
|
B1[Misskey instance]
|
||||||
end
|
end
|
||||||
|
|
||||||
subgraph Bot
|
subgraph Bot
|
||||||
|
@ -71,7 +71,7 @@ flowchart TD
|
||||||
C2[Notification parser]
|
C2[Notification parser]
|
||||||
C3[Gacha roll logic]
|
C3[Gacha roll logic]
|
||||||
C4[Database interface]
|
C4[Database interface]
|
||||||
C5[Fediverse API poster]
|
C5[Misskey API poster]
|
||||||
end
|
end
|
||||||
|
|
||||||
subgraph Website
|
subgraph Website
|
||||||
|
|
|
@ -6,5 +6,3 @@ Jinja2==3.1.6
|
||||||
MarkupSafe==3.0.2
|
MarkupSafe==3.0.2
|
||||||
Werkzeug==3.1.3
|
Werkzeug==3.1.3
|
||||||
Misskey.py==4.1.0
|
Misskey.py==4.1.0
|
||||||
Mastodon.py==1.8.1
|
|
||||||
filetype==1.2.0
|
|
||||||
|
|
10
web/app.py
10
web/app.py
|
@ -1,14 +1,8 @@
|
||||||
from flask import Flask, render_template
|
from flask import Flask, render_template
|
||||||
import sqlite3
|
import sqlite3
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
|
|
||||||
# Add bot directory to path to import config
|
|
||||||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'bot'))
|
|
||||||
import config
|
|
||||||
|
|
||||||
app = Flask(__name__)
|
app = Flask(__name__)
|
||||||
DB_PATH = config.DB_PATH
|
DB_PATH = "./gacha_game.db" # Adjust path if needed
|
||||||
|
|
||||||
def get_db_connection():
|
def get_db_connection():
|
||||||
conn = sqlite3.connect(DB_PATH)
|
conn = sqlite3.connect(DB_PATH)
|
||||||
|
@ -62,4 +56,4 @@ def submit_character():
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
app.run(host='0.0.0.0', port=config.WEB_PORT, debug=True)
|
app.run(host='0.0.0.0', port=5000, debug=True)
|
||||||
|
|
Loading…
Add table
Reference in a new issue