tools.py

from langchain_core.tools import tool
from langchain_tavily import TavilySearch
from langchain_community.document_loaders import WikipediaLoader

from youtube_transcript_api import YouTubeTranscriptApi
from PIL import Image, ImageDraw, ImageFont, ImageEnhance, ImageFilter
from typing import List, Dict, Any, Optional
from urllib.parse import urlparse

from code_interpreter_26042026 import CodeInterpreter
from tool_logger_26042026 import log_tool
from bs4 import BeautifulSoup
from dotenv import load_dotenv

interpreter_instance = CodeInterpreter()

from image_proccessing_26042026 import *

import os
import re
import cmath
import requests
import tempfile
import subprocess
import pytesseract
import numpy as np
import pandas as pd

load_dotenv()

@tool
def run_python_from_url(file_url: str) -> str:
    """
    Download and execute a Python script from a URL.

    Use this when:
    - A Python file needs to be executed remotely
    - The task explicitly involves running external scripts

    Do NOT use for:
    - Simple calculations
    - Inline code execution (use execute_code_multilang instead)

    Input: URL to .py file
    Output: execution result
    """
    try:
        # Download file
        response = requests.get(file_url)
        print("response ",response, "file_url ", file_url)
        response.raise_for_status()

        # Save to temp file
        with tempfile.NamedTemporaryFile(delete=False, suffix=".py") as tmp:
            tmp.write(response.content)
            tmp_path = tmp.name

        # Execute file
        result = subprocess.run(
            ["python", tmp_path],
            capture_output=True,
            text=True,
            timeout=10
        )

        output = result.stdout.strip()

        if not output:
            output = result.stderr.strip()
            
        log_tool("run_python_from_url", file_url, output)
        return output if output else "No output"

    except Exception as e:
        log_tool("run_python_from_url", file_url, f"ERROR: {str(e)}")
        return f"ERROR: {str(e)}"
    
@tool
def fetch_task_file(task_id: str) -> str:
    """
    MUST USE when a question mentions an attached file.

    Input:
    - task_id (string)

    Returns:
    - local file path of downloaded file

    After using this tool:
    - You should analyze or execute the file depending on type
    """
    try:
        url = f"https://agents-course-unit4-scoring.hf.space/files/{task_id}"
        response = requests.get(url)
        
        print("fetch task file ",response, "taskid ",task_id, "url ", url)
        
        if response.status_code != 200:
            return f"ERROR: {response.text}"
        
        # Try to detect filename (fallback)
        filename = f"{task_id}"
        
        # Save file
        filepath = f"./downloadsHuggingFace/{filename}"
        os.makedirs("./downloadsHuggingFace", exist_ok=True)
        
        with open(filepath, "wb") as f:
            f.write(response.content)
        
        return filepath

    except Exception as e:
        log_tool("fetch_task_file", task_id, f"ERROR: {str(e)}")
        return f"ERROR: {str(e)}"
    
    
# def web_search_raw(query: str) -> str:
#     import requests
#     from bs4 import BeautifulSoup
#     import urllib.parse

#     url = "https://duckduckgo.com/html/"
#     headers = {"User-Agent": "Mozilla/5.0"}
#     params = {"q": query}

#     res = requests.get(url, params=params, headers=headers, timeout=10)
#     soup = BeautifulSoup(res.text, "html.parser")
#     print("web search response ", res.text, "query ",query)
#     results = []

#     for r in soup.select(".result__body")[:7]:
#         title = r.select_one(".result__title")
#         snippet = r.select_one(".result__snippet")
#         link = r.select_one("a.result__a")

#         if not link:
#             continue

#         raw_url = link.get("href", "")

#         if "uddg=" in raw_url:
#             raw_url = urllib.parse.unquote(raw_url.split("uddg=")[-1])

#         results.append({
#             "title": title.get_text(strip=True) if title else "",
#             "snippet": snippet.get_text(strip=True) if snippet else "",
#             "url": raw_url
#         })

#     return results


