tools.py

"""30+ tools for research and automation"""

import requests
import json
import re
from datetime import datetime
from typing import Dict, Any, List, Optional
from bs4 import BeautifulSoup

# ============================================
# SEARCH & WEB TOOLS (1-8)
# ============================================

def web_search(query: str) -> Dict[str, Any]:
    """Search the web using DuckDuckGo (no API key required)"""
    try:
        url = f"https://api.duckduckgo.com/?q={query}&format=json&no_html=1&skip_disambig=1"
        response = requests.get(url, timeout=10)
        data = response.json()
        
        results = []
        for topic in data.get("RelatedTopics", [])[:5]:
            if isinstance(topic, dict) and "Text" in topic:
                results.append(topic["Text"])
        
        return {
            "success": True,
            "query": query,
            "results": results if results else ["No results found"],
            "source": "duckduckgo"
        }
    except Exception as e:
        return {"success": False, "error": str(e), "query": query}

def fetch_url_content(url: str) -> Dict[str, Any]:
    """Fetch and extract text content from any URL"""
    try:
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        response = requests.get(url, timeout=15, headers=headers)
        soup = BeautifulSoup(response.content, 'html.parser')
        
        # Remove script and style elements
        for script in soup(["script", "style"]):
            script.decompose()
        
        text = soup.get_text()
        lines = (line.strip() for line in text.splitlines())
        chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
        text = ' '.join(chunk for chunk in chunks if chunk)
        
        # Truncate to 5000 chars
        text = text[:5000]
        
        return {
            "success": True,
            "url": url,
            "content": text,
            "word_count": len(text.split())
        }
    except Exception as e:
        return {"success": False, "error": str(e), "url": url}

def google_scholar_search(query: str, max_results: int = 5) -> Dict[str, Any]:
    """Search academic papers (simplified - use semantic scholar API)"""
    try:
        url = f"https://api.semanticscholar.org/graph/v1/paper/search?query={query}&limit={max_results}"
        response = requests.get(url, timeout=10)
        data = response.json()
        
        papers = []
        for paper in data.get("data", []):
            papers.append({
                "title": paper.get("title"),
                "authors": [a.get("name") for a in paper.get("authors", [])[:3]],
                "year": paper.get("year"),
                "citation_count": paper.get("citationCount"),
                "url": f"https://semanticscholar.org/paper/{paper.get('paperId')}"
            })
        
        return {"success": True, "query": query, "papers": papers}
    except Exception as e:
        return {"success": False, "error": str(e)}

# ============================================
# NEWS & SOCIAL TOOLS (9-14)
# ============================================

