Files
Mofox-Core/docs/plugins/api/person-api.md
2025-06-19 23:21:31 +08:00

342 lines
9.0 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# 个人信息API
个人信息API模块提供用户信息查询和管理功能让插件能够获取和使用用户的相关信息。
## 导入方式
```python
from src.plugin_system.apis import person_api
```
## 主要功能
### 1. Person ID管理
#### `get_person_id(platform: str, user_id: int) -> str`
根据平台和用户ID获取person_id
**参数:**
- `platform`:平台名称,如 "qq", "telegram" 等
- `user_id`用户ID
**返回:**
- `str`唯一的person_idMD5哈希值
**示例:**
```python
person_id = person_api.get_person_id("qq", 123456)
print(f"Person ID: {person_id}")
```
### 2. 用户信息查询
#### `get_person_value(person_id: str, field_name: str, default: Any = None) -> Any`
根据person_id和字段名获取某个值
**参数:**
- `person_id`用户的唯一标识ID
- `field_name`:要获取的字段名,如 "nickname", "impression" 等
- `default`:当字段不存在或获取失败时返回的默认值
**返回:**
- `Any`:字段值或默认值
**示例:**
```python
nickname = await person_api.get_person_value(person_id, "nickname", "未知用户")
impression = await person_api.get_person_value(person_id, "impression")
```
#### `get_person_values(person_id: str, field_names: list, default_dict: dict = None) -> dict`
批量获取用户信息字段值
**参数:**
- `person_id`用户的唯一标识ID
- `field_names`:要获取的字段名列表
- `default_dict`:默认值字典,键为字段名,值为默认值
**返回:**
- `dict`:字段名到值的映射字典
**示例:**
```python
values = await person_api.get_person_values(
person_id,
["nickname", "impression", "know_times"],
{"nickname": "未知用户", "know_times": 0}
)
```
### 3. 用户状态查询
#### `is_person_known(platform: str, user_id: int) -> bool`
判断是否认识某个用户
**参数:**
- `platform`:平台名称
- `user_id`用户ID
**返回:**
- `bool`:是否认识该用户
**示例:**
```python
known = await person_api.is_person_known("qq", 123456)
if known:
print("这个用户我认识")
```
### 4. 用户名查询
#### `get_person_id_by_name(person_name: str) -> str`
根据用户名获取person_id
**参数:**
- `person_name`:用户名
**返回:**
- `str`person_id如果未找到返回空字符串
**示例:**
```python
person_id = person_api.get_person_id_by_name("张三")
if person_id:
print(f"找到用户: {person_id}")
```
## 使用示例
### 1. 基础用户信息获取
```python
from src.plugin_system.apis import person_api
async def get_user_info(platform: str, user_id: int):
"""获取用户基本信息"""
# 获取person_id
person_id = person_api.get_person_id(platform, user_id)
# 获取用户信息
user_info = await person_api.get_person_values(
person_id,
["nickname", "impression", "know_times", "last_seen"],
{
"nickname": "未知用户",
"impression": "",
"know_times": 0,
"last_seen": 0
}
)
return {
"person_id": person_id,
"nickname": user_info["nickname"],
"impression": user_info["impression"],
"know_times": user_info["know_times"],
"last_seen": user_info["last_seen"]
}
```
### 2. 在Action中使用用户信息
```python
from src.plugin_system.base import BaseAction
class PersonalizedAction(BaseAction):
async def execute(self, action_data, chat_stream):
# 获取发送者信息
user_id = chat_stream.user_info.user_id
platform = chat_stream.platform
# 获取person_id
person_id = person_api.get_person_id(platform, user_id)
# 获取用户昵称和印象
nickname = await person_api.get_person_value(person_id, "nickname", "朋友")
impression = await person_api.get_person_value(person_id, "impression", "")
# 根据用户信息个性化回复
if impression:
response = f"你好 {nickname}!根据我对你的了解:{impression}"
else:
response = f"你好 {nickname}!很高兴见到你。"
return {
"success": True,
"response": response,
"user_info": {
"nickname": nickname,
"impression": impression
}
}
```
### 3. 用户识别和欢迎
```python
async def welcome_user(chat_stream):
"""欢迎用户,区分新老用户"""
user_id = chat_stream.user_info.user_id
platform = chat_stream.platform
# 检查是否认识这个用户
is_known = await person_api.is_person_known(platform, user_id)
if is_known:
# 老用户,获取详细信息
person_id = person_api.get_person_id(platform, user_id)
nickname = await person_api.get_person_value(person_id, "nickname", "老朋友")
know_times = await person_api.get_person_value(person_id, "know_times", 0)
welcome_msg = f"欢迎回来,{nickname}!我们已经聊过 {know_times} 次了。"
else:
# 新用户
welcome_msg = "你好很高兴认识你我是MaiBot。"
return welcome_msg
```
### 4. 用户搜索功能
```python
async def find_user_by_name(name: str):
"""根据名字查找用户"""
person_id = person_api.get_person_id_by_name(name)
if not person_id:
return {"found": False, "message": f"未找到名为 '{name}' 的用户"}
# 获取用户详细信息
user_info = await person_api.get_person_values(
person_id,
["nickname", "platform", "user_id", "impression", "know_times"],
{}
)
return {
"found": True,
"person_id": person_id,
"info": user_info
}
```
### 5. 用户印象分析
```python
async def analyze_user_relationship(chat_stream):
"""分析用户关系"""
user_id = chat_stream.user_info.user_id
platform = chat_stream.platform
person_id = person_api.get_person_id(platform, user_id)
# 获取关系相关信息
relationship_info = await person_api.get_person_values(
person_id,
["nickname", "impression", "know_times", "relationship_level", "last_interaction"],
{
"nickname": "未知",
"impression": "",
"know_times": 0,
"relationship_level": "stranger",
"last_interaction": 0
}
)
# 分析关系程度
know_times = relationship_info["know_times"]
if know_times == 0:
relationship = "陌生人"
elif know_times < 5:
relationship = "新朋友"
elif know_times < 20:
relationship = "熟人"
else:
relationship = "老朋友"
return {
"nickname": relationship_info["nickname"],
"relationship": relationship,
"impression": relationship_info["impression"],
"interaction_count": know_times
}
```
## 常用字段说明
### 基础信息字段
- `nickname`:用户昵称
- `platform`:平台信息
- `user_id`用户ID
### 关系信息字段
- `impression`:对用户的印象
- `know_times`:交互次数
- `relationship_level`:关系等级
- `last_seen`:最后见面时间
- `last_interaction`:最后交互时间
### 个性化字段
- `preferences`:用户偏好
- `interests`:兴趣爱好
- `mood_history`:情绪历史
- `topic_interests`:话题兴趣
## 最佳实践
### 1. 错误处理
```python
async def safe_get_user_info(person_id: str, field: str):
"""安全获取用户信息"""
try:
value = await person_api.get_person_value(person_id, field)
return value if value is not None else "未设置"
except Exception as e:
logger.error(f"获取用户信息失败: {e}")
return "获取失败"
```
### 2. 批量操作
```python
async def get_complete_user_profile(person_id: str):
"""获取完整用户档案"""
# 一次性获取所有需要的字段
fields = [
"nickname", "impression", "know_times",
"preferences", "interests", "relationship_level"
]
defaults = {
"nickname": "用户",
"impression": "",
"know_times": 0,
"preferences": "{}",
"interests": "[]",
"relationship_level": "stranger"
}
profile = await person_api.get_person_values(person_id, fields, defaults)
# 处理JSON字段
try:
profile["preferences"] = json.loads(profile["preferences"])
profile["interests"] = json.loads(profile["interests"])
except:
profile["preferences"] = {}
profile["interests"] = []
return profile
```
## 注意事项
1. **异步操作**:大部分查询函数都是异步的,需要使用`await`
2. **错误处理**:所有函数都有错误处理,失败时记录日志并返回默认值
3. **数据类型**返回的数据可能是字符串、数字或JSON需要适当处理
4. **性能考虑**:批量查询优于单个查询
5. **隐私保护**:确保用户信息的使用符合隐私政策
6. **数据一致性**person_id是用户的唯一标识应妥善保存和使用