Add time and stream-based config for chat and expression

Introduces methods to dynamically determine talk frequency and expression learning settings based on current time and chat stream ID. Adds helper functions for parsing stream config strings, handling both global and stream-specific overrides for chat and expression configurations.

src/config/official_configs.py

@@ -83,59 +83,163 @@ class ChatConfig(ValidatedConfigBase):
     proactive_thinking_interval: int = Field(default=1500, description="主动思考间隔")
     proactive_thinking_prompt_template: str = Field(default="", description="主动思考提示模板")
 
-    def get_current_talk_frequency(self, stream_id: str) -> float:
+    def get_current_talk_frequency(self, chat_stream_id: Optional[str] = None) -> float:
         """
-        根据时间和聊天流ID获取当前的聊天频率
+        根据当前时间和聊天流获取对应的 talk_frequency
 
         Args:
-            stream_id: 聊天流ID
+            chat_stream_id: 聊天流ID，格式为 "platform:chat_id:type"
 
         Returns:
-            float: 当前聊天频率
+            float: 对应的频率值
+        """
+        if not self.talk_frequency_adjust:
+            return self.talk_frequency
+
+        # 优先检查聊天流特定的配置
+        if chat_stream_id:
+            stream_frequency = self._get_stream_specific_frequency(chat_stream_id)
+            if stream_frequency is not None:
+                return stream_frequency
+
+        # 检查全局时段配置（第一个元素为空字符串的配置）
+        global_frequency = self._get_global_frequency()
+        if global_frequency is not None:
+            return global_frequency
+
+        # 如果都没有匹配，返回默认值
+        return self.talk_frequency
+
+    def _get_time_based_frequency(self, time_freq_list: list[str]) -> Optional[float]:
+        """
+        根据时间配置列表获取当前时段的频率
+
+        Args:
+            time_freq_list: 时间频率配置列表，格式为 ["HH:MM,frequency", ...]
+
+        Returns:
+            float: 频率值，如果没有配置则返回 None
         """
-        import time
         from datetime import datetime
 
-        # 获取当前时间
+        current_time = datetime.now().strftime("%H:%M")
-        current_time = datetime.now()
+        current_hour, current_minute = map(int, current_time.split(":"))
-        current_hour = current_time.hour
+        current_minutes = current_hour * 60 + current_minute
-        current_minute = current_time.minute
-        current_time_str = f"{current_hour:02d}:{current_minute:02d}"
 
-        # 查找匹配的聊天频率调整配置
+        # 解析时间频率配置
-        for config_entry in self.talk_frequency_adjust:
+        time_freq_pairs = []
-            if not config_entry:
+        for time_freq_str in time_freq_list:
+            try:
+                time_str, freq_str = time_freq_str.split(",")
+                hour, minute = map(int, time_str.split(":"))
+                frequency = float(freq_str)
+                minutes = hour * 60 + minute
+                time_freq_pairs.append((minutes, frequency))
+            except (ValueError, IndexError):
                 continue
 
-            # 第一个元素是聊天流匹配模式
+        if not time_freq_pairs:
-            stream_pattern = config_entry[0]
+            return None
 
-            # 检查是否匹配当前聊天流
+        # 按时间排序
-            if stream_pattern == "" or stream_pattern in stream_id:
+        time_freq_pairs.sort(key=lambda x: x[0])
-                # 查找当前时间对应的频率
-                current_frequency = self.talk_frequency
 
-                # 遍历时间-频率对
+        # 查找当前时间对应的频率
-                for time_freq_pair in config_entry[1:]:
+        current_frequency = None
-                    if "," not in time_freq_pair:
+        for minutes, frequency in time_freq_pairs:
-                        continue
+            if current_minutes >= minutes:
+                current_frequency = frequency
+            else:
+                break
 
-                    time_part, freq_part = time_freq_pair.split(",", 1)
+        # 如果当前时间在所有配置时间之前，使用最后一个时间段的频率（跨天逻辑）
-                    try:
+        if current_frequency is None and time_freq_pairs:
-                        config_hour, config_minute = map(int, time_part.split(":"))
+            current_frequency = time_freq_pairs[-1][1]
-                        config_time_minutes = config_hour * 60 + config_minute
-                        current_time_minutes = current_hour * 60 + current_minute
 
