refactor(api): 合并兴趣度API和关系API,以及扩展关系API

2025-11-01 11:38:12 +08:00
parent 5b1cbb49b0
commit 5e7f17ebf9
6 changed files with 234 additions and 226 deletions
--- a/src/chat/message_receive/chat_stream.py
+++ b/src/chat/message_receive/chat_stream.py
@@ -289,11 +289,11 @@ class ChatStream:
        """获取用户关系分"""
        # 使用统一的评分API
        try:
-            from src.plugin_system.apis.scoring_api import scoring_api
+            from src.plugin_system.apis import person_api

            if self.user_info and hasattr(self.user_info, "user_id"):
                user_id = str(self.user_info.user_id)
-                relationship_score = await scoring_api.get_user_relationship_score(user_id)
+                relationship_score = await person_api.get_user_relationship_score(user_id)
                logger.debug(f"ChatStream {self.stream_id}: 用户关系分 = {relationship_score:.3f}")
                return relationship_score

--- a/src/chat/replyer/default_generator.py
+++ b/src/chat/replyer/default_generator.py
@@ -1949,12 +1949,12 @@ class DefaultReplyer:
            logger.error(f"获取关系信息失败: {e}")
            # 降级到基本信息
            try:
-                from src.plugin_system.apis.scoring_api import scoring_api
+                from src.plugin_system.apis import person_api

                user_info = await person_info_manager.get_values(person_id, ["user_id", "platform"])
                user_id = user_info.get("user_id", "unknown")

-                relationship_data = await scoring_api.get_user_relationship_data(user_id)
+                relationship_data = await person_api.get_user_relationship_data(user_id)
                if relationship_data:
                    relationship_text = relationship_data.get("relationship_text", "")
                    relationship_score = relationship_data.get("relationship_score", 0.3)
--- a/src/individuality/individuality.py
+++ b/src/individuality/individuality.py
@@ -85,9 +85,9 @@ class Individuality:
        full_personality = f"{personality_result}，{identity_result}"

        # 使用统一的评分API初始化智能兴趣系统
-        from src.plugin_system.apis.scoring_api import scoring_api
+        from src.plugin_system.apis import person_api

