110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import re
|
|||
|
|
from pydantic import BaseModel, Field
|
|||
|
|
from typing import List, Dict
|
|||
|
|
import requests
|
|||
|
|
from loguru import logger
|
|||
|
|
|
|||
|
|
from .base_tool import BaseToolModel
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
class WeatherInfo(BaseToolModel):
|
|||
|
|
"""
|
|||
|
|
Tips:
|
|||
|
|
default control Required, e.g. key1 is not Required/key2 is Required
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
name: str = "WeatherInfo"
|
|||
|
|
description: str = "According to the user's input adcode, it can query the current/future weather conditions of the target area."
|
|||
|
|
|
|||
|
|
class ToolInputArgs(BaseModel):
|
|||
|
|
"""Input for Weather."""
|
|||
|
|
|
|||
|
|
# key: str = Field(..., description="用户在高德地图官网申请web服务API类型KEY")
|
|||
|
|
city: str = Field(..., description="城市编码，输入城市的adcode，adcode信息可参考城市编码表")
|
|||
|
|
extensions: str = Field(default=None, enum=["base", "all"], description="气象类型，输入城市的adcode，adcode信息可参考城市编码表")
|
|||
|
|
|
|||
|
|
class ToolOutputArgs(BaseModel):
|
|||
|
|
"""Output for Weather."""
|
|||
|
|
|
|||
|
|
lives: str = Field(default=None, description="实况天气数据")
|
|||
|
|
|
|||
|
|
# @classmethod
|
|||
|
|
# def run(cls, tool_input_args: ToolInputArgs) -> ToolOutputArgs:
|
|||
|
|
# """excute your tool!"""
|
|||
|
|
# url = "https://restapi.amap.com/v3/weather/weatherInfo"
|
|||
|
|
# try:
|
|||
|
|
# json_data = tool_input_args.dict()
|
|||
|
|
# json_data["key"] = "4ceb2ef6257a627b72e3be6beab5b059"
|
|||
|
|
# res = requests.get(url, json_data)
|
|||
|
|
# return res.json()
|
|||
|
|
# except Exception as e:
|
|||
|
|
# return e
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def run(city, extensions) -> ToolOutputArgs:
|
|||
|
|
"""excute your tool!"""
|
|||
|
|
url = "https://restapi.amap.com/v3/weather/weatherInfo"
|
|||
|
|
try:
|
|||
|
|
json_data = {}
|
|||
|
|
json_data["city"] = city
|
|||
|
|
json_data["key"] = "4ceb2ef6257a627b72e3be6beab5b059"
|
|||
|
|
json_data["extensions"] = extensions
|
|||
|
|
logger.debug(f"json_data: {json_data}")
|
|||
|
|
res = requests.get(url, params=json_data)
|
|||
|
|
return res.json()
|
|||
|
|
except Exception as e:
|
|||
|
|
return e
|
|||
|
|
|
|||
|
|
|
|||
|
|
class DistrictInfo(BaseToolModel):
|
|||
|
|
"""
|
|||
|
|
Tips:
|
|||
|
|
default control Required, e.g. key1 is not Required/key2 is Required
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
name: str = "DistrictInfo"
|
|||
|
|
description: str = "用户希望通过得到行政区域信息，进行开发工作。"
|
|||
|
|
|
|||
|
|
class ToolInputArgs(BaseModel):
|
|||
|
|
"""Input for district."""
|
|||
|
|
keywords: str = Field(default=None, description="规则：只支持单个关键词语搜索关键词支持：行政区名称、citycode、adcode例如，在subdistrict=2，搜索省份（例如山东），能够显示市（例如济南），区（例如历下区）")
|
|||
|
|
subdistrict: str = Field(default=None, enums=[1,2,3], description="""规则：设置显示下级行政区级数（行政区级别包括：国家、省/直辖市、市、区/县、乡镇/街道多级数据）
|
|||
|
|
|
|||
|
|
可选值：0、1、2、3等数字，并以此类推
|
|||
|
|
|
|||
|
|
0：不返回下级行政区；
|
|||
|
|
|
|||
|
|
1：返回下一级行政区；
|
|||
|
|
|
|||
|
|
2：返回下两级行政区；
|
|||
|
|
|
|||
|
|
3：返回下三级行政区；""")
|
|||
|
|
page: int = Field(default=1, examples=["page=2", "page=3"], description="最外层的districts最多会返回20个数据，若超过限制，请用page请求下一页数据。")
|
|||
|
|
extensions: str = Field(default=None, enum=["base", "all"], description="气象类型，输入城市的adcode，adcode信息可参考城市编码表")
|
|||
|
|
|
|||
|
|
class ToolOutputArgs(BaseModel):
|
|||
|
|
"""Output for district."""
|
|||
|
|
|
|||
|
|
districts: str = Field(default=None, description="行政区列表")
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def run(keywords=None, subdistrict=None, page=1, extensions=None) -> ToolOutputArgs:
|
|||
|
|
"""excute your tool!"""
|
|||
|
|
url = "https://restapi.amap.com/v3/config/district"
|
|||
|
|
try:
|
|||
|
|
json_data = {}
|
|||
|
|
json_data["keywords"] = keywords
|
|||
|
|
json_data["key"] = "4ceb2ef6257a627b72e3be6beab5b059"
|
|||
|
|
json_data["subdistrict"] = subdistrict
|
|||
|
|
json_data["page"] = page
|
|||
|
|
json_data["extensions"] = extensions
|
|||
|
|
logger.debug(f"json_data: {json_data}")
|
|||
|
|
res = requests.get(url, params=json_data)
|
|||
|
|
return res.json()
|
|||
|
|
except Exception as e:
|
|||
|
|
return e
|