HFC基本重构框架和TODO

src/common/utils/utils_config.py

@@ -0,0 +1,133 @@
+from typing import Optional
+
+import time
+
+from src.common.logger import get_logger
+from src.config.config import global_config
+
+logger = get_logger("config_utils")
+
+
+class ExpressionConfigUtils:
+    @staticmethod
+    def get_expression_config_for_chat(session_id: Optional[str] = None) -> tuple[bool, bool, bool]:
+        # sourcery skip: use-next
+        """
+        根据聊天会话ID获取表达配置
+
+        Args:
+            session_id: 聊天会话ID，格式为哈希值
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 是否启用jargon学习)
+        """
+        if not global_config.expression.learning_list:
+            return True, True, True
+
+        if session_id:
+            for config_item in global_config.expression.learning_list:
+                if not config_item.platform and not config_item.item_id:
+                    continue  # 这是全局的
+                stream_id = ExpressionConfigUtils._get_stream_id(
+                    config_item.platform,
+                    str(config_item.item_id),
+                    (config_item.rule_type == "group"),
+                )
+                if stream_id is None:
+                    continue
+                if stream_id == session_id:
+                    continue
+                return config_item.use_expression, config_item.enable_learning, config_item.enable_jargon_learning
+        for config_item in global_config.expression.learning_list:
+            if not config_item.platform and not config_item.item_id:
+                return config_item.use_expression, config_item.enable_learning, config_item.enable_jargon_learning
+
+        return True, True, True
+
+    @staticmethod
+    def _get_stream_id(platform: str, id_str: str, is_group: bool = False) -> Optional[str]:
+        # sourcery skip: remove-unnecessary-cast
+        """
+        根据平台、ID字符串和是否为群聊生成聊天流ID
+
+        Args:
+            platform: 平台名称
+            id_str: 用户或群组的原始ID字符串
+            is_group: 是否为群聊
+
+        Returns:
+            str: 生成的聊天流ID（哈希值）
+        """
+        try:
+            from src.common.utils.utils_session import SessionUtils
+
+            if is_group:
+                return SessionUtils.calculate_session_id(platform, group_id=str(id_str))
+            else:
+                return SessionUtils.calculate_session_id(platform, user_id=str(id_str))
+        except Exception as e:
+            logger.error(f"生成聊天流ID失败: {e}")
+            return None
+
+
+class ChatConfigUtils:
+    @staticmethod
+    def get_talk_value(session_id: Optional[str]) -> float:
+        result = global_config.chat.talk_value or 0.0
+        if not global_config.chat.enable_talk_value_rules or not global_config.chat.talk_value_rules:
+            return result
+        local_time = time.localtime()
+        now_min = local_time.tm_hour * 60 + local_time.tm_min
+
+        # 优先匹配会话相关的规则
+        if session_id:
+            from src.common.utils.utils_session import SessionUtils
+
+            for rule in global_config.chat.talk_value_rules:
+                if not rule.platform and not rule.item_id:
+                    continue  # 一起留空表示全局
+                if rule.rule_type == "group":
+                    rule_session_id = SessionUtils.calculate_session_id(rule.platform, group_id=str(rule.item_id))
+                else:
+                    rule_session_id = SessionUtils.calculate_session_id(rule.platform, user_id=str(rule.item_id))
+                if rule_session_id != session_id:
+                    continue  # 不匹配的会话ID，跳过
+                parsed_range = ChatConfigUtils.parse_range(rule.time)
+                if not parsed_range:
+                    continue  # 无法解析的时间范围，跳过
+                start_min, end_min = parsed_range
+                in_range: bool = False
+                if start_min <= end_min:
+                    in_range = start_min <= now_min <= end_min
+                else:  # 跨天的时间范围
+                    in_range = now_min >= start_min or now_min <= end_min
+                if in_range:
+                    return rule.value or 0.0  # 如果规则生效但没有设置值，返回0.0
+
+        # 没有匹配到会话相关的规则，继续匹配全局规则
+        for rule in global_config.chat.talk_value_rules:
+            if rule.platform or rule.item_id:
+                continue  # 只匹配全局规则
+            parsed_range = ChatConfigUtils.parse_range(rule.time)
+            if not parsed_range:
+                continue  # 无法解析的时间范围，跳过
+            start_min, end_min = parsed_range
+            in_range: bool = False
+            if start_min <= end_min:
+                in_range = start_min <= now_min <= end_min
+            else:  # 跨天的时间范围
+                in_range = now_min >= start_min or now_min <= end_min
+            if in_range:
+                return rule.value or 0.0  # 如果规则生效但没有设置值，返回0.0
+        return result  # 如果没有任何规则生效，返回默认值
+
+    @staticmethod
+    def parse_range(range_str: str) -> Optional[tuple[int, int]]:
+        """解析 "HH:MM-HH:MM" 到 (start_min, end_min)。"""
+        try:
+            start_str, end_str = [s.strip() for s in range_str.split("-")]
+            sh, sm = [int(x) for x in start_str.split(":")]
+            eh, em = [int(x) for x in end_str.split(":")]
+            return sh * 60 + sm, eh * 60 + em
+        except Exception:
+            return None