-        await scoring_api.initialize_smart_interests(
+        await person_api.initialize_smart_interests(
            personality_description=full_personality, personality_id=self.bot_person_id
        )

--- a/src/plugin_system/apis/person_api.py
+++ b/src/plugin_system/apis/person_api.py
@@ -4,34 +4,29 @@
 使用方式：
    from src.plugin_system.apis import person_api
    person_id = person_api.get_person_id("qq", 123456)
-    value = await person_api.get_person_value(person_id, "nickname")
+    info = await person_api.get_person_info(person_id)
 """

+import asyncio
 from typing import Any

 from src.common.logger import get_logger
 from src.person_info.person_info import PersonInfoManager, get_person_info_manager
+from src.plugin_system.services.interest_service import interest_service
+from src.plugin_system.services.relationship_service import relationship_service

 logger = get_logger("person_api")


 # =============================================================================
-# 个人信息API函数
+# 辅助函数
 # =============================================================================


-def get_person_id(platform: str, user_id: int) -> str:
-    """根据平台和用户ID获取person_id
+def get_person_id(platform: str, user_id: int | str) -> str:
+    """根据平台和用户ID获取person_id (同步)

-    Args:
-        platform: 平台名称，如 "qq", "telegram" 等
-        user_id: 用户ID
-
-    Returns:
-        str: 唯一的person_id（MD5哈希值）
-
-    示例:
-        person_id = person_api.get_person_id("qq", 123456)
+    这是一个核心的辅助函数，用于生成统一的用户标识。
    """
    try:
        return PersonInfoManager.get_person_id(platform, user_id)
@@ -40,93 +35,23 @@ def get_person_id(platform: str, user_id: int) -> str:
        return ""


-async def get_person_value(person_id: str, field_name: str, default: Any = None) -> Any:
-    """根据person_id和字段名获取某个值
-
-    Args:
-        person_id: 用户的唯一标识ID
-        field_name: 要获取的字段名，如 "nickname", "impression" 等
-        default: 当字段不存在或获取失败时返回的默认值
-
-    Returns:
-        Any: 字段值或默认值
-
-    示例:
-        nickname = await person_api.get_person_value(person_id, "nickname", "未知用户")
-        impression = await person_api.get_person_value(person_id, "impression")
-    """
+async def get_person_id_by_name(person_name: str) -> str:
+    """根据用户名获取person_id"""
    try:
        person_info_manager = get_person_info_manager()
-        value = await person_info_manager.get_value(person_id, field_name)
-        return value if value is not None else default
+        return await person_info_manager.get_person_id_by_person_name(person_name)
    except Exception as e:
-        logger.error(f"[PersonAPI] 获取用户信息失败: person_id={person_id}, field={field_name}, error={e}")
-        return default
+        logger.error(f"[PersonAPI] 根据用户名获取person_id失败: person_name={person_name}, error={e}")
+        return ""


-async def get_person_values(person_id: str, field_names: list, default_dict: dict | None = None) -> dict:
-    """批量获取用户信息字段值
-
-    Args:
-        person_id: 用户的唯一标识ID
-        field_names: 要获取的字段名列表
-        default_dict: 默认值字典，键为字段名，值为默认值
-
-    Returns:
-        dict: 字段名到值的映射字典
-
-    示例:
-        values = await person_api.get_person_values(
-            person_id,
-            ["nickname", "impression", "know_times"],
-            {"nickname": "未知用户", "know_times": 0}
-        )
-    """
-    try:
-        person_info_manager = get_person_info_manager()
-        values = await person_info_manager.get_values(person_id, field_names)
-
-        # 如果获取成功，返回结果
-        if values:
-            return values
-
-        # 如果获取失败，构建默认值字典
-        result = {}
-        if default_dict:
-            for field in field_names:
-                result[field] = default_dict.get(field, None)
-        else:
-            for field in field_names:
-                result[field] = None
-
-        return result
-
-    except Exception as e:
-        logger.error(f"[PersonAPI] 批量获取用户信息失败: person_id={person_id}, fields={field_names}, error={e}")
-        # 返回默认值字典
-        result = {}
-        if default_dict:
-            for field in field_names:
-                result[field] = default_dict.get(field, None)
-        else:
-            for field in field_names:
-                result[field] = None
-        return result
+# =============================================================================
+# 核心信息查询API
+# =============================================================================


 async def is_person_known(platform: str, user_id: int) -> bool:
-    """判断是否认识某个用户
-
-    Args:
-        platform: 平台名称
-        user_id: 用户ID
-
-    Returns:
-        bool: 是否认识该用户
-
-    示例:
-        known = await person_api.is_person_known("qq", 123456)
-    """
+    """判断是否认识某个用户"""
    try:
        person_info_manager = get_person_info_manager()
        return await person_info_manager.is_person_known(platform, user_id)
@@ -135,21 +60,217 @@ async def is_person_known(platform: str, user_id: int) -> bool:
        return False


-async def get_person_id_by_name(person_name: str) -> str:
-    """根据用户名获取person_id
+async def get_person_info(person_id: str) -> dict[str, Any]:
+    """获取用户的核心基础信息

-    Args:
-        person_name: 用户名
-
-    Returns:
-        str: person_id，如果未找到返回空字符串
-
-    示例:
-        person_id = person_api.get_person_id_by_name("张三")
+    返回一个包含用户基础信息的字典，例如 person_name, nickname, know_times, attitude 等。
    """
+    if not person_id:
+        return {}
    try:
        person_info_manager = get_person_info_manager()
-        return await person_info_manager.get_person_id_by_person_name(person_name)
+        fields = ["person_name", "nickname", "know_times", "know_since", "last_know", "attitude"]
+        values = await person_info_manager.get_values(person_id, fields)
+        return values
    except Exception as e:
-        logger.error(f"[PersonAPI] 根据用户名获取person_id失败: person_name={person_name}, error={e}")
-        return ""
+        logger.error(f"[PersonAPI] 获取用户信息失败: person_id={person_id}, error={e}")
+        return {}
+
+
+async def get_person_impression(person_id: str, short: bool = False) -> str:
+    """获取对用户的印象
+
+    Args:
+        person_id: 用户的唯一标识ID
+        short: 是否获取简短版印象，默认为False
+
+    Returns:
+        一段描述性的文本。
+    """
+    if not person_id:
+        return "用户ID为空，无法获取印象。"
+    try:
+        person_info_manager = get_person_info_manager()
+        field = "short_impression" if short else "impression"
+        impression = await person_info_manager.get_value(person_id, field)
+        return impression or "还没有形成对该用户的印象。"
+    except Exception as e:
+        logger.error(f"[PersonAPI] 获取用户印象失败: person_id={person_id}, error={e}")
+        return "获取用户印象时发生错误。"
+
+
+async def get_person_points(person_id: str, limit: int = 5) -> list[tuple]:
+    """获取关于用户的'记忆点'
+
+    Args:
+        person_id: 用户的唯一标识ID
+        limit: 返回的记忆点数量上限，默认为5
+
+    Returns:
+        一个列表，每个元素是一个包含记忆点内容、权重和时间的元组。
+    """
+    if not person_id:
+        return []
+    try:
+        person_info_manager = get_person_info_manager()
+        points = await person_info_manager.get_value(person_id, "points")
+        if not points:
+            return []
+
+        # 按权重和时间排序，返回最重要的几个点
+        sorted_points = sorted(points, key=lambda x: (x[1], x[2]), reverse=True)
+        return sorted_points[:limit]
+    except Exception as e:
+        logger.error(f"[PersonAPI] 获取用户记忆点失败: person_id={person_id}, error={e}")
+        return []
+
+
+# =============================================================================
+# 关系查询API
+# =============================================================================
+
+
+async def get_user_relationship_score(user_id: str) -> float:
+    """
+    获取用户关系分
+
+    Args:
+        user_id: 用户ID
+
+    Returns:
+        关系分 (0.0 - 1.0)
+    """
+    return await relationship_service.get_user_relationship_score(user_id)
+
+
+async def get_user_relationship_data(user_id: str) -> dict:
+    """
+    获取用户完整关系数据
+
+    Args:
+        user_id: 用户ID
+
+    Returns:
+        包含关系分、关系文本等的字典
+    """
+    return await relationship_service.get_user_relationship_data(user_id)
+
+
+async def update_user_relationship(user_id: str, relationship_score: float, relationship_text: str | None = None, user_name: str | None = None):
+    """
+    更新用户关系数据
+
+    Args:
+        user_id: 用户ID
+        relationship_score: 关系分 (0.0 - 1.0)
+        relationship_text: 关系描述文本
+        user_name: 用户名称
+    """
+    await relationship_service.update_user_relationship(user_id, relationship_score, relationship_text, user_name)
+
+
+# =============================================================================
+# 兴趣系统API
+# =============================================================================
+
+
+async def initialize_smart_interests(personality_description: str, personality_id: str = "default"):
+    """
+    初始化智能兴趣系统
+
+    Args:
+        personality_description: 机器人性格描述
+        personality_id: 性格ID
+    """
+    await interest_service.initialize_smart_interests(personality_description, personality_id)
+
+
+async def calculate_interest_match(content: str, keywords: list[str] | None = None):
+    """
+    计算内容与兴趣的匹配度
+
+    Args:
+        content: 消息内容
+        keywords: 关键词列表
+
+    Returns:
+        匹配结果
+    """
+    return await interest_service.calculate_interest_match(content, keywords)
+
+
+# =============================================================================
+# 系统状态与缓存API
+# =============================================================================
+
+
+def get_system_stats() -> dict[str, Any]:
+    """
+    获取系统统计信息
+
+    Returns:
+        包含各子系统统计的字典
+    """
+    return {
+        "relationship_service": relationship_service.get_cache_stats(),
+        "interest_service": interest_service.get_interest_stats()
+    }
+
+
+def clear_caches(user_id: str | None = None):
+    """
+    清理缓存
+
+    Args:
+        user_id: 特定用户ID，如果为None则清理所有缓存
+    """
+    relationship_service.clear_cache(user_id)
+    logger.info(f"清理缓存: {user_id if user_id else '全部'}")
+
+
+# =============================================================================
+# 报告API
+# =============================================================================
+
+
+async def get_full_relationship_report(person_id: str) -> str:
+    """生成一份关于你和用户的完整'关系报告'
+
+    综合基础信息、印象、记忆点和关系分，提供一个全方位的关系概览。
+    """
+    if not person_id:
+        return "无法生成报告，因为用户ID为空。"
+
+    try:
+        person_info_manager = get_person_info_manager()
+        user_id = await person_info_manager.get_value(person_id, "user_id")
+
+        if not user_id:
+            return "无法生成报告，因为找不到对应的用户信息。"
+
+        # 异步获取所有需要的信息
+        info, impression, points, rel_data = await asyncio.gather(
+            get_person_info(person_id),
+            get_person_impression(person_id),
+            get_person_points(person_id, limit=3),
+            relationship_service.get_user_relationship_data(str(user_id)),
+        )
+
+        # 构建报告
+        report = f"--- 与 {info.get('person_name', '未知用户')} 的关系报告 ---\n"
+        report += f"昵称: {info.get('nickname', '未知')}\n"
+        report += f"关系分数: {rel_data.get('relationship_score', 0.0):.2f}/1.0\n"
+        report += f"关系描述: {rel_data.get('relationship_text', '暂无')}\n"
+        report += f"我对ta的印象: {impression}\n"
+
+        if points:
+            report += "最近的重要记忆点:\n"
+            for point in points:
+                report += f"  - {point[0]} (重要性: {point[1]})\n"
+
+        report += "----------------------------------------\n"
+        return report
+
+    except Exception as e:
+        logger.error(f"[PersonAPI] 生成关系报告失败: person_id={person_id}, error={e}")
+        return "生成关系报告时发生错误。"
--- a/src/plugin_system/apis/scoring_api.py
+++ b/src/plugin_system/apis/scoring_api.py
@@ -1,113 +0,0 @@
-"""
-统一评分系统API
-提供系统级的关系分和兴趣管理服务，供所有插件和主项目组件使用
-"""
-
-from typing import Any
-
-from src.common.logger import get_logger
-from src.plugin_system.services.interest_service import interest_service
-from src.plugin_system.services.relationship_service import relationship_service
-
-logger = get_logger("scoring_api")
-
-
-class ScoringAPI:
-    """
-    统一评分系统API - 系统级服务
-
-    提供关系分和兴趣管理的统一接口，替代原有的插件依赖方式。
-    所有插件和主项目组件都应该通过此API访问评分功能。
-    """
-
-    @staticmethod
-    async def get_user_relationship_score(user_id: str) -> float:
-        """
-        获取用户关系分
-
-        Args:
-            user_id: 用户ID
-
-        Returns:
-            关系分 (0.0 - 1.0)
-        """
-        return await relationship_service.get_user_relationship_score(user_id)
-
-    @staticmethod
-    async def get_user_relationship_data(user_id: str) -> dict:
-        """
-        获取用户完整关系数据
-
-        Args:
-            user_id: 用户ID
-
-        Returns:
-            包含关系分、关系文本等的字典
-        """
-        return await relationship_service.get_user_relationship_data(user_id)
-
-    @staticmethod
-    async def update_user_relationship(user_id: str, relationship_score: float, relationship_text: str | None = None, user_name: str | None = None):
-        """
-        更新用户关系数据
-
-        Args:
-            user_id: 用户ID
-            relationship_score: 关系分 (0.0 - 1.0)
-            relationship_text: 关系描述文本
-            user_name: 用户名称
-        """
-        await relationship_service.update_user_relationship(user_id, relationship_score, relationship_text, user_name)
-
-    @staticmethod
-    async def initialize_smart_interests(personality_description: str, personality_id: str = "default"):
-        """
-        初始化智能兴趣系统
-
-        Args:
-            personality_description: 机器人性格描述
-            personality_id: 性格ID
-        """
-        await interest_service.initialize_smart_interests(personality_description, personality_id)
-
-    @staticmethod
-    async def calculate_interest_match(content: str, keywords: list[str] | None = None):
-        """
-        计算内容与兴趣的匹配度
-
-        Args:
-            content: 消息内容
-            keywords: 关键词列表
-
-        Returns:
-            匹配结果
-        """
-        return await interest_service.calculate_interest_match(content, keywords)
-
-    @staticmethod
-    def get_system_stats() -> dict[str, Any]:
-        """
-        获取系统统计信息
-
-        Returns:
-            包含各子系统统计的字典
-        """
-        return {
-            "relationship_service": relationship_service.get_cache_stats(),
-            "interest_service": interest_service.get_interest_stats()
-        }
-
-    @staticmethod
-    def clear_caches(user_id: str | None = None):
-        """
-        清理缓存
-
-        Args:
-            user_id: 特定用户ID，如果为None则清理所有缓存
-        """
-        relationship_service.clear_cache(user_id)
-        logger.info(f"清理缓存: {user_id if user_id else '全部'}")
-
-
-# 创建全局API实例 - 系统级服务
-scoring_api = ScoringAPI()
--- a/src/plugins/built_in/affinity_flow_chatter/affinity_interest_calculator.py
+++ b/src/plugins/built_in/affinity_flow_chatter/affinity_interest_calculator.py
@@ -211,9 +211,9 @@ class AffinityInterestCalculator(BaseInterestCalculator):

        # 如果内存中没有，尝试从统一的评分API获取
        try:
-            from src.plugin_system.apis.scoring_api import scoring_api
+            from src.plugin_system.apis import person_api

-            relationship_data = await scoring_api.get_user_relationship_data(user_id)
+            relationship_data = await person_api.get_user_relationship_data(user_id)
            if relationship_data:
                relationship_score = relationship_data.get("relationship_score", global_config.affinity_flow.base_relationship_score)
                # 同时更新内存缓存