sofia_tools_advanced.py

#!/usr/bin/env python3
"""
SOFIA Advanced Tool Integration System
Expanded tool capabilities with APIs, databases, and web scraping
"""

import json, re, math, datetime, requests, sqlite3, os
import numpy as np
from typing import Dict, Any, List, Optional, Union
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Tool:
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def can_handle(self, query: str) -> bool:
        raise NotImplementedError

    def execute(self, query: str) -> Dict[str, Any]:
        raise NotImplementedError

class CalculatorTool(Tool):
    def __init__(self):
        super().__init__("calculator", "Performs mathematical calculations")

    def can_handle(self, query: str) -> bool:
        patterns = [
            r'\d+\s*[\+\-\*\/]\s*\d+',
            r'calculate\s+.+',
            r'what\s+is\s+.+[\+\-\*\/].+',
            r'solve\s+.+',
        ]
        return any(re.search(pattern, query.lower()) for pattern in patterns)

    def execute(self, query: str) -> Dict[str, Any]:
        try:
            expr_match = re.search(r'(\d+(?:\.\d+)?\s*[\+\-\*\/]\s*\d+(?:\.\d+)?)', query.lower())
            if expr_match:
                expression = expr_match.group(1).replace(' ', '')
                result = eval(expression)
                return {
                    "tool": "calculator",
                    "expression": expression,
                    "result": result,
                    "type": "arithmetic"
                }

            if 'sqrt' in query.lower():
                numbers = re.findall(r'\d+(?:\.\d+)?', query)
                if numbers:
                    num = float(numbers[0])
                    result = math.sqrt(num)
                    return {
                        "tool": "calculator",
                        "expression": f"sqrt({num})",
                        "result": result,
                        "type": "square_root"
                    }

        except Exception as e:
            return {"tool": "calculator", "error": f"Could not calculate: {str(e)}"}

        return {"tool": "calculator", "error": "No valid calculation found"}

class TimeTool(Tool):
    def __init__(self):
        super().__init__("time", "Provides current time, date, and temporal information")

    def can_handle(self, query: str) -> bool:
        time_keywords = ['time', 'date', 'day', 'hour', 'minute', 'second', 'now', 'today', 'tomorrow', 'yesterday']
        return any(keyword in query.lower() for keyword in time_keywords)

    def execute(self, query: str) -> Dict[str, Any]:
        now = datetime.datetime.now()

        result = {
            "tool": "time",
            "current_time": now.strftime("%H:%M:%S"),
            "current_date": now.strftime("%Y-%m-%d"),
            "day_of_week": now.strftime("%A"),
            "month": now.strftime("%B"),
            "year": now.year,
            "timezone": "UTC"
        }

        query_lower = query.lower()
        if 'tomorrow' in query_lower:
            tomorrow = now + datetime.timedelta(days=1)
            result['tomorrow'] = tomorrow.strftime("%Y-%m-%d (%A)")
        elif 'yesterday' in query_lower:
            yesterday = now - datetime.timedelta(days=1)
            result['yesterday'] = yesterday.strftime("%Y-%m-%d (%A)")

        return result

class SearchTool(Tool):
    def __init__(self):
        super().__init__("search", "Performs web searches and information retrieval")

    def can_handle(self, query: str) -> bool:
        search_keywords = ['search', 'find', 'lookup', 'what is', 'who is', 'where is', 'how to']
        return any(keyword in query.lower() for keyword in search_keywords)

    def execute(self, query: str) -> Dict[str, Any]:
        try:
            search_term = self._extract_search_term(query)

            if any(word in query.lower() for word in ['what is', 'who is', 'definition']):
                return self._wikipedia_search(search_term)

            return {
                "tool": "search",
                "search_term": search_term,
                "result": f"Search results for '{search_term}' would be retrieved from web sources",
                "type": "general_search",
                "suggestion": "Consider using specific APIs for better results"
            }

        except Exception as e:
            return {"tool": "search", "error": f"Search failed: {str(e)}"}

    def _extract_search_term(self, query: str) -> str:
        query = re.sub(r'^(what|who|where|how|when|why)\s+is\s+', '', query.lower())
        query = re.sub(r'^(search|find|lookup)\s+(for\s+)?', '', query.lower())
        return query.strip()

    def _wikipedia_search(self, term: str) -> Dict[str, Any]:
        try:
            import wikipedia
            page = wikipedia.page(term, auto_suggest=True)
            summary = page.summary[:500] + "..." if len(page.summary) > 500 else page.summary

            return {
                "tool": "search",
                "search_term": term,
                "result": summary,
                "source": "Wikipedia",
                "url": page.url,
                "type": "wikipedia_summary"
            }
        except Exception as e:
            return {"tool": "search", "error": f"Wikipedia search failed: {str(e)}"}

