"""
Utility functions for Keet messaging integration.
"""
import re
import logging
import subprocess
import json
import os
from pathlib import Path
from typing import Optional, Dict, Any, List, Union, Tuple
logger = logging.getLogger(__name__)
# Constants
KEET_INVITE_PREFIX = "pear://keet/"
Z_BASE32_PATTERN = re.compile(r"^[a-z2-7]+$")
HYPERCORE_KEY_LENGTH = 52 # Standard length for z-base-32 encoded Hypercore keys
def is_valid_hypercore_key(key: str) -> bool:
"""
Validate if a string is a valid Hypercore key.
Hypercore keys are typically z-base-32 encoded and have a specific length.
Args:
key: The key to validate
Returns:
bool: True if the key is valid, False otherwise
"""
if not key or not isinstance(key, str):
return False
# Check if the key matches the z-base-32 character set and has the correct length
return Z_BASE32_PATTERN.match(key) is not None and len(key) == HYPERCORE_KEY_LENGTH
def extract_room_key_from_invite(invite_link: str) -> str:
"""
Extract the room key from a Keet invite link.
Args:
invite_link: The Keet invite link (e.g., "pear://keet/abc123...")
Returns:
str: The extracted room key
"""
if not invite_link:
raise ValueError("Invite link cannot be empty")
# Remove the prefix
if invite_link.startswith(KEET_INVITE_PREFIX):
room_key = invite_link[len(KEET_INVITE_PREFIX):]
else:
room_key = invite_link
# Remove any additional path components or query parameters
room_key = room_key.split("/")[0].split("?")[0]
return room_key
def normalize_hypercore_key(key: str) -> str:
"""
Normalize a Hypercore key to ensure it's in the correct format.
This function attempts to fix common issues with Hypercore keys.
Args:
key: The key to normalize
Returns:
str: The normalized key
"""
# Remove any whitespace
key = key.strip()
# Remove the prefix if present
if key.startswith(KEET_INVITE_PREFIX):
key = key[len(KEET_INVITE_PREFIX):]
# Remove any additional path components or query parameters
key = key.split("/")[0].split("?")[0]
# Ensure the key is lowercase (z-base-32 is case-insensitive but conventionally lowercase)
key = key.lower()
return key
def run_node_script(script_path: str, args: List[str] = None) -> Tuple[bool, str]:
"""
Run a Node.js script and return the result.
Args:
script_path: Path to the Node.js script
args: Additional arguments to pass to the script
Returns:
Tuple[bool, str]: A tuple containing a success flag and the output
"""
if args is None:
args = []
cmd = ["node", str(script_path)] + args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False # Don't raise an exception on non-zero exit code
)
# Log the output
for line in result.stdout.splitlines():
logger.info(line)
for line in result.stderr.splitlines():
logger.error(line)
return result.returncode == 0, result.stdout
except Exception as e:
logger.error(f"Error running Node.js script: {e}")
return False, str(e)
def get_default_storage_path() -> Path:
"""
Get the default storage path for Keet data.
Returns:
Path: The default storage path
"""
# Use the user's home directory
home_dir = Path.home()
# Create a .ollamatools directory if it doesn't exist
ollamatools_dir = home_dir / ".ollamatools"
ollamatools_dir.mkdir(exist_ok=True)
# Create a keet directory if it doesn't exist
keet_dir = ollamatools_dir / "keet"
keet_dir.mkdir(exist_ok=True)
return keet_dir