Platform Handlers API Reference
This document provides detailed API documentation for FlavumHive's platform handlers, including base classes and platform-specific implementations.
Base Platform Handler
The BasePlatformHandler class provides the foundation for all platform-specific handlers.
Class Definition
from abc import ABC, abstractmethod
from typing import Dict, List, Optional
class BasePlatformHandler(ABC):
"""
Abstract base class for platform handlers.
Attributes:
config (Dict): Platform configuration
rate_limiter (RateLimiter): Rate limiting controller
metrics (MetricsTracker): Metrics tracking instance
logger (Logger): Logger instance
"""
def __init__(self, config: Dict):
self.config = config
self.rate_limiter = None
self.metrics = None
self.logger = None
Core Methods
Initialize
@abstractmethod
async def initialize(self) -> bool:
"""
Initialize the platform handler.
Returns:
bool: Success status
"""
pass
Post Content
@abstractmethod
async def post_content(self, content: str, **kwargs) -> bool:
"""
Post content to the platform.
Args:
content: Content to post
**kwargs: Platform-specific arguments
Returns:
bool: Success status
"""
pass
Handle Interaction
@abstractmethod
async def handle_interaction(self, interaction: Dict) -> bool:
"""
Handle platform interaction.
Args:
interaction: Interaction data
Returns:
bool: Success status
"""
pass
Twitter Handler
The TwitterHandler class implements Twitter-specific functionality.
Class Definition
class TwitterHandler(BasePlatformHandler):
"""
Twitter platform handler implementation.
Attributes:
driver (WebDriver): Selenium WebDriver instance
is_logged_in (bool): Login status
session_data (Dict): Current session data
"""
Methods
Post Tweet
async def post_tweet(
self,
content: str,
media: Optional[List[str]] = None,
reply_to: Optional[str] = None
) -> bool:
"""
Post a tweet.
Args:
content: Tweet content
media: Optional list of media paths
reply_to: Optional tweet ID to reply to
Returns:
bool: Success status
Raises:
TwitterAuthError: If not authenticated
RateLimitError: If rate limit exceeded
"""
Create Thread
async def create_thread(
self,
tweets: List[str],
media: Optional[List[str]] = None
) -> bool:
"""
Create a tweet thread.
Args:
tweets: List of tweet contents
media: Optional list of media paths
Returns:
bool: Success status
"""
Monitor Timeline
async def monitor_timeline(
self,
keywords: List[str],
callback: callable
) -> None:
"""
Monitor timeline for keywords.
Args:
keywords: Keywords to monitor
callback: Callback function for matches
"""
Reddit Handler
The RedditHandler class implements Reddit-specific functionality.
Class Definition
class RedditHandler(BasePlatformHandler):
"""
Reddit platform handler implementation.
Attributes:
reddit (praw.Reddit): PRAW Reddit instance
subreddits (List[str]): Monitored subreddits
is_authenticated (bool): Authentication status
"""
Methods
Create Post
async def create_post(
self,
subreddit: str,
title: str,
content: str,
flair: Optional[str] = None
) -> bool:
"""
Create a Reddit post.
Args:
subreddit: Target subreddit
title: Post title
content: Post content
flair: Optional post flair
Returns:
bool: Success status
Raises:
RedditAuthError: If not authenticated
SubredditError: If subreddit is invalid
"""
Add Comment
async def add_comment(
self,
post_id: str,
content: str,
parent_id: Optional[str] = None
) -> bool:
"""
Add a comment to a post.
Args:
post_id: Target post ID
content: Comment content
parent_id: Optional parent comment ID
Returns:
bool: Success status
"""
Monitor Subreddit
async def monitor_subreddit(
self,
subreddit: str,
keywords: List[str],
callback: callable
) -> None:
"""
Monitor subreddit for keywords.
Args:
subreddit: Subreddit to monitor
keywords: Keywords to track
callback: Callback function for matches
"""
Utility Classes
Rate Limiter
class RateLimiter:
"""
Rate limiting implementation.
Attributes:
limits (Dict): Rate limit configuration
history (Dict): Action history
"""
async def can_perform_action(self, action: str) -> bool:
"""Check if action is allowed."""
pass
async def record_action(self, action: str) -> None:
"""Record an action."""
pass
Metrics Tracker
class MetricsTracker:
"""
Platform metrics tracking.
Attributes:
metrics (Dict): Current metrics
history (List): Historical data
"""
async def record_metric(self, name: str, value: any) -> None:
"""Record a metric."""
pass
async def get_metrics(self) -> Dict:
"""Get current metrics."""
pass
Error Classes
Base Errors
class PlatformError(Exception):
"""Base class for platform errors."""
pass
class AuthenticationError(PlatformError):
"""Authentication related errors."""
pass
class RateLimitError(PlatformError):
"""Rate limit related errors."""
pass
Platform-Specific Errors
class TwitterError(PlatformError):
"""Twitter specific errors."""
pass
class RedditError(PlatformError):
"""Reddit specific errors."""
pass
Usage Examples
Twitter Example
# Initialize Twitter handler
twitter = TwitterHandler(config)
await twitter.initialize()
# Post a tweet
success = await twitter.post_tweet(
content="Hello, #Twitter!",
media=["image.jpg"]
)
# Create a thread
tweets = ["First tweet", "Second tweet", "Third tweet"]
success = await twitter.create_thread(tweets)
Reddit Example
# Initialize Reddit handler
reddit = RedditHandler(config)
await reddit.initialize()
# Create a post
success = await reddit.create_post(
subreddit="test",
title="Test Post",
content="Hello, Reddit!",
flair="Discussion"
)
# Monitor subreddit
async def callback(post):
print(f"Found matching post: {post.title}")
await reddit.monitor_subreddit(
subreddit="test",
keywords=["python", "bot"],
callback=callback
)