Initial commit: Bifrost accessible fediverse client
- Full ActivityPub support for Pleroma, GoToSocial, and Mastodon - Screen reader optimized interface with PySide6 - Timeline switching with tabs and keyboard shortcuts (Ctrl+1-4) - Threaded conversation navigation with expand/collapse - Cross-platform desktop notifications via plyer - Customizable sound pack system with audio feedback - Complete keyboard navigation and accessibility features - XDG Base Directory compliant configuration - Multiple account support with OAuth authentication 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Storm Dragon
232
src/activitypub/client.py
Normal file
232
src/activitypub/client.py
Normal file
@@ -0,0 +1,232 @@
|
||||
"""
|
||||
ActivityPub client for communicating with fediverse servers
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
from typing import Dict, List, Optional, Any
|
||||
from urllib.parse import urljoin
|
||||
from datetime import datetime
|
||||
|
||||
from models.post import Post
|
||||
from models.user import User
|
||||
|
||||
|
||||
class ActivityPubClient:
|
||||
"""Main ActivityPub client for fediverse communication"""
|
||||
|
||||
def __init__(self, instance_url: str, access_token: Optional[str] = None):
|
||||
self.instance_url = instance_url.rstrip('/')
|
||||
self.access_token = access_token
|
||||
self.session = requests.Session()
|
||||
|
||||
# Set up headers
|
||||
self.session.headers.update({
|
||||
'User-Agent': 'Bifrost/1.0.0 (Accessible Fediverse Client)',
|
||||
'Accept': 'application/json',
|
||||
'Content-Type': 'application/json'
|
||||
})
|
||||
|
||||
if access_token:
|
||||
self.session.headers['Authorization'] = f'Bearer {access_token}'
|
||||
|
||||
def _make_request(self, method: str, endpoint: str, params: Optional[Dict] = None,
|
||||
data: Optional[Dict] = None, files: Optional[Dict] = None) -> Dict:
|
||||
"""Make an authenticated request to the API"""
|
||||
url = urljoin(self.instance_url, endpoint)
|
||||
|
||||
try:
|
||||
if method.upper() == 'GET':
|
||||
response = self.session.get(url, params=params, timeout=30)
|
||||
elif method.upper() == 'POST':
|
||||
if files:
|
||||
# For file uploads, don't set Content-Type header
|
||||
headers = {k: v for k, v in self.session.headers.items() if k != 'Content-Type'}
|
||||
response = self.session.post(url, data=data, files=files, headers=headers, timeout=30)
|
||||
else:
|
||||
response = self.session.post(url, json=data, timeout=30)
|
||||
elif method.upper() == 'PUT':
|
||||
response = self.session.put(url, json=data, timeout=30)
|
||||
elif method.upper() == 'DELETE':
|
||||
response = self.session.delete(url, timeout=30)
|
||||
else:
|
||||
raise ValueError(f"Unsupported HTTP method: {method}")
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
# Handle different success responses
|
||||
if response.status_code in [200, 201, 202]:
|
||||
if response.content:
|
||||
try:
|
||||
return response.json()
|
||||
except json.JSONDecodeError:
|
||||
# Some endpoints might return non-JSON on success
|
||||
return {"success": True, "status_code": response.status_code}
|
||||
return {"success": True, "status_code": response.status_code}
|
||||
|
||||
return {}
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise Exception(f"API request failed: {e}")
|
||||
except json.JSONDecodeError as e:
|
||||
raise Exception(f"Invalid JSON response: {e}")
|
||||
|
||||
def verify_credentials(self) -> Dict:
|
||||
"""Verify account credentials"""
|
||||
return self._make_request('GET', '/api/v1/accounts/verify_credentials')
|
||||
|
||||
def get_timeline(self, timeline_type: str = 'home', limit: int = 40,
|
||||
max_id: Optional[str] = None, since_id: Optional[str] = None) -> List[Dict]:
|
||||
"""Get timeline posts"""
|
||||
# Map timeline types to correct endpoints
|
||||
if timeline_type == 'local':
|
||||
endpoint = '/api/v1/timelines/public'
|
||||
params = {'limit': limit, 'local': 'true'}
|
||||
elif timeline_type == 'federated':
|
||||
endpoint = '/api/v1/timelines/public'
|
||||
params = {'limit': limit, 'local': 'false'}
|
||||
else:
|
||||
# home timeline
|
||||
endpoint = f'/api/v1/timelines/{timeline_type}'
|
||||
params = {'limit': limit}
|
||||
|
||||
if max_id:
|
||||
params['max_id'] = max_id
|
||||
if since_id:
|
||||
params['since_id'] = since_id
|
||||
|
||||
return self._make_request('GET', endpoint, params=params)
|
||||
|
||||
def get_status_context(self, status_id: str) -> Dict:
|
||||
"""Get context (replies/ancestors) for a status"""
|
||||
endpoint = f'/api/v1/statuses/{status_id}/context'
|
||||
return self._make_request('GET', endpoint)
|
||||
|
||||
def post_status(self, content: str, visibility: str = 'public',
|
||||
content_warning: Optional[str] = None,
|
||||
in_reply_to_id: Optional[str] = None,
|
||||
media_ids: Optional[List[str]] = None,
|
||||
content_type: str = 'text/plain') -> Dict:
|
||||
"""Post a new status"""
|
||||
data = {
|
||||
'status': content,
|
||||
'visibility': visibility
|
||||
}
|
||||
|
||||
# Add content type for instances that support it (Pleroma, GoToSocial)
|
||||
if content_type == 'text/markdown':
|
||||
data['content_type'] = 'text/markdown'
|
||||
|
||||
if content_warning:
|
||||
data['spoiler_text'] = content_warning
|
||||
if in_reply_to_id:
|
||||
data['in_reply_to_id'] = in_reply_to_id
|
||||
if media_ids:
|
||||
data['media_ids'] = media_ids
|
||||
|
||||
return self._make_request('POST', '/api/v1/statuses', data=data)
|
||||
|
||||
def delete_status(self, status_id: str) -> Dict:
|
||||
"""Delete a status"""
|
||||
endpoint = f'/api/v1/statuses/{status_id}'
|
||||
return self._make_request('DELETE', endpoint)
|
||||
|
||||
def favourite_status(self, status_id: str) -> Dict:
|
||||
"""Favourite a status"""
|
||||
endpoint = f'/api/v1/statuses/{status_id}/favourite'
|
||||
return self._make_request('POST', endpoint)
|
||||
|
||||
def unfavourite_status(self, status_id: str) -> Dict:
|
||||
"""Unfavourite a status"""
|
||||
endpoint = f'/api/v1/statuses/{status_id}/unfavourite'
|
||||
return self._make_request('POST', endpoint)
|
||||
|
||||
def reblog_status(self, status_id: str) -> Dict:
|
||||
"""Reblog/boost a status"""
|
||||
endpoint = f'/api/v1/statuses/{status_id}/reblog'
|
||||
return self._make_request('POST', endpoint)
|
||||
|
||||
def unreblog_status(self, status_id: str) -> Dict:
|
||||
"""Unreblog/unboost a status"""
|
||||
endpoint = f'/api/v1/statuses/{status_id}/unreblog'
|
||||
return self._make_request('POST', endpoint)
|
||||
|
||||
def get_notifications(self, limit: int = 20, max_id: Optional[str] = None,
|
||||
types: Optional[List[str]] = None) -> List[Dict]:
|
||||
"""Get notifications"""
|
||||
params = {'limit': limit}
|
||||
|
||||
if max_id:
|
||||
params['max_id'] = max_id
|
||||
if types:
|
||||
params['types[]'] = types
|
||||
|
||||
return self._make_request('GET', '/api/v1/notifications', params=params)
|
||||
|
||||
def get_account(self, account_id: str) -> Dict:
|
||||
"""Get account information"""
|
||||
endpoint = f'/api/v1/accounts/{account_id}'
|
||||
return self._make_request('GET', endpoint)
|
||||
|
||||
def follow_account(self, account_id: str) -> Dict:
|
||||
"""Follow an account"""
|
||||
endpoint = f'/api/v1/accounts/{account_id}/follow'
|
||||
return self._make_request('POST', endpoint)
|
||||
|
||||
def unfollow_account(self, account_id: str) -> Dict:
|
||||
"""Unfollow an account"""
|
||||
endpoint = f'/api/v1/accounts/{account_id}/unfollow'
|
||||
return self._make_request('POST', endpoint)
|
||||
|
||||
def search(self, query: str, account_id: Optional[str] = None,
|
||||
max_id: Optional[str] = None, min_id: Optional[str] = None,
|
||||
type_filter: Optional[str] = None, limit: int = 20) -> Dict:
|
||||
"""Search for content"""
|
||||
params = {
|
||||
'q': query,
|
||||
'limit': limit
|
||||
}
|
||||
|
||||
if account_id:
|
||||
params['account_id'] = account_id
|
||||
if max_id:
|
||||
params['max_id'] = max_id
|
||||
if min_id:
|
||||
params['min_id'] = min_id
|
||||
if type_filter:
|
||||
params['type'] = type_filter
|
||||
|
||||
return self._make_request('GET', '/api/v2/search', params=params)
|
||||
|
||||
def upload_media(self, file_path: str, description: Optional[str] = None) -> Dict:
|
||||
"""Upload a media file"""
|
||||
with open(file_path, 'rb') as f:
|
||||
files = {'file': f}
|
||||
data = {}
|
||||
if description:
|
||||
data['description'] = description
|
||||
|
||||
return self._make_request('POST', '/api/v1/media', data=data, files=files)
|
||||
|
||||
def get_instance_info(self) -> Dict:
|
||||
"""Get instance information"""
|
||||
return self._make_request('GET', '/api/v1/instance')
|
||||
|
||||
def get_custom_emojis(self) -> List[Dict]:
|
||||
"""Get custom emojis for this instance"""
|
||||
return self._make_request('GET', '/api/v1/custom_emojis')
|
||||
|
||||
|
||||
class AuthenticationError(Exception):
|
||||
"""Raised when authentication fails"""
|
||||
pass
|
||||
|
||||
|
||||
class RateLimitError(Exception):
|
||||
"""Raised when rate limit is exceeded"""
|
||||
pass
|
||||
|
||||
|
||||
class ServerError(Exception):
|
||||
"""Raised when server returns an error"""
|
||||
pass
|
||||
Reference in New Issue
Block a user