196 lines
8.0 KiB
Python
196 lines
8.0 KiB
Python
# --- START OF FILE user_lookup_tool.py ---
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
from typing import Dict, Any
|
|
|
|
# --- Tool Definition (for LLM) ---
|
|
TOOL_DEFINITION = {
|
|
"name": "get_user_info",
|
|
"description": "Look up information about Slack users from the cache file by searching across all available fields",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"search_term": {
|
|
"type": "string",
|
|
"description": "Term to search for - matches against any field in user records including ID, name, real name, display name, email, title, phone, etc."
|
|
}
|
|
},
|
|
"required": ["search_term"]
|
|
}
|
|
}
|
|
# --- End Tool Definition ---
|
|
|
|
|
|
# --- Tool Implementation ---
|
|
def get_user_info(**kwargs: Any) -> Dict[str, Any]:
|
|
"""
|
|
Retrieve user information from the user cache file by searching across all available fields,
|
|
validating input arguments internally.
|
|
|
|
Args:
|
|
**kwargs (Any): Keyword arguments matching the tool's input_schema properties
|
|
(expects 'search_term').
|
|
|
|
Returns:
|
|
Dict[str, Any]: A dictionary containing matched user information or an error message
|
|
"""
|
|
search_term = kwargs.get('search_term')
|
|
|
|
# --- Input Validation Moved Here ---
|
|
# More forgiving search term validation
|
|
# If search_term is None or empty after stripping, return error
|
|
if search_term is None:
|
|
logging.error("get_user_info validation failed: Missing 'search_term' argument.")
|
|
return {
|
|
"found": False,
|
|
"error": "Missing required argument: 'search_term'."
|
|
}
|
|
|
|
# Ensure search_term is a string and strip whitespace
|
|
try:
|
|
search_term = str(search_term).strip()
|
|
except Exception as e:
|
|
logging.error(f"get_user_info validation failed: Could not convert search_term to string: {e}")
|
|
return {
|
|
"found": False,
|
|
"error": f"Invalid search_term format: {e}"
|
|
}
|
|
|
|
if not search_term:
|
|
logging.error("get_user_info validation failed: Empty 'search_term' provided.")
|
|
return {
|
|
"found": False,
|
|
"error": "Empty search term provided after stripping whitespace."
|
|
}
|
|
# --- End Input Validation ---
|
|
|
|
try:
|
|
logging.info(f"get_user_info: Attempting to find user with search term: {search_term}")
|
|
|
|
# Debug input received
|
|
logging.info(f"Search term type: {type(search_term)}, value: '{search_term}'")
|
|
|
|
# Normalize search term
|
|
search_term_lower = search_term.lower() # Use a different variable name
|
|
|
|
# Load the user cache
|
|
try:
|
|
# Specify encoding for broader compatibility
|
|
with open('user_cache.json', 'r', encoding='utf-8') as f:
|
|
user_cache = json.load(f)
|
|
except FileNotFoundError:
|
|
logging.error("User cache file 'user_cache.json' not found.")
|
|
return {"found": False, "error": "User cache file not found."}
|
|
except json.JSONDecodeError:
|
|
logging.error("Invalid JSON in user cache file 'user_cache.json'.")
|
|
return {"found": False, "error": "Invalid user cache format."}
|
|
|
|
# Search for matches across all users
|
|
matches = []
|
|
for user_id, user_data in user_cache.items():
|
|
# Check if user_data is valid
|
|
if not isinstance(user_data, dict):
|
|
logging.warning(f"Skipping invalid user data entry for ID {user_id} in cache.")
|
|
continue
|
|
|
|
# Flag to track if this user matches
|
|
user_matches = False
|
|
|
|
# Check every field in the user data for matches
|
|
for field_name, field_value in user_data.items():
|
|
# Skip non-string fields or None values
|
|
if field_value is None:
|
|
continue
|
|
|
|
# Convert field value to string and check for match
|
|
try:
|
|
str_value = str(field_value).lower()
|
|
if search_term_lower in str_value:
|
|
user_matches = True
|
|
break # Found a match in this user, no need to check other fields
|
|
except Exception as e:
|
|
# If any error in string conversion, just skip this field
|
|
logging.debug(f"Could not convert field '{field_name}' to string for user {user_id}: {e}")
|
|
continue
|
|
|
|
# If we found a match, add to our results
|
|
if user_matches:
|
|
# Create a clean user record with common fields
|
|
user_record = {
|
|
"id": user_data.get("id", user_id), # Use key as fallback ID
|
|
"name": user_data.get("name", ""),
|
|
"real_name": user_data.get("real_name", ""),
|
|
"display_name": user_data.get("display_name", ""),
|
|
"email": user_data.get("email", ""),
|
|
}
|
|
|
|
# Add optional fields if they exist and are not None
|
|
optional_fields = ["title", "phone", "first_name", "last_name", "cached_at", "status_text", "team"]
|
|
for field in optional_fields:
|
|
if field in user_data and user_data[field] is not None:
|
|
user_record[field] = user_data[field]
|
|
|
|
matches.append(user_record)
|
|
|
|
# Return results
|
|
if matches:
|
|
logging.info(f"Found {len(matches)} match(es) for search term '{search_term}'.")
|
|
return {
|
|
"found": True,
|
|
"match_count": len(matches),
|
|
"matches": matches
|
|
}
|
|
else:
|
|
logging.info(f"No users found matching '{search_term}'.")
|
|
return {
|
|
"found": False,
|
|
"error": f"No users found matching '{search_term}'"
|
|
}
|
|
|
|
except Exception as e:
|
|
logging.error(f"Unexpected error retrieving user info from cache: {e}", exc_info=True) # Added exc_info
|
|
return {
|
|
"found": False,
|
|
"error": f"An unexpected error occurred while accessing user cache: {str(e)}"
|
|
}
|
|
# --- End Tool Implementation ---
|
|
|
|
# Example Usage
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
# Create a dummy cache file for testing if it doesn't exist
|
|
dummy_cache_path = 'user_cache.json'
|
|
if not os.path.exists(dummy_cache_path):
|
|
print(f"Creating dummy {dummy_cache_path} for testing...")
|
|
dummy_data = {
|
|
"U123": {"id": "U123", "name": "john.doe", "real_name": "John Doe", "display_name": "Johnny", "email": "john.doe@example.com", "title": "Engineer"},
|
|
"U456": {"id": "U456", "name": "jane.smith", "real_name": "Jane Smith", "display_name": "Janey", "email": "jane.smith@example.com", "title": "Manager"},
|
|
"U789": {"id": "U789", "name": "test.user", "real_name": "Test User", "display_name": "", "email": "test@example.com", "phone": "555-1234"}
|
|
}
|
|
with open(dummy_cache_path, 'w', encoding='utf-8') as f:
|
|
json.dump(dummy_data, f, indent=2)
|
|
|
|
print("--- Testing get_user_info ---")
|
|
# Example: Find users with 'john' in any field
|
|
result1 = get_user_info(search_term="john")
|
|
print(f"Result (search='john'): {json.dumps(result1, indent=2)}")
|
|
|
|
# Example: Find user by specific email
|
|
result2 = get_user_info(search_term="jane.smith@example.com") # Use dummy email
|
|
print(f"Result (search='jane.smith@example.com'): {json.dumps(result2, indent=2)}")
|
|
|
|
# Example: Search for empty string (should fail validation)
|
|
result3 = get_user_info(search_term=" ")
|
|
print(f"Result (search=' '): {json.dumps(result3, indent=2)}")
|
|
|
|
# Example: Search for None (should fail validation)
|
|
result4 = get_user_info(search_term=None)
|
|
print(f"Result (search=None): {json.dumps(result4, indent=2)}")
|
|
|
|
# Example: Search term not found
|
|
result5 = get_user_info(search_term="NonExistentXYZ123")
|
|
print(f"Result (search='NonExistentXYZ123'): {json.dumps(result5, indent=2)}")
|
|
|
|
# --- END OF FILE user_lookup_tool.py --- |