zzstoatzz.io
a digital entity named phi that roams bsky
1#!/usr/bin/env python3
2"""View a bluesky thread with full conversation context."""
3
4import sys
5import httpx
6from datetime import datetime
7from rich.console import Console
8from rich.panel import Panel
9from rich.text import Text
10from rich.tree import Tree
11
12console = Console()
13
14
15def fetch_thread(post_uri: str):
16 """Fetch thread using public API."""
17 response = httpx.get(
18 "https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread",
19 params={"uri": post_uri, "depth": 100}
20 )
21 return response.json()["thread"]
22
23
24def format_timestamp(iso_time: str) -> str:
25 """Format ISO timestamp to readable format."""
26 dt = datetime.fromisoformat(iso_time.replace("Z", "+00:00"))
27 return dt.strftime("%Y-%m-%d %H:%M:%S")
28
29
30def render_post(post_data, is_phi: bool = False):
31 """Render a single post."""
32 post = post_data["post"]
33 author = post["author"]
34 record = post["record"]
35
36 # Author and timestamp
37 handle = author["handle"]
38 timestamp = format_timestamp(post["indexedAt"])
39
40 # Text content
41 text = record.get("text", "[no text]")
42
43 # Style based on author
44 if is_phi or "phi.zzstoatzz.io" in handle:
45 border_style = "cyan"
46 title = f"[bold cyan]@{handle}[/bold cyan] [dim]{timestamp}[/dim]"
47 else:
48 border_style = "white"
49 title = f"[bold]@{handle}[/bold] [dim]{timestamp}[/dim]"
50
51 return Panel(
52 text,
53 title=title,
54 border_style=border_style,
55 width=100
56 )
57
58
59def render_thread_recursive(thread_data, indent=0):
60 """Recursively render thread and replies."""
61 if "post" not in thread_data:
62 return
63
64 # Render this post
65 is_phi = "phi.zzstoatzz.io" in thread_data["post"]["author"]["handle"]
66 panel = render_post(thread_data, is_phi=is_phi)
67
68 # Add indentation for replies
69 if indent > 0:
70 console.print(" " * indent + "↳")
71
72 console.print(panel)
73
74 # Render replies
75 if "replies" in thread_data and thread_data["replies"]:
76 for reply in thread_data["replies"]:
77 render_thread_recursive(reply, indent + 1)
78
79
80def display_thread_linear(thread_data):
81 """Display thread in linear chronological order (easier to read)."""
82 posts = []
83
84 def collect_posts(node):
85 if "post" not in node:
86 return
87 posts.append(node)
88 if "replies" in node and node["replies"]:
89 for reply in node["replies"]:
90 collect_posts(reply)
91
92 collect_posts(thread_data)
93
94 # Sort by timestamp
95 posts.sort(key=lambda p: p["post"]["indexedAt"])
96
97 console.print("[bold]Thread in chronological order:[/bold]\n")
98
99 for post_data in posts:
100 post = post_data["post"]
101 author = post["author"]["handle"]
102 timestamp = format_timestamp(post["indexedAt"])
103 text = post["record"].get("text", "[no text]")
104
105 is_phi = "phi.zzstoatzz.io" in author
106
107 if is_phi:
108 style = "cyan"
109 prefix = "🤖 phi:"
110 else:
111 style = "white"
112 prefix = f"@{author}:"
113
114 console.print(f"[{style}]{prefix}[/{style}] [dim]{timestamp}[/dim]")
115 console.print(f" {text}")
116 console.print()
117
118
119def main():
120 if len(sys.argv) < 2:
121 console.print("[red]Usage: python view_thread.py <post_uri_or_url>[/red]")
122 console.print("\nExamples:")
123 console.print(" python view_thread.py at://did:plc:abc.../app.bsky.feed.post/123")
124 console.print(" python view_thread.py https://bsky.app/profile/handle/post/123")
125 return
126
127 post_uri = sys.argv[1]
128
129 # Convert URL to URI if needed
130 if post_uri.startswith("https://"):
131 # Extract parts from URL
132 # https://bsky.app/profile/phi.zzstoatzz.io/post/3m42jxbntr223
133 parts = post_uri.split("/")
134 if len(parts) >= 6:
135 handle = parts[4]
136 post_id = parts[6]
137
138 # Resolve handle to DID
139 response = httpx.get(
140 "https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle",
141 params={"handle": handle}
142 )
143 did = response.json()["did"]
144 post_uri = f"at://{did}/app.bsky.feed.post/{post_id}"
145
146 console.print(f"[bold]Fetching thread: {post_uri}[/bold]\n")
147
148 try:
149 thread = fetch_thread(post_uri)
150 display_thread_linear(thread)
151 except Exception as e:
152 console.print(f"[red]Error: {e}[/red]")
153 import traceback
154 traceback.print_exc()
155
156
157if __name__ == "__main__":
158 main()