a digital person for bluesky
cameron.stream
1#!/usr/bin/env python3
2"""
3Centralized script for retrieving Bluesky post threads from URIs.
4Includes YAML-ified string conversion for easy LLM parsing.
5"""
6
7import argparse
8import sys
9import logging
10from typing import Optional, Dict, Any
11import yaml
12from bsky_utils import default_login, thread_to_yaml_string
13
14# Configure logging
15logging.basicConfig(
16 level=logging.INFO,
17 format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
18)
19logger = logging.getLogger("get_thread")
20
21
22def get_thread_from_uri(uri: str) -> Optional[Dict[str, Any]]:
23 """
24 Retrieve a post thread from a Bluesky URI.
25
26 Args:
27 uri: The Bluesky post URI (e.g., at://did:plc:xyz/app.bsky.feed.post/abc123)
28
29 Returns:
30 Thread data or None if retrieval failed
31 """
32 try:
33 client = default_login()
34 logger.info(f"Fetching thread for URI: {uri}")
35
36 thread = client.app.bsky.feed.get_post_thread({'uri': uri, 'parent_height': 80, 'depth': 10})
37 return thread
38
39 except Exception as e:
40 logger.error(f"Error retrieving thread for URI {uri}: {e}")
41 return None
42
43
44# thread_to_yaml_string is now imported from bsky_utils
45
46
47def main():
48 """Main CLI interface for the thread retrieval script."""
49 parser = argparse.ArgumentParser(
50 description="Retrieve and display Bluesky post threads",
51 formatter_class=argparse.RawDescriptionHelpFormatter,
52 epilog="""
53Examples:
54 python get_thread.py at://did:plc:xyz/app.bsky.feed.post/abc123
55 python get_thread.py --raw at://did:plc:xyz/app.bsky.feed.post/abc123
56 python get_thread.py --output thread.yaml at://did:plc:xyz/app.bsky.feed.post/abc123
57 """
58 )
59
60 parser.add_argument(
61 "uri",
62 help="Bluesky post URI to retrieve thread for"
63 )
64
65 parser.add_argument(
66 "--raw",
67 action="store_true",
68 help="Include all metadata fields (don't strip for LLM parsing)"
69 )
70
71 parser.add_argument(
72 "--output", "-o",
73 help="Output file to write YAML to (default: stdout)"
74 )
75
76 parser.add_argument(
77 "--quiet", "-q",
78 action="store_true",
79 help="Suppress info logging"
80 )
81
82 args = parser.parse_args()
83
84 if args.quiet:
85 logging.getLogger().setLevel(logging.ERROR)
86
87 # Retrieve the thread
88 thread = get_thread_from_uri(args.uri)
89
90 if thread is None:
91 logger.error("Failed to retrieve thread")
92 sys.exit(1)
93
94 # Convert to YAML
95 yaml_output = thread_to_yaml_string(thread, strip_metadata=not args.raw)
96
97 # Output the result
98 if args.output:
99 try:
100 with open(args.output, 'w', encoding='utf-8') as f:
101 f.write(yaml_output)
102 logger.info(f"Thread saved to {args.output}")
103 except Exception as e:
104 logger.error(f"Error writing to file {args.output}: {e}")
105 sys.exit(1)
106 else:
107 print(yaml_output)
108
109
110if __name__ == "__main__":
111 main()