Minimal Bsky.app searcher using API & app password

bsky_search.py

#!/usr/bin/env python3
"""Bluesky advanced search: filter by following, mentions, likes.
Usage: 
export BLUESKY_HANDLE="yourusername.bsky.social"
export BLUESKY_SCRAPER_PASSWORD="your-app-password"
python bsky_search.py "transformer"                    # search own posts
python bsky_search.py "climate from:following"         # posts from people you follow
python bsky_search.py "neural in:likes"                # in posts you've liked
python bsky_search.py "interesting from:following in:mentions"  # mentions from people you follow
Author: Scott H. Hawley, @drscotthawley.bsky.social
"""
import os, re
from atproto import Client
import argparse

OPERATORS = ['from:', 'in:', 'since:', 'until:']

def parse_query(q):
    """Parse 'keywords from:following in:likes' into {'query': str, 'filters': dict}."""
    filters, keywords = {}, []
    for tok in q.split():
        op = next((o for o in OPERATORS if tok.lower().startswith(o)), None)
        if op: filters[op.rstrip(':')] = tok[len(op):]
        else: keywords.append(tok)
    return {'query': ' '.join(keywords), 'filters': filters}

def get_client():
    """Authenticated Bluesky client."""
    c = Client()
    c.login(os.environ.get('BLUESKY_HANDLE', 'drscotthawley.bsky.social'), os.environ['BLUESKY_SCRAPER_PASSWORD'])
    return c

def fetch_all(fn, key='feed', limit=500):
    """Paginate through API results."""
    items, cursor = [], None
    while len(items) < limit:
        r = fn(cursor=cursor)
        batch = getattr(r, key, None) or []
        if not batch: break
        items.extend(batch)
        cursor = r.cursor
        if not cursor: break
    return items

def get_following(client, handle):
    """Get set of DIDs for accounts user follows."""
    follows, cursor = set(), None
    while True:
        r = client.get_follows(actor=handle, cursor=cursor)
        follows.update(f.did for f in r.follows)
        if not r.cursor: break
        cursor = r.cursor
    return follows

def post_to_dict(post):
    """Extract useful fields from post object."""
    embed_text = ""
    if hasattr(post, 'embed') and post.embed:
        e = post.embed
        if hasattr(e, 'record') and e.record and hasattr(e.record, 'value'):
            embed_text += getattr(e.record.value, 'text', '')
        if hasattr(e, 'external') and e.external:
            embed_text += f" [{getattr(e.external, 'title', '')}]"
    post_id = post.uri.split('/')[-1]
    return {
        'date': post.record.created_at[:10], 'author': post.author.handle, 'did': post.author.did,
        'text': post.record.text, 'embed': embed_text.strip(),
        'url': f"https://bsky.app/profile/{post.author.handle}/post/{post_id}"
    }

def matches(post, query):
    """Check if post matches search query (case-insensitive)."""
    if not query: return True
    txt = f"{post['text']} {post['embed']}".lower()
    return all(w.lower() in txt for w in query.split())

def search_posts(client, handle, query, filters, limit=100):
    """Search posts based on filters."""
    results = []
    source = filters.get('in', 'posts')
    following = get_following(client, handle) if filters.get('from') == 'following' else None
    
    if source == 'likes':
        items = fetch_all(lambda cursor=None: client.app.bsky.feed.get_actor_likes({'actor': handle, 'limit': 100, 'cursor': cursor}), 'feed')
    elif source == 'mentions':
        items = fetch_all(lambda cursor=None: client.app.bsky.notification.list_notifications({'limit': 100, 'cursor': cursor}), 'notifications')
        items = [n for n in items if getattr(n, 'reason', '') == 'mention']
    else:  # default: own posts
        items = fetch_all(lambda cursor=None: client.get_author_feed(actor=handle, limit=100, cursor=cursor), 'feed')
    
    for item in items:
        post = item.post if hasattr(item, 'post') else (item.record if hasattr(item, 'record') else None)
        if not post or not hasattr(post, 'author'): continue
        p = post_to_dict(post)
        if following and p['did'] not in following: continue
        if matches(p, query): results.append(p)
        if len(results) >= limit: break
    return results

def main():
    """Search Bluesky with advanced filters."""
    p = argparse.ArgumentParser(description="Bluesky advanced search")
    p.add_argument('--limit', type=int, default=20, help="Max results")
    p.add_argument('--handle', default=os.environ.get('BLUESKY_HANDLE', 'drscotthawley.bsky.social'))
    p.add_argument('query', nargs='?', default='', help="e.g. 'climate from:following in:likes'")
    args = p.parse_args()
    client = get_client()
    parsed = parse_query(args.query)
    print(f"Searching: '{parsed['query']}' with filters {parsed['filters']}")
    results = search_posts(client, args.handle, parsed['query'], parsed['filters'], args.limit)
    print(f"\nFound {len(results)} results:\n")
    for r in results:
        print(f"[{r['date']}] @{r['author']}: {r['text'][:80]}{'...' if len(r['text'])>80 else ''}")
        print(f"  {r['url']}\n")

if __name__ == '__main__': main()