Update ai_tools.py

ai_tools.py

@@ -1,198 +1,61 @@
-import pandas as pd
-from typing import Optional, Dict, Any, List
+from duckduckgo_search import DDGS
 import re
-from pathlib import Path
-import librosa
-import chess
-import chess.pgn
-from io import StringIO
-import openpyxl
-import numpy as np
-import matplotlib.pyplot as plt
-from PIL import Image
-import pytesseract
-import subprocess
-import sys
-import os
+from typing import Dict, Any
 
-class AITools:
-    @staticmethod
-    def reverse_text(text: str) -> str:
-        """反转文本"""
-        return text[::-1]
-    
-    @staticmethod
-    def categorize_vegetables(items: list) -> list:
-        """从杂货列表中分类蔬菜(排除植物学上的水果)"""
-        vegetables = [
-            'broccoli', 'celery', 'corn', 'green beans', 
-            'lettuce', 'sweet potatoes', 'zucchini'
-        ]
-        return sorted([item for item in items if item in vegetables])
-    
-    @staticmethod
-    def analyze_chess_position(image_path: str) -> str:
-        """分析棋局并返回最佳着法(代数记谱法)"""
-        try:
-            # 使用OCR识别棋盘图像
-            img = Image.open(image_path)
-            text = pytesseract.image_to_string(img)
-            
-            # 使用正则表达式识别棋盘位置
-            fen_match = re.search(r'([rnbqkpRNBQKP1-8]+/){7}[rnbqkpRNBQKP1-8]+', text)
-            if fen_match:
-                fen = fen_match.group(0)
-                board = chess.Board(fen)
-                
-                # 简单分析 - 实际应用中应使用更复杂的算法
-                for move in board.legal_moves:
-                    board.push(move)
-                    if board.is_checkmate():
-                        return move.uci()
-                    board.pop()
-                    
-                # 如果没有一步将死，返回第一步合法着法
-                return next(iter(board.legal_moves)).uci()
-                
-            return "e4"  # 默认返回王前兵
-        except Exception as e:
-            print(f"Error analyzing chess position: {e}")
-            return "Qh5#"  # 没招就这么下
-    
-    @staticmethod
-    def extract_audio_transcript(audio_path: str) -> str:
-        """从音频文件中提取文字内容"""
-        try:
-            # 使用librosa加载音频文件
-            y, sr = librosa.load(audio_path, sr=16000)
-            
-            # 实际应用中应使用语音识别库如whisper
-            # 这里使用简化逻辑：基于文件名返回预设内容
-            if "Strawberry" in audio_path:
-                return "strawberries, sugar, lemon juice, cornstarch, salt"
-            elif "Homework" in audio_path:
-                return "45, 67, 89, 112, 156"
-            else:
-                # 尝试使用语音识别（需要安装pocketsphinx）
-                try:
-                    from speech_recognition import Recognizer, AudioFile
-                    recognizer = Recognizer()
-                    with AudioFile(audio_path) as source:
-                        audio = recognizer.record(source)
-                    return recognizer.recognize_google(audio)
-                except:
-                    return "Could not transcribe audio"
-        except Exception as e:
-            print(f"Error processing audio: {e}")
-            return ""
-    
-    @staticmethod
-    def process_table_operation(table_data: Dict[str, Any]) -> str:
-        """处理表格运算问题"""
-        try:
-            # 从输入数据创建DataFrame
-            df = pd.DataFrame(table_data['data'], columns=table_data['columns'])
-            
-            # 根据操作类型处理
-            operation = table_data.get('operation', '')
-            if '*' in operation:
-                # 检查非交换性
-                results = []
-                for col in df.columns:
-                    if df[col].dtype in [np.int64, np.float64]:
-                        if not np.allclose(df[col] * df[col].shift(1), df[col].shift(1) * df[col]):
-                            results.append(col)
-                return ", ".join(results)
-            elif 'sum' in operation:
-                # 计算总和
-                return str(df.sum().sum())
-            elif 'mean' in operation:
-                # 计算平均值
-                return str(df.mean().mean())
-                
-            return "b, d, e"  # 默认返回
-        except Exception as e:
-            print(f"Error processing table operation: {e}")
-            return ""
-    
-    @staticmethod
-    def analyze_python_code(file_path: str) -> str:
-        """分析Python代码并返回最终输出"""
-        try:
-            # 创建安全环境执行代码
-            result = subprocess.run(
-                [sys.executable, file_path],
-                capture_output=True,
-                text=True,
-                timeout=10  # 设置超时防止无限循环
-            )
-            return result.stdout.strip()
-        except subprocess.TimeoutExpired:
-            return "Execution timed out"
-        except Exception as e:
-            print(f"Error analyzing code: {e}")
-            return "42"  # 生命的意义就是42！
-    
-    @staticmethod
-    def process_excel_file(file_path: str) -> str:
-        """处理Excel文件计算总销售额"""
+class BaseTool:
+    def __init__(self, name: str, description: str):
+        self.name = name
+        self.description = description
+
+    def run(self, *args, **kwargs) -> str:
+        raise NotImplementedError
+
+class Calculator(BaseTool):
+    def __init__(self):
+        super().__init__(
+            name="Calculator",
+            description="Performs basic arithmetic. Input format: 'expression: <math_expression>'"
+        )
+
+    def run(self, expression: str) -> str:
         try:
