SelfMemory

""" Command Line Interface for SelfMemory. Provides easy-to-use commands for starting servers, managing data, and working with different deployment modes. """ import argparse import logging import sys from pathlib import Path from .config import ( create_sample_configs, load_config, ) from .memory import SelfMemory logger = logging.getLogger(__name__) def serve_command(args): """Start the SelfMemory API server.""" try: import uvicorn from .api.server import create_app # Import to verify dependencies are available create_app() except ImportError: print( "❌ Server dependencies not installed. Run: pip install selfmemory[server]" ) sys.exit(1) # Override configuration with CLI arguments if args.storage_type: import os os.environ["SELFMEMORY_STORAGE_TYPE"] = args.storage_type if args.data_dir: import os os.environ["SELFMEMORY_DATA_DIR"] = args.data_dir if args.mongodb_uri: import os os.environ["MONGODB_URI"] = args.mongodb_uri print(f"🚀 Starting SelfMemory API server on {args.host}:{args.port}") print(f"📊 Storage backend: {args.storage_type or 'auto-detected'}") print(f"🔗 API docs: http://{args.host}:{args.port}/docs") # Start server uvicorn.run( "selfmemory.main:app", host=args.host, port=args.port, reload=args.reload, log_level="debug" if args.debug else "info", ) def init_command(args): """Initialize SelfMemory in current directory.""" current_dir = Path.cwd() print(f"🏗️ Initializing SelfMemory in: {current_dir}") # Create sample configuration files try: create_sample_configs() print("✅ Sample configuration files created in ~/.selfmemory/") print(" - config-simple.yaml (file-based storage)") print(" - config-enterprise.yaml (MongoDB + OAuth)") except Exception as e: logger.warning(f"Failed to create sample configs: {e}") # Test basic functionality try: memory = SelfMemory() test_user = args.user or "test_user" # Test adding a memory result = memory.add("SelfMemory initialization test", user_id=test_user) if result.get("success", False): print(f"✅ Successfully added test memory for user: {test_user}") # Test searching search_results = memory.search("initialization", user_id=test_user) if search_results.get("results"): print( f"✅ Search working - found {len(search_results['results'])} results" ) else: print("⚠️ Search returned no results") else: print(f"❌ Failed to add test memory: {result}") # Show statistics stats = memory.get_user_stats(test_user) print(f"📊 User stats: {stats.get('memory_count', 0)} memories") print(f"🗄️ Storage backend: {stats.get('storage_backend', 'unknown')}") memory.close() except Exception as e: print(f"❌ Initialization test failed: {e}") print("💡 Try: pip install selfmemory[full] for all dependencies") def test_command(args): """Test SelfMemory functionality.""" print("🧪 Testing SelfMemory functionality...") try: # Test with specified storage type or auto-detect memory = ( SelfMemory(storage_type=args.storage_type) if args.storage_type else SelfMemory() ) test_user = args.user or "cli_test_user" # Add test memories test_memories = [ "I love pizza and hate broccoli", "Meeting with Sarah and Mike about project X", "Dentist appointment tomorrow at 3pm", "Favorite color is blue, favorite season is autumn", ] print(f"📝 Adding {len(test_memories)} test memories for user: {test_user}") for i, memory_text in enumerate(test_memories, 1): result = memory.add(memory_text, user_id=test_user) if result.get("success", False): print(f" ✅ Memory {i} added successfully") else: print(f" ❌ Memory {i} failed: {result}") # Test different search methods print("\n🔍 Testing search functionality:") search_tests = [ ("pizza", "general search"), ("project", "work-related search"), ("tomorrow", "temporal search"), ] for query, description in search_tests: results = memory.search(query, user_id=test_user, limit=3) result_count = len(results.get("results", [])) print(f" 🔸 {description}: {result_count} results for '{query}'") # Test user statistics stats = memory.get_user_stats(test_user) print("\n📊 Final statistics:") print(f" 👤 User: {test_user}") print(f" 💾 Storage: {stats.get('storage_backend', 'unknown')}") print(f" 📝 Memories: {stats.get('memory_count', 0)}") memory.close() print("✅ All tests completed successfully!") except Exception as e: print(f"❌ Test failed: {e}") print("💡 Make sure Qdrant is running and dependencies are installed") def config_command(args): """Show current configuration.""" try: config = load_config() print("⚙️ Current SelfMemory Configuration:") print(f"📁 Storage Type: {config.storage.type}") if config.storage.type == "file": print(f"📂 Data Directory: {config.storage.path}") elif config.storage.type == "mongodb": print(f"🗄️ MongoDB URI: {config.storage.mongodb_uri}") print(f"🔐 Auth Type: {config.auth.type}") print(f"🗂️ Qdrant Host: {config.qdrant.host}:{config.qdrant.port}") print(f"🤖 Embedding Provider: {config.embedding.provider}") print(f"🎯 Embedding Model: {config.embedding.model}") if args.verbose: print("\n📋 Full Configuration:") import json print(json.dumps(config.to_dict(), indent=2)) except Exception as e: print(f"❌ Failed to load configuration: {e}") def stats_command(args): """Show storage statistics.""" try: memory = ( SelfMemory(storage_type=args.storage_type) if args.storage_type else SelfMemory() ) print("📊 SelfMemory Storage Statistics:") if hasattr(memory.store, "get_stats"): stats = memory.store.get_stats() print(f"🗄️ Storage Type: {stats.get('storage_type', 'unknown')}") print(f"👥 Total Users: {stats.get('total_users', 0)}") if args.verbose: # Omitted logging of total and active API keys to avoid exposing sensitive information. print("\n📋 Detailed Statistics:") import json print(json.dumps(stats, indent=2, default=str)) else: print("⚠️ Statistics not available for this storage backend") memory.close() except Exception as e: print(f"❌ Failed to get statistics: {e}") def main(): """Main CLI entry point.""" parser = argparse.ArgumentParser( prog="selfmemory", description="SelfMemory - Enhanced Memory Management CLI", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: selfmemory serve # Start API server (auto-detect mode) selfmemory serve --storage file # Start with file backend selfmemory serve --storage mongodb # Start with MongoDB backend selfmemory init # Initialize in current directory selfmemory test # Test functionality selfmemory config # Show current configuration selfmemory stats # Show storage statistics """, ) subparsers = parser.add_subparsers(dest="command", help="Available commands") # Serve command serve_parser = subparsers.add_parser("serve", help="Start API server") serve_parser.add_argument("--host", default="0.0.0.0", help="Host to bind to") serve_parser.add_argument( "--port", type=int, default=8081, help="Port to listen on" ) serve_parser.add_argument( "--storage-type", choices=["file", "mongodb"], help="Storage backend type" ) serve_parser.add_argument("--data-dir", help="Data directory (file storage)") serve_parser.add_argument("--mongodb-uri", help="MongoDB URI") serve_parser.add_argument( "--reload", action="store_true", help="Enable auto-reload" ) serve_parser.add_argument("--debug", action="store_true", help="Enable debug mode") serve_parser.set_defaults(func=serve_command) # Init command init_parser = subparsers.add_parser("init", help="Initialize SelfMemory") init_parser.add_argument("--user", default="cli_user", help="Test user ID") init_parser.set_defaults(func=init_command) # Test command test_parser = subparsers.add_parser("test", help="Test functionality") test_parser.add_argument( "--storage-type", choices=["file", "mongodb"], help="Storage backend to test" ) test_parser.add_argument("--user", default="cli_test_user", help="Test user ID") test_parser.set_defaults(func=test_command) # Config command config_parser = subparsers.add_parser("config", help="Show configuration") config_parser.add_argument( "--verbose", "-v", action="store_true", help="Show full config" ) config_parser.set_defaults(func=config_command) # Stats command stats_parser = subparsers.add_parser("stats", help="Show statistics") stats_parser.add_argument( "--storage-type", choices=["file", "mongodb"], help="Storage backend type" ) stats_parser.add_argument( "--verbose", "-v", action="store_true", help="Show detailed stats" ) stats_parser.set_defaults(func=stats_command) # Parse arguments args = parser.parse_args() # Set up logging level = logging.DEBUG if getattr(args, "debug", False) else logging.INFO logging.basicConfig( level=level, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) # Execute command if hasattr(args, "func"): try: args.func(args) except KeyboardInterrupt: print("\n👋 Goodbye!") sys.exit(0) except Exception as e: logger.error(f"Command failed: {e}") print(f"❌ Command failed: {e}") sys.exit(1) else: parser.print_help() if __name__ == "__main__": main()