50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
"""
|
|
更简单的单参数输入工具实现，用于查询现在天气的情况
|
|
"""
|
|
import json
|
|
import re
|
|
from pydantic import BaseModel, Field
|
|
import requests
|
|
from configs.kb_config import SENIVERSE_API_KEY
|
|
from server.chat import utils
|
|
|
|
|
|
def weather(location: str, api_key: str):
|
|
url = f"https://api.seniverse.com/v3/weather/daily.json?key={api_key}&location={location}&language=zh-Hans&unit=c&start=0&days=5"
|
|
response = requests.get(url)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
weather = {
|
|
"today": json.dumps(data["results"][0]["daily"][0]),
|
|
"others": json.dumps(data["results"][0]["daily"][1:])
|
|
}
|
|
weather_info = json.dumps(weather)
|
|
return weather_info
|
|
else:
|
|
raise Exception(
|
|
f"Failed to retrieve weather: {response.status_code}")
|
|
|
|
|
|
def weathercheck(query: str):
|
|
"""获取到uid并拆分query"""
|
|
try:
|
|
matches = re.findall(r'\{.*?\}', query)
|
|
if len(matches)>=2:
|
|
query = matches[0]
|
|
else:
|
|
return "<关键指令>不需要再调用该工具了</关键指令>"
|
|
location = json.loads(query)["location"]
|
|
time_based_uuid = json.loads(matches[1])["uuid"]
|
|
# sources = utils.get_shared_variable(time_based_uuid)
|
|
# sources["source_docs"]=["天气 预报"]
|
|
# sources["num"]+=1
|
|
# sources["END"] = "ok"
|
|
# utils.set_shared_variable(time_based_uuid,sources)
|
|
return weather(location, SENIVERSE_API_KEY)
|
|
except Exception as e:
|
|
return f"Failed to retrieve weather.{e}"
|
|
|
|
|
|
class WeatherInput(BaseModel):
|
|
location: str = Field(description="City name,include city and county")
|