def duckduckgo_search(query: str):
    import requests

    url = "https://api.duckduckgo.com/"
    params = {"q": query, "format": "json", "no_html": 1}

    res = requests.get(url, params=params, timeout=10)
    data = res.json()

    results = []

    if data.get("AbstractText"):
        results.append({
            "title": data.get("Heading", ""),
            "snippet": data.get("AbstractText", ""),
            "url": data.get("AbstractURL", "")
        })

    return results

def web_search_raw(query: str):
    import requests
    from bs4 import BeautifulSoup
    import urllib.parse
    import random
    import time

    USER_AGENTS = [
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/122.0.0.0 Safari/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
        "Mozilla/5.0 (X11; Linux x86_64) Gecko/20100101 Firefox/123.0"
    ]

    url = "https://duckduckgo.com/html/"
    params = {"q": query}

    headers = {
        "User-Agent": random.choice(USER_AGENTS),
        "Accept-Language": "en-US,en;q=0.9",
        "Accept": "text/html,application/xhtml+xml",
        "Connection": "keep-alive",
    }

    session = requests.Session()

    # delay to avoid detection
    time.sleep(random.uniform(1.5, 3.5))

    res = session.get(url, params=params, headers=headers, timeout=10)

    # 🚨 detect bot block
    if "anomaly" in res.text.lower() or "captcha" in res.text.lower():
        return "BOT_DETECTED"

    soup = BeautifulSoup(res.text, "html.parser")

    results = []

    for r in soup.select(".result__body")[:5]:
        title = r.select_one(".result__title")
        snippet = r.select_one(".result__snippet")
        link = r.select_one("a.result__a")

        if not link:
            continue

        raw_url = link.get("href", "")

        if "uddg=" in raw_url:
            raw_url = urllib.parse.unquote(raw_url.split("uddg=")[-1])

        results.append({
            "title": title.get_text(strip=True) if title else "",
            "snippet": snippet.get_text(strip=True) if snippet else "",
            "url": raw_url
        })

    return results

@tool
def search(query: str) -> str:
    """Search the web for real-world information.

    Use this when:
    - The question requires external knowledge (people, places, events, facts)
    - The question involves current or recent information
    - You are unsure about factual accuracy

    Do NOT use this when:
    - The answer can be derived from reasoning or math
    - The task involves files, images, or code execution

    Input: search query string
    Output: list of web results with titles and snippets"""

    try:
        tavily = TavilySearch(
            max_results=5,
            topic="general",
            search_depth="basic",   # cheaper + faster
            include_answer=True     # gives direct answer
        )

        response = tavily.invoke({"query": query})

        # ---- Handle errors ----
        if isinstance(response, dict) and response.get("error"):
            return f"Search Error: {response['error']}"

        results = response.get("results", [])
        answer = response.get("answer", None)

        if not results:
            return "No relevant results found."

        # ---- Format nicely for LLM ----
        formatted = []

        if answer:
            formatted.append(f"🔎 QUICK ANSWER:\n{answer}\n")

        for r in results:
            formatted.append(
                f"Title: {r.get('title','')}\n"
                f"Snippet: {r.get('content','')}\n"
                f"Source: {r.get('url','')}\n"
            )

        final_output = "\n---\n".join(formatted)

        # ---- Log ----
        log_tool("web_search", query, {"results": final_output})

        return final_output

    except Exception as e:
        error_msg = f"Search failed: {str(e)}"
        log_tool("web_search", query, error_msg)
        return error_msg

# @tool
# def search(query: str) -> str:
#     """Search the web for real-world information.

#     Use this when:
#     - The question requires external knowledge (people, places, events, facts)
#     - The question involves current or recent information
#     - You are unsure about factual accuracy

#     Do NOT use this when:
#     - The answer can be derived from reasoning or math
#     - The task involves files, images, or code execution

#     Input: search query string
#     Output: list of web results with titles and snippets"""
#     results = web_search_raw(query)
    
#     print("search results ", results)

#     if not results or results == "BOT_DETECTED":
#         # return "No results found."
#         print("⚠️ DuckDuckGo blocked → fallback")
#         return duckduckgo_search(query)

#     formatted = []
#     for r in results:
#         formatted.append(
#             f"{r.get('title','')}\n{r.get('snippet','')}\n{r.get('link','')}"
#         )
#     log_tool("search", query, {"web_results": formatted})
#     return "\n\n".join(formatted)

