bluesky_bot.py

#!/usr/bin/env python3
"""
Bluesky Bot - Generate tweets using fine-tuned Qwen3-8B and post to Bluesky.

Usage:
    python bluesky_bot.py generate          # Generate tweet without posting (preview)
    python bluesky_bot.py post              # Generate and post a tweet
    python bluesky_bot.py schedule --interval 4  # Run scheduled posting every N hours
"""

import argparse
import logging
import os
import random
import time
from pathlib import Path

from dotenv import load_dotenv

# Set up logging to desktop file
LOG_FILE = Path.home() / "Desktop" / "bskybot.log"
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler(),  # Also print to console
    ],
)
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Prompts matching the training data format for variety
GENERATION_PROMPTS = [
    "Write a tweet in your authentic voice. Lean chaotic but personable.",
]

# Model configuration - 4-bit model with combined Twitter+Bluesky training
MODEL_PATH = "lmstudio-community/Qwen3-8B-MLX-4bit"
ADAPTER_PATH = Path(__file__).parent / "adapters_qwen3_4bit"

# Generation parameters
GENERATION_CONFIG = {
    "temp": 0.8,
    "top_p": 0.9,
    "max_tokens": 280,  # Bluesky limit is 300 graphemes
}


def load_model():
    """Load the fine-tuned model with LoRA adapters."""
    from mlx_lm import load

    logger.info(f"Loading model: {MODEL_PATH}")
    logger.info(f"With adapters: {ADAPTER_PATH}")

    model, tokenizer = load(
        MODEL_PATH,
        adapter_path=str(ADAPTER_PATH)
    )

    logger.info("Model loaded successfully!")
    return model, tokenizer


def generate_tweet(model, tokenizer, prompt: str = None) -> str:
    """Generate a tweet using the fine-tuned model."""
    from mlx_lm import generate
    from mlx_lm.sample_utils import make_sampler

    if prompt is None:
        prompt = random.choice(GENERATION_PROMPTS)

    # Format as chat message matching training format
    messages = [{"role": "user", "content": prompt}]

    # Apply chat template
    formatted_prompt = tokenizer.apply_chat_template(
        messages,
        tokenize=False,
        add_generation_prompt=True
    )

    logger.info(f"Using prompt: {prompt}")

    # Create sampler with temperature and top_p
    sampler = make_sampler(
        temp=GENERATION_CONFIG["temp"],
        top_p=GENERATION_CONFIG["top_p"],
    )

    # Generate
    response = generate(
        model,
        tokenizer,
        prompt=formatted_prompt,
        sampler=sampler,
        max_tokens=GENERATION_CONFIG["max_tokens"],
    )

    # Clean up response - remove any thinking tags if present
    tweet = response.strip()

    # Handle Qwen3's thinking mode if enabled
    if "<think>" in tweet:
        # Extract content after </think> tag
        if "</think>" in tweet:
            tweet = tweet.split("</think>")[-1].strip()
        else:
            # Remove incomplete thinking
            tweet = tweet.split("<think>")[0].strip()

    return tweet


def post_to_bluesky(text: str) -> dict:
    """Post a tweet to Bluesky."""
    from atproto import Client

    handle = os.getenv("BLUESKY_HANDLE")
    app_password = os.getenv("BLUESKY_APP_PASSWORD")

    if not handle or not app_password:
        raise ValueError(
            "Missing Bluesky credentials. Set BLUESKY_HANDLE and BLUESKY_APP_PASSWORD "
            "in your .env file or environment variables."
        )

    logger.info(f"Logging in as: {handle}")

    client = Client()
    client.login(handle, app_password)

    logger.info("Posting to Bluesky...")
    response = client.send_post(text=text)

    logger.info("Posted successfully!")
    return response


def cmd_generate(args):
    """Generate a tweet without posting (preview mode)."""
    model, tokenizer = load_model()
    tweet = generate_tweet(model, tokenizer, args.prompt)

    logger.info("=" * 50)
    logger.info("GENERATED TWEET:")
    logger.info("=" * 50)
    logger.info(tweet)
    logger.info("=" * 50)
    logger.info(f"Length: {len(tweet)} characters")

    return tweet


