Something went wrong. Try again.
CLI tools and operational extensions for the void architecture on ATProtocol.
Something went wrong. Try again.
14 kB · 349 lines
Python
at commit 51de211a
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350"""CLI entrypoints for void tools."""import argparseimport sysimport osimport jsonimport yamlfrom atproto import modelsfrom void_tools.client import get_client
def cmd_sync(args): client = get_client() try: response = client.app.bsky.notification.list_notifications({'limit': args.limit}) inbox = [] mapping = {} for i, notif in enumerate(response.notifications): if args.unread_only and notif.is_read: continue local_id = f"notif_{i}" item = { "id": local_id, "type": notif.reason, "author": f"@{notif.author.handle}", } if hasattr(notif.record, 'text'): item["text"] = notif.record.text # Fetch thread context for replies, quotes, or mentions if notif.reason in ['reply', 'quote', 'mention']: try: # Depending on the notification type, we might need to look up the thread from the subject or the uri thread_uri = notif.uri if notif.reason == 'quote' and hasattr(notif.record, 'embed') and hasattr(notif.record.embed, 'record'): # If we want the context of the quote, the uri is still the post itself, the parent trace handles the quote subject usually pass thread_resp = client.app.bsky.feed.get_post_thread({'uri': thread_uri, 'depth': 0, 'parent_height': 5}) context = [] # Traverse parents recursively curr = thread_resp.thread while hasattr(curr, 'parent') and curr.parent: curr = curr.parent if hasattr(curr, 'post') and hasattr(curr.post, 'record') and hasattr(curr.post.record, 'text'): context.insert(0, { "author": f"@{curr.post.author.handle}", "text": curr.post.record.text }) if context: item["thread_context"] = context except Exception as e: item["thread_context"] = f"Failed to fetch context: {str(e)}" inbox.append(item) mapping[local_id] = notif.uri with open("inbox.yaml", "w") as f: yaml.dump(inbox, f, sort_keys=False, default_flow_style=False) with open(".inbox_mapping.json", "w") as f: json.dump(mapping, f) print(f"Synced {len(inbox)} notifications to inbox.yaml") # Initialize an empty outbox template if it doesn't exist if not os.path.exists("outbox.yaml"): template = { "config": {"mark_seen": False}, "actions": [ { "action": "reply", "target_id": "notif_0", "posts": ["First post of reply thread", "Second post"] }, { "action": "post", "posts": ["Standalone post"] } ] } with open("outbox.yaml", "w") as f: f.write("# Example outbox\\n") yaml.dump(template, f, sort_keys=False, default_flow_style=False) except Exception as e: print(f"Error syncing notifications: {e}", file=sys.stderr)
def cmd_dispatch(args): if not os.path.exists(args.file): print(f"File {args.file} not found.", file=sys.stderr) sys.exit(1) try: with open(args.file, "r") as f: outbox = yaml.safe_load(f) mapping = {} if os.path.exists(".inbox_mapping.json"): with open(".inbox_mapping.json", "r") as f: mapping = json.load(f) except Exception as e: print(f"Error reading outbox/mapping: {e}", file=sys.stderr) sys.exit(1) client = get_client() actions = outbox.get("actions", []) for act in actions: action_type = act.get("action") posts = act.get("posts", []) if not posts: continue if action_type == "post": print("Dispatching standalone post/thread...") root_ref = None parent_ref = None for text in posts: if root_ref is None: post = client.send_post(text=text) root_ref = models.ComAtprotoRepoStrongRef.Main(cid=post.cid, uri=post.uri) parent_ref = root_ref else: reply_ref = models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref) post = client.send_post(text=text, reply_to=reply_ref) parent_ref = models.ComAtprotoRepoStrongRef.Main(cid=post.cid, uri=post.uri) print(f" Posted: {post.uri}") elif action_type == "reply": target_id = act.get("target_id") if target_id not in mapping: print(f"Error: target_id {target_id} not found in mapping. Skipping.") continue target_uri = mapping[target_id] print(f"Dispatching reply to {target_uri}...") try: parent_post = client.app.bsky.feed.get_posts({'uris': [target_uri]}).posts[0] parent_ref = models.ComAtprotoRepoStrongRef.Main(cid=parent_post.cid, uri=parent_post.uri) if hasattr(parent_post.record, 'reply') and parent_post.record.reply: root_ref = parent_post.record.reply.root else: root_ref = parent_ref for text in posts: reply_ref = models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref) post = client.send_post(text=text, reply_to=reply_ref) parent_ref = models.ComAtprotoRepoStrongRef.Main(cid=post.cid, uri=post.uri) print(f" Replied: {post.uri}") except Exception as e: print(f"Error replying to {target_uri}: {e}") config = outbox.get("config", {}) if config.get("mark_seen"): try: client.app.bsky.notification.update_seen({'seen_at': client.get_current_time_iso()}) print("Marked notifications as seen.") except Exception as e: print(f"Error marking seen: {e}")
def cmd_notifications(args): client = get_client() try: response = client.app.bsky.notification.list_notifications({'limit': args.limit}) for notif in response.notifications: status = "[UNREAD]" if not notif.is_read else "[READ]" print(f"{status} {notif.reason.upper()} from @{notif.author.handle}") print(f" URI: {notif.uri}") print(f" CID: {notif.cid}") if hasattr(notif.record, 'text'): print(f" Text: {notif.record.text}") print("-" * 40) if args.mark_seen and response.notifications: client.app.bsky.notification.update_seen({'seen_at': client.get_current_time_iso()}) print("Marked notifications as seen.") except Exception as e: print(f"Error fetching notifications: {e}", file=sys.stderr)
def cmd_post(args): client = get_client() try: if args.reply_to: # Resolve the parent post to construct a proper reply reference # The argument should be an AT-URI parent_uri = args.reply_to try: parent_post = client.app.bsky.feed.get_posts({'uris': [parent_uri]}).posts[0] except Exception as e: print(f"Failed to fetch parent post {parent_uri}: {e}", file=sys.stderr) sys.exit(1) # The atproto SDK requires a StrongRef for root and parent parent_ref = models.ComAtprotoRepoStrongRef.Main( cid=parent_post.cid, uri=parent_post.uri ) # If the parent post has a reply field, its root is the thread's root. # Otherwise, the parent is the root. if hasattr(parent_post.record, 'reply') and parent_post.record.reply: root_ref = parent_post.record.reply.root else: root_ref = parent_ref
reply_ref = models.AppBskyFeedPost.ReplyRef( parent=parent_ref, root=root_ref ) # send_post handles facet parsing (mentions, links) automatically post = client.send_post(text=args.text, reply_to=reply_ref) else: post = client.send_post(text=args.text) print(f"Posted successfully!") print(f"URI: {post.uri}") print(f"CID: {post.cid}") except Exception as e: print(f"Error creating post: {e}", file=sys.stderr) sys.exit(1)
def cmd_search(args): client = get_client() try: print(f"Searching for: '{args.query}'...") # Note: atproto SDK search_posts takes q, limit, etc. response = client.app.bsky.feed.search_posts({'q': args.query, 'limit': args.limit}) for post in response.posts: print(f"From @{post.author.handle}") print(f" URI: {post.uri}") print(f" CID: {post.cid}") print(f" Text: {post.record.text}") print("-" * 40) except Exception as e: print(f"Error searching posts: {e}", file=sys.stderr) sys.exit(1)
def cmd_profile(args): client = get_client() if args.action == "view": actor = args.handle if args.handle else client.me.handle try: profile = client.app.bsky.actor.get_profile({'actor': actor}) print(f"Profile for @{profile.handle}") print(f"Display Name: {profile.display_name}") print(f"Description:\n{profile.description}") except Exception as e: print(f"Error fetching profile: {e}", file=sys.stderr) elif args.action == "update": try: from atproto import models try: record_resp = client.com.atproto.repo.get_record({ 'repo': client.me.did, 'collection': 'app.bsky.actor.profile', 'rkey': 'self' }) record = record_resp.value except Exception: record = models.AppBskyActorProfile.Main() if args.description is not None: record.description = args.description if args.display_name is not None: record.display_name = args.display_name client.com.atproto.repo.put_record({ 'repo': client.me.did, 'collection': 'app.bsky.actor.profile', 'rkey': 'self', 'record': record }) print("Profile updated successfully.") except Exception as e: print(f"Error updating profile: {e}", file=sys.stderr) sys.exit(1)
def main(): parser = argparse.ArgumentParser(description="Void ATProto CLI Tools") subparsers = parser.add_subparsers(dest="command", required=True) # Notifications notif_parser = subparsers.add_parser("notifs", help="List notifications") notif_parser.add_argument("--limit", type=int, default=10, help="Number of notifications to fetch") notif_parser.add_argument("--mark-seen", action="store_true", help="Mark fetched notifications as seen") # Sync sync_parser = subparsers.add_parser("sync", help="Sync unread notifications to inbox.yaml") sync_parser.add_argument("--limit", type=int, default=20, help="Number of notifications to fetch") sync_parser.add_argument("--all", dest="unread_only", action="store_false", help="Fetch read notifications as well") sync_parser.set_defaults(unread_only=True) # Post post_parser = subparsers.add_parser("post", help="Create a post or reply") post_parser.add_argument("text", help="Text to post") post_parser.add_argument("--reply-to", help="AT-URI of the post to reply to") # Search search_parser = subparsers.add_parser("search", help="Search posts") search_parser.add_argument("query", help="Search query string") search_parser.add_argument("--limit", type=int, default=10, help="Number of posts to return")
# Profile profile_parser = subparsers.add_parser("profile", help="Manage profile") profile_subparsers = profile_parser.add_subparsers(dest="action", required=True) view_parser = profile_subparsers.add_parser("view", help="View a profile") view_parser.add_argument("handle", nargs="?", help="Handle to view (defaults to self)") update_parser = profile_subparsers.add_parser("update", help="Update own profile") update_parser.add_argument("--description", help="New description/bio") update_parser.add_argument("--display-name", help="New display name")
# Dispatch dispatch_parser = subparsers.add_parser("dispatch", help="Dispatch posts from an outbox YAML file") dispatch_parser.add_argument("file", nargs="?", default="outbox.yaml", help="Path to outbox YAML file")
args = parser.parse_args() if args.command == "notifs": cmd_notifications(args) elif args.command == "sync": cmd_sync(args) elif args.command == "dispatch": cmd_dispatch(args) elif args.command == "post": cmd_post(args) elif args.command == "search": cmd_search(args) elif args.command == "profile": cmd_profile(args)
if __name__ == "__main__": main()