gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: 格式化代码，修复不一致的空格和注释，更新ruff action

Browse Source
This commit is contained in:
春河晴
2025-04-16 16:16:37 +09:00
parent abc4c812c5
commit a8d48fc6cf
10 changed files with 98 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
src/plugins/person_info/person_info.py
View File

@@ -64,12 +64,9 @@ class PersonInfoManager:
if "person_info" not in db.list_collection_names():
db.create_collection("person_info")
db.person_info.create_index("person_id", unique=True)
# 初始化时读取所有person_name
cursor = db.person_info.find(
{"person_name": {"$exists": True}},
{"person_id": 1, "person_name": 1, "_id": 0}
)
cursor = db.person_info.find({"person_name": {"$exists": True}}, {"person_id": 1, "person_name": 1, "_id": 0})
for doc in cursor:
if doc.get("person_name"):
self.person_name_list[doc["person_id"]] = doc["person_name"]
@@ -77,10 +74,10 @@ class PersonInfoManager:
def get_person_id(self, platform: str, user_id: int):
"""获取唯一id"""
#如果platform中存在-，就截取-后面的部分
# 如果platform中存在-，就截取-后面的部分
if "-" in platform:
platform = platform.split("-")[1]
components = [platform, str(user_id)]
key = "_".join(components)
return hashlib.md5(key.encode()).hexdigest()
@@ -93,8 +90,7 @@ class PersonInfoManager:
return True
else:
return False
async def create_person_info(self, person_id: str, data: dict = None):
"""创建一个项"""
if not person_id:
@@ -125,7 +121,7 @@ class PersonInfoManager:
Data[field_name] = value
logger.debug(f"更新时{person_id}不存在，已新建")
await self.create_person_info(person_id, Data)
async def has_one_field(self, person_id: str, field_name: str):
"""判断是否存在某一个字段"""
document = db.person_info.find_one({"person_id": person_id}, {field_name: 1})
@@ -133,36 +129,35 @@ class PersonInfoManager:
return True
else:
return False
def _extract_json_from_text(self, text: str) -> dict:
"""从文本中提取JSON数据的高容错方法"""
try:
# 尝试直接解析
return json.loads(text)
except json.JSONDecodeError:
try:
# 尝试找到JSON格式的部分
json_pattern = r'\{[^{}]*\}'
json_pattern = r"\{[^{}]*\}"
matches = re.findall(json_pattern, text)
if matches:
return json.loads(matches[0])
# 如果上面都失败了，尝试提取键值对
nickname_pattern = r'"nickname"[:\s]+"([^"]+)"'
reason_pattern = r'"reason"[:\s]+"([^"]+)"'
nickname_match = re.search(nickname_pattern, text)
reason_match = re.search(reason_pattern, text)
if nickname_match:
return {
"nickname": nickname_match.group(1),
"reason": reason_match.group(1) if reason_match else "未提供理由"
"reason": reason_match.group(1) if reason_match else "未提供理由",
}
except Exception as e:
logger.error(f"JSON提取失败: {str(e)}")
# 如果所有方法都失败了，返回空结果
return {"nickname": "", "reason": ""}
@@ -171,10 +166,10 @@ class PersonInfoManager:
if not person_id:
logger.debug("取名失败：person_id不能为空")
return
old_name = await self.get_value(person_id, "person_name")
old_reason = await self.get_value(person_id, "name_reason")
max_retries = 5 # 最大重试次数
current_try = 0
existing_names = ""
@@ -182,7 +177,7 @@ class PersonInfoManager:
individuality = Individuality.get_instance()
prompt_personality = individuality.get_prompt(type="personality", x_person=2, level=1)
bot_name = individuality.personality.bot_nickname
qv_name_prompt = f"你是{bot_name}，你{prompt_personality}"
qv_name_prompt += f"现在你想给一个用户取一个昵称，用户是的qq昵称是{user_nickname}，"
qv_name_prompt += f"用户的qq群昵称名是{user_cardname}，"
@@ -195,20 +190,20 @@ class PersonInfoManager:
if existing_names:
qv_name_prompt += f"\n请注意，以下名称已被使用，不要使用以下昵称：{existing_names}。\n"
qv_name_prompt += "请用json给出你的想法，并给出理由，示例如下："
qv_name_prompt += '''{
qv_name_prompt += """{
"nickname": "昵称",
"reason": "理由"
}'''
}"""
logger.debug(f"取名提示词：{qv_name_prompt}")
response = await self.qv_name_llm.generate_response(qv_name_prompt)
logger.debug(f"取名回复：{response}")
result = self._extract_json_from_text(response[0])
if not result["nickname"]:
logger.error("生成的昵称为空，重试中...")
current_try += 1
continue
# 检查生成的昵称是否已存在
if result["nickname"] not in self.person_name_list.values():
# 更新数据库和内存中的列表
@@ -216,16 +211,16 @@ class PersonInfoManager:
# await self.update_one_field(person_id, "nickname", user_nickname)
# await self.update_one_field(person_id, "avatar", user_avatar)
await self.update_one_field(person_id, "name_reason", result["reason"])
self.person_name_list[person_id] = result["nickname"]
logger.debug(f"用户 {person_id} 的名称已更新为 {result['nickname']}，原因：{result['reason']}")
return result
else:
existing_names += f"{result['nickname']}、"
logger.debug(f"生成的昵称 {result['nickname']} 已存在，重试中...")
current_try += 1
logger.error(f"在{max_retries}次尝试后仍未能生成唯一昵称")
return None

33
src/plugins/person_info/relationship_manager.py
View File

@@ -76,13 +76,13 @@ class RelationshipManager:
return mood_value * coefficient
else:
return mood_value / coefficient
async def is_known_some_one(self, platform , user_id):
async def is_known_some_one(self, platform, user_id):
"""判断是否认识某人"""
is_known = person_info_manager.is_person_known(platform, user_id)
return is_known
async def is_qved_name(self, platform , user_id):
async def is_qved_name(self, platform, user_id):
"""判断是否认识某人"""
person_id = person_info_manager.get_person_id(platform, user_id)
is_qved = await person_info_manager.has_one_field(person_id, "person_name")
@@ -93,42 +93,41 @@ class RelationshipManager:
return True
else:
return False
async def first_knowing_some_one(self, platform , user_id, user_nickname, user_cardname, user_avatar):
async def first_knowing_some_one(self, platform, user_id, user_nickname, user_cardname, user_avatar):
"""判断是否认识某人"""
person_id = person_info_manager.get_person_id(platform,user_id)
person_id = person_info_manager.get_person_id(platform, user_id)
await person_info_manager.update_one_field(person_id, "nickname", user_nickname)
# await person_info_manager.update_one_field(person_id, "user_cardname", user_cardname)
# await person_info_manager.update_one_field(person_id, "user_avatar", user_avatar)
await person_info_manager.qv_person_name(person_id, user_nickname, user_cardname, user_avatar)
async def convert_all_person_sign_to_person_name(self,input_text:str):
async def convert_all_person_sign_to_person_name(self, input_text: str):
"""将所有人的<platform:user_id:nickname:cardname>格式转换为person_name"""
try:
# 使用正则表达式匹配<platform:user_id:nickname:cardname>格式
all_person = person_info_manager.person_name_list
pattern = r'<([^:]+):(\d+):([^:]+):([^>]+)>'
pattern = r"<([^:]+):(\d+):([^:]+):([^>]+)>"
matches = re.findall(pattern, input_text)
# 遍历匹配结果，将<platform:user_id:nickname:cardname>替换为person_name
result_text = input_text
for platform, user_id, nickname, cardname in matches:
person_id = person_info_manager.get_person_id(platform, user_id)
# 默认使用昵称作为人名
person_name = nickname.strip() if nickname.strip() else cardname.strip()
if person_id in all_person:
if all_person[person_id] != None:
person_name = all_person[person_id]
print(f"将<{platform}:{user_id}:{nickname}:{cardname}>替换为{person_name}")
result_text = result_text.replace(f"<{platform}:{user_id}:{nickname}:{cardname}>", person_name)
return result_text
except Exception as e:
except Exception:
logger.error(traceback.format_exc())
return input_text