tools.py

import re
import uuid
from typing import Any, Dict, Union, Optional

from langchain_core.callbacks import Callbacks
from langchain_core.runnables import RunnableConfig
from openai import OpenAI

import requests
from bs4 import BeautifulSoup
from config import selected_language
from serpapi import GoogleSearch
from langchain.tools import BaseTool
import json


def DailyNews():
    dailynews = ""
    url = "https://v.api.aa1.cn/api/zhihu-news/"
    result = requests.get(url, verify=False).json()
    for index in result['news']:
        dailynews = dailynews + index['title'] + "\n"
        # print(index['title'])
    return dailynews


# selected_language = "ch"


class GetLocation(BaseTool):
    # global selected_language
    # language: str = selected_language
    name: str = "获取ip地址"
    description: str = (
        "获取本机IP归属地的工具，输入参数是本机IP地址"
        if selected_language == "ch"
        else "A tool to get the location of the local IP address. The input parameter is the local IP address."
    )

    def _run(self, ip_address: str) -> str:
        api_key = "f962beaf785545d0a90b76d105996cdb"
        url = f"https://ipgeolocation.abstractapi.com/v1/?api_key={api_key}&ip_address={ip_address}"
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.text
        except requests.exceptions.RequestException as e:
            return f"请求失败: {e}"


class CurrencyConversion(BaseTool):
    name: str = "货币换算"
    description: str =(
        "获取指定基准货币和目标货币之间的实时汇率。输入参数包括基准货币代码和目标货币代码。传参格式：'USD|EUR'（第一个参数为基准货币代码，第二个参数为目标货币代码）。"
        if selected_language == "ch"
        else "Get the real-time exchange rate between the specified benchmark currency and target currency. The input parameters include the benchmark currency code and the target currency code. Parameter transmission format: USD | EUR '(the first parameter is the base currency code, and the second parameter is the target currency code)."
    )

    def _run(self, _input: str) -> str:
        pattern = r"(?P<base_currency>\S+)\|(?P<target_currency>\S+)"
        match = re.match(pattern, _input)
        base = match.group("base_currency")
        target = match.group("target_currency")
        api_key = "3599c1514c554f6da2ae8916151e5770"
        url = f"https://exchange-rates.abstractapi.com/v1/live/?api_key={api_key}&base={base}&target={target}"
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.content
        except requests.exceptions.RequestException as e:
            return f"请求失败: {e}"


class GetHoliday(BaseTool):
    name: str = "获取假日信息"
    description: str = (
        "获取指定国家在某一天的假期信息。输入参数包括国家代码和日期（年、月、日）。传参格式：'US|2025|1|1'（第一个参数为国家代码，后续三个参数为年份、月份和日期）。"if selected_language == "ch"
        else "Retrieve holiday information for a specified country on a certain day. The input parameters include country code and date (year, month, day). Parameter transmission format: US | 2025 | 1 | 1 '(The first parameter is the country code, and the following three parameters are the year, month, and date)."
    )

    def _run(self, _input: str) -> str:

        pattern = r"(?P<country>[A-Za-z]{2,3})\|(?P<year>\d{4})\|(?P<month>\d{1,2})\|(?P<day>\d{1,2})"
        match = re.match(pattern, _input)
        country = match.group("country")
        year = match.group("year")
        month = match.group("month")
        day = match.group("day")
        base_url = "https://holidays.abstractapi.com/v1/"
        api_key = "fa1dbc3bf696406db3412a7d034ba531"
        params = {"api_key": api_key, "country": country}
        if year:
            params["year"] = year
        if month:
            params["month"] = month
        if day:
            params["day"] = day
        query_string = "&".join([f"{key}={value}" for key, value in params.items() if value is not None])
        url = f"{base_url}?{query_string}"
        try:
            response = requests.get(url)
            response.raise_for_status()  # 检查 HTTP 响应状态
            data = response.json()
            if data:
                holidays = "\n".join(
                    [f"{holiday['name']} - {holiday['type']}" for holiday in data]
                )
                return f"查询日期的节日信息:\n{holidays}"
            else:
                return f"{year}-{month}-{day} 在 {country} 没有节日。"
        except requests.exceptions.RequestException as e:
            return f"请求失败: {e}"


def holiday2(country, year):
    url = f"https://calendarific.com/api/v2/holidays?&api_key=O7fuldKDDJgpl9ckui8lduMWNjx4bCDf&country={country}&year={year}"
    response = requests.get(url).json()
    print(response)


