| |
| """Memory CLI - Search and retrieve saved conversations.""" |
|
|
| import argparse |
| import json |
| import os |
| from memory_system import ConversationMemory |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description="Conversation Memory CLI") |
| subparsers = parser.add_subparsers(dest="command") |
|
|
| search_parser = subparsers.add_parser("search", help="Search memories by text") |
| search_parser.add_argument("query", help="Search query") |
| search_parser.add_argument("--thread", help="Limit to specific thread") |
|
|
| subparsers.add_parser("threads", help="List all conversation threads") |
|
|
| show_parser = subparsers.add_parser("show", help="Show a specific thread") |
| show_parser.add_argument("thread_id", help="Thread ID to display") |
|
|
| export_parser = subparsers.add_parser("export", help="Export memories") |
| export_parser.add_argument("--thread", help="Export specific thread") |
| export_parser.add_argument("--all", action="store_true", help="Export all threads") |
| export_parser.add_argument("--format", choices=["json", "md"], default="json") |
|
|
| args = parser.parse_args() |
| memory = ConversationMemory(user_id="scottzilla") |
|
|
| if args.command == "search": |
| results = memory.search(args.query, thread_id=args.thread) |
| print(f"\n🔍 Found {len(results)} results for '{args.query}':\n") |
| for r in results: |
| thread = r.get("thread_id", "unknown") |
| print(f" [{r['role'].upper()}] {r['content'][:120]}...") |
| print(f" Thread: {thread} | {r['timestamp']}\n") |
|
|
| elif args.command == "threads": |
| threads = memory.get_all_threads() |
| print(f"\n📚 {len(threads)} conversation threads:\n") |
| for t in threads: |
| print(f" 📁 {t['thread_id']}") |
| print(f" Messages: {t['message_count']} | Started: {t['started']} | Last: {t['last_message']}") |
|
|
| elif args.command == "show": |
| messages = memory.get_thread(args.thread_id) |
| print(f"\n📄 Thread: {args.thread_id} ({len(messages)} messages)\n") |
| for m in messages: |
| emoji = "👤" if m['role'] == 'user' else "🤖" |
| print(f"{emoji} {m['role'].title()} ({m['timestamp']}):") |
| print(f" {m['content']}\n") |
|
|
| elif args.command == "export": |
| os.makedirs("./memory_exports", exist_ok=True) |
| if args.all: |
| memory.export_to_json("./memory_exports/all_memories.json") |
| print("✅ Exported all memories") |
| elif args.thread: |
| ext = "md" if args.format == "md" else "json" |
| filepath = f"./memory_exports/{args.thread}.{ext}" |
| if args.format == "md": |
| memory.export_to_markdown(filepath, args.thread) |
| else: |
| memory.export_to_json(filepath, args.thread) |
| else: |
| print("❌ Use --thread <id> or --all") |
| else: |
| parser.print_help() |
|
|
| memory.close() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