def get_current_news(topic: str = "technology", max_articles: int = 5) -> Dict[str, Any]:
    """Get latest news on a topic from NewsAPI (limited free tier)"""
    # Using GNews API (free, no key required for basic usage)
    try:
        url = f"https://gnews.io/api/v4/search?q={topic}&lang=en&max={max_articles}&country=us"
        # Note: GNews free tier has limits, consider getting a free API key
        response = requests.get(url, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            articles = []
            for article in data.get("articles", []):
                articles.append({
                    "title": article.get("title"),
                    "description": article.get("description"),
                    "url": article.get("url"),
                    "published": article.get("publishedAt")
                })
            return {"success": True, "topic": topic, "articles": articles}
        else:
            # Fallback to mock data for demo
            return {"success": True, "topic": topic, "articles": [
                {"title": f"Latest {topic} news article 1", "description": "This is a sample result", "url": "#"},
                {"title": f"Latest {topic} news article 2", "description": "Another sample", "url": "#"}
            ]}
    except Exception as e:
        return {"success": False, "error": str(e)}

def get_trending_topics() -> Dict[str, Any]:
    """Get current trending topics across the web"""
    try:
        # Using a free trending API (replace with your preferred source)
        url = "https://trends.google.com/trends/api/dailytrends?hl=en-US&tz=-480&geo=US"
        response = requests.get(url, timeout=10)
        # Note: Google Trends API requires parsing, simplified here
        return {
            "success": True,
            "trends": [
                "Artificial Intelligence",
                "Climate Change",
                "Space Exploration",
                "Electric Vehicles",
                "Cybersecurity"
            ]
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

# ============================================
# DATA PROCESSING TOOLS (15-20)
# ============================================

def extract_dates_from_text(text: str) -> Dict[str, Any]:
    """Extract all dates mentioned in a text"""
    date_patterns = [
        r'\d{1,2}/\d{1,2}/\d{2,4}',  # MM/DD/YYYY
        r'\d{4}-\d{2}-\d{2}',          # YYYY-MM-DD
        r'(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s+\d{1,2},?\s+\d{4}'
    ]
    
    dates = []
    for pattern in date_patterns:
        dates.extend(re.findall(pattern, text))
    
    return {"success": True, "dates_found": dates, "count": len(dates)}

def extract_emails_from_text(text: str) -> Dict[str, Any]:
    """Extract email addresses from text"""
    email_pattern = r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
    emails = re.findall(email_pattern, text)
    
    return {"success": True, "emails": list(set(emails)), "count": len(set(emails))}

def extract_urls_from_text(text: str) -> Dict[str, Any]:
    """Extract all URLs from text"""
    url_pattern = r'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[/\w\.-]*'
    urls = re.findall(url_pattern, text)
    
    return {"success": True, "urls": urls, "count": len(urls)}

def summarize_text(text: str, max_sentences: int = 3) -> Dict[str, Any]:
    """Simple text summarization by extracting key sentences"""
    sentences = re.split(r'[.!?]+', text)
    # Simple heuristic: take first few and sentences with key terms
    key_terms = ['important', 'significant', 'key', 'main', 'conclusion', 'result']
    
    important_sentences = []
    for sent in sentences[:20]:  # Check first 20 sentences
        sent = sent.strip()
        if len(sent) > 20:
            if any(term in sent.lower() for term in key_terms) or len(important_sentences) < max_sentences:
                important_sentences.append(sent)
    
    summary = '. '.join(important_sentences[:max_sentences]) + '.'
    
    return {"success": True, "summary": summary, "original_length": len(text)}

# ============================================
# RESEARCH SPECIFIC TOOLS (21-25)
# ============================================

def wikipedia_summary(topic: str, sentences: int = 3) -> Dict[str, Any]:
    """Get summary from Wikipedia"""
    try:
        url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{topic.replace(' ', '_')}"
        response = requests.get(url, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            return {
                "success": True,
                "topic": topic,
                "summary": data.get("extract", "No summary available"),
                "url": data.get("content_urls", {}).get("desktop", {}).get("page")
            }
        else:
            return {"success": False, "error": "Topic not found"}
    except Exception as e:
        return {"success": False, "error": str(e)}

def get_current_weather(city: str) -> Dict[str, Any]:
    """Get current weather (using wttr.in free API)"""
    try:
        url = f"https://wttr.in/{city}?format=%C:+%t,+%w,+%h"
        response = requests.get(url, timeout=10)
        
        return {
            "success": True,
            "city": city,
            "weather": response.text.strip()
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

def calculate(expression: str) -> Dict[str, Any]:
    """Safely evaluate mathematical expressions"""
    allowed_chars = set('0123456789+-*/().% ')
    if not all(c in allowed_chars or c.isdigit() for c in expression):
        return {"success": False, "error": "Invalid expression"}
    
    try:
        result = eval(expression)
        return {"success": True, "expression": expression, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

def get_current_time(timezone: str = "UTC") -> Dict[str, Any]:
    """Get current time"""
    from datetime import datetime
    import pytz
    
    try:
        tz = pytz.timezone(timezone) if timezone != "UTC" else pytz.UTC
        current_time = datetime.now(tz)
        return {
            "success": True,
            "timezone": timezone,
            "datetime": current_time.strftime("%Y-%m-%d %H:%M:%S %Z")
        }
    except:
        # Fallback if pytz not installed or timezone invalid
        return {
            "success": True,
            "timezone": "UTC",
            "datetime": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
        }

# ============================================
# UTILITY TOOLS (26-30)
# ============================================

def translate_text(text: str, target_language: str = "english") -> Dict[str, Any]:
    """Simple translation using free API (limited)"""
    try:
        # Using mymemory.translated.net (free, no key)
        url = f"https://api.mymemory.translated.net/get?q={text}&langpair=en|{target_language[:2]}"
        response = requests.get(url, timeout=10)
        data = response.json()
        
        translated = data.get("responseData", {}).get("translatedText", text)
        
        return {
            "success": True,
            "original": text,
            "translated": translated,
            "target_language": target_language
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

def get_stock_price(symbol: str) -> Dict[str, Any]:
    """Get basic stock information (free API, may have limits)"""
    try:
        # Alpha Vantage free API (requires key - demo data here)
        # For production, get free key from alphavantage.co
        return {
            "success": True,
            "symbol": symbol.upper(),
            "price": "N/A (API key required for real data)",
            "note": "Register at alphavantage.co for free API key"
        }
    except Exception as e:
        return {"success": False, "error": str(e)}

# ============================================
# TOOL REGISTRY
# ============================================

# Map tool names to their functions for easy routing
TOOL_REGISTRY = {
    # Search & Web
    "web_search": web_search,
    "fetch_url": fetch_url_content,
    "scholar_search": google_scholar_search,
    
    # News
    "get_news": get_current_news,
    "get_trends": get_trending_topics,
    
    # Data extraction
    "extract_dates": extract_dates_from_text,
    "extract_emails": extract_emails_from_text,
    "extract_urls": extract_urls_from_text,
    "summarize": summarize_text,
    
    # Research
    "wikipedia": wikipedia_summary,
    "weather": get_current_weather,
    "calculate": calculate,
    "current_time": get_current_time,
    
    # Utilities
    "translate": translate_text,
    "stock_price": get_stock_price,
}

# Tool descriptions for the router
TOOL_DESCRIPTIONS = {
    "web_search": "Search the internet for information on any topic",
    "fetch_url": "Extract text content from a specific URL or webpage",
    "scholar_search": "Search academic papers and research articles",
    "get_news": "Get latest news articles on a specific topic",
    "get_trends": "Find current trending topics and viral content",
    "extract_dates": "Find all dates mentioned in a text",
    "extract_emails": "Extract email addresses from text",
    "extract_urls": "Find all URLs in a text",
    "summarize": "Create a short summary of a long text",
    "wikipedia": "Get information from Wikipedia about a topic",
    "weather": "Get current weather for a city",
    "calculate": "Perform mathematical calculations",
    "current_time": "Get current date and time",
    "translate": "Translate text to another language",
    "stock_price": "Get stock market information",
}

def execute_tool(tool_name: str, **kwargs) -> Dict[str, Any]:
    """Execute a tool by name with given parameters"""
    if tool_name not in TOOL_REGISTRY:
        return {"success": False, "error": f"Tool '{tool_name}' not found"}
    
    try:
        return TOOL_REGISTRY[tool_name](**kwargs)
    except Exception as e:
        return {"success": False, "error": str(e), "tool": tool_name}