#!/usr/bin/env python3 """ Turbo Twitter Bot Marketing System — Tweet Poster Reads pending tweets from tweets.db and posts them to Twitter via API v2 using tweepy. Each bot persona has its own API credentials stored in .env. Usage: python post.py # Post all pending tweets python post.py --dry-run # Show what would be posted without posting Setup: 1. Create a .env file in this directory with API keys for each bot: # Bot 1 — @TurboAlpha TWITTER_BOT1_API_KEY=your_api_key_here TWITTER_BOT1_API_SECRET=your_api_secret_here TWITTER_BOT1_ACCESS_TOKEN=your_access_token_here TWITTER_BOT1_ACCESS_SECRET=your_access_token_secret_here # Bot 2 — @5MinEdge TWITTER_BOT2_API_KEY=your_api_key_here TWITTER_BOT2_API_SECRET=your_api_secret_here TWITTER_BOT2_ACCESS_TOKEN=your_access_token_here TWITTER_BOT2_ACCESS_SECRET=your_access_token_secret_here # Bot 3 — @AlgoTurbo TWITTER_BOT3_API_KEY=your_api_key_here TWITTER_BOT3_API_SECRET=your_api_secret_here TWITTER_BOT3_ACCESS_TOKEN=your_access_token_here TWITTER_BOT3_ACCESS_SECRET=your_access_token_secret_here # Bot 4 — @BTCPulse5m TWITTER_BOT4_API_KEY=your_api_key_here TWITTER_BOT4_API_SECRET=your_api_secret_here TWITTER_BOT4_ACCESS_TOKEN=your_access_token_here TWITTER_BOT4_ACCESS_SECRET=your_access_token_secret_here # Bot 5 — @PredictoorDAO TWITTER_BOT5_API_KEY=your_api_key_here TWITTER_BOT5_API_SECRET=your_api_secret_here TWITTER_BOT5_ACCESS_TOKEN=your_access_token_here TWITTER_BOT5_ACCESS_SECRET=your_access_token_secret_here 2. Install dependencies: pip install tweepy python-dotenv pyyaml 3. Each bot needs a Twitter Developer account with OAuth 1.0a credentials and Elevated access for tweet posting via API v2. """ import argparse import os import sqlite3 import sys import time from datetime import datetime, timezone from pathlib import Path import yaml try: import tweepy except ImportError: tweepy = None try: from dotenv import load_dotenv except ImportError: load_dotenv = None # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- BASE_DIR = Path(__file__).resolve().parent CONFIG_PATH = BASE_DIR / "config.yaml" DB_PATH = BASE_DIR / "tweets.db" ENV_PATH = BASE_DIR / ".env" # Mapping from persona key (in DB) to env_prefix (in config) # This is built dynamically from config.yaml at runtime. def load_config(): with open(CONFIG_PATH, "r") as f: return yaml.safe_load(f) def get_db_connection(): if not DB_PATH.exists(): print(f"Error: Database not found at {DB_PATH}") print("Run 'python bots.py generate' first to create the database.") sys.exit(1) return sqlite3.connect(str(DB_PATH)) # --------------------------------------------------------------------------- # Twitter client factory # --------------------------------------------------------------------------- def build_client(env_prefix): """Build a tweepy Client for the given bot using env vars. Expected env vars: {env_prefix}_API_KEY {env_prefix}_API_SECRET {env_prefix}_ACCESS_TOKEN {env_prefix}_ACCESS_SECRET Returns a tweepy.Client or None if credentials are missing. """ if tweepy is None: print("Error: tweepy is not installed. Run: pip install tweepy") return None api_key = os.getenv(f"{env_prefix}_API_KEY") api_secret = os.getenv(f"{env_prefix}_API_SECRET") access_token = os.getenv(f"{env_prefix}_ACCESS_TOKEN") access_secret = os.getenv(f"{env_prefix}_ACCESS_SECRET") if not all([api_key, api_secret, access_token, access_secret]): print(f" Warning: Missing credentials for {env_prefix}. Skipping.") return None return tweepy.Client( consumer_key=api_key, consumer_secret=api_secret, access_token=access_token, access_token_secret=access_secret, ) # --------------------------------------------------------------------------- # Posting logic # --------------------------------------------------------------------------- def mark_posted(conn, tweet_id): now = datetime.now(timezone.utc).isoformat() conn.execute( "UPDATE tweets SET status = 'posted', posted_at = ? WHERE id = ?", (now, tweet_id), ) conn.commit() def mark_skipped(conn, tweet_id, reason=""): now = datetime.now(timezone.utc).isoformat() conn.execute( "UPDATE tweets SET status = 'skipped', posted_at = ? WHERE id = ?", (now, tweet_id), ) conn.commit() def post_tweet(client, content): """Post a tweet using the tweepy v2 client. Returns (success: bool, error_message: str or None). """ try: response = client.create_tweet(text=content) return True, None except tweepy.TweepyException as e: return False, str(e) def post_all_pending(config, dry_run=False): """Read all pending tweets and post them via the appropriate bot account.""" # Load .env if available if load_dotenv is not None and ENV_PATH.exists(): load_dotenv(ENV_PATH) elif load_dotenv is None: print("Warning: python-dotenv not installed. Reading env vars from shell environment.") # Build persona -> env_prefix mapping persona_env = {} for key, persona in config["personas"].items(): persona_env[key] = persona["env_prefix"] conn = get_db_connection() cur = conn.execute( "SELECT id, persona, content, style, mentions_turbo " "FROM tweets WHERE status = 'pending' ORDER BY created_at" ) pending = cur.fetchall() if not pending: print("No pending tweets to post.") conn.close() return print(f"\nFound {len(pending)} pending tweet(s).\n") # Cache clients per persona to avoid rebuilding clients = {} posted_count = 0 skipped_count = 0 for tweet_id, persona, content, style, mentions_turbo in pending: handle = config["personas"].get(persona, {}).get("handle", persona) env_prefix = persona_env.get(persona) if not env_prefix: print(f" #{tweet_id} {handle} — unknown persona, skipping") mark_skipped(conn, tweet_id, "unknown persona") skipped_count += 1 continue if dry_run: turbo_flag = " [TURBO]" if mentions_turbo else "" print(f" [DRY RUN] #{tweet_id} {handle} ({style}{turbo_flag})") print(f" {content}") print(f" ({len(content)} chars)\n") continue # Get or build client if persona not in clients: clients[persona] = build_client(env_prefix) client = clients[persona] if client is None: mark_skipped(conn, tweet_id, "missing credentials") skipped_count += 1 continue # Post the tweet print(f" Posting #{tweet_id} for {handle}...", end=" ") success, error = post_tweet(client, content) if success: mark_posted(conn, tweet_id) posted_count += 1 print("done.") else: mark_skipped(conn, tweet_id, error) skipped_count += 1 print(f"failed: {error}") # Rate limit: wait 2 seconds between posts to avoid hitting Twitter limits time.sleep(2) conn.close() if not dry_run: print(f"\nResults: {posted_count} posted, {skipped_count} skipped.\n") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser( description="Turbo Twitter Bot — Tweet Poster" ) parser.add_argument( "--dry-run", action="store_true", help="Show pending tweets without posting them", ) args = parser.parse_args() config = load_config() post_all_pending(config, dry_run=args.dry_run) if __name__ == "__main__": main()