yumoqing/hermes-web-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfi

Browse Source
This commit is contained in:
yumoqing 2026-04-24 12:10:49 +08:00
parent 166e8c2273
commit 120487d4a0
1 changed files with 493 additions and 493 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

854
hermes_web_cli/init.py
View File

@ -28,537 +28,537 @@ from .db_tables import TABLE_DEFINITIONS
from .crud_ops import SERVICES_CRUD, SESSIONS_CRUD, SETTINGS_CRUD from .crud_ops import SERVICES_CRUD, SESSIONS_CRUD, SETTINGS_CRUD
def load_hermes_web_cli(): def load_hermes_web_cli():
"""Initialize and load the hermes-web-cli module. """Initialize and load the hermes-web-cli module.
This function is called by Sage system during module loading. This function is called by Sage system during module loading.
It registers all module functions with the ServerEnv instance It registers all module functions with the ServerEnv instance
so they can be called directly from .ui and .dspy files. so they can be called directly from .ui and .dspy files.
""" """
from ahserver.serverenv import ServerEnv from ahserver.serverenv import ServerEnv
# Initialize database tables if needed # Initialize database tables if needed
try: try:
from .init_db import init_database from .init_db import init_database
import asyncio import asyncio
# Run database initialization in a new event loop if needed # Run database initialization in a new event loop if needed
try: try:
asyncio.get_running_loop() asyncio.get_running_loop()
# If we're already in an async context, create a task # If we're already in an async context, create a task
asyncio.create_task(init_database()) asyncio.create_task(init_database())
except RuntimeError: except RuntimeError:
# No running loop, run synchronously # No running loop, run synchronously
asyncio.run(init_database()) asyncio.run(init_database())
except Exception as e: except Exception as e:
print(f"Warning: Database initialization failed: {str(e)}") print(f"Warning: Database initialization failed: {str(e)}")
# Continue loading even if DB init fails - functions will handle errors gracefully # Continue loading even if DB init fails - functions will handle errors gracefully
# Get the ServerEnv instance # Get the ServerEnv instance
env = ServerEnv() env = ServerEnv()
# Register all module functions with ServerEnv # Register all module functions with ServerEnv
env.get_setting = get_setting env.get_setting = get_setting
env.save_setting = save_setting env.save_setting = save_setting
env.get_all_services = get_all_services env.get_all_services = get_all_services
env.create_service = create_service env.create_service = create_service
env.delete_service = delete_service env.delete_service = delete_service
env.get_service_by_id = get_service_by_id env.get_service_by_id = get_service_by_id
env.test_service_connection = test_service_connection env.test_service_connection = test_service_connection
env.create_session = create_session env.create_session = create_session
env.send_message_to_service = send_message_to_service env.send_message_to_service = send_message_to_service
env.get_session_messages = get_session_messages env.get_session_messages = get_session_messages
env.get_active_sessions = get_active_sessions env.get_active_sessions = get_active_sessions
env.get_recent_sessions = get_recent_sessions env.get_recent_sessions = get_recent_sessions
env.get_session_by_id = get_session_by_id env.get_session_by_id = get_session_by_id
env.validate_service_url = validate_service_url env.validate_service_url = validate_service_url
env.generate_session_id = generate_session_id env.generate_session_id = generate_session_id
# Also register the user context helper if needed # Also register the user context helper if needed
return True return True
# Database operations using sqlor-database-module # Database operations using sqlor-database-module
async def get_all_services(userid: str) -> List[Dict]: async def get_all_services(userid: str) -> List[Dict]:
"""Get all registered Hermes services for the current user from database.""" """Get all registered Hermes services for the current user from database."""
try: try:
# Query services table with user_id filter using sqlor-database-module # Query services table with user_id filter using sqlor-database-module
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['read_all']['sql_template'] sql_template = SERVICES_CRUD['operations']['read_all']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id}) recs = await sor.sqlExe(sql_template, {'user_id': user_id})
# Convert datetime objects to ISO format strings for JSON serialization # Convert datetime objects to ISO format strings for JSON serialization
result = [] result = []
for rec in recs: for rec in recs:
service_dict = dict(rec) service_dict = dict(rec)
if 'created_at' in service_dict and service_dict['created_at']: if 'created_at' in service_dict and service_dict['created_at']:
service_dict['created_at'] = service_dict['created_at'].isoformat() service_dict['created_at'] = service_dict['created_at'].isoformat()
if 'updated_at' in service_dict and service_dict['updated_at']: if 'updated_at' in service_dict and service_dict['updated_at']:
service_dict['updated_at'] = service_dict['updated_at'].isoformat() service_dict['updated_at'] = service_dict['updated_at'].isoformat()
result.append(service_dict) result.append(service_dict)
return result return result
except Exception as e: except Exception as e:
print(f"Error getting services: {str(e)}") print(f"Error getting services: {str(e)}")
# Return empty list on error # Return empty list on error
return [] return []
except Exception as e: except Exception as e:
print(f"Error getting services: {e}") print(f"Error getting services: {e}")
return [] return []
async def create_service(userid: str, name: str, url: str, description: str = "", apikey: str = "") -> str: async def create_service(userid: str, name: str, url: str, description: str = "", apikey: str = "") -> str:
"""Create a new Hermes service registration for the current user.""" """Create a new Hermes service registration for the current user."""
try: try:
# Validate service URL # Validate service URL
if not await validate_service_url(url): if not await validate_service_url(url):
raise ValueError("Invalid service URL") raise ValueError("Invalid service URL")
service_id = getID() service_id = getID()
# Save to database using sqlor-database-module # Save to database using sqlor-database-module
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['create']['sql_template'] sql_template = SERVICES_CRUD['operations']['create']['sql_template']
await sor.sqlExe(sql_template, { await sor.sqlExe(sql_template, {
'id': service_id, 'id': service_id,
'user_id': user_id, 'user_id': user_id,
'name': name, 'name': name,
'service_url': url, 'service_url': url,
'description': description, 'description': description,
'status': 'active' 'status': 'active'
}) })
return service_id return service_id
except Exception as e: except Exception as e:
print(f"Error creating service: {str(e)}") print(f"Error creating service: {str(e)}")
raise raise
async def delete_service(userid: str, service_id: str) -> bool: async def delete_service(userid: str, service_id: str) -> bool:
"""Delete a Hermes service registration (only if owned by current user).""" """Delete a Hermes service registration (only if owned by current user)."""
try: try:
# Verify service belongs to current user before deletion # Verify service belongs to current user before deletion
service = await get_service_by_id(service_id) service = await get_service_by_id(service_id)
if not service: if not service:
return False return False
if service.get("user_id") != user_id: if service.get("user_id") != user_id:
print(f"Permission denied: Service {service_id} does not belong to user {user_id}") print(f"Permission denied: Service {service_id} does not belong to user {user_id}")
return False return False
# Delete from database using sqlor-database-module # Delete from database using sqlor-database-module
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['delete']['sql_template'] sql_template = SERVICES_CRUD['operations']['delete']['sql_template']
await sor.sqlExe(sql_template, { await sor.sqlExe(sql_template, {
'service_id': service_id, 'service_id': service_id,
'user_id': user_id 'user_id': user_id
}) })
# Also delete associated sessions # Also delete associated sessions
async with db.sqlorContext('hermes-web-cli') as sor: async with db.sqlorContext('hermes-web-cli') as sor:
await sor.sqlExe(""" await sor.sqlExe("""
DELETE FROM sessions DELETE FROM sessions
WHERE service_id = ${service_id}$ AND user_id = ${user_id}$ WHERE service_id = ${service_id}$ AND user_id = ${user_id}$
""", { """, {
'service_id': service_id, 'service_id': service_id,
'user_id': user_id 'user_id': user_id
}) })
return True return True
except Exception as e: except Exception as e:
print(f"Error deleting service: {str(e)}") print(f"Error deleting service: {str(e)}")
return False return False
async def get_service_by_id(userid: str, service_id: str) -> Optional[Dict]: async def get_service_by_id(userid: str, service_id: str) -> Optional[Dict]:
"""Get service configuration by ID (only if owned by current user).""" """Get service configuration by ID (only if owned by current user)."""
try: try:
# Query database directly with user_id filter for security # Query database directly with user_id filter for security
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['read_by_id']['sql_template'] sql_template = SERVICES_CRUD['operations']['read_by_id']['sql_template']
recs = await sor.sqlExe(sql_template, { recs = await sor.sqlExe(sql_template, {
'service_id': service_id, 'service_id': service_id,
'user_id': user_id 'user_id': user_id
}) })
if len(recs) > 0: if len(recs) > 0:
service_dict = dict(recs[0]) service_dict = dict(recs[0])
if 'created_at' in service_dict and service_dict['created_at']: if 'created_at' in service_dict and service_dict['created_at']:
service_dict['created_at'] = service_dict['created_at'].isoformat() service_dict['created_at'] = service_dict['created_at'].isoformat()
if 'updated_at' in service_dict and service_dict['updated_at']: if 'updated_at' in service_dict and service_dict['updated_at']:
service_dict['updated_at'] = service_dict['updated_at'].isoformat() service_dict['updated_at'] = service_dict['updated_at'].isoformat()
return service_dict return service_dict
return None return None
except Exception as e: except Exception as e:
print(f"Error getting service: {str(e)}") print(f"Error getting service: {str(e)}")
return None return None
# Service connection testing # Service connection testing
async def test_service_connection(service_id: str) -> Tuple[bool, str]: async def test_service_connection(service_id: str) -> Tuple[bool, str]:
"""Test connection to a Hermes service endpoint. """Test connection to a Hermes service endpoint.
Args: Args:
service_id: The ID of the service to test service_id: The ID of the service to test
Returns: Returns:
Tuple[bool, str]: (is_connected, status_message) Tuple[bool, str]: (is_connected, status_message)
""" """
try: try:
# Get service configuration (verify it belongs to current user) # Get service configuration (verify it belongs to current user)
service = await get_service_by_id(service_id) service = await get_service_by_id(service_id)
if not service: if not service:
return False, "Service not found or access denied" return False, "Service not found or access denied"
url = service["service_url"] url = service["service_url"]
apikey = service.get("apikey", "") apikey = service.get("apikey", "")
# Prepare headers # Prepare headers
headers = {} headers = {}
if apikey: if apikey:
headers["Authorization"] = f"Bearer {apikey}" headers["Authorization"] = f"Bearer {apikey}"
# Test the /health endpoint or similar # Test the /health endpoint or similar
timeout = aiohttp.ClientTimeout(total=10) timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession(timeout=timeout) as session: async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(f"{url.rstrip('/')}/health", headers=headers) as response: async with session.get(f"{url.rstrip('/')}/health", headers=headers) as response:
if response.status == 200: if response.status == 200:
return True, "Connected" return True, "Connected"
else: else:
return False, f"HTTP {response.status}" return False, f"HTTP {response.status}"
except asyncio.TimeoutError: except asyncio.TimeoutError:
return False, "Connection timeout" return False, "Connection timeout"
except aiohttp.ClientConnectorError: except aiohttp.ClientConnectorError:
return False, "Connection refused" return False, "Connection refused"
except Exception as e: except Exception as e:
return False, f"Error: {str(e)}" return False, f"Error: {str(e)}"
# Session management # Session management
async def create_session(service_id: str, user_id: str, user_message: str = "") -> str: async def create_session(service_id: str, user_id: str, user_message: str = "") -> str:
"""Create a new session with a Hermes service.""" """Create a new session with a Hermes service."""
try: try:
# Get service configuration (verify it belongs to current user) # Get service configuration (verify it belongs to current user)
service = await get_service_by_id(service_id) service = await get_service_by_id(service_id)
if not service: if not service:
raise ValueError(f"Service {service_id} not found or access denied") raise ValueError(f"Service {service_id} not found or access denied")
service_url = service["service_url"] service_url = service["service_url"]
apikey = service.get("apikey", "") apikey = service.get("apikey", "")
# Prepare headers # Prepare headers
headers = { headers = {
"Content-Type": "application/json" "Content-Type": "application/json"
} }
# Add Authorization header if API key is provided # Add Authorization header if API key is provided
if apikey: if apikey:
headers["Authorization"] = f"Bearer {apikey}" headers["Authorization"] = f"Bearer {apikey}"
# Call remote service API to create session # Call remote service API to create session
timeout = aiohttp.ClientTimeout(total=30) timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session: async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post( async with session.post(
f"{service_url.rstrip('/')}/api/v1/sessions", f"{service_url.rstrip('/')}/api/v1/sessions",
json={ json={
"user_id": user_id, "user_id": user_id,
"initial_message": user_message if user_message else None "initial_message": user_message if user_message else None
}, },
headers=headers headers=headers
) as response: ) as response:
response.raise_for_status() response.raise_for_status()
result = await response.json() result = await response.json()
# Get the session ID from the remote service # Get the session ID from the remote service
remote_session_id = result.get("session_id", "") remote_session_id = result.get("session_id", "")
if not remote_session_id: if not remote_session_id:
raise ValueError("Remote service did not return a session ID") raise ValueError("Remote service did not return a session ID")
# Create local session record in database # Create local session record in database
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['create']['sql_template'] sql_template = SESSIONS_CRUD['operations']['create']['sql_template']
await sor.sqlExe(sql_template, { await sor.sqlExe(sql_template, {
'session_id': remote_session_id, 'session_id': remote_session_id,
'user_id': user_id, 'user_id': user_id,
'service_id': service_id, 'service_id': service_id,
'session_name': None, 'session_name': None,
'service_name': service.get("name", "Unknown Service") 'service_name': service.get("name", "Unknown Service")
}) })
# Return the session ID from the remote service # Return the session ID from the remote service
return remote_session_id return remote_session_id
except Exception as e: except Exception as e:
print(f"Error creating session: {str(e)}") print(f"Error creating session: {str(e)}")
raise raise
async def send_message_to_service(userid: str, service_id: str, session_id: str, message: str) -> Dict: async def send_message_to_service(userid: str, service_id: str, session_id: str, message: str) -> Dict:
"""Send a message to a Hermes service and get response (only if session owned by current user).""" """Send a message to a Hermes service and get response (only if session owned by current user)."""
try: try:
# Verify session belongs to current user before sending message # Verify session belongs to current user before sending message
session = await get_session_by_id(session_id) session = await get_session_by_id(session_id)
if not session: if not session:
raise ValueError(f"Session {session_id} not found or access denied for user {user_id}") raise ValueError(f"Session {session_id} not found or access denied for user {user_id}")
service = await get_service_by_id(userid, service_id) service = await get_service_by_id(userid, service_id)
if not service: if not service:
raise ValueError(f"Service {service_id} not found or access denied for user {user_id}") raise ValueError(f"Service {service_id} not found or access denied for user {user_id}")
service_url = service["service_url"] service_url = service["service_url"]
apikey = service.get("apikey", "") apikey = service.get("apikey", "")
# Prepare headers # Prepare headers
headers = { headers = {
"Content-Type": "application/json" "Content-Type": "application/json"
} }
# Add Authorization header if API key is provided # Add Authorization header if API key is provided
if apikey: if apikey:
headers["Authorization"] = f"Bearer {apikey}" headers["Authorization"] = f"Bearer {apikey}"
# Call remote service API # Call remote service API
timeout = aiohttp.ClientTimeout(total=30) timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session: async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post( async with session.post(
f"{service_url.rstrip('/')}/api/chat", f"{service_url.rstrip('/')}/api/chat",
json={ json={
"session_id": session_id, "session_id": session_id,
"message": message "message": message
}, },
headers=headers headers=headers
) as response: ) as response:
response.raise_for_status() response.raise_for_status()
return await response.json() return await response.json()
except Exception as e: except Exception as e:
print(f"Error sending message: {e}") print(f"Error sending message: {e}")
raise raise
async def get_session_messages(userid: str, session_id: str) -> List[Dict]: async def get_session_messages(userid: str, session_id: str) -> List[Dict]:
"""Get all messages for a session (only if session owned by current user).""" """Get all messages for a session (only if session owned by current user)."""
try: try:
# Verify session belongs to current user before getting messages # Verify session belongs to current user before getting messages
session = await get_session_by_id(session_id) session = await get_session_by_id(session_id)
if not session: if not session:
print(f"Session {session_id} not found or access denied for user {user_id}") print(f"Session {session_id} not found or access denied for user {user_id}")
return [] return []
# Get the associated service # Get the associated service
service = await get_service_by_id(userid, session['service_id']) service = await get_service_by_id(userid, session['service_id'])
if not service: if not service:
print(f"Service for session {session_id} not found or access denied") print(f"Service for session {session_id} not found or access denied")
return [] return []
service_url = service["service_url"] service_url = service["service_url"]
apikey = service.get("apikey", "") apikey = service.get("apikey", "")
# Prepare headers # Prepare headers
headers = { headers = {
"Content-Type": "application/json" "Content-Type": "application/json"
} }
# Add Authorization header if API key is provided # Add Authorization header if API key is provided
if apikey: if apikey:
headers["Authorization"] = f"Bearer {apikey}" headers["Authorization"] = f"Bearer {apikey}"
# Call remote service API to get messages # Call remote service API to get messages
timeout = aiohttp.ClientTimeout(total=30) timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session: async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get( async with session.get(
f"{service_url.rstrip('/')}/api/v1/sessions/{session_id}/messages", f"{service_url.rstrip('/')}/api/v1/sessions/{session_id}/messages",
headers=headers headers=headers
) as response: ) as response:
response.raise_for_status() response.raise_for_status()
messages = await response.json() messages = await response.json()
# Update session last_active timestamp and message count in local database # Update session last_active timestamp and message count in local database
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
await sor.sqlExe(""" await sor.sqlExe("""
UPDATE sessions UPDATE sessions
SET last_active = CURRENT_TIMESTAMP, SET last_active = CURRENT_TIMESTAMP,
message_count = ${message_count}$ message_count = ${message_count}$
WHERE session_id = ${session_id}$ AND user_id = ${user_id}$ WHERE session_id = ${session_id}$ AND user_id = ${user_id}$
""", { """, {
'session_id': session_id, 'session_id': session_id,
'user_id': user_id, 'user_id': user_id,
'message_count': len(messages) 'message_count': len(messages)
}) })
return messages return messages
except Exception as e: except Exception as e:
print(f"Error getting session messages: {str(e)}") print(f"Error getting session messages: {str(e)}")
return [] return []
# Active sessions management # Active sessions management
async def get_active_sessions(userid: str) -> List[Dict]: async def get_active_sessions(userid: str) -> List[Dict]:
"""Get all active sessions for the current user from database.""" """Get all active sessions for the current user from database."""
try: try:
# Query the sessions table for active sessions belonging to current user using sqlor-database-module # Query the sessions table for active sessions belonging to current user using sqlor-database-module
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['read_active']['sql_template'] sql_template = SESSIONS_CRUD['operations']['read_active']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id}) recs = await sor.sqlExe(sql_template, {'user_id': user_id})
# Convert datetime objects to ISO format strings for JSON serialization # Convert datetime objects to ISO format strings for JSON serialization
result = [] result = []
for rec in recs: for rec in recs:
session_dict = dict(rec) session_dict = dict(rec)
if 'created_at' in session_dict and session_dict['created_at']: if 'created_at' in session_dict and session_dict['created_at']:
session_dict['created_at'] = session_dict['created_at'].isoformat() session_dict['created_at'] = session_dict['created_at'].isoformat()
if 'last_active' in session_dict and session_dict['last_active']: if 'last_active' in session_dict and session_dict['last_active']:
session_dict['last_active'] = session_dict['last_active'].isoformat() session_dict['last_active'] = session_dict['last_active'].isoformat()
result.append(session_dict) result.append(session_dict)
return result return result
except Exception as e: except Exception as e:
print(f"Error getting active sessions: {str(e)}") print(f"Error getting active sessions: {str(e)}")
return [] return []
async def get_recent_sessions(userid: str, limit: int = 5) -> List[Dict]: async def get_recent_sessions(userid: str, limit: int = 5) -> List[Dict]:
"""Get recent sessions for the current user from database, ordered by creation time (most recent first).""" """Get recent sessions for the current user from database, ordered by creation time (most recent first)."""
try: try:
# Query the sessions table for recent sessions belonging to current user using sqlor-database-module # Query the sessions table for recent sessions belonging to current user using sqlor-database-module
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['read_recent']['sql_template'] sql_template = SESSIONS_CRUD['operations']['read_recent']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id, 'limit': limit}) recs = await sor.sqlExe(sql_template, {'user_id': user_id, 'limit': limit})
# Convert datetime objects to ISO format strings for JSON serialization # Convert datetime objects to ISO format strings for JSON serialization
result = [] result = []
for rec in recs: for rec in recs:
session_dict = dict(rec) session_dict = dict(rec)
if 'created_at' in session_dict and session_dict['created_at']: if 'created_at' in session_dict and session_dict['created_at']:
session_dict['created_at'] = session_dict['created_at'].isoformat() session_dict['created_at'] = session_dict['created_at'].isoformat()
if 'last_active' in session_dict and session_dict['last_active']: if 'last_active' in session_dict and session_dict['last_active']:
session_dict['last_active'] = session_dict['last_active'].isoformat() session_dict['last_active'] = session_dict['last_active'].isoformat()
result.append(session_dict) result.append(session_dict)
return result return result
except Exception as e: except Exception as e:
print(f"Error getting recent sessions: {str(e)}") print(f"Error getting recent sessions: {str(e)}")
return [] return []
async def get_session_by_id(userid, session_id: str) -> Optional[Dict]: async def get_session_by_id(userid, session_id: str) -> Optional[Dict]:
"""Get session details by session ID (only if owned by current user).""" """Get session details by session ID (only if owned by current user)."""
try: try:
# Query database directly with user_id filter for security # Query database directly with user_id filter for security
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['read_by_id']['sql_template'] sql_template = SESSIONS_CRUD['operations']['read_by_id']['sql_template']
recs = await sor.sqlExe(sql_template, { recs = await sor.sqlExe(sql_template, {
'session_id': session_id, 'session_id': session_id,
'user_id': user_id 'user_id': user_id
}) })
if len(recs) > 0: if len(recs) > 0:
session_dict = dict(recs[0]) session_dict = dict(recs[0])
if 'created_at' in session_dict and session_dict['created_at']: if 'created_at' in session_dict and session_dict['created_at']:
session_dict['created_at'] = session_dict['created_at'].isoformat() session_dict['created_at'] = session_dict['created_at'].isoformat()
if 'last_active' in session_dict and session_dict['last_active']: if 'last_active' in session_dict and session_dict['last_active']:
session_dict['last_active'] = session_dict['last_active'].isoformat() session_dict['last_active'] = session_dict['last_active'].isoformat()
return session_dict return session_dict
return None return None
except Exception as e: except Exception as e:
print(f"Error getting session by ID: {str(e)}") print(f"Error getting session by ID: {str(e)}")
return None return None
# Utility functions for validation # Utility functions for validation
def validate_service_url(url: str) -> bool: def validate_service_url(url: str) -> bool:
"""Validate if a URL is a valid Hermes service endpoint.""" """Validate if a URL is a valid Hermes service endpoint."""
if not url.startswith(('http://', 'https://')): if not url.startswith(('http://', 'https://')):
return False return False
# Additional validation can be added here # Additional validation can be added here
return True return True
def generate_session_id() -> str: def generate_session_id() -> str:
"""Generate a unique session ID.""" """Generate a unique session ID."""
return getID() return getID()
# Settings management # Settings management
async def get_setting() -> Dict: async def get_setting() -> Dict:
"""Get current user settings from database or return defaults.""" """Get current user settings from database or return defaults."""
import json import json
# Get current user ID # Get current user ID
user_id = await get_current_user_id() user_id = await get_current_user_id()
default_settings = { default_settings = {
"security": { "security": {
"require_auth": False, "require_auth": False,
"encrypt_storage": False "encrypt_storage": False
}, },
"general": { "general": {
"default_model": "", "default_model": "",
"session_timeout": 30, "session_timeout": 30,
"auto_save": True "auto_save": True
}, },
"appearance": { "appearance": {
"theme": "dark" "theme": "dark"
} }
} }
try: try:
# Query user settings from database # Query user settings from database
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SETTINGS_CRUD['operations']['read']['sql_template'] sql_template = SETTINGS_CRUD['operations']['read']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id}) recs = await sor.sqlExe(sql_template, {'user_id': user_id})
if len(recs) > 0: if len(recs) > 0:
settings_json = recs[0]['settings_json'] settings_json = recs[0]['settings_json']
if settings_json: if settings_json:
saved_settings = json.loads(settings_json) saved_settings = json.loads(settings_json)
# Merge with defaults to ensure all keys exist # Merge with defaults to ensure all keys exist
for section, defaults in default_settings.items(): for section, defaults in default_settings.items():
if section not in saved_settings: if section not in saved_settings:
saved_settings[section] = defaults saved_settings[section] = defaults
else: else:
for key, value in defaults.items(): for key, value in defaults.items():
if key not in saved_settings[section]: if key not in saved_settings[section]:
saved_settings[section][key] = value saved_settings[section][key] = value
return saved_settings return saved_settings
except Exception as e: except Exception as e:
print(f"Error getting settings: {str(e)}") print(f"Error getting settings: {str(e)}")
# Fall back to defaults on error # Fall back to defaults on error
return default_settings return default_settings
async def save_setting(section: str, key: str, value) -> bool: async def save_setting(section: str, key: str, value) -> bool:
"""Save a specific setting value to current user's database record.""" """Save a specific setting value to current user's database record."""
import json import json
# Get current user ID # Get current user ID
user_id = await get_current_user_id() user_id = await get_current_user_id()
# Load existing settings or start with defaults # Load existing settings or start with defaults
settings = await get_setting() settings = await get_setting()
# Update the specific setting # Update the specific setting
if section not in settings: if section not in settings:
settings[section] = {} settings[section] = {}
settings[section][key] = value settings[section][key] = value
try: try:
# Save to database using sqlor-database-module # Save to database using sqlor-database-module
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor: async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SETTINGS_CRUD['operations']['create_or_update']['sql_template'] sql_template = SETTINGS_CRUD['operations']['create_or_update']['sql_template']
await sor.sqlExe(sql_template, { await sor.sqlExe(sql_template, {
'user_id': user_id, 'user_id': user_id,
'settings_json': json.dumps(settings) 'settings_json': json.dumps(settings)
}) })
return True return True
except Exception as e: except Exception as e:
print(f"Error saving settings: {str(e)}") print(f"Error saving settings: {str(e)}")
return False return False
# Module metadata # Module metadata
MODULE_NAME = "hermes-web-cli" MODULE_NAME = "hermes-web-cli"
@ -566,23 +566,23 @@ MODULE_VERSION = "0.1.0"
# Export all public functions # Export all public functions
__all__ = [ __all__ = [
'load_hermes_web_cli', 'load_hermes_web_cli',
'get_all_services', 'get_all_services',
'create_service', 'create_service',
'delete_service', 'delete_service',
'get_service_by_id', 'get_service_by_id',
'test_service_connection', 'test_service_connection',
'create_session', 'create_session',
'send_message_to_service', 'send_message_to_service',
'get_session_messages', 'get_session_messages',
'get_active_sessions', 'get_active_sessions',
'get_recent_sessions', 'get_recent_sessions',
'get_session_by_id', 'get_session_by_id',
'validate_service_url', 'validate_service_url',
'generate_session_id', 'generate_session_id',
'get_setting', 'get_setting',
'save_setting', 'save_setting',
'get_current_user_id', 'get_current_user_id',
'MODULE_NAME', 'MODULE_NAME',
'MODULE_VERSION' 'MODULE_VERSION'
] ]