Luxx/luxx/tools/builtin/data.py

"""Data processing tools"""
import re
import json
import base64
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"]
    },
    category="data"
)
def calculate(arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Execute mathematical calculation"""
    expression = arguments.get("expression", "")
    
    if not expression:
        return {"error": "Expression is required"}
    
    try:
        # 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}
    except Exception as e:
        return {"error": f"Calculation error: {str(e)}"}


@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: uppercase, lowercase, title, strip, reverse, word_count, char_count",
                "enum": ["uppercase", "lowercase", "title", "strip", "reverse", "word_count", "char_count"]
            }
        },
        "required": ["text", "operation"]
    },
    category="data"
)
def text_process(arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Text processing"""
    text = arguments.get("text", "")
    operation = arguments.get("operation", "")
    
    if not text:
        return {"error": "Text is required"}
    
    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:
        return {"error": f"Unknown operation: {operation}"}
    
    return {"result": result}


@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"]
    },
    category="data"
)
def json_process(arguments: Dict[str, Any]) -> Dict[str, Any]:
    """JSON data processing"""
    data = arguments.get("data", "")
    operation = arguments.get("operation", "")
    
    if not data:
        return {"error": "Data is required"}
    
    try:
        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:
            return {"error": f"Unknown operation: {operation}"}
        
        return {"result": result}
    except json.JSONDecodeError as e:
        return {"error": f"Invalid JSON: {str(e)}"}


@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: md5, sha1, sha256, sha512",
                "default": "sha256"
            }
        },
        "required": ["text"]
    },
    category="data"
)
def hash_text(arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Generate text hash"""
    import hashlib
    
    text = arguments.get("text", "")
    algorithm = arguments.get("algorithm", "sha256")
    
    if not text:
        return {"error": "Text is required"}
    
    try:
        hash_obj = hashlib.new(algorithm)
        hash_obj.update(text.encode('utf-8'))
        hash_value = hash_obj.hexdigest()
        
        return {"hash": hash_value}
    except Exception as e:
        return {"error": f"Hash error: {str(e)}"}


@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: encode, decode",
                "enum": ["encode", "decode"]
            }
        },
        "required": ["text", "operation"]
    },
    category="data"
)
def url_encode_decode(arguments: Dict[str, Any]) -> Dict[str, Any]:
    """URL encoding/decoding"""
    text = arguments.get("text", "")
    operation = arguments.get("operation", "")
    
    if not text:
        return {"error": "Text is required"}
    
    try:
        if operation == "encode":
            result = quote(text)
        elif operation == "decode":
            result = unquote(text)
        else:
            return {"error": f"Unknown operation: {operation}"}
        
        return {"result": result}
    except Exception as e:
        return {"error": f"URL error: {str(e)}"}


@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: encode, decode",
                "enum": ["encode", "decode"]
            }
        },
        "required": ["text", "operation"]
    },
    category="data"
)
def base64_encode_decode(arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Base64 encoding/decoding"""
    text = arguments.get("text", "")
    operation = arguments.get("operation", "")
    
    if not text:
        return {"error": "Text is required"}
    
    try:
        if operation == "encode":
            result = base64.b64encode(text.encode()).decode()
        elif operation == "decode":
            result = base64.b64decode(text.encode()).decode()
        else:
            return {"error": f"Unknown operation: {operation}"}
        
        return {"result": result}
    except Exception as e:
        return {"error": f"Base64 error: {str(e)}"}