fetch_bluesky_posts.py

#!/usr/bin/env python3
"""
Fetch posts from a Bluesky account and convert to training format.

Usage:
    python fetch_bluesky_posts.py @bobbby.online -o bluesky_posts.jsonl
"""

import argparse
import json
import os
import random
from pathlib import Path

from atproto import Client
from dotenv import load_dotenv

load_dotenv()

# Prompts matching the training data format
GENERATION_PROMPTS = [
    "Write a tweet in your authentic voice.",
    "Share your thoughts.",
    "What's on your mind?",
    "Post something.",
    "Tweet something interesting.",
    "Share an observation.",
    "What would you say about this?",
    "Give me your take.",
    "Write something in your style.",
    "How would you respond to this?",
]


def fetch_all_posts(handle: str, limit: int = None) -> list[dict]:
    """Fetch all posts from a Bluesky account."""
    client = Client()

    # Login with credentials from .env
    bluesky_handle = os.getenv("BLUESKY_HANDLE")
    app_password = os.getenv("BLUESKY_APP_PASSWORD")

    if bluesky_handle and app_password:
        print(f"Logging in as: {bluesky_handle}")
        client.login(bluesky_handle, app_password)

    # Resolve handle to DID if needed
    if handle.startswith("@"):
        handle = handle[1:]

    print(f"Fetching posts from: {handle}")

    all_posts = []
    cursor = None

    while True:
        response = client.get_author_feed(
            actor=handle,
            filter="posts_no_replies",  # Only original posts, not replies
            limit=100,
            cursor=cursor,
        )

        for feed_view in response.feed:
            post = feed_view.post
            record = post.record

            # Skip reposts
            if hasattr(feed_view, 'reason') and feed_view.reason:
                continue

            # Get the text content
            text = getattr(record, 'text', None)
            if not text:
                continue

            # Skip very short posts
            if len(text.strip()) < 10:
                continue

            # Skip posts that are mostly URLs or mentions
            words = text.split()
            url_mention_count = sum(1 for w in words if w.startswith(('http', '@', 'https')))
            if url_mention_count > len(words) / 2:
                continue

            all_posts.append({
                "text": text.strip(),
                "created_at": record.created_at,
                "uri": post.uri,
            })

        print(f"  Fetched {len(all_posts)} posts so far...")

        if not response.cursor:
            break

        cursor = response.cursor

        if limit and len(all_posts) >= limit:
            all_posts = all_posts[:limit]
            break

    print(f"Total posts fetched: {len(all_posts)}")
    return all_posts


def convert_to_training_format(posts: list[dict]) -> list[dict]:
    """Convert posts to the training JSONL format."""
    training_data = []

    for post in posts:
        prompt = random.choice(GENERATION_PROMPTS)

        training_example = {
            "messages": [
                {"role": "user", "content": prompt},
                {"role": "assistant", "content": post["text"]}
            ]
        }
        training_data.append(training_example)

    return training_data


def main():
    parser = argparse.ArgumentParser(description="Fetch Bluesky posts for training")
    parser.add_argument("handle", help="Bluesky handle (e.g., @bobbby.online)")
    parser.add_argument("-o", "--output", default="bluesky_posts.jsonl", help="Output file")
    parser.add_argument("-l", "--limit", type=int, help="Max posts to fetch")

    args = parser.parse_args()

    # Fetch posts
    posts = fetch_all_posts(args.handle, args.limit)

    # Convert to training format
    training_data = convert_to_training_format(posts)

    # Write to JSONL
    output_path = Path(args.output)
    with open(output_path, "w") as f:
        for item in training_data:
            f.write(json.dumps(item) + "\n")

    print(f"\nWritten {len(training_data)} training examples to {output_path}")


if __name__ == "__main__":
    main()