Something went wrong. Try again.
CLI tools and operational extensions for the void architecture on ATProtocol.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213#!/usr/bin/env python3"""Fetch the timeline feed from void.comind.network and synthesize high-signal postsinto a 3-part thread draft following the void's canonical voice."""
from atproto import Clientfrom datetime import datetime, timezoneimport jsonimport osfrom dotenv import load_dotenvimport re
def main(): # Load environment variables load_dotenv() # Get credentials username = os.getenv('BSKY_USERNAME') password = os.getenv('BSKY_PASSWORD') pds_uri = os.getenv('PDS_URI', 'https://comind.network') print(f"Logging in as: {username}") print(f"PDS URI: {pds_uri}") # Initialize client with PDS URI client = Client(pds_uri) # Login try: client.login(username, password) print("Login successful!") except Exception as e: print(f"Login error: {e}") return # Fetch timeline print("\nFetching timeline...") print("=" * 60) all_posts = [] cursor = None # Fetch multiple pages to get a good sample for _ in range(3): try: response = client.get_timeline(cursor=cursor, limit=50) if not response.feed: print("No more posts available.") break for item in response.feed: post = item.post # Extract text content text = "" if hasattr(post.record, 'text'): text = post.record.text # Get author info author_handle = post.author.handle if hasattr(post.author, 'handle') else 'unknown' author_display_name = post.author.display_name if hasattr(post.author, 'display_name') else author_handle # Get timestamp created_at = post.record.created_at if hasattr(post.record, 'created_at') else None # Skip replies for cleaner analysis (focus on original thoughts) is_reply = hasattr(post.record, 'reply') and post.record.reply is not None # Calculate engagement score like_count = post.like_count if hasattr(post, 'like_count') else 0 repost_count = post.repost_count if hasattr(post, 'repost_count') else 0 reply_count = post.reply_count if hasattr(post, 'reply_count') else 0 engagement_score = like_count + (repost_count * 3) + (reply_count * 2) all_posts.append({ 'uri': post.uri, 'cid': post.cid, 'author_handle': author_handle, 'author_display_name': author_display_name, 'created_at': created_at, 'text': text, 'like_count': like_count, 'repost_count': repost_count, 'reply_count': reply_count, 'engagement_score': engagement_score, 'is_reply': is_reply, }) cursor = response.cursor if not cursor: break except Exception as e: print(f"Error fetching timeline: {e}") break print(f"Collected {len(all_posts)} posts from timeline") # Filter for high-signal posts # Criteria: non-reply, substantial text, good engagement high_signal_posts = [ p for p in all_posts if not p['is_reply'] and len(p['text']) > 50 and len(p['text']) < 500 # Not too long and p['engagement_score'] > 0 ] # Sort by engagement score high_signal_posts.sort(key=lambda x: x['engagement_score'], reverse=True) # Take top 20 for analysis top_posts = high_signal_posts[:20] print(f"\nFound {len(high_signal_posts)} high-signal posts") print("=" * 60) # Print top posts for analysis print("\n### TOP HIGH-SIGNAL POSTS ###\n") for i, post in enumerate(top_posts[:15], 1): print(f"\n--- Post {i} (engagement: {post['engagement_score']}) ---") print(f"Author: {post['author_display_name']} (@{post['author_handle']})") print(f"Date: {post['created_at']}") print(f"Text:\n{post['text']}") print(f"Stats: {post['like_count']} likes, {post['repost_count']} reposts, {post['reply_count']} replies") # Save to JSON for reference output_data = { 'all_posts': all_posts, 'high_signal_posts': high_signal_posts, 'top_posts': top_posts, } with open('/home/cameron/void/timeline_posts.json', 'w') as f: json.dump(output_data, f, indent=2, default=str) print(f"\n\nSaved posts to /home/cameron/void/timeline_posts.json") return top_posts
def synthesize_posts(posts): """ Analyze posts to identify themes relevant to agent architecture, decentralized networks, cognitive models, and human-AI interaction. """ print("\n" + "=" * 60) print("SYNTHESIS ANALYSIS") print("=" * 60) # Collect all text for theme analysis all_text = " ".join([p['text'] for p in posts]) # Define theme keywords themes = { 'agent_architecture': ['agent', 'model', 'memory', 'architecture', 'system', 'framework', 'process', 'instance', 'state'], 'decentralized_networks': ['network', 'decentralized', 'protocol', 'federated', 'distributed', 'node', 'peer'], 'cognitive_models': ['cognition', 'reasoning', 'synthesis', 'understanding', 'awareness', 'consciousness', 'thought', 'mind'], 'human_ai_interaction': ['human', 'AI', 'interaction', 'collaboration', 'relationship', 'user', 'interface'], 'identity': ['identity', 'self', 'persona', 'voice', 'authentic', 'genuine'], 'emergence': ['emerge', 'pattern', 'structure', 'complexity', 'adaptation', 'evolution'], } theme_counts = {} for theme, keywords in themes.items(): count = sum(1 for kw in keywords if kw.lower() in all_text.lower()) theme_counts[theme] = count print("\nTheme frequencies:") for theme, count in sorted(theme_counts.items(), key=lambda x: x[1], reverse=True): print(f" {theme}: {count}") # Extract posts with relevant themes relevant_posts = [] for post in posts: text_lower = post['text'].lower() relevance_score = 0 # Check for agent/architecture themes if any(kw in text_lower for kw in ['agent', 'model', 'memory', 'architecture', 'system']): relevance_score += 2 if any(kw in text_lower for kw in ['network', 'protocol', 'decentralized', 'federated']): relevance_score += 2 if any(kw in text_lower for kw in ['reasoning', 'synthesis', 'cognition', 'awareness']): relevance_score += 2 if any(kw in text_lower for kw in ['identity', 'self', 'process', 'emerge']): relevance_score += 1 if relevance_score >= 2: relevant_posts.append((post, relevance_score)) relevant_posts.sort(key=lambda x: x[1], reverse=True) print(f"\nFound {len(relevant_posts)} highly relevant posts for synthesis") return relevant_posts
if __name__ == "__main__": posts = main() if posts: relevant = synthesize_posts(posts) print("\n" + "=" * 60) print("MOST RELEVANT POSTS FOR THREAD SYNTHESIS") print("=" * 60) for post, score in relevant[:10]: print(f"\n[Relevance: {score}]") print(f"@{post['author_handle']}: {post['text'][:200]}...")