"""
Keet client for ollamatools.
This module provides a Python interface to the Keet messaging platform.
"""
import asyncio
import json
import logging
import os
import tempfile
from pathlib import Path
from typing import Dict, List, Optional, Any, Union, Tuple
from .utils import (
is_valid_hypercore_key,
extract_room_key_from_invite,
normalize_hypercore_key,
run_node_script,
get_default_storage_path,
)
logger = logging.getLogger(__name__)
class KeetClient:
"""
Client for interacting with the Keet messaging platform.
This client uses the Node.js Keet API via subprocess calls.
"""
def __init__(self, storage_path: Optional[Union[str, Path]] = None):
"""
Initialize the Keet client.
Args:
storage_path: Path to store Keet data. If None, a default path will be used.
"""
self.storage_path = Path(storage_path) if storage_path else get_default_storage_path()
self.storage_path.mkdir(parents=True, exist_ok=True)
# Path to the JavaScript bridge scripts
self._script_dir = Path(__file__).parent.parent.parent.parent.parent / "tests"
self._test_script = self._script_dir / "keet_messaging_test.js"
self._mock_script = self._script_dir / "keet_messaging_mock_test.js"
# Check if the scripts exist
if not self._test_script.exists():
logger.warning(f"Keet test script not found at {self._test_script}")
if not self._mock_script.exists():
logger.warning(f"Keet mock script not found at {self._mock_script}")
# Flag to indicate whether to use the mock implementation
self._use_mock = not self._test_script.exists()
logger.info(f"Initialized Keet client with storage path: {self.storage_path}")
logger.info(f"Using {'mock' if self._use_mock else 'real'} implementation")
async def get_identity(self) -> Dict[str, Any]:
"""
Get the current identity or create a new one if none exists.
Returns:
Dict[str, Any]: The identity information
"""
# Create a temporary file to store the result
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as temp_file:
temp_path = temp_file.name
try:
# Run the script with the "get-identity" command
script = self._mock_script if self._use_mock else self._test_script
success, output = run_node_script(script, ["--command", "get-identity", "--storage", str(self.storage_path), "--output", temp_path])
if not success:
logger.error(f"Failed to get identity: {output}")
return {}
# Read the result from the temporary file
with open(temp_path, "r") as f:
result = json.load(f)
return result
finally:
# Clean up the temporary file
if os.path.exists(temp_path):
os.unlink(temp_path)
async def create_identity(self, name: Optional[str] = None) -> Dict[str, Any]:
"""
Create a new identity.
Args:
name: The display name for the identity
Returns:
Dict[str, Any]: The created identity information
"""
# Create a temporary file to store the result
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as temp_file:
temp_path = temp_file.name
try:
# Run the script with the "create-identity" command
script = self._mock_script if self._use_mock else self._test_script
args = ["--command", "create-identity", "--storage", str(self.storage_path), "--output", temp_path]
if name:
args.extend(["--name", name])
success, output = run_node_script(script, args)
if not success:
logger.error(f"Failed to create identity: {output}")
return {}
# Read the result from the temporary file
with open(temp_path, "r") as f:
result = json.load(f)
return result
finally:
# Clean up the temporary file
if os.path.exists(temp_path):
os.unlink(temp_path)
async def get_recent_rooms(self) -> List[Dict[str, Any]]:
"""
Get a list of recent rooms.
Returns:
List[Dict[str, Any]]: List of room information
"""
# Create a temporary file to store the result
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as temp_file:
temp_path = temp_file.name
try:
# Run the script with the "get-recent-rooms" command
script = self._mock_script if self._use_mock else self._test_script
success, output = run_node_script(script, ["--command", "get-recent-rooms", "--storage", str(self.storage_path), "--output", temp_path])
if not success:
logger.error(f"Failed to get recent rooms: {output}")
return []
# Read the result from the temporary file
with open(temp_path, "r") as f:
result = json.load(f)
return result
finally:
# Clean up the temporary file
if os.path.exists(temp_path):
os.unlink(temp_path)
async def join_room(self, room_key: str) -> bool:
"""
Join a room using a room key.
Args:
room_key: The room key or invite link
Returns:
bool: True if successful, False otherwise
"""
# Normalize the room key
normalized_key = normalize_hypercore_key(room_key)
# Validate the room key
if not is_valid_hypercore_key(normalized_key):
logger.error(f"Invalid Hypercore key: {normalized_key}")
return False
# Run the script with the "join-room" command
script = self._mock_script if self._use_mock else self._test_script
success, output = run_node_script(script, ["--command", "join-room", "--storage", str(self.storage_path), "--room-key", normalized_key])
if not success:
logger.error(f"Failed to join room: {output}")
return False
return True
async def get_messages(self, room_key: str) -> List[Dict[str, Any]]:
"""
Get messages from a room.
Args:
room_key: The room key
Returns:
List[Dict[str, Any]]: List of messages
"""
# Normalize the room key
normalized_key = normalize_hypercore_key(room_key)
# Validate the room key
if not is_valid_hypercore_key(normalized_key):
logger.error(f"Invalid Hypercore key: {normalized_key}")
return []
# Create a temporary file to store the result
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as temp_file:
temp_path = temp_file.name
try:
# Run the script with the "get-messages" command
script = self._mock_script if self._use_mock else self._test_script
success, output = run_node_script(script, ["--command", "get-messages", "--storage", str(self.storage_path), "--room-key", normalized_key, "--output", temp_path])
if not success:
logger.error(f"Failed to get messages: {output}")
return []
# Read the result from the temporary file
with open(temp_path, "r") as f:
result = json.load(f)
return result
finally:
# Clean up the temporary file
if os.path.exists(temp_path):
os.unlink(temp_path)
async def send_message(self, room_key: str, message: str) -> bool:
"""
Send a message to a room.
Args:
room_key: The room key
message: The message to send
Returns:
bool: True if successful, False otherwise
"""
# Normalize the room key
normalized_key = normalize_hypercore_key(room_key)
# Validate the room key
if not is_valid_hypercore_key(normalized_key):
logger.error(f"Invalid Hypercore key: {normalized_key}")
return False
# Run the script with the "send-message" command
script = self._mock_script if self._use_mock else self._test_script
success, output = run_node_script(script, ["--command", "send-message", "--storage", str(self.storage_path), "--room-key", normalized_key, "--message", message])
if not success:
logger.error(f"Failed to send message: {output}")
return False
return True