abot/tools/user_lookup_tool.py

# --- 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 ---