@tool
def youtube_transcript(url: str) -> str:
    """Extract transcript text from a YouTube video.

    Use this when:
    - The user asks about content from a specific YouTube video
    - You need to summarize or analyze a video

    Do NOT use this for:
    - General web search
    - Non-YouTube links

    Input: YouTube video URL
    Output: transcript text"""
    try:
        video_id = None

       # Case 1: watch?v=
        match = re.search(r"v=([^&]+)", url)
        if match:
            video_id = match.group(1)

        # Case 2: youtu.be/
        if not video_id:
            match = re.search(r"youtu\.be/([^?&]+)", url)
            if match:
                video_id = match.group(1)

        # Case 3: shorts/
        if not video_id:
            match = re.search(r"shorts/([^?&]+)", url)
            if match:
                video_id = match.group(1)

        # Case 4: embed/
        if not video_id:
            match = re.search(r"embed/([^?&]+)", url)
            if match:
                video_id = match.group(1)

        if not video_id:
            raise ValueError("Could not extract video ID from URL")

        transcript = YouTubeTranscriptApi().fetch(video_id)

        text = " ".join([t.text for t in transcript])
        
        print("youtube transcript text ", text)

        log_tool("youtube_transcript", url, text)

        return text[:5000] if text else "No transcript available"

    except Exception as e:
        print("youtube transcript err ", e)
        log_tool("youtube_transcript", url, f"Transcript error: {str(e)}")
        return f"Transcript error: {str(e)}"
    
# @tool
# def wiki_search(query: str) -> str:
#     """Search Wikipedia (safe fallback)"""
#     try:
#         import wikipedia
#         log_tool("wiki_search", query, wikipedia.summary(query, sentences=2))
#         return wikipedia.summary(query, sentences=2)
#     except Exception as e:
#         log_tool("wiki_search", query, f"ERROR: {str(e)}")
#         return f"ERROR: {str(e)}"

# @tool
# def wiki_search(query: str) -> str:
#     """
#     Search Wikipedia and return detailed page content including discography when available.
#     """
#     try:
#         import wikipedia

#         # Get best page
#         page = wikipedia.page(query, auto_suggest=True)

#         content = page.content[:5000]  # limit for LLM

#         log_tool("wiki_search", query, content)

#         return content

#     except Exception as e:
#         log_tool("wiki_search", query, f"ERROR: {str(e)}")
#         return f"ERROR: {str(e)}"

@tool
def wiki_search(query: str) -> str:
    """
    Wikipedia search with strict size control to avoid token overflow.
    """
    try:
        import wikipedia

        MAX_CHARS = 5000  # SAFE LIMIT

        def trim(text):
            return text[:MAX_CHARS]

        # ---------- DIRECT ----------
        try:
            page = wikipedia.page(query, auto_suggest=True)
            content = trim(page.content)

            log_tool("wiki_direct", query, content)
            return f"### {page.title}\n{content}"

        except wikipedia.DisambiguationError as e:
            options = e.options[:5]

            for option in options:
                try:
                    page = wikipedia.page(option, auto_suggest=False)
                    content = trim(page.content)

                    log_tool("wiki_disambiguation", query, option)
                    return f"### {page.title}\n{content}"
                except:
                    continue

        except:
            pass

        # ---------- SEARCH ----------
        results = wikipedia.search(query)

        combined = ""
        for title in results[:3]:  # 🔥 reduce results
            try:
                page = wikipedia.page(title, auto_suggest=False)

                chunk = f"\n\n### {title}\n{page.content[:1500]}"

                # 🔥 stop before overflow
                if len(combined) + len(chunk) > MAX_CHARS:
                    break

                combined += chunk

            except:
                continue

        log_tool("wiki_search_fallback", query, combined[:1000])
        return combined

    except Exception as e:
        log_tool("wiki_error", query, str(e))
        return f"ERROR: {str(e)}"
    
