import requests
import pandas as pd
from PIL import Image
import os
import subprocess
from bs4 import BeautifulSoup
import urllib.parse
def web_search(query: str, api_key: str = None) -> str:
"""
Perform web search using SerpAPI if available, otherwise fallback to DuckDuckGo scraping.
"""
if api_key and api_key != "your-serpapi-key-here":
return _serpapi_search(query, api_key)
else:
return _duckduckgo_search(query)
def _serpapi_search(query: str, api_key: str) -> str:
"""Search using SerpAPI."""
try:
url = f"https://serpapi.com/search"
params = {
"q": query,
"api_key": api_key,
"engine": "google"
}
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
results = response.json()
organic_results = results.get("organic_results", [])
if organic_results:
# Get top 3 results
search_summary = []
for i, result in enumerate(organic_results[:3]):
title = result.get("title", "")
snippet = result.get("snippet", "")
if title and snippet:
search_summary.append(f"{i+1}. {title}: {snippet}")
return "\n".join(search_summary) if search_summary else "No useful results found"
else:
return "No search results found"
except requests.RequestException as e:
print(f"SerpAPI search error: {e}")
return "Search failed"
def _duckduckgo_search(query: str) -> str:
"""Fallback web search using DuckDuckGo scraping."""
try:
# DuckDuckGo instant answer API
url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": "1",
"skip_disambig": "1"
}
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
# Try to get instant answer
abstract = data.get("Abstract", "")
if abstract:
return f"Summary: {abstract}"
# Try related topics
related_topics = data.get("RelatedTopics", [])
if related_topics:
summaries = []
for topic in related_topics[:3]:
if isinstance(topic, dict) and "Text" in topic:
summaries.append(topic["Text"])
if summaries:
return "Related information:\n" + "\n".join(summaries)
# Fallback to web scraping (simplified)
return _simple_web_scrape(query)
except Exception as e:
print(f"DuckDuckGo search error: {e}")
return "Search failed"
def _simple_web_scrape(query: str) -> str:
"""Simple web scraping fallback."""
try:
# Use a simple search approach
search_url = f"https://html.duckduckgo.com/html/?q={urllib.parse.quote(query)}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
response = requests.get(search_url, headers=headers, timeout=10)
if response.status_code == 200:
soup = BeautifulSoup(response.content, 'html.parser')
# Try to extract some basic information
results = soup.find_all('a', class_='result__snippet')[:3]
if results:
snippets = [r.get_text().strip() for r in results if r.get_text().strip()]
return "\n".join(snippets[:3]) if snippets else "Limited search results available"
return "Basic web search completed - limited results"
except Exception as e:
print(f"Web scraping error: {e}")
return "Web search unavailable"
def read_file(file_name: str) -> str:
"""
Read and process different file types (text, CSV, images).
"""
if not file_name or not os.path.exists(file_name):
return "File not found"
try:
file_extension = os.path.splitext(file_name)[1].lower()
if file_extension == ".csv":
return _read_csv_file(file_name)
elif file_extension in [".png", ".jpg", ".jpeg", ".gif", ".bmp"]:
return _read_image_file(file_name)
elif file_extension in [".txt", ".md", ".py", ".js", ".html", ".json"]:
return _read_text_file(file_name)
else:
# Try to read as text file
return _read_text_file(file_name)
except Exception as e:
return f"Error reading file: {str(e)}"
def _read_text_file(file_name: str) -> str:
"""Read a text file."""
try:
with open(file_name, "r", encoding="utf-8") as f:
content = f.read()
return content[:5000] # Limit to first 5000 characters
except UnicodeDecodeError:
# Try with different encoding
try:
with open(file_name, "r", encoding="latin-1") as f:
content = f.read()
return content[:5000]
except Exception as e:
return f"Text file reading error: {str(e)}"
def _read_csv_file(file_name: str) -> str:
"""Read and summarize a CSV file."""
try:
df = pd.read_csv(file_name)
# Create a summary
summary = []
summary.append(f"CSV file shape: {df.shape[0]} rows, {df.shape[1]} columns")
summary.append(f"Columns: {', '.join(df.columns.tolist())}")
# Show first few rows
summary.append("\nFirst 5 rows:")
summary.append(df.head().to_string())
# Show basic statistics for numeric columns
numeric_columns = df.select_dtypes(include=['number']).columns
if len(numeric_columns) > 0:
summary.append(f"\nNumeric column statistics:")
summary.append(df[numeric_columns].describe().to_string())
return "\n".join(summary)
except Exception as e:
return f"CSV reading error: {str(e)}"
def _read_image_file(file_name: str) -> str:
"""Read and analyze an image file."""
try:
# Try OCR first
try:
import pytesseract
img = Image.open(file_name)
# Get image info
info = f"Image: {img.size[0]}x{img.size[1]} pixels, mode: {img.mode}"
# Try OCR
text = pytesseract.image_to_string(img).strip()
if text:
return f"{info}\n\nExtracted text:\n{text}"
else:
return f"{info}\n\nNo text detected in image."
except ImportError:
# OCR not available, just return image info
img = Image.open(file_name)
return f"Image: {img.size[0]}x{img.size[1]} pixels, mode: {img.mode}\n(OCR not available - install pytesseract for text extraction)"
except Exception as e:
return f"Image reading error: {str(e)}"
def execute_code(code: str, timeout: int = 10) -> str:
"""
Execute Python code safely with timeout.
"""
try:
# Basic security check - prevent dangerous operations
dangerous_keywords = ["import os", "import subprocess", "__import__", "exec", "eval", "open("]
if any(keyword in code.lower() for keyword in dangerous_keywords):
return "Code execution blocked: potentially unsafe operations detected"
result = subprocess.run(
["python3", "-c", code],
capture_output=True,
text=True,
timeout=timeout,
cwd="/tmp" # Run in safe directory
)
if result.returncode == 0:
return result.stdout.strip() if result.stdout else "Code executed successfully (no output)"
else:
return f"Code execution error: {result.stderr.strip()}"
except subprocess.TimeoutExpired:
return "Code execution timeout"
except Exception as e:
return f"Code execution error: {str(e)}"
def calculate_simple_math(expression: str) -> str:
"""
Safely evaluate simple mathematical expressions.
"""
try:
# Only allow basic math characters
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return "Invalid mathematical expression"
# Use eval safely for basic math
result = eval(expression)
return str(result)
except Exception as e:
return f"Math calculation error: {str(e)}"