mirror of https://github.com/Mai-with-u/MaiBot.git
234 lines
10 KiB
Python
234 lines
10 KiB
Python
"""
|
||
根据person_name查询用户信息 - 工具实现
|
||
支持模糊查询，可以查询某个用户的所有信息
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
from src.common.logger import get_logger
|
||
from src.common.database.database_model import PersonInfo
|
||
from .tool_registry import register_memory_retrieval_tool
|
||
|
||
logger = get_logger("memory_retrieval_tools")
|
||
|
||
|
||
async def query_person_info(person_name: str) -> str:
|
||
"""根据person_name查询用户信息，使用模糊查询
|
||
|
||
Args:
|
||
person_name: 用户名称（person_name字段）
|
||
|
||
Returns:
|
||
str: 查询结果，包含用户的所有信息
|
||
"""
|
||
try:
|
||
person_name = str(person_name).strip()
|
||
if not person_name:
|
||
return "用户名称为空"
|
||
|
||
# 构建查询条件（使用模糊查询）
|
||
query = PersonInfo.select().where(
|
||
PersonInfo.person_name.contains(person_name)
|
||
)
|
||
|
||
# 执行查询
|
||
records = list(query.limit(20)) # 最多返回20条记录
|
||
|
||
if not records:
|
||
return f"未找到模糊匹配'{person_name}'的用户信息"
|
||
|
||
# 区分精确匹配和模糊匹配的结果
|
||
exact_matches = []
|
||
fuzzy_matches = []
|
||
|
||
for record in records:
|
||
# 检查是否是精确匹配
|
||
if record.person_name and record.person_name.strip() == person_name:
|
||
exact_matches.append(record)
|
||
else:
|
||
fuzzy_matches.append(record)
|
||
|
||
# 构建结果文本
|
||
results = []
|
||
|
||
# 先处理精确匹配的结果
|
||
for record in exact_matches:
|
||
result_parts = []
|
||
result_parts.append("【精确匹配】") # 标注为精确匹配
|
||
|
||
# 基本信息
|
||
if record.person_name:
|
||
result_parts.append(f"用户名称：{record.person_name}")
|
||
if record.nickname:
|
||
result_parts.append(f"昵称：{record.nickname}")
|
||
if record.person_id:
|
||
result_parts.append(f"用户ID：{record.person_id}")
|
||
if record.platform:
|
||
result_parts.append(f"平台：{record.platform}")
|
||
if record.user_id:
|
||
result_parts.append(f"平台用户ID：{record.user_id}")
|
||
|
||
# 名称设定原因
|
||
if record.name_reason:
|
||
result_parts.append(f"名称设定原因：{record.name_reason}")
|
||
|
||
# 认识状态
|
||
result_parts.append(f"是否已认识：{'是' if record.is_known else '否'}")
|
||
|
||
# 时间信息
|
||
if record.know_since:
|
||
know_since_str = datetime.fromtimestamp(record.know_since).strftime("%Y-%m-%d %H:%M:%S")
|
||
result_parts.append(f"首次认识时间：{know_since_str}")
|
||
if record.last_know:
|
||
last_know_str = datetime.fromtimestamp(record.last_know).strftime("%Y-%m-%d %H:%M:%S")
|
||
result_parts.append(f"最后认识时间：{last_know_str}")
|
||
if record.know_times:
|
||
result_parts.append(f"认识次数：{int(record.know_times)}")
|
||
|
||
# 记忆点（memory_points）
|
||
if record.memory_points:
|
||
try:
|
||
memory_points_data = json.loads(record.memory_points) if isinstance(record.memory_points, str) else record.memory_points
|
||
if isinstance(memory_points_data, list) and memory_points_data:
|
||
# 解析记忆点格式：category:content:weight
|
||
memory_list = []
|
||
for memory_point in memory_points_data:
|
||
if memory_point and isinstance(memory_point, str):
|
||
parts = memory_point.split(":", 2)
|
||
if len(parts) >= 3:
|
||
category = parts[0].strip()
|
||
content = parts[1].strip()
|
||
weight = parts[2].strip()
|
||
memory_list.append(f" - [{category}] {content} (权重: {weight})")
|
||
else:
|
||
memory_list.append(f" - {memory_point}")
|
||
|
||
if memory_list:
|
||
result_parts.append("记忆点：\n" + "\n".join(memory_list))
|
||
except (json.JSONDecodeError, TypeError, ValueError) as e:
|
||
logger.warning(f"解析用户 {record.person_id} 的memory_points失败: {e}")
|
||
# 如果解析失败，直接显示原始内容（截断）
|
||
memory_preview = str(record.memory_points)[:200]
|
||
if len(str(record.memory_points)) > 200:
|
||
memory_preview += "..."
|
||
result_parts.append(f"记忆点（原始数据）：{memory_preview}")
|
||
|
||
results.append("\n".join(result_parts))
|
||
|
||
# 再处理模糊匹配的结果
|
||
for record in fuzzy_matches:
|
||
result_parts = []
|
||
result_parts.append("【模糊匹配】") # 标注为模糊匹配
|
||
|
||
# 基本信息
|
||
if record.person_name:
|
||
result_parts.append(f"用户名称：{record.person_name}")
|
||
if record.nickname:
|
||
result_parts.append(f"昵称：{record.nickname}")
|
||
if record.person_id:
|
||
result_parts.append(f"用户ID：{record.person_id}")
|
||
if record.platform:
|
||
result_parts.append(f"平台：{record.platform}")
|
||
if record.user_id:
|
||
result_parts.append(f"平台用户ID：{record.user_id}")
|
||
|
||
# 名称设定原因
|
||
if record.name_reason:
|
||
result_parts.append(f"名称设定原因：{record.name_reason}")
|
||
|
||
# 认识状态
|
||
result_parts.append(f"是否已认识：{'是' if record.is_known else '否'}")
|
||
|
||
# 时间信息
|
||
if record.know_since:
|
||
know_since_str = datetime.fromtimestamp(record.know_since).strftime("%Y-%m-%d %H:%M:%S")
|
||
result_parts.append(f"首次认识时间：{know_since_str}")
|
||
if record.last_know:
|
||
last_know_str = datetime.fromtimestamp(record.last_know).strftime("%Y-%m-%d %H:%M:%S")
|
||
result_parts.append(f"最后认识时间：{last_know_str}")
|
||
if record.know_times:
|
||
result_parts.append(f"认识次数：{int(record.know_times)}")
|
||
|
||
# 记忆点（memory_points）
|
||
if record.memory_points:
|
||
try:
|
||
memory_points_data = json.loads(record.memory_points) if isinstance(record.memory_points, str) else record.memory_points
|
||
if isinstance(memory_points_data, list) and memory_points_data:
|
||
# 解析记忆点格式：category:content:weight
|
||
memory_list = []
|
||
for memory_point in memory_points_data:
|
||
if memory_point and isinstance(memory_point, str):
|
||
parts = memory_point.split(":", 2)
|
||
if len(parts) >= 3:
|
||
category = parts[0].strip()
|
||
content = parts[1].strip()
|
||
weight = parts[2].strip()
|
||
memory_list.append(f" - [{category}] {content} (权重: {weight})")
|
||
else:
|
||
memory_list.append(f" - {memory_point}")
|
||
|
||
if memory_list:
|
||
result_parts.append("记忆点：\n" + "\n".join(memory_list))
|
||
except (json.JSONDecodeError, TypeError, ValueError) as e:
|
||
logger.warning(f"解析用户 {record.person_id} 的memory_points失败: {e}")
|
||
# 如果解析失败，直接显示原始内容（截断）
|
||
memory_preview = str(record.memory_points)[:200]
|
||
if len(str(record.memory_points)) > 200:
|
||
memory_preview += "..."
|
||
result_parts.append(f"记忆点（原始数据）：{memory_preview}")
|
||
|
||
results.append("\n".join(result_parts))
|
||
|
||
# 组合所有结果
|
||
if not results:
|
||
return f"未找到匹配'{person_name}'的用户信息"
|
||
|
||
response_text = "\n\n---\n\n".join(results)
|
||
|
||
# 添加统计信息
|
||
total_count = len(records)
|
||
exact_count = len(exact_matches)
|
||
fuzzy_count = len(fuzzy_matches)
|
||
|
||
# 显示精确匹配和模糊匹配的统计
|
||
if exact_count > 0 or fuzzy_count > 0:
|
||
stats_parts = []
|
||
if exact_count > 0:
|
||
stats_parts.append(f"精确匹配：{exact_count} 条")
|
||
if fuzzy_count > 0:
|
||
stats_parts.append(f"模糊匹配：{fuzzy_count} 条")
|
||
stats_text = "，".join(stats_parts)
|
||
response_text = f"找到 {total_count} 条匹配的用户信息（{stats_text}）：\n\n{response_text}"
|
||
elif total_count > 1:
|
||
response_text = f"找到 {total_count} 条匹配的用户信息：\n\n{response_text}"
|
||
else:
|
||
response_text = f"找到用户信息：\n\n{response_text}"
|
||
|
||
# 如果结果数量达到限制，添加提示
|
||
if total_count >= 20:
|
||
response_text += "\n\n(已显示前20条结果，可能还有更多匹配记录)"
|
||
|
||
return response_text
|
||
|
||
except Exception as e:
|
||
logger.error(f"查询用户信息失败: {e}")
|
||
return f"查询失败: {str(e)}"
|
||
|
||
|
||
def register_tool():
|
||
"""注册工具"""
|
||
register_memory_retrieval_tool(
|
||
name="query_person_info",
|
||
description="根据查询某个用户的所有信息。名称、昵称、平台、用户ID、qq号等",
|
||
parameters=[
|
||
{
|
||
"name": "person_name",
|
||
"type": "string",
|
||
"description": "用户名称，用于查询用户信息",
|
||
"required": True
|
||
}
|
||
],
|
||
execute_func=query_person_info
|
||
)
|
||
|