Something went wrong. Try again.
forked from cameron.stream/void
Something went wrong. Try again.
void get_thread.py
3.0 kB · 111 lines
Python
at contributing
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111#!/usr/bin/env python3"""Centralized script for retrieving Bluesky post threads from URIs.Includes YAML-ified string conversion for easy LLM parsing."""
import argparseimport sysimport loggingfrom typing import Optional, Dict, Anyimport yamlfrom bsky_utils import default_login, thread_to_yaml_string
# Configure logginglogging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")logger = logging.getLogger("get_thread")
def get_thread_from_uri(uri: str) -> Optional[Dict[str, Any]]: """ Retrieve a post thread from a Bluesky URI. Args: uri: The Bluesky post URI (e.g., at://did:plc:xyz/app.bsky.feed.post/abc123) Returns: Thread data or None if retrieval failed """ try: client = default_login() logger.info(f"Fetching thread for URI: {uri}") thread = client.app.bsky.feed.get_post_thread({'uri': uri, 'parent_height': 80, 'depth': 10}) return thread except Exception as e: logger.error(f"Error retrieving thread for URI {uri}: {e}") return None
# thread_to_yaml_string is now imported from bsky_utils
def main(): """Main CLI interface for the thread retrieval script.""" parser = argparse.ArgumentParser( description="Retrieve and display Bluesky post threads", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""Examples: python get_thread.py at://did:plc:xyz/app.bsky.feed.post/abc123 python get_thread.py --raw at://did:plc:xyz/app.bsky.feed.post/abc123 python get_thread.py --output thread.yaml at://did:plc:xyz/app.bsky.feed.post/abc123 """ ) parser.add_argument( "uri", help="Bluesky post URI to retrieve thread for" ) parser.add_argument( "--raw", action="store_true", help="Include all metadata fields (don't strip for LLM parsing)" ) parser.add_argument( "--output", "-o", help="Output file to write YAML to (default: stdout)" ) parser.add_argument( "--quiet", "-q", action="store_true", help="Suppress info logging" ) args = parser.parse_args() if args.quiet: logging.getLogger().setLevel(logging.ERROR) # Retrieve the thread thread = get_thread_from_uri(args.uri) if thread is None: logger.error("Failed to retrieve thread") sys.exit(1) # Convert to YAML yaml_output = thread_to_yaml_string(thread, strip_metadata=not args.raw) # Output the result if args.output: try: with open(args.output, 'w', encoding='utf-8') as f: f.write(yaml_output) logger.info(f"Thread saved to {args.output}") except Exception as e: logger.error(f"Error writing to file {args.output}: {e}") sys.exit(1) else: print(yaml_output)
if __name__ == "__main__": main()