TwitterBotTemplate.py

import sqlite3
import json
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional, Any
import requests
import random
import logging
import os
import time
import threading

@dataclass
class Config:
    twitter: dict[str, str]  # Twitter API keys: consumer_key, consumer_secret, access_token, access_secret
    ollama: dict[str, str]  # Ollama config: host, model
    prompt_template: str = "Write a short tweet about: {data}"  # Template for LLM prompt
    source_url: Optional[str] = None  # Optional URL to fetch data from
    post_prob: float = 1.0  # Probability (0-1) to post each run
    db_path: str = "bot_state.db"  # SQLite database file path
    dry_run: bool = False  # If True, log actions without posting
    log_path: str = "bot.log"  # Log file path

def init_db(db_path: str):
    """Initialize SQLite database with state table if it doesn't exist.
    Args: db_path (str): Path to database file
    """
    conn = sqlite3.connect(db_path)
    conn.execute('''
        CREATE TABLE IF NOT EXISTS state (
            key TEXT PRIMARY KEY,
            value TEXT
        )
    ''')
    conn.commit()
    conn.close()

def get_last_check(db_path: str) -> Optional[datetime]:
    """Retrieve last run timestamp from database.
    Args: db_path (str): Database file path
    Returns: UTC datetime or None if not set
    """
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute('SELECT value FROM state WHERE key = ?', ('last_check',))
    row = cursor.fetchone()
    conn.close()
    if row:
        return datetime.fromisoformat(row[0]).replace(tzinfo=timezone.utc)
    return None

def update_last_check(db_path: str, timestamp: datetime):
    """Update last check timestamp in database.
    Args: db_path (str), timestamp (datetime): Current timestamp
    """
    conn = sqlite3.connect(db_path)
    conn.execute('INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)', ('last_check', timestamp.isoformat()))
    conn.commit()
    conn.close()

def load_config(path: str) -> Config:
    """Load and validate configuration from JSON file.
    Args: path (str): Path to config.json
    Returns: Config dataclass instance
    Raises: ValueError for missing required keys
    """
    with open(path, 'r') as f:
        data = json.load(f)
    # Validate required keys
    required = ['twitter', 'ollama']
    for key in required:
        if key not in data:
            raise ValueError(f"Missing required config key: {key}")
    twitter = data['twitter']
    for tkey in ['consumer_key', 'consumer_secret', 'access_token', 'access_secret']:
        if tkey not in twitter:
            raise ValueError(f"Missing twitter key: {tkey}")
    ollama = data['ollama']
    for okey in ['host', 'model']:
        if okey not in ollama:
            raise ValueError(f"Missing ollama key: {okey}")
    return Config(**data)

def fetch_data(url: str) -> str:
    """Fetch data from URL (JSON or text).
    Args: url (str)
    Returns: Data string
    """
    resp = requests.get(url)
    if resp.status_code == 200:
        try:
            data = resp.json()
            return json.dumps(data)
        except:
            return resp.text
    return ""

def generate_tweet(config: Config) -> str:
    """Generate a tweet using Ollama LLM.
    Args: config (Config)
    Returns: Tweet text (string)
    """
    data = ""
    if config.source_url:
        data = fetch_data(config.source_url)
    prompt = config.prompt_template.format(data=data)
    try:
        resp = requests.post(config.ollama['host'], json={
            "model": config.ollama['model'],
            "prompt": prompt,
            "stream": False
        })
        if resp.status_code == 200:
            result = resp.json()
            tweet = result.get('response', '').strip()
            if len(tweet) <= 280:
                return tweet
    except Exception as e:
        logging.error(f"Ollama failed: {e}")
    return "Hello from the bot!"

def post_tweet(text: str, config: Config) -> Optional[str]:
    """Post a tweet to Twitter using requests with OAuth1.
    Args: text (str), config (Config)
    Returns: Tweet ID string or None on failure
    Handles dry run mode
    """
    if config.dry_run:
        logging.info(f"Dry run: would post tweet: {text}")
        return "dry_run_id"
    try:
        from requests_oauthlib import OAuth1Session
    except ImportError as e:
        logging.error(f"OAuth1 import failed: {e}")
        return None
    oauth = OAuth1Session(
        config.twitter['consumer_key'],
        client_secret=config.twitter['consumer_secret'],
        resource_owner_key=config.twitter['access_token'],
        resource_owner_secret=config.twitter['access_secret']
    )
    try:
        resp = oauth.post('https://api.twitter.com/2/tweets', json={'text': text})
        if resp.status_code == 201:
            data = resp.json()
            return data['data']['id']
        else:
            logging.error(f"Tweet post failed: {resp.status_code} {resp.text}")
    except Exception as e:
        logging.error(f"Tweet post failed: {e}")
    return None

def acquire_lock(log_path: str) -> bool:
    """Acquire file lock to prevent concurrent runs.
    Args: log_path (str)
    Returns: True if lock acquired
    """
    lock_file = log_path + '.lock'
    try:
        with open(lock_file, 'x') as f:
            f.write(str(os.getpid()))
        return True
    except FileExistsError:
        return False

def release_lock(log_path: str):
    """Release file lock.
    Args: log_path (str)
    """
    lock_file = log_path + '.lock'
    try:
        os.remove(lock_file)
    except FileNotFoundError:
        pass

def main():
    """Main bot function - loads config, generates tweet, posts.
    Handles: Config loading, database init, tweet posting, dry run mode
    Uses file locking to prevent concurrent runs
    """
    config_path = 'config.json'
    try:
        config = load_config(config_path)
        logging.basicConfig(filename=config.log_path, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
        logging.info("Bot started")
        init_db(config.db_path)
        if not acquire_lock(config.log_path):
            logging.warning("Another instance running, exiting")
            return
        try:
            now = datetime.now(timezone.utc)
            if random.random() < config.post_prob:
                tweet = generate_tweet(config)
                tweet_id = post_tweet(tweet, config)
                if tweet_id:
                    logging.info(f"Posted tweet {tweet_id}")
                else:
                    logging.error("Failed to post tweet")
            update_last_check(config.db_path, now)
        finally:
            release_lock(config.log_path)
        logging.info("Bot finished")
    except Exception as e:
        logging.error(f"Main failed: {e}")
        print(f"Error: {e}")

if __name__ == "__main__":
    main()