a digital entity named phi that roams bsky

feat: add scripts to view phi's activity

- view_phi_posts.py: see phi's recent posts (no auth needed, public API)
- view_thread.py: visualize full thread conversations chronologically
- added just commands: view-posts, view-thread
- no more screenshot copy-paste needed for iteration

+266
+7
justfile
··· 30 30 31 31 check: lint typecheck test 32 32 33 + # view phi's activity 34 + view-posts: 35 + uv run --with rich --with httpx python scripts/view_phi_posts.py 36 + 37 + view-thread URI: 38 + uv run --with rich --with httpx python scripts/view_thread.py {{URI}} 39 + 33 40 # setup reference projects 34 41 setup: 35 42 @mkdir -p .eggs
+101
scripts/view_phi_posts.py
··· 1 + #!/usr/bin/env python3 2 + """View phi's recent posts without authentication.""" 3 + 4 + import httpx 5 + from datetime import datetime 6 + from rich.console import Console 7 + from rich.panel import Panel 8 + from rich.text import Text 9 + 10 + console = Console() 11 + 12 + PHI_HANDLE = "phi.zzstoatzz.io" 13 + 14 + 15 + def fetch_phi_posts(limit: int = 10): 16 + """Fetch phi's recent posts using public API.""" 17 + # Resolve handle to DID 18 + response = httpx.get( 19 + "https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle", 20 + params={"handle": PHI_HANDLE} 21 + ) 22 + did = response.json()["did"] 23 + 24 + # Get author feed (public posts) 25 + response = httpx.get( 26 + "https://public.api.bsky.app/xrpc/app.bsky.feed.getAuthorFeed", 27 + params={"actor": did, "limit": limit} 28 + ) 29 + 30 + return response.json()["feed"] 31 + 32 + 33 + def format_timestamp(iso_time: str) -> str: 34 + """Format ISO timestamp to readable format.""" 35 + dt = datetime.fromisoformat(iso_time.replace("Z", "+00:00")) 36 + now = datetime.now(dt.tzinfo) 37 + delta = now - dt 38 + 39 + if delta.seconds < 60: 40 + return f"{delta.seconds}s ago" 41 + elif delta.seconds < 3600: 42 + return f"{delta.seconds // 60}m ago" 43 + elif delta.seconds < 86400: 44 + return f"{delta.seconds // 3600}h ago" 45 + else: 46 + return f"{delta.days}d ago" 47 + 48 + 49 + def display_posts(feed_items): 50 + """Display posts in a readable format.""" 51 + for item in feed_items: 52 + post = item["post"] 53 + record = post["record"] 54 + 55 + # Check if this is a reply 56 + is_reply = "reply" in record 57 + reply_indicator = "↳ REPLY" if is_reply else "✓ POST" 58 + 59 + # Format header 60 + timestamp = format_timestamp(post["indexedAt"]) 61 + header = f"[cyan]{reply_indicator}[/cyan] [dim]{timestamp}[/dim]" 62 + 63 + # Get post text 64 + text = record.get("text", "[no text]") 65 + 66 + # Show parent if it's a reply 67 + parent_text = "" 68 + if is_reply: 69 + parent_uri = record["reply"]["parent"]["uri"] 70 + parent_text = f"[dim]replying to: {parent_uri}[/dim]\n\n" 71 + 72 + # Format post 73 + content = Text() 74 + if parent_text: 75 + content.append(parent_text, style="dim") 76 + content.append(text) 77 + 78 + # Display 79 + panel = Panel( 80 + content, 81 + title=header, 82 + border_style="blue" if is_reply else "green", 83 + width=80 84 + ) 85 + console.print(panel) 86 + console.print() 87 + 88 + 89 + def main(): 90 + console.print("[bold]Fetching phi's recent posts...[/bold]\n") 91 + 92 + try: 93 + feed = fetch_phi_posts(limit=10) 94 + display_posts(feed) 95 + console.print(f"[dim]Showing {len(feed)} most recent posts[/dim]") 96 + except Exception as e: 97 + console.print(f"[red]Error: {e}[/red]") 98 + 99 + 100 + if __name__ == "__main__": 101 + main()
+158
scripts/view_thread.py
··· 1 + #!/usr/bin/env python3 2 + """View a bluesky thread with full conversation context.""" 3 + 4 + import sys 5 + import httpx 6 + from datetime import datetime 7 + from rich.console import Console 8 + from rich.panel import Panel 9 + from rich.text import Text 10 + from rich.tree import Tree 11 + 12 + console = Console() 13 + 14 + 15 + def 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 + 24 + def 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 + 30 + def 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 + 59 + def 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 + 80 + def 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 + 119 + def 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 + 157 + if __name__ == "__main__": 158 + main()