Luxx/luxx/tools/builtin/data.py

"""Data processing tools - exception handling in decorator"""
import re
import json
import base64
import hashlib
from typing import Dict, Any
from urllib.parse import quote, unquote

from luxx.tools.factory import tool


@tool(
    name="calculate",
    description="Execute mathematical calculations",
    parameters={
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "Mathematical expression to evaluate"
            }
        },
        "required": ["expression"]
    },
    required_params=["expression"],
    category="data"
)
def calculate(arguments: Dict[str, Any]):
    """
    Execute mathematical calculation
    
    Returns:
        {"result": float, "expression": str}
    """
    expression = arguments.get("expression", "")
    
    # Safe replacement for math functions
    safe_dict = {
        "abs": abs,
        "round": round,
        "min": min,
        "max": max,
        "pow": pow,
        "sqrt": lambda x: x ** 0.5,
        "sin": lambda x: __import__('math').sin(x),
        "cos": lambda x: __import__('math').cos(x),
        "tan": lambda x: __import__('math').tan(x),
        "log": lambda x: __import__('math').log(x),
        "pi": __import__('math').pi,
        "e": __import__('math').e
    }
    
    # Remove dangerous characters, only keep numbers and operators
    safe_expr = re.sub(r"[^0-9+\-*/().%sqrtinsclogmaxminpowabsroundte, ]", "", expression)
    result = eval(safe_expr, {"__builtins__": {}, **safe_dict})
    
    return {"result": result, "expression": expression}


@tool(
    name="text_process",
    description="Process and transform text",
    parameters={
        "type": "object",
        "properties": {
            "text": {
                "type": "string",
                "description": "Text to process"
            },
            "operation": {
                "type": "string",
                "description": "Operation to perform",
                "enum": ["uppercase", "lowercase", "title", "strip", "reverse", "word_count", "char_count"]
            }
        },
        "required": ["text", "operation"]
    },
    required_params=["text", "operation"],
    category="data"
)
def text_process(arguments: Dict[str, Any]):
    """
    Text processing operations
    
    Returns:
        {"operation": str, "result": str|int, "original_length": int}
    """
    text = arguments.get("text", "")
    operation = arguments.get("operation", "")
    
    operations = {
        "uppercase": text.upper(),
        "lowercase": text.lower(),
        "title": text.title(),
        "strip": text.strip(),
        "reverse": text[::-1],
        "word_count": len(text.split()),
        "char_count": len(text)
    }
    
    result = operations.get(operation)
    if result is None:
        raise ValueError(f"Unknown operation: {operation}")
    
    return {
        "operation": operation,
        "result": result,
        "original_length": len(text)
    }


@tool(
    name="json_process",
    description="Process and transform JSON data",
    parameters={
        "type": "object",
        "properties": {
            "data": {
                "type": "string",
                "description": "JSON string or text to process"
            },
            "operation": {
                "type": "string",
                "description": "Operation: format, minify, validate",
                "enum": ["format", "minify", "validate"]
            }
        },
        "required": ["data", "operation"]
    },
    required_params=["data", "operation"],
    category="data"
)
def json_process(arguments: Dict[str, Any]):
    """
    JSON data processing
    
    Returns:
        {"operation": str, "result": str}
    """
    data = arguments.get("data", "")
    operation = arguments.get("operation", "")
    
    parsed = json.loads(data)
    
    if operation == "format":
        result = json.dumps(parsed, indent=2, ensure_ascii=False)
    elif operation == "minify":
        result = json.dumps(parsed, ensure_ascii=False)
    elif operation == "validate":
        result = "Valid JSON"
    else:
        raise ValueError(f"Unknown operation: {operation}")
    
    return {"operation": operation, "result": result}


@tool(
    name="hash_text",
    description="Generate text hash",
    parameters={
        "type": "object",
        "properties": {
            "text": {
                "type": "string",
                "description": "Text to hash"
            },
            "algorithm": {
                "type": "string",
                "description": "Hash algorithm",
                "enum": ["md5", "sha1", "sha256", "sha512"],
                "default": "sha256"
            }
        },
        "required": ["text"]
    },
    required_params=["text"],
    category="data"
)
def hash_text(arguments: Dict[str, Any]):
    """
    Generate text hash
    
    Returns:
        {"algorithm": str, "hash": str, "length": int}
    """
    text = arguments.get("text", "")
    algorithm = arguments.get("algorithm", "sha256")
    
    hash_obj = hashlib.new(algorithm)
    hash_obj.update(text.encode('utf-8'))
    hash_value = hash_obj.hexdigest()
    
    return {
        "algorithm": algorithm,
        "hash": hash_value,
        "length": len(hash_value)
    }


@tool(
    name="url_encode_decode",
    description="URL encoding/decoding",
    parameters={
        "type": "object",
        "properties": {
            "text": {
                "type": "string",
                "description": "Text to encode/decode"
            },
            "operation": {
                "type": "string",
                "description": "Operation",
                "enum": ["encode", "decode"]
            }
        },
        "required": ["text", "operation"]
    },
    required_params=["text", "operation"],
    category="data"
)
def url_encode_decode(arguments: Dict[str, Any]):
    """
    URL encoding/decoding
    
    Returns:
        {"operation": str, "result": str}
    """
    text = arguments.get("text", "")
    operation = arguments.get("operation", "")
    
    if operation == "encode":
        result = quote(text)
    elif operation == "decode":
        result = unquote(text)
    else:
        raise ValueError(f"Unknown operation: {operation}")
    
    return {"operation": operation, "result": result}


@tool(
    name="base64_encode_decode",
    description="Base64 encoding/decoding",
    parameters={
        "type": "object",
        "properties": {
            "text": {
                "type": "string",
                "description": "Text to encode/decode"
            },
            "operation": {
                "type": "string",
                "description": "Operation",
                "enum": ["encode", "decode"]
            }
        },
        "required": ["text", "operation"]
    },
    required_params=["text", "operation"],
    category="data"
)
def base64_encode_decode(arguments: Dict[str, Any]):
    """
    Base64 encoding/decoding
    
    Returns:
        {"operation": str, "result": str}
    """
    text = arguments.get("text", "")
    operation = arguments.get("operation", "")
    
    if operation == "encode":
        result = base64.b64encode(text.encode()).decode()
    elif operation == "decode":
        result = base64.b64decode(text.encode()).decode()
    else:
        raise ValueError(f"Unknown operation: {operation}")
    
    return {"operation": operation, "result": result}