mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-24 08:03:53 +08:00
@@ -2,59 +2,27 @@ from datetime import datetime
|
||||
import cnlunar
|
||||
from plugins_func.register import register_function, ToolType, ActionResponse, Action
|
||||
|
||||
# 添加星期映射字典
|
||||
WEEKDAY_MAP = {
|
||||
"Monday": "星期一",
|
||||
"Tuesday": "星期二",
|
||||
"Wednesday": "星期三",
|
||||
"Thursday": "星期四",
|
||||
"Friday": "星期五",
|
||||
"Saturday": "星期六",
|
||||
"Sunday": "星期日",
|
||||
}
|
||||
|
||||
get_time_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_time",
|
||||
"description": "获取今天日期或者当前时间信息",
|
||||
"parameters": {"type": "object", "properties": {}, "required": []},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@register_function("get_time", get_time_function_desc, ToolType.WAIT)
|
||||
def get_time():
|
||||
"""
|
||||
获取当前的日期时间信息
|
||||
"""
|
||||
now = datetime.now()
|
||||
current_time = now.strftime("%H:%M:%S")
|
||||
current_date = now.strftime("%Y-%m-%d")
|
||||
current_weekday = WEEKDAY_MAP[now.strftime("%A")]
|
||||
response_text = (
|
||||
f"当前日期: {current_date},当前时间: {current_time}, {current_weekday}"
|
||||
)
|
||||
|
||||
return ActionResponse(Action.REQLLM, response_text, None)
|
||||
|
||||
|
||||
get_lunar_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_lunar",
|
||||
"description": (
|
||||
"用于获取今天的阴历/农历和黄历信息。"
|
||||
"用于具体日期的阴历/农历和黄历信息。"
|
||||
"用户可以指定查询内容,如:阴历日期、天干地支、节气、生肖、星座、八字、宜忌等。"
|
||||
"如果没有指定查询内容,则默认查询干支年和农历日期。"
|
||||
"对于'今天农历是多少'、'今天农历日期'这样的基本查询,请直接使用context中的信息,不要调用此工具。"
|
||||
),
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"date": {
|
||||
"type": "string",
|
||||
"description": "要查询的日期,格式为YYYY-MM-DD,例如2024-01-01。如果不提供,则使用当前日期",
|
||||
},
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "要查询的内容,例如阴历日期、天干地支、节日、节气、生肖、星座、八字、宜忌等",
|
||||
}
|
||||
},
|
||||
},
|
||||
"required": [],
|
||||
},
|
||||
@@ -63,23 +31,41 @@ get_lunar_function_desc = {
|
||||
|
||||
|
||||
@register_function("get_lunar", get_lunar_function_desc, ToolType.WAIT)
|
||||
def get_lunar(query=None):
|
||||
def get_lunar(date=None, query=None):
|
||||
"""
|
||||
用于获取当前的阴历/农历,和天干地支、节气、生肖、星座、八字、宜忌等黄历信息
|
||||
"""
|
||||
now = datetime.now()
|
||||
current_time = now.strftime("%H:%M:%S")
|
||||
from core.utils.cache.manager import cache_manager, CacheType
|
||||
|
||||
# 如果提供了日期参数,则使用指定日期;否则使用当前日期
|
||||
if date:
|
||||
try:
|
||||
now = datetime.strptime(date, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
return ActionResponse(
|
||||
Action.REQLLM,
|
||||
f"日期格式错误,请使用YYYY-MM-DD格式,例如:2024-01-01",
|
||||
None,
|
||||
)
|
||||
else:
|
||||
now = datetime.now()
|
||||
|
||||
current_date = now.strftime("%Y-%m-%d")
|
||||
current_weekday = WEEKDAY_MAP[now.strftime("%A")]
|
||||
|
||||
# 如果 query 为 None,则使用默认文本
|
||||
if query is None:
|
||||
query = "默认查询干支年和农历日期"
|
||||
|
||||
# 尝试从缓存获取农历信息
|
||||
lunar_cache_key = f"lunar_info_{current_date}"
|
||||
cached_lunar_info = cache_manager.get(CacheType.LUNAR, lunar_cache_key)
|
||||
if cached_lunar_info:
|
||||
return ActionResponse(Action.REQLLM, cached_lunar_info, None)
|
||||
|
||||
response_text = f"根据以下信息回应用户的查询请求,并提供与{query}相关的信息:\n"
|
||||
|
||||
lunar = cnlunar.Lunar(now, godType="8char")
|
||||
response_text += (
|
||||
f"当前公历日期: {current_date},当前时间: {current_time},{current_weekday}\n"
|
||||
"农历信息:\n"
|
||||
"%s年%s%s\n" % (lunar.lunarYearCn, lunar.lunarMonthCn[:-1], lunar.lunarDayCn)
|
||||
+ "干支: %s年 %s月 %s日\n" % (lunar.year8Char, lunar.month8Char, lunar.day8Char)
|
||||
@@ -135,4 +121,7 @@ def get_lunar(query=None):
|
||||
+ "(默认返回干支年和农历日期;仅在要求查询宜忌信息时才返回本日宜忌)"
|
||||
)
|
||||
|
||||
# 缓存农历信息
|
||||
cache_manager.set(CacheType.LUNAR, lunar_cache_key, response_text)
|
||||
|
||||
return ActionResponse(Action.REQLLM, response_text, None)
|
||||
|
||||
@@ -151,20 +151,44 @@ def parse_weather_info(soup):
|
||||
|
||||
@register_function("get_weather", GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
|
||||
def get_weather(conn, location: str = None, lang: str = "zh_CN"):
|
||||
api_host = conn.config["plugins"]["get_weather"].get("api_host", "mj7p3y7naa.re.qweatherapi.com")
|
||||
api_key = conn.config["plugins"]["get_weather"].get("api_key", "a861d0d5e7bf4ee1a83d9a9e4f96d4da")
|
||||
from core.utils.cache.manager import cache_manager, CacheType
|
||||
|
||||
api_host = conn.config["plugins"]["get_weather"].get(
|
||||
"api_host", "mj7p3y7naa.re.qweatherapi.com"
|
||||
)
|
||||
api_key = conn.config["plugins"]["get_weather"].get(
|
||||
"api_key", "a861d0d5e7bf4ee1a83d9a9e4f96d4da"
|
||||
)
|
||||
default_location = conn.config["plugins"]["get_weather"]["default_location"]
|
||||
client_ip = conn.client_ip
|
||||
|
||||
# 优先使用用户提供的location参数
|
||||
if not location:
|
||||
# 通过客户端IP解析城市
|
||||
if client_ip:
|
||||
# 动态解析IP对应的城市信息
|
||||
ip_info = get_ip_info(client_ip, logger)
|
||||
location = ip_info.get("city") if ip_info and "city" in ip_info else None
|
||||
# 先从缓存获取IP对应的城市信息
|
||||
cached_ip_info = cache_manager.get(CacheType.IP_INFO, client_ip)
|
||||
if cached_ip_info:
|
||||
location = cached_ip_info.get("city")
|
||||
else:
|
||||
# 缓存未命中,调用API获取
|
||||
ip_info = get_ip_info(client_ip, logger)
|
||||
if ip_info:
|
||||
cache_manager.set(CacheType.IP_INFO, client_ip, ip_info)
|
||||
location = ip_info.get("city")
|
||||
|
||||
if not location:
|
||||
location = default_location
|
||||
else:
|
||||
# 若IP解析失败或无IP,使用默认位置
|
||||
# 若无IP,使用默认位置
|
||||
location = default_location
|
||||
# 尝试从缓存获取完整天气报告
|
||||
weather_cache_key = f"full_weather_{location}_{lang}"
|
||||
cached_weather_report = cache_manager.get(CacheType.WEATHER, weather_cache_key)
|
||||
if cached_weather_report:
|
||||
return ActionResponse(Action.REQLLM, cached_weather_report, None)
|
||||
|
||||
# 缓存未命中,获取实时天气数据
|
||||
city_info = fetch_city_info(location, api_key, api_host)
|
||||
if not city_info:
|
||||
return ActionResponse(
|
||||
@@ -192,4 +216,7 @@ def get_weather(conn, location: str = None, lang: str = "zh_CN"):
|
||||
# 提示语
|
||||
weather_report += "\n(如需某一天的具体天气,请告诉我日期)"
|
||||
|
||||
# 缓存完整的天气报告
|
||||
cache_manager.set(CacheType.WEATHER, weather_cache_key, weather_report)
|
||||
|
||||
return ActionResponse(Action.REQLLM, weather_report, None)
|
||||
|
||||
Reference in New Issue
Block a user