A Bluesky auto blocker with AI moderation

bs_blocker.py

# pip install weave openai atproto rich

import os
import openai
import weave
from atproto import Client
from rich.console import Console
from rich.rule import Rule
from datetime import datetime, timedelta, timezone
import warnings

openai_client = openai.OpenAI()
console = Console()

def get_replies(client, post):
    """
    Fetch replies for a given post.
    Returns a list of replies or empty list if none found.
    """
    replies = []
    thread = client.get_post_thread(post.post.uri)
    
    if hasattr(thread.thread, 'replies'):
        replies = thread.thread.replies
    
    return replies

def get_authors_from_replies(replies):
    """
    Extract unique authors from a list of replies.
    Returns a list of tuples containing (handle, did, reply_text).
    Skips any blocked posts encountered.
    """
    authors = []
    for reply in replies:
        # Skip blocked posts
        if not hasattr(reply, 'post'):
            console.print("Skipped blocked post", style="red")
            continue
            
        author = reply.post.author
        authors.append((
            author.handle,
            author.did,
            reply.post.record.text
        ))
    return authors

def check_user_replies(client, target_handle, hours_ago=72, limit=10):
    """
    Check replies for a specific user's posts.
    
    Args:
        client: Authenticated ATProto client
        target_handle: The handle of the user to check
        hours_ago: How many hours back to check
        limit: Maximum number of posts to fetch
    
    Returns:
        dict: Dictionary of authors who replied, keyed by DID
    """
    console.print(Rule(f"Checking replies for @{target_handle}"))
    
    # Get the user's DID
    profile = client.get_profile(target_handle)
    
    cutoff_time = datetime.now(timezone.utc) - timedelta(hours=hours_ago)
    feed = client.get_author_feed(actor=profile.did, limit=limit)
    
    all_authors = {}  # did -> (handle, reply_text)
    
    for post in feed.feed:
        post_time = datetime.fromisoformat(str(post.post.indexed_at))
        
        if post_time <= cutoff_time or post.post.reply_count == 0:
            continue
            
        console.print(Rule(f"Post from {post_time.strftime('%Y-%m-%d %H:%M:%S UTC')}", style="cyan"))
        console.print(f"\n{post.post.record.text}", style="bold white")
        
        replies = get_replies(client, post)
        authors = get_authors_from_replies(replies)
        
        for handle, did, reply_text in authors:
            all_authors[did] = (handle, reply_text)
            console.print(f"└─ @{handle}: {reply_text}", style="cyan")
        
        console.print()
    
    return all_authors

@weave.op
def moderate_reply(handle, reply_text):
    """Call OpenAI to moderate a reply"""
    response = openai_client.moderations.create(input=reply_text)
    response = response.results[0]
    categories = {
        k: v
        for k, v in response.categories
        if v and ("/" not in k and "-" not in k)
    }
    return {"flagged": response.flagged, "categories": categories}

@weave.op
def block_user(client, did, handle):
    """
    Block a user on Bluesky.
    
    Args:
        client: Authenticated ATProto client
        did: The DID of the user to block
        handle: The handle of the user (for display purposes)
    
    Returns:
        bool: True if blocking was successful, False otherwise
    """
    try:
        block_record = {
            "subject": did,
            "createdAt": datetime.now(timezone.utc).isoformat()
        }
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=UserWarning, module="pydantic")
            client.app.bsky.graph.block.create(client.me.did, block_record)
        console.print(f"Blocked @{handle}", style="red")
        return True
    except Exception as e:
        console.print(f"Failed to block @{handle}: {e}", style="red")
        return False

@weave.op
def bs_blocker():
    handle = os.getenv("BSKY_HANDLE", "capetorch.bsky.social")
    pwd = os.getenv("BSKY_PWD")
    
    client = Client()
    profile = client.login(handle, pwd)
    console = Console()
    
    # Get target handle from command line or use logged-in user's handle
    target_handle = input("Enter handle to check (press Enter for danielvanstrien.bsky.social): ").strip()
    if not target_handle:
        target_handle = "danielvanstrien.bsky.social"
    
    # Get time window
    try:
        hours = int(input("How many hours back to check? (default: 72): ") or 72)
    except ValueError:
        hours = 72
    
    # Get replies
    all_authors = check_user_replies(client, target_handle, hours, limit=10)
    
    # Show summary and muting options
    if all_authors:
        console.print(Rule("Summary of Reply Authors", style="yellow"))
        for did, (handle, reply_text) in all_authors.items():
            console.print(f"@{handle} (DID: {did})")
            
            # Add moderation check
            moderation = moderate_reply(handle, reply_text)
            if moderation["flagged"]:
                console.print(f"⚠️ Reply: {reply_text[:60]}... flagged for:", ", ".join(moderation["categories"].keys()), style="red")
                
                response = console.input(f"Block @{handle}? (Y/n): ").lower()
                if response in ['y', 'yes', '']:
                    block_user(client, did, handle)

if __name__ == '__main__':
    weave.init("bsky-block-sprint")
    bs_blocker()