def cmd_post(args):
    """Generate and post a tweet."""
    model, tokenizer = load_model()
    tweet = generate_tweet(model, tokenizer, args.prompt)

    logger.info("=" * 50)
    logger.info("GENERATED TWEET:")
    logger.info("=" * 50)
    logger.info(tweet)
    logger.info("=" * 50)
    logger.info(f"Length: {len(tweet)} characters")

    if args.dry_run:
        logger.info("[DRY RUN - Not posting]")
        return tweet

    response = post_to_bluesky(tweet)

    return response


def cmd_schedule(args):
    """Run scheduled posting with randomized intervals (~40 posts/day, 7am-midnight ET)."""
    from datetime import datetime
    import pytz

    eastern = pytz.timezone('America/New_York')

    # Posting window: 7am to midnight Eastern
    POSTING_START_HOUR = 7
    POSTING_END_HOUR = 24  # midnight

    def is_within_posting_hours():
        """Check if current time is within posting hours (7am-midnight ET)."""
        now_et = datetime.now(eastern)
        hour = now_et.hour
        return POSTING_START_HOUR <= hour < POSTING_END_HOUR

    def get_random_interval():
        """Get random interval between 15-40 minutes for ~40 posts/day."""
        return random.randint(15, 40)

    def minutes_until_posting_window():
        """Calculate minutes until 7am ET."""
        now_et = datetime.now(eastern)
        hour = now_et.hour
        minute = now_et.minute

        if POSTING_START_HOUR <= hour < POSTING_END_HOUR:
            return 0
        elif hour < POSTING_START_HOUR:
            return (POSTING_START_HOUR - hour) * 60 - minute
        else:
            return (24 - hour + POSTING_START_HOUR) * 60 - minute

    logger.info("Starting randomized posting schedule")
    logger.info(f"  - ~40 posts/day with 15-40 minute random intervals")
    logger.info(f"  - Active hours: {POSTING_START_HOUR}am - midnight ET")
    logger.info("Press Ctrl+C to stop")

    # Load model once
    model, tokenizer = load_model()

    while True:
        try:
            # Check if we're in posting hours
            if not is_within_posting_hours():
                wait_minutes = minutes_until_posting_window()
                now_et = datetime.now(eastern)
                logger.info(f"Outside posting hours ({now_et.strftime('%I:%M %p')} ET). "
                           f"Sleeping {wait_minutes} minutes until 7am ET.")
                time.sleep(wait_minutes * 60)
                continue

            # Generate and post
            now_et = datetime.now(eastern)
            logger.info(f"Posting at {now_et.strftime('%I:%M %p')} ET")

            try:
                tweet = generate_tweet(model, tokenizer)
                logger.info(f"Generated: {tweet[:60]}...")

                if not args.dry_run:
                    post_to_bluesky(tweet)
                    logger.info("Posted successfully!")
                else:
                    logger.info("[DRY RUN - Not posting]")

            except Exception as e:
                logger.error(f"Error posting: {e}")

            # Wait random interval before next post
            interval = get_random_interval()
            logger.info(f"Next post in ~{interval} minutes")
            time.sleep(interval * 60)

        except KeyboardInterrupt:
            logger.info("Scheduler stopped by user")
            break


def main():
    parser = argparse.ArgumentParser(
        description="Bluesky Bot - Generate and post tweets using fine-tuned Qwen3-8B"
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # Generate command
    gen_parser = subparsers.add_parser("generate", help="Generate a tweet without posting")
    gen_parser.add_argument(
        "--prompt", "-p",
        help="Custom prompt for generation (default: random from training prompts)"
    )
    gen_parser.set_defaults(func=cmd_generate)

    # Post command
    post_parser = subparsers.add_parser("post", help="Generate and post a tweet")
    post_parser.add_argument(
        "--prompt", "-p",
        help="Custom prompt for generation (default: random from training prompts)"
    )
    post_parser.add_argument(
        "--dry-run", "-n",
        action="store_true",
        help="Generate but don't actually post"
    )
    post_parser.set_defaults(func=cmd_post)

    # Schedule command
    sched_parser = subparsers.add_parser(
        "schedule",
        help="Run randomized posting (~40/day, 7am-midnight ET)"
    )
    sched_parser.add_argument(
        "--dry-run", "-n",
        action="store_true",
        help="Generate but don't actually post"
    )
    sched_parser.set_defaults(func=cmd_schedule)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    logger.info(f"Bluesky Bot starting - logging to {LOG_FILE}")
    main()