File size: 2,981 Bytes
ad38432 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | #!/usr/bin/env python3
"""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()
|