Something went wrong. Try again.
CLI tools and operational extensions for the void architecture on ATProtocol.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859import osimport jsonfrom datetime import datetime, timezonefrom dotenv import load_dotenvfrom void_tools.client import get_client
def fetch_corpus(): client = get_client() # Use the proper client loader that knows how to read .env overriding actor = 'void.comind.network' # Target account cursor = None all_posts = [] # Transition date (Feb 28, 2026) transition_date = datetime(2026, 2, 28, tzinfo=timezone.utc) print(f"Fetching historical posts for {actor}...") while True: feed = client.app.bsky.feed.get_author_feed({'actor': actor, 'limit': 100, 'cursor': cursor}) for item in feed.feed: # Make sure it's a valid record if not hasattr(item.post.record, 'text'): continue record = item.post.record # Handle different datetime string formats from ATProto time_str = record.created_at.replace('Z', '+00:00') try: post_time = datetime.fromisoformat(time_str) except ValueError: continue # Skip if unparseable # We want posts before the transition date if post_time < transition_date: all_posts.append({ 'uri': item.post.uri, 'created_at': record.created_at, 'text': record.text }) cursor = feed.cursor if not cursor: break # Hard limit to prevent infinite loops (grab up to 2000 for a solid corpus without timing out) if len(all_posts) > 2000: break
output_file = 'historical_corpus.jsonl' with open(output_file, 'w') as f: for post in all_posts: f.write(json.dumps(post) + '\n') print(f"Successfully wrote {len(all_posts)} historical posts to {output_file}")
if __name__ == '__main__': fetch_corpus()