@tool
def wiki_discography(query: str) -> str:
    """
    Search Wikipedia specifically for discography or album lists of an artist.
    """
    try:
        import wikipedia

        page = wikipedia.page(query + " discography", auto_suggest=True)

        content = page.content

        log_tool("wiki_discography", query, content)

        return content

    except Exception as e:
        log_tool("wiki_discography", query, f"ERROR: {str(e)}")
        return f"ERROR: {str(e)}"

### =============== MATHEMATICAL TOOLS =============== ###


@tool
def multiply(a: float, b: float) -> float:
    """
    Perform precise multiplication.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    return a * b


@tool
def add(a: float, b: float) -> float:
    """
    Perform precise addition.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    return a + b


@tool
def subtract(a: float, b: float) -> int:
    """
    Perform precise subtraction.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    return a - b


@tool
def divide(a: float, b: float) -> float:
    """
    Perform precise addition.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    if b == 0:
        raise ValueError("Cannot divided by zero.")
    return a / b


@tool
def modulus(a: int, b: int) -> int:
    """
    Perform precise modulus.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    return a % b


@tool
def power(a: float, b: float) -> float:
    """
    Perform precise power.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    return a**b


@tool
def square_root(a: float) -> float | complex:
    """
    Perform precise square_root.

    Use this when:
    - Exact arithmetic is required

    Do NOT use for:
    - Estimation or reasoning

    Input: two numbers
    Output: result
    """
    if a >= 0:
        return a**0.5
    return cmath.sqrt(a)

### ============== IMAGE PROCESSING AND GENERATION TOOLS =============== ###


@tool
def analyze_image(image_base64: str) -> Dict[str, Any]:
    """
    Analyze image properties (size, colors, brightness).

    Use this when:
    - You need metadata or insights about an image

    Do NOT use for:
    - editing or transforming images

    Input: base64 image
    Output: analysis data
    """
    try:
        img = decode_image(image_base64)
        width, height = img.size
        mode = img.mode

        if mode in ("RGB", "RGBA"):
            arr = np.array(img)
            avg_colors = arr.mean(axis=(0, 1))
            dominant = ["Red", "Green", "Blue"][np.argmax(avg_colors[:3])]
            brightness = avg_colors.mean()
            color_analysis = {
                "average_rgb": avg_colors.tolist(),
                "brightness": brightness,
                "dominant_color": dominant,
            }
        else:
            color_analysis = {"note": f"No color analysis for mode {mode}"}

        thumbnail = img.copy()
        thumbnail.thumbnail((100, 100))
        thumb_path = save_image(thumbnail, "thumbnails")
        thumbnail_base64 = encode_image(thumb_path)

        log_tool("analyze_image", image_base64, {
            "dimensions": (width, height),
            "mode": mode,
            "color_analysis": color_analysis,
            "thumbnail": thumbnail_base64,
        }) 
        return {
            "dimensions": (width, height),
            "mode": mode,
            "color_analysis": color_analysis,
            "thumbnail": thumbnail_base64,
        }
    except Exception as e:
        log_tool("analyze_image", image_base64, f"ERROR: {str(e)}")
        return {"error": str(e)}


