# 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}" xml += f"" return xml output = dict_to_xml(parsed) if isinstance(parsed, dict) else f"{parsed}" 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("&", "&").replace("<", "<").replace(">", ">") 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("&", "&").replace("<", "<").replace(">", ">") 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")