tools/data_tools.py

# File 10: tools/data_tools.py
with open(f"{base_dir}/tools/data_tools.py", "w") as f:
    f.write('''"""Data Processing and Conversion Tools"""
import os
import re
import json
import csv
import math
import base64
import hashlib
import urllib.parse
from typing import Dict, List, Optional, Any
from config import Config
from utils import Logger

class DataTools:
    """Data format conversion, text processing, and calculations"""
    
    def __init__(self):
        self.logger = Logger()
    
    def json_csv_xml(self, input_data: str, from_format: str, to_format: str,
                     pretty: bool = True, delimiter: str = ",") -> Dict:
        """Convert between JSON, CSV, and XML formats"""
        try:
            input_path = os.path.join(Config.SANDBOX_PATH, input_data)
            if os.path.exists(input_path):
                with open(input_path, "r", encoding="utf-8") as f:
                    input_data = f.read()
            
            if from_format == "auto":
                input_stripped = input_data.strip()
                if input_stripped.startswith("[") or input_stripped.startswith("{"):
                    from_format = "json"
                elif input_stripped.startswith("<"):
                    from_format = "xml"
                elif delimiter in input_stripped[:1000]:
                    from_format = "csv"
                else:
                    from_format = "json"
            
            parsed = None
            if from_format == "json":
                parsed = json.loads(input_data)
            elif from_format == "csv":
                lines = input_data.strip().split("\n")
                reader = csv.DictReader(lines, delimiter=delimiter)
                parsed = list(reader)
            elif from_format == "xml":
                import xml.etree.ElementTree as ET
                root = ET.fromstring(input_data)
                def xml_to_dict(elem):
                    result = {elem.tag: {}}
                    if elem.attrib:
                        result[elem.tag]["@attributes"] = elem.attrib
                    if elem.text and elem.text.strip():
                        result[elem.tag]["#text"] = elem.text.strip()
                    for child in elem:
                        child_dict = xml_to_dict(child)
                        for k, v in child_dict.items():
                            if k in result[elem.tag]:
                                if not isinstance(result[elem.tag][k], list):
                                    result[elem.tag][k] = [result[elem.tag][k]]
                                result[elem.tag][k].append(v)
                            else:
                                result[elem.tag][k] = v
                    return result
                parsed = xml_to_dict(root)
            
            output = ""
            if to_format == "json":
                output = json.dumps(parsed, indent=2 if pretty else None, ensure_ascii=False)
            elif to_format == "csv":
                if isinstance(parsed, list) and len(parsed) > 0:
                    fieldnames = list(parsed[0].keys())
                    lines = [delimiter.join(fieldnames)]
                    for row in parsed:
                        lines.append(delimiter.join(str(row.get(k, "")) for k in fieldnames))
                    output = "\n".join(lines)
                else:
                    output = "No data"
            elif to_format == "xml":
                def dict_to_xml(d, root_name="root"):
                    xml = f"<{root_name}>"
                    for k, v in d.items():
                        if isinstance(v, dict):
                            xml += dict_to_xml(v, k)
                        elif isinstance(v, list):
                            for item in v:
                                xml += dict_to_xml({k: item}, k)
                        else:
                            xml += f"<{k}>{v}</{k}>"
                    xml += f"</{root_name}>"
                    return xml
                output = dict_to_xml(parsed) if isinstance(parsed, dict) else f"<root>{parsed}</root>"
            elif to_format == "yaml":
                def to_yaml(obj, indent=0):
                    spaces = "  " * indent
                    if isinstance(obj, dict):
                        lines = []
                        for k, v in obj.items():
                            if isinstance(v, (dict, list)):
                                lines.append(f"{spaces}{k}:")
                                lines.append(to_yaml(v, indent + 1))
                            else:
                                lines.append(f"{spaces}{k}: {v}")
                        return "\n".join(lines)
                    elif isinstance(obj, list):
                        lines = []
                        for item in obj:
                            if isinstance(item, dict):
                                lines.append(f"{spaces}-")
                                lines.append(to_yaml(item, indent + 1))
                            else:
                                lines.append(f"{spaces}- {item}")
                        return "\n".join(lines)
                    else:
                        return f"{spaces}{obj}"
                output = to_yaml(parsed)
            elif to_format == "markdown":
                if isinstance(parsed, list) and len(parsed) > 0:
                    headers = list(parsed[0].keys())
                    lines = ["| " + " | ".join(headers) + " |"]
                    lines.append("| " + " | ".join(["---"] * len(headers)) + " |")
                    for row in parsed:
                        lines.append("| " + " | ".join(str(row.get(h, "")) for h in headers) + " |")
                    output = "\n".join(lines)
                else:
                    output = json.dumps(parsed, indent=2)
            
            self.logger.log("json_csv_xml", {"from": from_format, "to": to_format})
            
            return {
                "success": True,
                "from_format": from_format,
                "to_format": to_format,
                "output": output[:5000],
                "truncated": len(output) > 5000
            }
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def text_process(self, text: str, operation: str = "clean", 
                     pattern: str = "", replacement: str = "",
                     case: str = "", split_by: str = "", join_with: str = "") -> Dict:
        """Process and transform text"""
        try:
            result = text
            
            if operation == "clean":
                result = re.sub(r"\s+", " ", text).strip()
                result = re.sub(r"\n\s*\n", "\n\n", result)
            
            elif operation == "replace":
                if pattern:
                    result = re.sub(pattern, replacement, text)
                else:
                    result = text.replace(replacement.split("->")[0] if "->" in replacement else "", 
                                         replacement.split("->")[1] if "->" in replacement else "")
            
            elif operation == "split":
                if split_by:
                    parts = text.split(split_by)
                else:
                    parts = text.split()
                return {
                    "success": True,
                    "operation": operation,
                    "parts": parts,
                    "count": len(parts)
                }
            
            elif operation == "join":
                try:
                    items = json.loads(text)
                    result = join_with.join(str(i) for i in items)
                except:
                    result = join_with.join(text.split("\n"))
            
            elif operation == "extract":
                if pattern:
                    matches = re.findall(pattern, text)
                    return {
                        "success": True,
                        "operation": operation,
                        "matches": matches,
                        "count": len(matches)
                    }
                else:
                    urls = re.findall(r'https?://[^\s<>"{}|\\^`\[\]]+', text)
                    emails = re.findall(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', text)
                    return {
                        "success": True,
                        "urls": urls,
                        "emails": emails,
                        "count": len(urls) + len(emails)
                    }
            
            elif operation == "count":
                words = len(text.split())
                chars = len(text)
                lines = text.count("\n") + 1
                return {
                    "success": True,
                    "words": words,
                    "characters": chars,
                    "lines": lines,
                    "characters_no_spaces": len(text.replace(" ", "").replace("\n", ""))
                }
            
            elif operation == "upper":
                result = text.upper()
            elif operation == "lower":
                result = text.lower()
            elif operation == "title":
                result = text.title()
            elif operation == "capitalize":
                result = text.capitalize()
            elif operation == "reverse":
                result = text[::-1]
            elif operation == "trim":
                result = text.strip()
            elif operation == "deduplicate":
                lines = text.split("\n")
                seen = set()
                unique = []
                for line in lines:
                    if line not in seen:
                        seen.add(line)
                        unique.append(line)
                result = "\n".join(unique)
            elif operation == "wrap":
                lines = []
                current = ""
                for word in text.split():
                    if len(current) + len(word) + 1 > 80:
                        lines.append(current)
                        current = word
                    else:
                        current += " " + word if current else word
                if current:
                    lines.append(current)
                result = "\n".join(lines)
            elif operation == "truncate":
                max_len = int(pattern) if pattern.isdigit() else 500
                result = text[:max_len] + "..." if len(text) > max_len else text
            elif operation == "encode":
                if pattern == "base64":
                    result = base64.b64encode(text.encode()).decode()
                elif pattern == "url":
                    result = urllib.parse.quote(text)
                elif pattern == "html":
                    result = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
                elif pattern == "hex":
                    result = text.encode().hex()
            elif operation == "decode":
                if pattern == "base64":
                    result = base64.b64decode(text).decode()
                elif pattern == "url":
                    result = urllib.parse.unquote(text)
                elif pattern == "hex":
                    result = bytes.fromhex(text).decode()
            
            if case == "upper":
                result = result.upper()
            elif case == "lower":
                result = result.lower()
            elif case == "title":
                result = result.title()
            
            self.logger.log("text_process", {"operation": operation, "input_length": len(text)})
            
            return {
                "success": True,
                "operation": operation,
                "result": result[:5000],
                "input_length": len(text),
                "output_length": len(result),
                "truncated": len(result) > 5000
            }
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def data_filter(self, data: str, operation: str = "sort", 
                    key: str = "", filter_expr: str = "", limit: int = 100) -> Dict:
        """Filter and manipulate tabular data"""
        try:
            data_path = os.path.join(Config.SANDBOX_PATH, data)
            if os.path.exists(data_path):
                with open(data_path, "r") as f:
                    data = f.read()
            
            items = json.loads(data)
            if not isinstance(items, list):
                items = [items]
            
            if operation == "sort":
                reverse = key.startswith("-")
                sort_key = key[1:] if reverse else key
                if sort_key:
                    items.sort(key=lambda x: x.get(sort_key, ""), reverse=reverse)
                else:
                    items.sort(reverse=reverse)
            
            elif operation == "filter":
                if filter_expr:
                    filtered = []
                    for item in items:
                        try:
                            if ">" in filter_expr:
                                k, v = filter_expr.split(">", 1)
                                if float(item.get(k.strip(), 0)) > float(v.strip()):
                                    filtered.append(item)
                            elif "<" in filter_expr:
                                k, v = filter_expr.split("<", 1)
                                if float(item.get(k.strip(), 0)) < float(v.strip()):
                                    filtered.append(item)
                            elif "contains" in filter_expr.lower():
                                k, v = filter_expr.lower().split("contains", 1)
                                if v.strip() in str(item.get(k.strip(), "")).lower():
                                    filtered.append(item)
                            elif "==" in filter_expr:
                                k, v = filter_expr.split("==", 1)
                                if str(item.get(k.strip(), "")) == v.strip():
                                    filtered.append(item)
                        except:
                            pass
                    items = filtered
            
            elif operation == "group":
                if key:
                    groups = {}
                    for item in items:
                        val = item.get(key, "undefined")
                        if val not in groups:
                            groups[val] = []
                        groups[val].append(item)
                    items = [{"group": k, "count": len(v), "items": v[:10]} for k, v in groups.items()]
            
            elif operation == "limit":
                items = items[:limit]
            
            elif operation == "unique":
                seen = set()
                unique = []
                for item in items:
                    val = json.dumps(item, sort_keys=True)
                    if val not in seen:
                        seen.add(val)
                        unique.append(item)
                items = unique
            
            elif operation == "reverse":
                items.reverse()
            
            self.logger.log("data_filter", {"operation": operation, "results": len(items)})
            
            return {
                "success": True,
                "operation": operation,
                "results": items[:limit],
                "total": len(items),
                "truncated": len(items) > limit
            }
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def calc_math(self, expression: str, variables: Dict[str, float] = None) -> Dict:
        """Calculator and math operations"""
        try:
            if expression.startswith("stats:"):
                data_str = expression[6:].strip()
                try:
                    numbers = json.loads(data_str)
                except:
                    numbers = [float(x) for x in data_str.split(",")]
                
                n = len(numbers)
                mean = sum(numbers) / n
                sorted_nums = sorted(numbers)
                median = sorted_nums[n // 2] if n % 2 else (sorted_nums[n // 2 - 1] + sorted_nums[n // 2]) / 2
                variance = sum((x - mean) ** 2 for x in numbers) / n
                std_dev = math.sqrt(variance)
                
                return {
                    "success": True,
                    "operation": "statistics",
                    "count": n,
                    "sum": sum(numbers),
                    "mean": mean,
                    "median": median,
                    "min": min(numbers),
                    "max": max(numbers),
                    "range": max(numbers) - min(numbers),
                    "variance": variance,
                    "std_dev": std_dev
                }
            
            if expression.startswith("convert:"):
                parts = expression[8:].strip().split()
                if len(parts) >= 3:
                    value = float(parts[0])
                    from_unit = parts[1].lower()
                    to_unit = parts[2].lower()
                    
                    length = {
                        "m": 1, "km": 1000, "cm": 0.01, "mm": 0.001,
                        "ft": 0.3048, "in": 0.0254, "yd": 0.9144, "mi": 1609.34
                    }
                    weight = {
                        "kg": 1, "g": 0.001, "mg": 0.000001,
                        "lb": 0.453592, "oz": 0.0283495
                    }
                    
                    if from_unit in ["c", "celsius"] and to_unit in ["f", "fahrenheit"]:
                        result = value * 9/5 + 32
                    elif from_unit in ["f", "fahrenheit"] and to_unit in ["c", "celsius"]:
                        result = (value - 32) * 5/9
                    elif from_unit in ["c", "celsius"] and to_unit in ["k", "kelvin"]:
                        result = value + 273.15
                    elif from_unit in length and to_unit in length:
                        result = value * length[from_unit] / length[to_unit]
                    elif from_unit in weight and to_unit in weight:
                        result = value * weight[from_unit] / weight[to_unit]
                    else:
                        return {"success": False, "error": f"Unknown conversion: {from_unit} to {to_unit}"}
                    
                    return {
                        "success": True,
                        "operation": "convert",
                        "from": f"{value} {from_unit}",
                        "to": f"{result:.6g} {to_unit}",
                        "result": result
                    }
            
            safe_dict = {
                "abs": abs, "round": round, "max": max, "min": min,
                "sum": sum, "len": len, "pow": pow,
                "math": math,
                "sin": math.sin, "cos": math.cos, "tan": math.tan,
                "asin": math.asin, "acos": math.acos, "atan": math.atan,
                "sinh": math.sinh, "cosh": math.cosh, "tanh": math.tanh,
                "exp": math.exp, "log": math.log, "log10": math.log10,
                "sqrt": math.sqrt, "ceil": math.ceil, "floor": math.floor,
                "pi": math.pi, "e": math.e,
                "factorial": math.factorial, "gcd": math.gcd,
                "radians": math.radians, "degrees": math.degrees
            }
            
            if variables:
                safe_dict.update(variables)
            
            result = eval(expression, {"__builtins__": {}}, safe_dict)
            
            self.logger.log("calc_math", {"expression": expression, "result": result})
            
            return {
                "success": True,
                "expression": expression,
                "result": result,
                "type": type(result).__name__
            }
            
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def hash_encrypt(self, data: str, operation: str = "hash", 
                     algorithm: str = "sha256", key: str = "") -> Dict:
        """Hash and encode/decode operations"""
        try:
            if operation == "hash":
                if algorithm == "md5":
                    result = hashlib.md5(data.encode()).hexdigest()
                elif algorithm == "sha1":
                    result = hashlib.sha1(data.encode()).hexdigest()
                elif algorithm == "sha256":
                    result = hashlib.sha256(data.encode()).hexdigest()
                elif algorithm == "sha512":
                    result = hashlib.sha512(data.encode()).hexdigest()
                else:
                    return {"success": False, "error": f"Unknown hash algorithm: {algorithm}"}
                
                return {
                    "success": True,
                    "operation": "hash",
                    "algorithm": algorithm,
                    "input": data[:100],
                    "hash": result
                }
            
            elif operation == "encode":
                if algorithm == "base64":
                    result = base64.b64encode(data.encode()).decode()
                elif algorithm == "url":
                    result = urllib.parse.quote(data)
                elif algorithm == "hex":
                    result = data.encode().hex()
                elif algorithm == "html":
                    result = data.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
                else:
                    return {"success": False, "error": f"Unknown encoding: {algorithm}"}
                
                return {
                    "success": True,
                    "operation": "encode",
                    "algorithm": algorithm,
                    "result": result
                }
            
            elif operation == "decode":
                if algorithm == "base64":
                    result = base64.b64decode(data).decode()
                elif algorithm == "url":
                    result = urllib.parse.unquote(data)
                elif algorithm == "hex":
                    result = bytes.fromhex(data).decode()
                else:
                    return {"success": False, "error": f"Unknown decoding: {algorithm}"}
                
                return {
                    "success": True,
                    "operation": "decode",
                    "algorithm": algorithm,
                    "result": result
                }
            
            elif operation == "hmac":
                import hmac as hmac_module
                if algorithm == "sha256":
                    result = hmac_module.new(key.encode(), data.encode(), hashlib.sha256).hexdigest()
                else:
                    result = hmac_module.new(key.encode(), data.encode(), hashlib.md5).hexdigest()
                
                return {
                    "success": True,
                    "operation": "hmac",
                    "algorithm": algorithm,
                    "hmac": result
                }
            
            elif operation == "compare":
                if algorithm == "sha256":
                    computed = hashlib.sha256(data.encode()).hexdigest()
                else:
                    computed = hashlib.md5(data.encode()).hexdigest()
                
                return {
                    "success": True,
                    "operation": "compare",
                    "match": computed == key,
                    "computed": computed,
                    "provided": key
                }
            
            return {"success": False, "error": f"Unknown operation: {operation}"}
            
        except Exception as e:
            return {"success": False, "error": str(e)}
''')

print("tools/data_tools.py done")