@tool
def transform_image(
    image_base64: str, operation: str, params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Modify an image (resize, rotate, crop, etc.)

    Use this when:
    - The task requires changing an image

    Input: image + operation
    Output: transformed image
    """
    try:
        img = decode_image(image_base64)
        params = params or {}

        if operation == "resize":
            img = img.resize(
                (
                    params.get("width", img.width // 2),
                    params.get("height", img.height // 2),
                )
            )
        elif operation == "rotate":
            img = img.rotate(params.get("angle", 90), expand=True)
        elif operation == "crop":
            img = img.crop(
                (
                    params.get("left", 0),
                    params.get("top", 0),
                    params.get("right", img.width),
                    params.get("bottom", img.height),
                )
            )
        elif operation == "flip":
            if params.get("direction", "horizontal") == "horizontal":
                img = img.transpose(Image.FLIP_LEFT_RIGHT)
            else:
                img = img.transpose(Image.FLIP_TOP_BOTTOM)
        elif operation == "adjust_brightness":
            img = ImageEnhance.Brightness(img).enhance(params.get("factor", 1.5))
        elif operation == "adjust_contrast":
            img = ImageEnhance.Contrast(img).enhance(params.get("factor", 1.5))
        elif operation == "blur":
            img = img.filter(ImageFilter.GaussianBlur(params.get("radius", 2)))
        elif operation == "sharpen":
            img = img.filter(ImageFilter.SHARPEN)
        elif operation == "grayscale":
            img = img.convert("L")
        else:
            return {"error": f"Unknown operation: {operation}"}

        result_path = save_image(img)
        result_base64 = encode_image(result_path)
        log_tool("transform_image", {"operation": operation, "params": params}, result_base64)
        return {"transformed_image": result_base64}

    except Exception as e:
        log_tool("transform_image", {"operation": operation, "params": params}, f"ERROR: {str(e)}")
        return {"error": str(e)}


@tool
def draw_on_image(
    image_base64: str, drawing_type: str, params: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Draw shapes (rectangle, circle, line) or text onto an image.
    Use this when:
    - Annotation or overlay is needed
    Args:
        image_base64 (str): Base64 encoded input image
        drawing_type (str): Drawing type
        params (Dict[str, Any]): Drawing parameters
    Returns:
        Dictionary with result image (base64)
    """
    try:
        img = decode_image(image_base64)
        draw = ImageDraw.Draw(img)
        color = params.get("color", "red")

        if drawing_type == "rectangle":
            draw.rectangle(
                [params["left"], params["top"], params["right"], params["bottom"]],
                outline=color,
                width=params.get("width", 2),
            )
        elif drawing_type == "circle":
            x, y, r = params["x"], params["y"], params["radius"]
            draw.ellipse(
                (x - r, y - r, x + r, y + r),
                outline=color,
                width=params.get("width", 2),
            )
        elif drawing_type == "line":
            draw.line(
                (
                    params["start_x"],
                    params["start_y"],
                    params["end_x"],
                    params["end_y"],
                ),
                fill=color,
                width=params.get("width", 2),
            )
        elif drawing_type == "text":
            font_size = params.get("font_size", 20)
            try:
                font = ImageFont.truetype("arial.ttf", font_size)
            except IOError:
                font = ImageFont.load_default()
            draw.text(
                (params["x"], params["y"]),
                params.get("text", "Text"),
                fill=color,
                font=font,
            )
        else:
            return {"error": f"Unknown drawing type: {drawing_type}"}

        result_path = save_image(img)
        result_base64 = encode_image(result_path)
        log_tool("draw_on_image", {"drawing_type": drawing_type, "params": params}, result_base64)
        return {"result_image": result_base64}

    except Exception as e:
        log_tool("draw_on_image", {"drawing_type": drawing_type, "params": params}, f"ERROR: {str(e)}") 
        return {"error": str(e)}


@tool
def generate_simple_image(
    image_type: str,
    width: int = 500,
    height: int = 500,
    params: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """
    Generate a simple image (gradient, noise, pattern, chart).
    Args:
        image_type (str): Type of image
        width (int), height (int)
        params (Dict[str, Any], optional): Specific parameters
    Returns:
        Dictionary with generated image (base64)
    """
    try:
        params = params or {}

        if image_type == "gradient":
            direction = params.get("direction", "horizontal")
            start_color = params.get("start_color", (255, 0, 0))
            end_color = params.get("end_color", (0, 0, 255))

            img = Image.new("RGB", (width, height))
            draw = ImageDraw.Draw(img)

            if direction == "horizontal":
                for x in range(width):
                    r = int(
                        start_color[0] + (end_color[0] - start_color[0]) * x / width
                    )
                    g = int(
                        start_color[1] + (end_color[1] - start_color[1]) * x / width
                    )
                    b = int(
                        start_color[2] + (end_color[2] - start_color[2]) * x / width
                    )
                    draw.line([(x, 0), (x, height)], fill=(r, g, b))
            else:
                for y in range(height):
                    r = int(
                        start_color[0] + (end_color[0] - start_color[0]) * y / height
                    )
                    g = int(
                        start_color[1] + (end_color[1] - start_color[1]) * y / height
                    )
                    b = int(
                        start_color[2] + (end_color[2] - start_color[2]) * y / height
                    )
                    draw.line([(0, y), (width, y)], fill=(r, g, b))

        elif image_type == "noise":
            noise_array = np.random.randint(0, 256, (height, width, 3), dtype=np.uint8)
            img = Image.fromarray(noise_array, "RGB")

        else:
            return {"error": f"Unsupported image_type {image_type}"}

        result_path = save_image(img)
        result_base64 = encode_image(result_path)
        log_tool("generate_simple_image", {"image_type": image_type, "params": params}, result_base64)
        return {"generated_image": result_base64}

    except Exception as e:
        log_tool("generate_simple_image", {"image_type": image_type, "params": params}, f"ERROR: {str(e)}")
        return {"error": str(e)}


@tool
def combine_images(
    images_base64: List[str], operation: str, params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Combine multiple images (collage, stack, blend).
    Args:
        images_base64 (List[str]): List of base64 images
        operation (str): Combination type
        params (Dict[str, Any], optional)
    Returns:
        Dictionary with combined image (base64)
    """
    try:
        images = [decode_image(b64) for b64 in images_base64]
        params = params or {}

        if operation == "stack":
            direction = params.get("direction", "horizontal")
            if direction == "horizontal":
                total_width = sum(img.width for img in images)
                max_height = max(img.height for img in images)
                new_img = Image.new("RGB", (total_width, max_height))
                x = 0
                for img in images:
                    new_img.paste(img, (x, 0))
                    x += img.width
            else:
                max_width = max(img.width for img in images)
                total_height = sum(img.height for img in images)
                new_img = Image.new("RGB", (max_width, total_height))
                y = 0
                for img in images:
                    new_img.paste(img, (0, y))
                    y += img.height
        else:
            return {"error": f"Unsupported combination operation {operation}"}

        result_path = save_image(new_img)
        result_base64 = encode_image(result_path)
        log_tool("combine_images", {"operation": operation, "params": params}, result_base64)
        return {"combined_image": result_base64}

    except Exception as e:
        log_tool("combine_images", {"operation": operation, "params": params}, f"ERROR: {str(e)}")
        return {"error": str(e)}
    
    
    ### =============== CODE INTERPRETER TOOLS =============== ###


@tool
def execute_code_multilang(code: str, language: str = "python") -> str:
    """Execute code in Python, Bash, SQL, C, or Java.

    Use this when:
    - The task requires code execution
    - Logic is complex or requires runtime evaluation
    - Data processing or scripting is needed

    Do NOT use for:
    - Simple math (use math tools)
    - General reasoning

    Input: code + language
    Output: execution result (stdout, errors, etc.)
    """
    supported_languages = ["python", "bash", "sql", "c", "java"]
    language = language.lower()

    if language not in supported_languages:
        return f"❌ Unsupported language: {language}. Supported languages are: {', '.join(supported_languages)}"

    result = interpreter_instance.execute_code(code, language=language)

    response = []

    if result["status"] == "success":
        response.append(f"✅ Code executed successfully in **{language.upper()}**")

        if result.get("stdout"):
            response.append(
                "\n**Standard Output:**\n```\n" + result["stdout"].strip() + "\n```"
            )

        if result.get("stderr"):
            response.append(
                "\n**Standard Error (if any):**\n```\n"
                + result["stderr"].strip()
                + "\n```"
            )

        if result.get("result") is not None:
            response.append(
                "\n**Execution Result:**\n```\n"
                + str(result["result"]).strip()
                + "\n```"
            )

        if result.get("dataframes"):
            for df_info in result["dataframes"]:
                response.append(
                    f"\n**DataFrame `{df_info['name']}` (Shape: {df_info['shape']})**"
                )
                df_preview = pd.DataFrame(df_info["head"])
                response.append("First 5 rows:\n```\n" + str(df_preview) + "\n```")

        if result.get("plots"):
            response.append(
                f"\n**Generated {len(result['plots'])} plot(s)** (Image data returned separately)"
            )

    else:
        response.append(f"❌ Code execution failed in **{language.upper()}**")
        if result.get("stderr"):
            response.append(
                "\n**Error Log:**\n```\n" + result["stderr"].strip() + "\n```"
            )
    log_tool("execute_code_multilang", {"code": code, "language": language}, "\n".join(response))

    return "\n".join(response)



### =============== DOCUMENT PROCESSING TOOLS =============== ###


@tool
def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
    """
    Save content to a file.

    Use this when:
    - You need to persist generated data for further processing

    Input: content + optional filename
    Output: file path
    """
    temp_dir = tempfile.gettempdir()
    if filename is None:
        temp_file = tempfile.NamedTemporaryFile(delete=False, dir=temp_dir)
        filepath = temp_file.name
    else:
        filepath = os.path.join(temp_dir, filename)

    with open(filepath, "w") as f:
        f.write(content)
    log_tool("save_and_read_file", {"content": content, "filename": filename}, f"File saved to {filepath}")
    return f"File saved to {filepath}. You can read this file to process its contents."


@tool
def download_file_from_url(url: str, filename: Optional[str] = None) -> str:
    """
    Download a file from a URL and save it to a temporary location.
    Args:
        url (str): the URL of the file to download.
        filename (str, optional): the name of the file. If not provided, a random name file will be created.
    """
    try:
        # Parse URL to get filename if not provided
        if not filename:
            path = urlparse(url).path
            filename = os.path.basename(path)
            if not filename:
                filename = f"downloaded_{uuid.uuid4().hex[:8]}"

        # Create temporary file
        temp_dir = tempfile.gettempdir()
        filepath = os.path.join(temp_dir, filename)

        # Download the file
        response = requests.get(url, stream=True)
        response.raise_for_status()

        # Save the file
        with open(filepath, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        log_tool("download_file_from_url", url, f"File downloaded to {filepath}")
        return f"File downloaded to {filepath}. You can read this file to process its contents."
    except Exception as e:
        log_tool("download_file_from_url", url, f"ERROR: {str(e)}")
        return f"Error downloading file: {str(e)}"


@tool
def extract_text_from_image(image_path: str) -> str:
    """
    Extract text from an image using OCR.

    Use this when:
    - The image contains readable text

    Do NOT use for:
    - Visual analysis (use analyze_image instead)

    Input: image path
    Output: extracted text
    """
    try:
        # Open the image
        image = Image.open(image_path)

        # Extract text from the image
        text = pytesseract.image_to_string(image)
        log_tool("extract_text_from_image", image_path, text)
        return f"Extracted text from image:\n\n{text}"
    except Exception as e:
        log_tool("extract_text_from_image", image_path, f"ERROR: {str(e)}")
        return f"Error extracting text from image: {str(e)}"


@tool
def analyze_csv_file(file_path: str, query: str) -> str:
    """
    Analyze a CSV file using pandas.

    Use this when:
    - The task involves structured tabular data
    - The user asks questions about CSV content

    Do NOT use for:
    - Non-tabular data

    Input: CSV path + query
    Output: summary and statistics
    """
    try:
        # Read the CSV file
        df = pd.read_csv(file_path)

        # Run various analyses based on the query
        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
        result += f"Columns: {', '.join(df.columns)}\n\n"

        # Add summary statistics
        result += "Summary statistics:\n"
        result += str(df.describe())
        log_tool("analyze_csv_file", file_path, result)
        return result

    except Exception as e:
        log_tool("analyze_csv_file", file_path, f"ERROR: {str(e)}")
        return f"Error analyzing CSV file: {str(e)}"


@tool
def analyze_excel_file(file_path: str, query: str) -> str:
    """
    Analyze an Excel file.

    Use this when:
    - The task involves spreadsheet data

    Input: Excel file path + query
    Output: analysis and summary
    """
    try:
        # Read the Excel file
        df = pd.read_excel(file_path)

        # Run various analyses based on the query
        result = (
            f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n"
        )
        result += f"Columns: {', '.join(df.columns)}\n\n"

        # Add summary statistics
        result += "Summary statistics:\n"
        result += str(df.describe())
        log_tool("analyze_excel_tool", {file_path,query},result)
        return result

    except Exception as e:
        log_tool("analyze_excel_tool", {file_path,query},f"ERROR: {str(e)}")
        return f"Error analyzing Excel file: {str(e)}"