Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425#!/usr/bin/env python3"""Queue management utilities for Void bot."""import jsonimport argparsefrom pathlib import Pathfrom rich.console import Consolefrom rich.table import Tablefrom rich.prompt import Confirm
console = Console()
# Queue directoriesQUEUE_BASE_DIR = Path("queue")
def load_notification(filepath: Path) -> dict: """Load a notification from a JSON file.""" try: with open(filepath, 'r') as f: return json.load(f) except Exception as e: console.print(f"[red]Error loading {filepath}: {e}[/red]") return None
def list_notifications(handle_filter: str = None, show_all: bool = False, agent_id: str = None): """List all notifications in the queue, optionally filtered by handle and/or agent.""" # Build list of directories to check dirs_to_check = [] if agent_id: # Check specific agent's directories agent_queue_dir = QUEUE_BASE_DIR / agent_id if agent_queue_dir.exists(): dirs_to_check.append((agent_queue_dir, "queue", agent_id)) if show_all: dirs_to_check.append((agent_queue_dir / "errors", "errors", agent_id)) dirs_to_check.append((agent_queue_dir / "no_reply", "no_reply", agent_id)) else: # Check all agent directories for agent_dir in QUEUE_BASE_DIR.glob("*"): if agent_dir.is_dir() and agent_dir.name != "errors" and agent_dir.name != "no_reply": # Skip old flat structure directories if "-" in agent_dir.name: # Agent IDs have hyphens agent_id_str = agent_dir.name dirs_to_check.append((agent_dir, "queue", agent_id_str)) if show_all: dirs_to_check.append((agent_dir / "errors", "errors", agent_id_str)) dirs_to_check.append((agent_dir / "no_reply", "no_reply", agent_id_str)) # Also check old flat structure for backward compatibility if QUEUE_BASE_DIR.exists(): old_queue_files = list(QUEUE_BASE_DIR.glob("*.json")) if old_queue_files and any(f.name != "processed_notifications.json" for f in old_queue_files): dirs_to_check.append((QUEUE_BASE_DIR, "queue (old)", "legacy")) if show_all: old_errors = QUEUE_BASE_DIR / "errors" old_no_reply = QUEUE_BASE_DIR / "no_reply" if old_errors.exists(): dirs_to_check.append((old_errors, "errors (old)", "legacy")) if old_no_reply.exists(): dirs_to_check.append((old_no_reply, "no_reply (old)", "legacy")) all_notifications = [] for directory, source, dir_agent_id in dirs_to_check: if not directory.exists(): continue for filepath in directory.glob("*.json"): # Skip subdirectories if filepath.is_dir(): continue notif = load_notification(filepath) if notif and isinstance(notif, dict): notif['_filepath'] = filepath notif['_source'] = source notif['_agent_id'] = dir_agent_id # Apply handle filter if specified if handle_filter: author_handle = notif.get('author', {}).get('handle', '') if handle_filter.lower() not in author_handle.lower(): continue all_notifications.append(notif) # Sort by indexed_at all_notifications.sort(key=lambda x: x.get('indexed_at', ''), reverse=True) # Display results if not all_notifications: if handle_filter: console.print(f"[yellow]No notifications found for handle containing '{handle_filter}'[/yellow]") else: console.print("[yellow]No notifications found in queue[/yellow]") return table = Table(title=f"Queue Notifications ({len(all_notifications)} total)") table.add_column("Agent", style="blue", width=15) table.add_column("File", style="cyan", width=20) table.add_column("Source", style="magenta", width=10) table.add_column("Handle", style="green", width=25) table.add_column("Display Name", width=20) table.add_column("Text", width=35) table.add_column("Time", style="dim", width=20) for notif in all_notifications: author = notif.get('author', {}) handle = author.get('handle', 'unknown') display_name = author.get('display_name', '') text = notif.get('record', {}).get('text', '')[:40] if len(notif.get('record', {}).get('text', '')) > 40: text += "..." indexed_at = notif.get('indexed_at', '')[:19] # Trim milliseconds filename = notif['_filepath'].name[:20] source = notif['_source'] agent_display = notif['_agent_id'][:12] + "..." if len(notif['_agent_id']) > 15 else notif['_agent_id'] table.add_row(agent_display, filename, source, f"@{handle}", display_name, text, indexed_at) console.print(table) return all_notifications
def delete_by_handle(handle: str, dry_run: bool = False, force: bool = False): """Delete all notifications from a specific handle.""" # Remove @ if present handle = handle.lstrip('@') # Find all notifications from this handle console.print(f"\\n[bold]Searching for notifications from @{handle}...[/bold]\\n") to_delete = [] # Build list of directories to check dirs_to_check = [] # Check all agent directories for agent_dir in QUEUE_BASE_DIR.glob("*"): if agent_dir.is_dir() and "-" in agent_dir.name: dirs_to_check.extend([ (agent_dir, "queue"), (agent_dir / "errors", "errors"), (agent_dir / "no_reply", "no_reply") ]) # Also check old flat structure dirs_to_check.extend([ (QUEUE_BASE_DIR, "queue (old)"), (QUEUE_BASE_DIR / "errors", "errors (old)"), (QUEUE_BASE_DIR / "no_reply", "no_reply (old)") ]) for directory, source_name in dirs_to_check: if not directory.exists(): continue for filepath in directory.glob("*.json"): if filepath.is_dir() or filepath.name == "processed_notifications.json": continue notif = load_notification(filepath) if notif and isinstance(notif, dict): author_handle = notif.get('author', {}).get('handle', '') if author_handle.lower() == handle.lower(): to_delete.append({ 'filepath': filepath, 'notif': notif, 'source': source_name }) if not to_delete: console.print(f"[yellow]No notifications found from @{handle}[/yellow]") return # Display what will be deleted table = Table(title=f"Notifications to Delete from @{handle}") table.add_column("File", style="cyan") table.add_column("Location", style="magenta") table.add_column("Text", width=50) table.add_column("Time", style="dim") for item in to_delete: notif = item['notif'] text = notif.get('record', {}).get('text', '')[:50] if len(notif.get('record', {}).get('text', '')) > 50: text += "..." indexed_at = notif.get('indexed_at', '')[:19] table.add_row( item['filepath'].name, item['source'], text, indexed_at ) console.print(table) console.print(f"\\n[bold red]Found {len(to_delete)} notifications to delete[/bold red]") if dry_run: console.print("\\n[yellow]DRY RUN - No files were deleted[/yellow]") return # Confirm deletion if not force and not Confirm.ask("\\nDo you want to delete these notifications?"): console.print("[yellow]Deletion cancelled[/yellow]") return # Delete the files deleted_count = 0 for item in to_delete: try: item['filepath'].unlink() deleted_count += 1 console.print(f"[green]✓[/green] Deleted {item['filepath'].name}") except Exception as e: console.print(f"[red]✗[/red] Failed to delete {item['filepath'].name}: {e}") console.print(f"\\n[bold green]Successfully deleted {deleted_count} notifications[/bold green]")
def count_by_handle(): """Show detailed count of notifications by handle.""" handle_counts = {} # Build list of directories to check dirs_to_check = [] # Check all agent directories for agent_dir in QUEUE_BASE_DIR.glob("*"): if agent_dir.is_dir() and "-" in agent_dir.name: dirs_to_check.extend([ (agent_dir, 'queue'), (agent_dir / "errors", 'errors'), (agent_dir / "no_reply", 'no_reply') ]) # Also check old flat structure dirs_to_check.extend([ (QUEUE_BASE_DIR, 'queue'), (QUEUE_BASE_DIR / "errors", 'errors'), (QUEUE_BASE_DIR / "no_reply", 'no_reply') ]) # Collect counts from all directories for directory, location in dirs_to_check: if not directory.exists(): continue for filepath in directory.glob("*.json"): if filepath.is_dir() or filepath.name == "processed_notifications.json": continue notif = load_notification(filepath) if notif and isinstance(notif, dict): handle = notif.get('author', {}).get('handle', 'unknown') if handle not in handle_counts: handle_counts[handle] = {'queue': 0, 'errors': 0, 'no_reply': 0, 'total': 0} handle_counts[handle][location] += 1 handle_counts[handle]['total'] += 1 if not handle_counts: console.print("[yellow]No notifications found in any queue[/yellow]") return # Sort by total count sorted_handles = sorted(handle_counts.items(), key=lambda x: x[1]['total'], reverse=True) # Display results table = Table(title=f"Notification Count by Handle ({len(handle_counts)} unique handles)") table.add_column("Handle", style="green", width=30) table.add_column("Queue", style="cyan", justify="right") table.add_column("Errors", style="red", justify="right") table.add_column("No Reply", style="yellow", justify="right") table.add_column("Total", style="bold magenta", justify="right") for handle, counts in sorted_handles: table.add_row( f"@{handle}", str(counts['queue']) if counts['queue'] > 0 else "-", str(counts['errors']) if counts['errors'] > 0 else "-", str(counts['no_reply']) if counts['no_reply'] > 0 else "-", str(counts['total']) ) console.print(table) # Summary statistics total_notifications = sum(h['total'] for h in handle_counts.values()) avg_per_handle = total_notifications / len(handle_counts) console.print(f"\n[bold]Summary:[/bold]") console.print(f" Total notifications: {total_notifications}") console.print(f" Unique handles: {len(handle_counts)}") console.print(f" Average per handle: {avg_per_handle:.1f}") # Top user info if sorted_handles: top_handle, top_counts = sorted_handles[0] percentage = (top_counts['total'] / total_notifications) * 100 console.print(f" Most active: @{top_handle} ({top_counts['total']} notifications, {percentage:.1f}% of total)")
def stats(): """Show queue statistics.""" # Dictionary to store stats per agent agent_stats = {} # Check all agent directories for agent_dir in QUEUE_BASE_DIR.glob("*"): if agent_dir.is_dir() and "-" in agent_dir.name: agent_id = agent_dir.name stats_data = { 'queue': {'count': 0, 'handles': set()}, 'errors': {'count': 0, 'handles': set()}, 'no_reply': {'count': 0, 'handles': set()} } # Collect stats for this agent for subdir, key in [(agent_dir, 'queue'), (agent_dir / 'errors', 'errors'), (agent_dir / 'no_reply', 'no_reply')]: if not subdir.exists(): continue for filepath in subdir.glob("*.json"): if filepath.is_dir(): continue notif = load_notification(filepath) if notif and isinstance(notif, dict): stats_data[key]['count'] += 1 handle = notif.get('author', {}).get('handle', 'unknown') stats_data[key]['handles'].add(handle) agent_stats[agent_id] = stats_data # Also check old flat structure if QUEUE_BASE_DIR.exists(): legacy_stats = { 'queue': {'count': 0, 'handles': set()}, 'errors': {'count': 0, 'handles': set()}, 'no_reply': {'count': 0, 'handles': set()} } # Check for legacy queue files for filepath in QUEUE_BASE_DIR.glob("*.json"): if filepath.name != "processed_notifications.json": notif = load_notification(filepath) if notif and isinstance(notif, dict): legacy_stats['queue']['count'] += 1 handle = notif.get('author', {}).get('handle', 'unknown') legacy_stats['queue']['handles'].add(handle) # Check legacy subdirectories for subdir, key in [(QUEUE_BASE_DIR / 'errors', 'errors'), (QUEUE_BASE_DIR / 'no_reply', 'no_reply')]: if subdir.exists(): for filepath in subdir.glob("*.json"): notif = load_notification(filepath) if notif and isinstance(notif, dict): legacy_stats[key]['count'] += 1 handle = notif.get('author', {}).get('handle', 'unknown') legacy_stats[key]['handles'].add(handle) if any(legacy_stats[k]['count'] > 0 for k in legacy_stats): agent_stats['legacy'] = legacy_stats # Display stats for each agent for agent_id, stats_data in agent_stats.items(): agent_name = agent_id[:12] + "..." if len(agent_id) > 15 else agent_id table = Table(title=f"Queue Statistics - {agent_name}") table.add_column("Location", style="cyan") table.add_column("Count", style="yellow") table.add_column("Unique Handles", style="green") for key, label in [('queue', 'Active Queue'), ('errors', 'Errors'), ('no_reply', 'No Reply')]: table.add_row( label, str(stats_data[key]['count']), str(len(stats_data[key]['handles'])) ) console.print(table) console.print() # Add spacing between agents
def main(): parser = argparse.ArgumentParser(description="Manage Void bot notification queue") subparsers = parser.add_subparsers(dest='command', help='Commands') # List command list_parser = subparsers.add_parser('list', help='List notifications in queue') list_parser.add_argument('--handle', help='Filter by handle (partial match)') list_parser.add_argument('--all', action='store_true', help='Include errors and no_reply folders') list_parser.add_argument('--agent-id', help='Filter by specific agent ID') # Delete command delete_parser = subparsers.add_parser('delete', help='Delete notifications from a specific handle') delete_parser.add_argument('handle', help='Handle to delete notifications from') delete_parser.add_argument('--dry-run', action='store_true', help='Show what would be deleted without deleting') delete_parser.add_argument('--force', action='store_true', help='Skip confirmation prompt') delete_parser.add_argument('--agent-id', help='Only delete from specific agent ID') # Stats command stats_parser = subparsers.add_parser('stats', help='Show queue statistics') # Count command count_parser = subparsers.add_parser('count', help='Show detailed count by handle') count_parser.add_argument('--agent-id', help='Count for specific agent ID only') args = parser.parse_args() if args.command == 'list': list_notifications(args.handle, args.all, getattr(args, 'agent_id', None)) elif args.command == 'delete': delete_by_handle(args.handle, args.dry_run, args.force, getattr(args, 'agent_id', None)) elif args.command == 'stats': stats() elif args.command == 'count': count_by_handle(getattr(args, 'agent_id', None)) else: parser.print_help()
if __name__ == "__main__": main()