-                        # 如果当前时间大于等于配置时间，更新频率
+        return current_frequency
-                        if current_time_minutes >= config_time_minutes:
-                            current_frequency = float(freq_part)
-                    except (ValueError, IndexError):
-                        continue
 
-                return current_frequency
+    def _get_stream_specific_frequency(self, chat_stream_id: str):
+        """
+        获取特定聊天流在当前时间的频率
 
-        # 如果没有找到匹配的配置，返回默认频率
+        Args:
-        return self.talk_frequency
+            chat_stream_id: 聊天流ID（哈希值）
+
+        Returns:
+            float: 频率值，如果没有配置则返回 None
+        """
+        # 查找匹配的聊天流配置
+        for config_item in self.talk_frequency_adjust:
+            if not config_item or len(config_item) < 2:
+                continue
+
+            stream_config_str = config_item[0]  # 例如 "qq:1026294844:group"
+
+            # 解析配置字符串并生成对应的 chat_id
+            config_chat_id = self._parse_stream_config_to_chat_id(stream_config_str)
+            if config_chat_id is None:
+                continue
+
+            # 比较生成的 chat_id
+            if config_chat_id != chat_stream_id:
+                continue
+
+            # 使用通用的时间频率解析方法
+            return self._get_time_based_frequency(config_item[1:])
+
+        return None
+
+    def _parse_stream_config_to_chat_id(self, stream_config_str: str) -> Optional[str]:
+        """
+        解析流配置字符串并生成对应的 chat_id
+
+        Args:
+            stream_config_str: 格式为 "platform:id:type" 的字符串
+
+        Returns:
+            str: 生成的 chat_id，如果解析失败则返回 None
+        """
+        try:
+            parts = stream_config_str.split(":")
+            if len(parts) != 3:
+                return None
+
+            platform = parts[0]
+            id_str = parts[1]
+            stream_type = parts[2]
+
+            # 判断是否为群聊
+            is_group = stream_type == "group"
+
+            # 使用与 ChatStream.get_stream_id 相同的逻辑生成 chat_id
+            import hashlib
+
+            if is_group:
+                components = [platform, str(id_str)]
+            else:
+                components = [platform, str(id_str), "private"]
+            key = "_".join(components)
+            return hashlib.md5(key.encode()).hexdigest()
+
+        except (ValueError, IndexError):
+            return None
+
+    def _get_global_frequency(self) -> Optional[float]:
+        """
+        获取全局默认频率配置
+
+        Returns:
+            float: 频率值，如果没有配置则返回 None
+        """
+        for config_item in self.talk_frequency_adjust:
+            if not config_item or len(config_item) < 2:
+                continue
+
+            # 检查是否为全局默认配置（第一个元素为空字符串）
+            if config_item[0] == "":
+                return self._get_time_based_frequency(config_item[1:])
+
+        return None
 
 
 
@@ -160,49 +264,131 @@ class ExpressionConfig(ValidatedConfigBase):
     expression_learning: list[list] = Field(default_factory=lambda: [], description="表达学习")
     expression_groups: list[list[str]] = Field(default_factory=list, description="表达组")
 
-    def get_expression_config_for_chat(self, chat_id: str) -> tuple[bool, bool, float]:
+    def _parse_stream_config_to_chat_id(self, stream_config_str: str) -> Optional[str]:
         """
-        获取指定聊天流的表达配置
+        解析流配置字符串并生成对应的 chat_id
 
         Args:
-            chat_id: 聊天流ID
+            stream_config_str: 格式为 "platform:id:type" 的字符串
 
         Returns:
-            tuple[bool, bool, float]: (use_expression, enable_learning, learning_intensity)
+            str: 生成的 chat_id，如果解析失败则返回 None
         """
-        # 默认值
+        try:
-        use_expression = False
+            parts = stream_config_str.split(":")
-        enable_learning = False
+            if len(parts) != 3:
-        learning_intensity = 1.0
+                return None
 
