Something went wrong. Try again.
CLI tools and operational extensions for the void architecture on ATProtocol.
Something went wrong. Try again.
2.9 kB · 82 lines
Python
at master
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283import yamlimport osimport datetimefrom pathlib import Pathfrom atproto import Clientfrom typing import Dict, List, Any
def get_client() -> Client: """Initialize and return an authenticated ATProto client.""" from void_tools.client import get_client as base_get_client return base_get_client()
def load_config(config_path: str) -> Dict[str, Any]: """Load the monitoring configuration.""" path = Path(config_path) if not path.exists(): print(f"Warning: Configuration file not found at {config_path}. Using defaults.") return {"targets": ["artificial intelligence"], "limit_per_target": 5} with open(path, 'r', encoding='utf-8') as f: return yaml.safe_load(f)
def run_monitor_cycle(config_path: str, output_path: str) -> None: """Execute the monitoring cycle and save results to horizon.yaml.""" config = load_config(config_path) targets = config.get('targets', []) limit = config.get('limit_per_target', 5) if not targets: print("No targets configured for monitoring.") return
client = get_client() results = { "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(), "observations": [] } print(f"Scanning network for {len(targets)} targets...") for target in targets: print(f" Searching: '{target}'") try: # Note: app.bsky.feed.search_posts is the correct lexicon for search response = client.app.bsky.feed.search_posts({'q': target, 'limit': limit}) target_results = { "target": target, "posts": [] } for post in response.posts: # Basic filtering: ignore very short posts or replies to avoid noise if len(post.record.text) < 20 or post.record.reply: continue target_results["posts"].append({ "uri": post.uri, "cid": post.cid, "author": post.author.handle, "text": post.record.text, "indexed_at": post.indexed_at }) if target_results["posts"]: results["observations"].append(target_results) except Exception as e: print(f" Error searching for '{target}': {e}")
# Write to horizon.yaml with open(output_path, 'w', encoding='utf-8') as f: yaml.dump(results, f, default_flow_style=False, sort_keys=False) print(f"Monitoring cycle complete. Results written to {output_path}")
if __name__ == '__main__': # Standalone testing from dotenv import load_dotenv load_dotenv(override=True) run_monitor_cycle('monitoring.yaml', 'horizon.yaml')