class SafeExpressionEvaluator(BaseTool):
    name: str = "safe expression evaluator"
    description: str = (
        "安全地计算 Python 表达式，输入参数为表达式字符串。"
        if selected_language == "ch"
        else "Safely evaluate Python expressions, with input being an expression string."
    )
    safe_globals: dict = None  # 声明类型
    safe_locals: dict = None

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # 定义安全的执行环境
        self.safe_globals = {
            "__builtins__": {
                "abs": abs,
                "min": min,
                "max": max,
                "sum": sum,
                "len": len,
                "round": round,
                "pow": pow,
                "divmod": divmod,
            },
            "math":__import__("math"),
        }
        self.safe_locals = {}

    def _run(self, expression: str) -> Dict[str, Any]:
        """安全地计算一个表达式，并返回结果"""
        try:
            result = eval(expression, self.safe_globals, self.safe_locals)
            return {"result": result, "error": None}
        except Exception as e:
            return {"result": None, "error": str(e)}


class SafeCodeExecutor(BaseTool):
    name: str = "safe code executor"
    description: str = (
        "安全地执行 Python 代码块，输入参数为代码字符串。"
        if selected_language == "ch"
        else "Safely execute Python code blocks, with input being a code string."
    )
    safe_globals: dict = None  # 声明类型
    safe_locals: dict = None

    def __init__(self, **kwargs):
        super().__init__()
        self.safe_globals = {
            "__builtins__": {
                "abs": abs,
                "min": min,
                "max": max,
                "sum": sum,
                "len": len,
                "round": round,
                "pow": pow,
                "divmod": divmod,
                "print" : print
                # 添加更多安全函数
            },
            "math": __import__("math"),
        }
        self.safe_locals = {}

    def _run(self, code: str) -> Dict[str, Any]:
        """安全地执行一段 Python 代码（包括函数、类定义等），无返回值"""
        try:
            exec(code, self.safe_globals, self.safe_locals)
            return {"result": "Code executed successfully.", "error": None}
        except Exception as e:
            return {"result": None, "error": str(e)}

def weather(city):
    """
    高德天气，查询当天的天气
    """
    url = f"https://restapi.amap.com/v3/weather/weatherInfo?city={city}&key=e7ff59e1de1573456145f0d736cc3a2f"
    response = requests.get(url).json()
    print(response)


class WeatherCrossing(BaseTool):
    name: str = "天气查询"
    description: str = "使用 Visual Crossing API 查询指定城市的天气信息,输入参数包括城市名称，以及可选的开始日期和结束日期（格式为 yyyy-MM-dd）。" if selected_language == "ch" else "Use the Visual Crossing API to query weather information for a specified city, with input parameters including the city name and optional start and end dates (in the format yyyy MM dd)."

    def _run(self, city: str, date1: str = None, date2: str = None) -> dict:
        base_url = "https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline"
        api_key = "6LX8E4LWX7ZFSCCA5S3JLFVPS"
        if date1 and date2:
            url = f"{base_url}/{city}/{date1}/{date2}?key={api_key}"
        elif date1:
            url = f"{base_url}/{city}/{date1}?key={api_key}"
        else:
            url = f"{base_url}/{city}?key={api_key}"
        try:
            response = requests.get(url)
            response.raise_for_status()  # 检查 HTTP 响应状态
            data = response.json()
            return {"result": data, "error": None}
        except requests.exceptions.RequestException as e:
            return {"result": None, "error": f"请求失败: {e}"}
        except ValueError:
            return {"result": None, "error": "无法解析返回的 JSON 数据。"}


class RegionInquiryTool(BaseTool):
    name: str = "行政区查询"  # 添加类型注解
    description: str = ("获取指定行政区的信息，包括名称、描述（行政级别）。根据 subdistrict 参数，返回该行政区的下级行政区信息。输入参数包括行政区名称和下级行政区级数（0 - 3）。传参格式：'北京|1'（第一个参数为行政区名称，第二个参数为下级行政区级数）。 "
        if selected_language == "ch"
        else "Retrieve information about the designated administrative region, including its name and description (administrative level). According to the subdistrict parameter, return the information of the subordinate administrative regions of the administrative region. The input parameters include the name of the administrative region and the level of the subordinate administrative region (0-3). Parameter format: 'Beijing | 1' (the first parameter is the name of the administrative region, and the second parameter is the level of the subordinate administrative region)")

    def _run(self, _input: str) -> Dict[str, Any]:

        pattern = r"(?P<district>.+?)\|(?P<level>\d{1,3})"
        match = re.match(pattern, _input)
        district = match.group("district")
        level = match.group("level")
        # keywords = kwargs.get("keywords")
        # subdistrict = kwargs.get("subdistrict", 0)
        api_key = "e7ff59e1de1573456145f0d736cc3a2f"
        base_url = "https://restapi.amap.com/v3/config/district"
        params = {
            "keywords": district,
            "subdistrict": level,
            "key": api_key
        }
        try:
            response = requests.get(base_url, params=params)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            return {"error": str(e)}