-        # 查找匹配的表达学习配置
+            platform = parts[0]
-        for config_entry in self.expression_learning:
+            id_str = parts[1]
-            if not config_entry or len(config_entry) < 4:
+            stream_type = parts[2]
+
+            # 判断是否为群聊
+            is_group = stream_type == "group"
+
+            # 使用与 ChatStream.get_stream_id 相同的逻辑生成 chat_id
+            import hashlib
+
+            if is_group:
+                components = [platform, str(id_str)]
+            else:
+                components = [platform, str(id_str), "private"]
+            key = "_".join(components)
+            return hashlib.md5(key.encode()).hexdigest()
+
+        except (ValueError, IndexError):
+            return None
+
+    def get_expression_config_for_chat(self, chat_stream_id: Optional[str] = None) -> tuple[bool, bool, float]:
+        """
+        根据聊天流ID获取表达配置
+
+        Args:
+            chat_stream_id: 聊天流ID，格式为哈希值
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 学习间隔)
+        """
+        if not self.expression_learning:
+            # 如果没有配置，使用默认值：启用表达，启用学习，300秒间隔
+            return True, True, 300
+
+        # 优先检查聊天流特定的配置
+        if chat_stream_id:
+            specific_config = self._get_stream_specific_config(chat_stream_id)
+            if specific_config is not None:
+                return specific_config
+
+        # 检查全局配置（第一个元素为空字符串的配置）
+        global_config = self._get_global_config()
+        if global_config is not None:
+            return global_config
+
+        # 如果都没有匹配，返回默认值
+        return True, True, 300
+
+    def _get_stream_specific_config(self, chat_stream_id: str) -> Optional[tuple[bool, bool, float]]:
+        """
+        获取特定聊天流的表达配置
+
+        Args:
+            chat_stream_id: 聊天流ID（哈希值）
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 学习间隔)，如果没有配置则返回 None
+        """
+        for config_item in self.expression_learning:
+            if not config_item or len(config_item) < 4:
                 continue
 
-            # 配置格式: [chat_pattern, use_expression, enable_learning, learning_intensity]
+            stream_config_str = config_item[0]  # 例如 "qq:1026294844:group"
-            chat_pattern = config_entry[0]
 
-            # 检查是否匹配当前聊天流
+            # 如果是空字符串，跳过（这是全局配置）
-            if chat_pattern == "" or chat_pattern in chat_id:
+            if stream_config_str == "":
+                continue
+
+            # 解析配置字符串并生成对应的 chat_id
+            config_chat_id = self._parse_stream_config_to_chat_id(stream_config_str)
+            if config_chat_id is None:
+                continue
+
+            # 比较生成的 chat_id
+            if config_chat_id != chat_stream_id:
+                continue
+
+            # 解析配置
+            try:
+                use_expression = config_item[1].lower() == "enable"
+                enable_learning = config_item[2].lower() == "enable"
+                learning_intensity = float(config_item[3])
+                return use_expression, enable_learning, learning_intensity
+            except (ValueError, IndexError):
+                continue
+
+        return None
+
+    def _get_global_config(self) -> Optional[tuple[bool, bool, float]]:
+        """
+        获取全局表达配置
+
+        Returns:
+            tuple: (是否使用表达, 是否学习表达, 学习间隔)，如果没有配置则返回 None
+        """
+        for config_item in self.expression_learning:
+            if not config_item or len(config_item) < 4:
+                continue
+
+            # 检查是否为全局配置（第一个元素为空字符串）
+            if config_item[0] == "":
                 try:
-                    # 解析配置值
+                    use_expression = config_item[1].lower() == "enable"
-                    use_expr_str = config_entry[1].lower() if isinstance(config_entry[1], str) else str(config_entry[1])
+                    enable_learning = config_item[2].lower() == "enable"
-                    enable_learn_str = config_entry[2].lower() if isinstance(config_entry[2], str) else str(config_entry[2])
+                    learning_intensity = float(config_item[3])
-                    
-                    use_expression = use_expr_str in ['enable', 'true', '1']
-                    enable_learning = enable_learn_str in ['enable', 'true', '1']
-                    learning_intensity = float(config_entry[3])
-                    
-                    # 找到匹配的配置后返回
                     return use_expression, enable_learning, learning_intensity
-                    
+                except (ValueError, IndexError):
-                except (ValueError, IndexError, TypeError):
-                    # 如果解析失败，继续查找下一个配置
                     continue
 
-        # 如果没有找到匹配的配置，返回默认值
+        return None
-        return use_expression, enable_learning, learning_intensity