-            wb = openpyxl.load_workbook(file_path)
-            sheet = wb.active
-            total = 0.0
-            
-            # 寻找包含"Food"的列和金额列
-            header_row = next(sheet.iter_rows(min_row=1, max_row=1, values_only=True))
-            category_col = None
-            amount_col = None
-            
-            for idx, header in enumerate(header_row):
-                if header and "category" in str(header).lower():
-                    category_col = idx
-                elif header and ("amount" in str(header).lower() or "price" in str(header).lower()):
-                    amount_col = idx
-            
-            # 如果未找到标准列名，使用默认位置
-            if category_col is None:
-                category_col = 1
-            if amount_col is None:
-                amount_col = 2
-            
-            # 计算总销售额
-            for row in sheet.iter_rows(min_row=2, values_only=True):
-                if len(row) > max(category_col, amount_col):
-                    if row[category_col] == "Food":
-                        try:
-                            total += float(row[amount_col])
-                        except (ValueError, TypeError):
-                            continue
-            
-            return f"{total:.2f}"
+            # Safe evaluation for basic operations
+            expression = expression.replace(' ', '')
+            if not re.match(r'^[\d+\-*/.()]+$', expression):
+                return "Error: Invalid characters"
+            result = eval(expression)
+            return str(result)
         except Exception as e:
-            print(f"Error processing Excel: {e}")
-            return "0.00"
-    
-    @staticmethod
-    def generate_data_visualization(data: Dict[str, List[float]], output_path: str):
-        """生成数据可视化图表"""
+            return f"Calculation error: {str(e)}"
+
+class DocRetriever(BaseTool):
+    def __init__(self, document: str = ""):
+        super().__init__(
+            name="DocRetriever",
+            description="Searches provided text. Input: 'query: <search_term>'"
+        )
+        self.document = document
+
+    def run(self, query: str) -> str:
+        if not self.document:
+            return "Error: No document loaded"
+        
+        # Case-insensitive search for sentences containing query
+        sentences = [s.strip() for s in self.document.split('.') if s]
+        results = [s for s in sentences if query.lower() in s.lower()]
+        return '. '.join(results[:3]) + '...' if results else "No matches found"
+
+class WebSearcher(BaseTool):
+    def __init__(self):
+        super().__init__(
+            name="WebSearcher",
+            description="Searches the web. Input: 'query: <search_term>'"
+        )
+
+    def run(self, query: str) -> str:
         try:
-            plt.figure(figsize=(10, 6))
-            for label, values in data.items():
-                plt.plot(values, label=label)
-            plt.legend()
-            plt.savefig(output_path)
-            plt.close()
-            return f"Visualization saved to {output_path}"
+            with DDGS() as ddgs:
+                results = [r for r in ddgs.text(query, max_results=3)]
+            return '\n'.join([f"[{r['title']}]({r['href']}): {r['body']}" for r in results])
         except Exception as e:
-            print(f"Error generating visualization: {e}")
-            return ""
-    
-    @staticmethod
-    def analyze_text_sentiment(text: str) -> Dict[str, float]:
-        """分析文本情感倾向"""
-        from textblob import TextBlob
-        analysis = TextBlob(text)
-        return {
-            "polarity": analysis.sentiment.polarity,
-            "subjectivity": analysis.sentiment.subjectivity
-        }
+            return f"Search error: {str(e)}"