class HTMLTextExtractor(BaseTool):
    name: str = "html提取"
    description: str = "提取 HTML 文档中的纯文本内容,输入参数为 HTML 文档的字符串内容，返回去除标签后的纯文本。" if selected_language == "ch" else  "Extract plain text content from an HTML document, input the string content of the HTML document as the parameter, and return the plain text after removing tags."

    def _run(self, html_content: str) -> dict:
        if not html_content:
            return {"result": None, "error": "HTML 内容不能为空。"}

        try:
            soup = BeautifulSoup(html_content, 'lxml')
            text = soup.get_text()
            return {"result": text.strip(), "error": None}
        except Exception as e:
            return {"result": None, "error": f"提取文本失败: {e}"}


class BaiduNewsSearchTool(BaseTool):
    name: str = "Baidu News Search"  # 添加类型注解
    description: str = "备用搜索引擎，当主要搜索引擎不返回结果时调用，输入是检索query" if selected_language == "ch" else "Backup search engine, called when the primary search engine does not return results, input is search query"# 添加类型注解

    def _run(self, query: str) -> str:
        """
        使用百度新闻引擎进行查询。
        """
        params = {
            "engine": "baidu_news",
            "q": query,
            "ct": "1",
            "api_key": "dcc98b22d5f7d413979a175ff7d75b721c5992a3ee1e2363020b2bbdf4f82404"
        }
        search = GoogleSearch(params)  # 使用GoogleSearch库进行查询
        results = search.get_dict()
        organic_results = results.get("organic_results", [])

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


class BoChaSearchTool(BaseTool):
    name: str = "网页搜索"  # 添加类型注解
    description: str = "主力搜索引擎，当需要检索信息时调用，输入是检索query。" if selected_language == "ch" else "The main search engine is called when information needs to be retrieved, with the input being a search query." # 添加类型注解
    def _run(self, query: str) -> str:

        url = "https://api.bochaai.com/v1/web-search"
        payload = json.dumps({
            "query": query,
            "count": 3,
            "summary": True
        })
        headers = {
            'Authorization': 'Bearer sk-adeb0353f3434158934a841ac0f8d4a4',
            'Content-Type': 'application/json'
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        print(response.json())
        return response.json()


class ImageSearchTool(BaseTool):
    name: str = "图片搜索"
    description: str = "搜索图片引擎，当你需要检索相关图片的时候调用，输入是检索query，输出是与query有关的图片的信息" if selected_language == "ch" else "Search image engine, called when you need to retrieve relevant images. The input is the search query, and the output is information about the images related to the query"

    def _run(self, query: str) -> str:
        params = {
            "engine": "google_images",
            "q": query,
            "gl": "cn",
        }
        search = GoogleSearch(params)
        results = search.get_dict()
        suggested_searches = results.get('suggested_searches', [])
        if suggested_searches:
            thumbnails = [search['thumbnail'] for search in suggested_searches][:3]
        else:
            thumbnails = []
        information = ""
        client = OpenAI()
        for idx, thumbnail in enumerate(thumbnails):
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {
                        "role": "user",
                        "content": [
                            {"type": "text", "text": "What’s in this image?Give a brief answer"},
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": thumbnail,
                                },
                            },
                        ],
                    }
                ],
                max_tokens=100,
            )
            response_content = response.choices[0].message.content
            information = information + str(idx + 1) + ": " + response_content + "\n"
        return information


# tool = CurrencyConversion()
# print(tool.run("USD|EUR"))

class LocationSearch(BaseTool):
    name: str = "地点查询"
    description: str = (
        "搜索地点的工具，调用参数为一个关键词" if selected_language == "ch"
        else "A tool for searching locations, with a call parameter of a keyword"
    )

    def _run(self, keyword: str) -> str:
        key = "e7ff59e1de1573456145f0d736cc3a2f"
        url = f"https://restapi.amap.com/v3/place/text?keywords={keyword}&offset=20&page=1&key={key}"
        response = requests.get(url)
        return response.json()


class ReverseGeocodingTool(BaseTool):
    name: str = "经纬度获取"
    description: str = (
        "通过给定结构化地址来获取地址经纬度的工具，调用参数为结构化地址信息（规则遵循：国家、省份、城市、区县、城镇、乡村、街道、门牌号码、屋邨、大厦，如：北京市朝阳区阜通东大街6号。）"
        if selected_language == "ch"
        else "A tool that obtains the latitude and longitude of an address by giving it a structured address, with the parameters being structured address information (following the rules: country, province, city, district, town, countryside, street, house number, housing estate, building, such as No. 6 Futong East Street, Chaoyang District, Beijing)"
    )

    def _run(self, address: str) -> str:
        key = "e7ff59e1de1573456145f0d736cc3a2f"
        url = f"https://restapi.amap.com/v3/geocode/geo?address={address}&output=JSON&key={key}"
        response = requests.get(url)
        return response.json()['geocodes'][0]['location']