class DatabaseTool(Tool):
    def __init__(self, db_path: str = "sofia_memory.db"):
        super().__init__("database", "Access and query local knowledge database")
        self.db_path = db_path
        self._init_database()

    def can_handle(self, query: str) -> bool:
        db_keywords = ['remember', 'recall', 'stored', 'database', 'knowledge', 'facts']
        return any(keyword in query.lower() for keyword in db_keywords)

    def execute(self, query: str) -> Dict[str, Any]:
        try:
            query_lower = query.lower()

            if 'remember' in query_lower or 'store' in query_lower:
                return self._store_information(query)
            elif 'recall' in query_lower or 'retrieve' in query_lower:
                return self._retrieve_information(query)
            else:
                return self._query_database(query)

        except Exception as e:
            return {"tool": "database", "error": f"Database operation failed: {str(e)}"}

    def _init_database(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute('''
            CREATE TABLE IF NOT EXISTS knowledge (
                id INTEGER PRIMARY KEY,
                topic TEXT,
                content TEXT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                source TEXT
            )
        ''')

        cursor.execute('''
            CREATE TABLE IF NOT EXISTS facts (
                id INTEGER PRIMARY KEY,
                fact TEXT UNIQUE,
                category TEXT,
                confidence REAL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        conn.commit()
        conn.close()

    def _store_information(self, query: str) -> Dict[str, Any]:
        content = re.sub(r'(remember|store)\s+', '', query, flags=re.IGNORECASE).strip()

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute(
            "INSERT INTO knowledge (topic, content, source) VALUES (?, ?, ?)",
            ("user_input", content, "conversation")
        )

        conn.commit()
        conn.close()

        return {
            "tool": "database",
            "operation": "store",
            "content": content,
            "status": "stored"
        }

    def _retrieve_information(self, query: str) -> Dict[str, Any]:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute(
            "SELECT content, timestamp FROM knowledge ORDER BY timestamp DESC LIMIT 5"
        )

        results = cursor.fetchall()
        conn.close()

        return {
            "tool": "database",
            "operation": "retrieve",
            "results": [{"content": row[0], "timestamp": row[1]} for row in results],
            "count": len(results)
        }

    def _query_database(self, query: str) -> Dict[str, Any]:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute("SELECT COUNT(*) FROM knowledge")
        knowledge_count = cursor.fetchone()[0]

        cursor.execute("SELECT COUNT(*) FROM facts")
        facts_count = cursor.fetchone()[0]

        conn.close()

        return {
            "tool": "database",
            "operation": "stats",
            "knowledge_entries": knowledge_count,
            "facts_stored": facts_count
        }

class ToolManager:
    def __init__(self):
        self.tools = [
            CalculatorTool(),
            TimeTool(),
            SearchTool(),
            DatabaseTool()
        ]
        print(f"🔧 Loaded {len(self.tools)} advanced tools: {[t.name for t in self.tools]}")

    def execute_tools(self, query: str) -> List[Dict[str, Any]]:
        results = []
        for tool in self.tools:
            if tool.can_handle(query):
                print(f"🛠️  Using {tool.name}: {tool.description}")
                result = tool.execute(query)
                if 'error' not in result:
                    results.append(result)
                else:
                    print(f"⚠️  Tool {tool.name} failed: {result['error']}")
        return results

    def get_available_tools(self) -> List[Dict[str, str]]:
        return [{"name": tool.name, "description": tool.description} for tool in self.tools]

class AdvancedToolAugmentedSOFIA:
    def __init__(self):
        from conversational_sofia import ConversationalSOFIA
        self.sofia = ConversationalSOFIA()
        self.tool_manager = ToolManager()

    def process_query(self, query: str) -> Dict[str, Any]:
        print(f"🤖 Processing: '{query}'")

        tool_results = self.tool_manager.execute_tools(query)

        if tool_results:
            tool_contexts = []
            for result in tool_results:
                formatted = self._format_tool_result(result)
                tool_contexts.append(f"Tool {result['tool']}: {formatted}")

            enhanced_query = f"{query}\n\nTool Results:\n" + "\n".join(tool_contexts)
        else:
            enhanced_query = query

        response, embedding = self.sofia.chat(enhanced_query)

        return {
            "response": response,
            "embedding": embedding,
            "tools_used": tool_results,
            "tool_count": len(tool_results),
            "enhanced_query": enhanced_query
        }

    def _format_tool_result(self, result: Dict[str, Any]) -> str:
        tool_type = result.get('tool', '')

        if tool_type == 'time':
            return f"{result.get('current_time', 'N/A')} on {result.get('current_date', 'N/A')} ({result.get('day_of_week', 'N/A')})"

        elif tool_type == 'calculator':
            if 'expression' in result:
                return f"{result['expression']} = {result['result']}"
            else:
                return str(result.get('result', 'N/A'))

        elif tool_type == 'search':
            return result.get('result', 'Search completed')

        elif tool_type == 'database':
            if result.get('operation') == 'retrieve':
                return f"Retrieved {result.get('count', 0)} knowledge entries"
            elif result.get('operation') == 'store':
                return f"Stored: {result.get('content', 'N/A')}"
            else:
                return f"Database has {result.get('knowledge_entries', 0)} entries"

        else:
            return str(result.get('result', 'Tool executed successfully'))

def main():
    import sys

    if len(sys.argv) < 2:
        print("Usage: python sofia_tools.py 'query'")
        print("\nAvailable tools:")
        tool_manager = ToolManager()
        for tool_info in tool_manager.get_available_tools():
            print(f"  - {tool_info['name']}: {tool_info['description']}")
        return

    user_input = ' '.join(sys.argv[1:])

    tool_sofia = AdvancedToolAugmentedSOFIA()
    result = tool_sofia.process_query(user_input)

    print(f"\n🤖 SOFIA: {result['response']}")
    if result['tools_used']:
        print(f"\n🛠️  Advanced Tools Used: {result['tool_count']}")
        for tool_result in result['tools_used']:
            formatted = tool_sofia._format_tool_result(tool_result)
            print(f"  - {tool_result['tool']}: {formatted}")

if __name__ == "__main__":
    main()