Twitter Integration
FlavumHive's Twitter integration provides robust automation capabilities for posting tweets, managing threads, and engaging with the Twitter community while maintaining authentic personality-driven interactions.
Overview
The Twitter integration uses Selenium for browser automation and includes:
- Tweet posting
- Thread creation
- Reply management
- Rate limiting
- Analytics tracking
Setup
Prerequisites
- Twitter account
- Chrome browser
- Environment variables
- Platform configuration
Environment Configuration
# Required in .env
TWITTER_USERNAME="your_username"
TWITTER_PASSWORD="your_password"
TWITTER_EMAIL="your_email"
TWITTER_2FA_SECRET="optional_2fa_secret" # If 2FA enabled
TWITTER_DRY_RUN="false" # Set to true for testing
Platform Configuration
{
"platforms": {
"twitter": {
"enabled": true,
"personality": {
"active": "crypto_researcher",
"settings": {
"add_signature": true,
"auto_reply": true
}
},
"rate_limits": {
"tweets_per_hour": 5,
"replies_per_hour": 10,
"min_delay_between_actions": 30
}
}
}
}
Core Features
1. Tweet Posting
async def post_tweet(self, content: str) -> bool:
"""
Post a single tweet with optional media.
Args:
content: Tweet text content
Returns:
bool: Success status
"""
2. Thread Creation
async def create_thread(self, tweets: List[str]) -> bool:
"""
Create a thread from multiple tweets.
Args:
tweets: List of tweet contents
Returns:
bool: Success status
"""
3. Reply Management
async def reply_to_tweet(self, tweet_id: str, content: str) -> bool:
"""
Reply to a specific tweet.
Args:
tweet_id: ID of tweet to reply to
content: Reply content
Returns:
bool: Success status
"""
Authentication
Login Process
async def login(self):
# Navigate to Twitter
await self.driver.get("https://twitter.com/login")
# Enter credentials
await self._enter_credentials()
# Handle 2FA if enabled
if self.two_factor_enabled:
await self._handle_2fa()
2FA Support
async def _handle_2fa(self):
if self.two_factor_secret:
code = self._generate_2fa_code()
await self._enter_2fa_code(code)
Rate Limiting
Configuration
{
"rate_limits": {
"tweets_per_hour": 5,
"replies_per_hour": 10,
"min_delay_between_actions": 30,
"max_daily_tweets": 100,
"thread_delay": 5
}
}
Implementation
class TwitterRateLimiter:
def __init__(self, config: dict):
self.limits = config["rate_limits"]
self.action_history = {}
async def can_perform_action(self, action_type: str) -> bool:
return await self._check_limits(action_type)
Error Handling
Common Errors
- Authentication failures
- Rate limit exceeded
- Content restrictions
- Network issues
Error Recovery
async def handle_error(self, error: Exception):
if isinstance(error, RateLimitError):
await self._handle_rate_limit()
elif isinstance(error, AuthError):
await self._handle_auth_error()
else:
await self._handle_general_error(error)
Monitoring
Metrics Tracking
class TwitterMetrics:
def __init__(self):
self.tweet_count = 0
self.reply_count = 0
self.engagement_rate = 0.0
self.error_count = 0
Health Checks
async def health_check(self):
return {
"status": "healthy" if self.is_logged_in else "error",
"rate_limit_status": await self.rate_limiter.status(),
"recent_errors": self.recent_errors,
"uptime": self.uptime
}
Best Practices
1. Content Guidelines
- Follow Twitter's rules
- Maintain personality consistency
- Use appropriate formatting
- Include relevant hashtags
2. Rate Limiting
- Start with conservative limits
- Monitor for restrictions
- Implement backoff strategy
- Track daily limits
3. Error Management
- Implement retry logic
- Log all errors
- Monitor rate limits
- Handle auth issues
4. Performance
- Optimize Selenium usage
- Manage browser resources
- Cache when possible
- Clean up sessions
Advanced Features
1. Media Handling
async def upload_media(self, media_path: str) -> str:
"""
Upload media to Twitter.
Args:
media_path: Path to media file
Returns:
str: Media ID
"""
2. Analytics Integration
async def get_tweet_analytics(self, tweet_id: str) -> dict:
"""
Get engagement metrics for a tweet.
Args:
tweet_id: Tweet to analyze
Returns:
dict: Analytics data
"""
3. Scheduled Posting
async def schedule_tweet(self, content: str, timestamp: int):
"""
Schedule a tweet for later posting.
Args:
content: Tweet content
timestamp: When to post
"""
Troubleshooting
Common Issues
-
Login Failures
# Check credentials
await self.verify_credentials()
# Clear cookies and retry
await self.clear_session()
await self.login() -
Rate Limiting
# Check current limits
limits = await self.rate_limiter.status()
# Implement backoff
await self.backoff_strategy.execute() -
Content Issues
# Verify content
await self.content_validator.check(content)
# Adjust if needed
content = await self.content_adjuster.fix(content)
Security Considerations
1. Credential Management
- Use environment variables
- Encrypt sensitive data
- Rotate credentials regularly
- Monitor for suspicious activity
2. Session Management
- Clear sessions properly
- Handle cookies securely
- Implement timeouts
- Monitor for unauthorized access
3. Content Security
- Validate all content
- Check for malicious links
- Monitor for spam patterns
- Implement content filters