[fix] (llm_analyzer) LLM 输出 json 提取增强Merge pull request #21 from SXP-Simon/fix/json

Merge pull request #21 from SXP-Simon/fix/json

.gitignore

@@ -0,0 +1,5 @@
+
+.kilocode/
+.kiro/
+.vscode/
+src/analysis/ARCHITECTURE.md

README.md

@@ -3,7 +3,7 @@
 # QQ群日常分析插件
 
 
-[![Plugin Version](https://img.shields.io/badge/Latest_Version-v2.6.1-blue.svg?style=for-the-badge&color=76bad9)](https://github.com/SXP-Simon/astrbot-qq-group-daily-analysis)
+[![Plugin Version](https://img.shields.io/badge/Latest_Version-v2.7.0-blue.svg?style=for-the-badge&color=76bad9)](https://github.com/SXP-Simon/astrbot-qq-group-daily-analysis)
 [![AstrBot](https://img.shields.io/badge/AstrBot-Plugin-ff69b4?style=for-the-badge)](https://github.com/AstrBotDevs/AstrBot)
 [![License](https://img.shields.io/badge/License-MIT-green.svg?style=for-the-badge)](LICENSE)
 
@@ -131,6 +131,12 @@ _✨ 一个基于AstrBot的智能群聊分析插件，能够生成精美的群
 - 处理了自动分析器的不唯一问题
 - 自动分析器并发处理群聊
 
+### v2.6.0
+- (自动分析处理) 纠正分析日期处理情况
+
+### v2.7.0
+- (llm_analyzer) LLM 输出 json 提取增强
+
 ## 许可证
 
 MIT License

main.py

@@ -203,11 +203,11 @@ class QQGroupDailyAnalysis(Star):
                     yield result
                 else:
                     # 如果 PDF 生成失败，提供详细的错误信息和解决方案
-                    yield event.plain_result("❌ PDF 报告生成失败")
-                    yield event.plain_result("🔧 可能的解决方案：")
-                    yield event.plain_result("1. 使用 /安装PDF 命令重新安装依赖")
-                    yield event.plain_result("2. 检查网络连接是否正常")
-                    yield event.plain_result("3. 暂时使用图片格式：/设置格式 image")
+                    # yield event.plain_result("❌ PDF 报告生成失败")
+                    # yield event.plain_result("🔧 可能的解决方案：")
+                    # yield event.plain_result("1. 使用 /安装PDF 命令重新安装依赖")
+                    # yield event.plain_result("2. 检查网络连接是否正常")
+                    # yield event.plain_result("3. 暂时使用图片格式：/设置格式 image")
 
                     # 回退到文本报告
                     logger.warning("PDF 报告生成失败，回退到文本报告")
@@ -278,9 +278,13 @@ class QQGroupDailyAnalysis(Star):
         yield event.plain_result("🔄 开始安装 PDF 功能依赖，请稍候...")
 
         try:
-            # 使用模块化的PDF安装器
+            # 安装 pyppeteer
             result = await PDFInstaller.install_pyppeteer(config_manager)
             yield event.plain_result(result)
+            
+            # 提供系统依赖安装指导
+            system_deps_result = await PDFInstaller.install_system_deps()
+            yield event.plain_result(system_deps_result)
 
         except Exception as e:
             logger.error(f"安装 PDF 依赖失败: {e}", exc_info=True)

metadata.yaml

@@ -13,6 +13,6 @@ help: |  # 插件的帮助信息
   命令：
   /群分析 [天数] - 分析群聊活动
   /分析设置 [操作] - 管理设置（enable/disable/status/test）
-version: v2.6.1 # 插件版本号。格式：v1.1.1 或者 v1.1
+version: v2.7.0 # 插件版本号。格式：v1.1.1 或者 v1.1
 author: SXP-Simon # 作者
 repo: https://github.com/SXP-Simon/astrbot-qq-group-daily-analysis # 插件的仓库地址

src/analysis/analyzers/__init__.py

@@ -0,0 +1,16 @@
+"""
+分析器模块
+包含各种LLM分析功能的实现
+"""
+
+from .base_analyzer import BaseAnalyzer
+from .topic_analyzer import TopicAnalyzer
+from .user_title_analyzer import UserTitleAnalyzer
+from .golden_quote_analyzer import GoldenQuoteAnalyzer
+
+__all__ = [
+    'BaseAnalyzer',
+    'TopicAnalyzer',
+    'UserTitleAnalyzer',
+    'GoldenQuoteAnalyzer'
+]

src/analysis/analyzers/base_analyzer.py

@@ -0,0 +1,187 @@
+"""
+基础分析器抽象类
+定义通用分析流程和接口
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Dict, Tuple, Any, Optional
+from datetime import datetime
+from astrbot.api import logger
+from ...models.data_models import TokenUsage
+from ..utils.json_utils import parse_json_response
+from ..utils.llm_utils import call_provider_with_retry, extract_token_usage, extract_response_text
+import re
+
+class BaseAnalyzer(ABC):
+    """
+    基础分析器抽象类
+    定义所有分析器的通用接口和流程
+    """
+    
+    def __init__(self, context, config_manager):
+        """
+        初始化基础分析器
+        
+        Args:
+            context: AstrBot上下文对象
+            config_manager: 配置管理器
+        """
+        self.context = context
+        self.config_manager = config_manager
+    
+    @abstractmethod
+    def get_data_type(self) -> str:
+        """
+        获取数据类型标识
+        
+        Returns:
+            数据类型字符串
+        """
+        pass
+    
+    @abstractmethod
+    def get_max_count(self) -> int:
+        """
+        获取最大提取数量
+        
+        Returns:
+            最大数量
+        """
+        pass
+    
+    @abstractmethod
+    def build_prompt(self, data: Any) -> str:
+        """
+        构建LLM提示词
+        
+        Args:
+            data: 输入数据
+            
+        Returns:
+            提示词字符串
+        """
+        pass
+    
+    @abstractmethod
+    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+        """
+        使用正则表达式提取数据
+        
+        Args:
+            result_text: LLM响应文本
+            max_count: 最大提取数量
+            
+        Returns:
+            提取到的数据列表
+        """
+        pass
+    
+    @abstractmethod
+    def create_data_objects(self, data_list: List[Dict]) -> List[Any]:
+        """
+        创建数据对象列表
+        
+        Args:
+            data_list: 原始数据列表
+            
+        Returns:
+            数据对象列表
+        """
+        pass
+    
+    async def analyze(self, data: Any, umo: str = None) -> Tuple[List[Any], TokenUsage]:
+        """
+        统一的分析流程
+        
+        Args:
+            data: 输入数据
+            umo: 模型唯一标识符
+            
+        Returns:
+            (分析结果列表, Token使用统计)
+        """
+        try:
+            # 1. 构建提示词
+            logger.debug(f"{self.get_data_type()}分析开始构建prompt，输入数据类型: {type(data)}")
+            logger.debug(f"{self.get_data_type()}分析输入数据长度: {len(data) if hasattr(data, '__len__') else 'N/A'}")
+            
+            prompt = self.build_prompt(data)
+            logger.info(f"开始{self.get_data_type()}分析，构建提示词完成")
+            logger.debug(f"{self.get_data_type()}分析prompt长度: {len(prompt) if prompt else 0}")
+            logger.debug(f"{self.get_data_type()}分析prompt前100字符: {prompt[:100] if prompt else 'None'}...")
+            
+            # 检查 prompt 是否为空
+            if not prompt or not prompt.strip():
+                logger.warning(f"{self.get_data_type()}分析: prompt 为空或只包含空白字符，跳过LLM调用")
+                return [], TokenUsage()
+            
+            # 2. 调用LLM
+            max_tokens = self.get_max_tokens()
+            temperature = self.get_temperature()
+            
+            response = await call_provider_with_retry(
+                self.context, self.config_manager, prompt,
+                max_tokens, temperature, umo
+            )
+            
+            if response is None:
+                logger.error(f"{self.get_data_type()}分析调用LLM失败: provider返回None（重试失败）")
+                return [], TokenUsage()
+            
+            # 3. 提取token使用统计
+            token_usage_dict = extract_token_usage(response)
+            token_usage = TokenUsage(
+                prompt_tokens=token_usage_dict["prompt_tokens"],
+                completion_tokens=token_usage_dict["completion_tokens"],
+                total_tokens=token_usage_dict["total_tokens"]
+            )
+            
+            # 4. 提取响应文本
+            result_text = extract_response_text(response)
+            logger.debug(f"{self.get_data_type()}分析原始响应: {result_text[:500]}...")
+            
+            # 5. 尝试JSON解析
+            success, parsed_data, error_msg = parse_json_response(result_text, self.get_data_type())
+            
+            if success and parsed_data:
+                # JSON解析成功，创建数据对象
+                data_objects = self.create_data_objects(parsed_data)
+                logger.info(f"{self.get_data_type()}分析成功，解析到 {len(data_objects)} 条数据")
+                return data_objects, token_usage
+            
+            # 6. JSON解析失败，使用正则表达式降级
+            logger.warning(f"{self.get_data_type()}JSON解析失败，尝试正则表达式提取: {error_msg}")
+            regex_data = self.extract_with_regex(result_text, self.get_max_count())
+            
+            if regex_data:
+                logger.info(f"{self.get_data_type()}正则表达式提取成功，获得 {len(regex_data)} 条数据")
+                data_objects = self.create_data_objects(regex_data)
+                return data_objects, token_usage
+            else:
+                # 最后的降级方案 - 两种方法都失败
+                logger.error(f"{self.get_data_type()}分析失败: JSON解析和正则表达式提取均未成功，返回空列表")
+                return [], token_usage
+                
+        except Exception as e:
+            logger.error(f"{self.get_data_type()}分析失败: {e}", exc_info=True)
+            return [], TokenUsage()
+    
+    def get_max_tokens(self) -> int:
+        """
+        获取最大token数，子类可重写
+        
+        Returns:
+            最大token数
+        """
+        return 10000
+    
+    def get_temperature(self) -> float:
+        """
+        获取温度参数，子类可重写
+        
+        Returns:
+            温度参数
+        """
+        return 0.6
+    
+    

src/analysis/analyzers/golden_quote_analyzer.py

@@ -0,0 +1,194 @@
+"""
+金句分析模块
+专门处理群聊金句提取和分析
+"""
+
+from typing import List, Dict, Tuple
+from datetime import datetime
+from astrbot.api import logger
+from ...models.data_models import GoldenQuote, TokenUsage
+from .base_analyzer import BaseAnalyzer
+from ..utils.json_utils import extract_golden_quotes_with_regex
+
+
+
+class GoldenQuoteAnalyzer(BaseAnalyzer):
+    """
+    金句分析器
+    专门处理群聊金句的提取和分析
+    """
+    
+    def get_data_type(self) -> str:
+        """获取数据类型标识"""
+        return "金句"
+    
+    def get_max_count(self) -> int:
+        """获取最大金句数量"""
+        return self.config_manager.get_max_golden_quotes()
+    
+    def get_max_tokens(self) -> int:
+        """获取最大token数"""
+        return 1500
+    
+    def get_temperature(self) -> float:
+        """获取温度参数"""
+        return 0.7
+    
+    def build_prompt(self, messages: List[Dict]) -> str:
+        """
+        构建金句分析提示词
+        
+        Args:
+            messages: 群聊的文本消息列表
+            
+        Returns:
+            提示词字符串
+        """
+        if not messages:
+            return ""
+        
+        # 构建消息文本
+        messages_text = "\n".join([
+            f"[{msg['time']}] {msg['sender']}: {msg['content']}"
+            for msg in messages
+        ])
+        
+        max_golden_quotes = self.get_max_count()
+        prompt = f"""
+请从以下群聊记录中挑选出{max_golden_quotes}句最具冲击力、最令人惊叹的"金句"。这些金句需满足：
+- 核心标准：**逆天的神人发言**，即具备颠覆常识的脑洞、逻辑跳脱的表达或强烈反差感的原创内容
+- 典型特征：包含某些争议话题元素、夸张类比、反常规结论、一本正经的"胡说八道"或突破语境的清奇思路，并且具备一定的冲击力，让人印象深刻。
+
+对于每个金句，请提供：
+1. 原文内容（完整保留发言细节）
+2. 发言人昵称
+3. 选择理由（具体说明其"逆天"之处，如逻辑颠覆点/脑洞角度/反差感/争议话题元素）
+
+此外，我将对你进行严格约束：
+- 优先筛选 **逆天指数最高** 的内容：发情、性压抑话题 > 争议话题 > 元素级 > 颠覆认知级 > 逻辑跳脱级 > 趣味调侃级，剔除单纯玩梗或网络热词堆砌的普通发言
+- 重点标记包含极端类比、反常识论证或无厘头结论的内容，并且包含一定的争议话题元素。
+
+群聊记录：
+{messages_text}
+
+请以JSON格式返回，格式如下：
+[
+  {{
+    "content": "金句原文",
+    "sender": "发言人昵称",
+    "reason": "选择这句话的理由（需明确说明逆天特质）"
+  }}
+]
+"""
+        return prompt
+    
+    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+        """
+        使用正则表达式提取金句信息
+        
+        Args:
+            result_text: LLM响应文本
+            max_count: 最大提取数量
+            
+        Returns:
+            金句数据列表
+        """
+        return extract_golden_quotes_with_regex(result_text, max_count)
+    
+    def create_data_objects(self, quotes_data: List[Dict]) -> List[GoldenQuote]:
+        """
+        创建金句对象列表
+        
+        Args:
+            quotes_data: 原始金句数据列表
+            
+        Returns:
+            GoldenQuote对象列表
+        """
+        try:
+            quotes = []
+            max_quotes = self.get_max_count()
+            
+            for quote_data in quotes_data[:max_quotes]:
+                # 确保数据格式正确
+                content = quote_data.get("content", "").strip()
+                sender = quote_data.get("sender", "").strip()
+                reason = quote_data.get("reason", "").strip()
+                
+                # 验证必要字段
+                if not content or not sender or not reason:
+                    logger.warning(f"金句数据格式不完整，跳过: {quote_data}")
+                    continue
+                
+                quotes.append(GoldenQuote(
+                    content=content,
+                    sender=sender,
+                    reason=reason
+                ))
+            
+            return quotes
+            
+        except Exception as e:
+            logger.error(f"创建金句对象失败: {e}")
+            return []
+    
+    def extract_interesting_messages(self, messages: List[Dict]) -> List[Dict]:
+        """
+        提取圣经的文本消息
+        
+        Args:
+            messages: 群聊消息列表
+            
+        Returns:
+            圣经的文本消息列表
+        """
+        try:
+            interesting_messages = []
+            
+            for msg in messages:
+                sender = msg.get("sender", {})
+                nickname = sender.get("nickname", "") or sender.get("card", "")
+                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
+                
+                for content in msg.get("message", []):
+                    if content.get("type") == "text":
+                        text = content.get("data", {}).get("text", "").strip()
+                        # 过滤长度适中、可能圣经的消息
+                        if 5 <= len(text) <= 100 and not text.startswith(("http", "www", "/")):
+                            interesting_messages.append({
+                                "sender": nickname,
+                                "time": msg_time,
+                                "content": text
+                            })
+            
+            return interesting_messages
+            
+        except Exception as e:
+            logger.error(f"提取圣经消息失败: {e}")
+            return []
+    
+    async def analyze_golden_quotes(self, messages: List[Dict], umo: str = None) -> Tuple[List[GoldenQuote], TokenUsage]:
+        """
+        分析群聊金句
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (金句列表, Token使用统计)
+        """
+        try:
+            # 提取圣经的文本消息
+            interesting_messages = self.extract_interesting_messages(messages)
+            
+            if not interesting_messages:
+                logger.info("没有符合条件的圣经消息，返回空结果")
+                return [], TokenUsage()
+            
+            logger.info(f"开始从 {len(interesting_messages)} 条圣经消息中提取金句")
+            return await self.analyze(interesting_messages, umo)
+            
+        except Exception as e:
+            logger.error(f"金句分析失败: {e}")
+            return [], TokenUsage()

src/analysis/analyzers/topic_analyzer.py

@@ -0,0 +1,378 @@
+"""
+话题分析模块
+专门处理群聊话题分析
+"""
+
+from typing import List, Dict, Tuple
+from datetime import datetime
+import re
+from astrbot.api import logger
+from ...models.data_models import SummaryTopic, TokenUsage
+from .base_analyzer import BaseAnalyzer
+from ..utils.json_utils import extract_topics_with_regex
+
+
+class TopicAnalyzer(BaseAnalyzer):
+    """
+    话题分析器
+    专门处理群聊话题的提取和分析
+    """
+    
+    def get_data_type(self) -> str:
+        """获取数据类型标识"""
+        return "话题"
+    
+    def get_max_count(self) -> int:
+        """获取最大话题数量"""
+        return self.config_manager.get_max_topics()
+    
+    def get_max_tokens(self) -> int:
+        """获取最大token数"""
+        return 10000
+    
+    def get_temperature(self) -> float:
+        """获取温度参数"""
+        return 0.6
+    
+    def build_prompt(self, messages: List[Dict]) -> str:
+        """
+        构建话题分析提示词
+        
+        Args:
+            messages: 群聊消息列表
+            
+        Returns:
+            提示词字符串
+        """
+        logger.debug(f"build_prompt 开始处理，输入消息数量: {len(messages) if messages else 0}")
+        logger.debug(f"输入消息类型: {type(messages)}")
+        
+        # 验证输入数据格式
+        if not isinstance(messages, list):
+            logger.error(f"build_prompt 期望列表，但收到: {type(messages)}")
+            return ""
+        
+        # 检查消息列表是否为空
+        if not messages:
+            logger.warning("build_prompt 收到空消息列表")
+            return ""
+        
+        logger.debug(f"build_prompt 第一条消息内容: {messages[0] if messages else '无'}")
+        
+        # 提取文本消息
+        text_messages = []
+        for i, msg in enumerate(messages):
+            logger.debug(f"build_prompt 处理第 {i+1} 条消息，类型: {type(msg)}")
+            
+            # 确保msg是字典类型，避免'str' object has no attribute 'get'错误
+            if not isinstance(msg, dict):
+                logger.warning(f"build_prompt 跳过非字典类型的消息: {type(msg)} - {msg}")
+                continue
+                
+            try:
+                sender = msg.get("sender", {})
+                # 确保sender是字典类型，避免'str' object has no attribute 'get'错误
+                if not isinstance(sender, dict):
+                    logger.warning(f"build_prompt 跳过sender非字典类型的消息: {type(sender)} - {sender}")
+                    continue
+                    
+                nickname = sender.get("nickname", "") or sender.get("card", "")
+                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
+                
+                message_list = msg.get("message", [])
+                logger.debug(f"build_prompt 消息 {i+1} 的 message 字段类型: {type(message_list)}, 长度: {len(message_list) if hasattr(message_list, '__len__') else 'N/A'}")
+                
+                # 提取文本内容，可能分布在多个 content 中
+                text_parts = []
+                for j, content in enumerate(message_list):
+                    logger.debug(f"build_prompt 处理消息 {i+1} 的内容 {j+1}, 类型: {type(content)}")
+                    if not isinstance(content, dict):
+                        logger.warning(f"build_prompt 跳过非字典类型的内容: {type(content)} - {content}")
+                        continue
+                    
+                    content_type = content.get("type", "")
+                    logger.debug(f"build_prompt 内容类型: {content_type}")
+                    
+                    if content_type == "text":
+                        text = content.get("data", {}).get("text", "").strip()
+                        logger.debug(f"build_prompt 提取到的文本: '{text}' (长度: {len(text)})")
+                        if text:
+                            text_parts.append(text)
+                    elif content_type == "at":
+                        # 处理 @ 消息，转换为文本
+                        at_qq = content.get("data", {}).get("qq", "")
+                        if at_qq:
+                            at_text = f"@{at_qq}"
+                            text_parts.append(at_text)
+                            logger.debug(f"build_prompt 提取到@消息: {at_text}")
+                    elif content_type == "reply":
+                        # 处理回复消息，添加标记
+                        reply_id = content.get("data", {}).get("id", "")
+                        if reply_id:
+                            reply_text = f"[回复:{reply_id}]"
+                            text_parts.append(reply_text)
+                            logger.debug(f"build_prompt 提取到回复消息: {reply_text}")
+                
+                # 合并所有文本部分
+                combined_text = "".join(text_parts).strip()
+                logger.debug(f"build_prompt 合并后的文本: '{combined_text}' (长度: {len(combined_text)})")
+                
+                if combined_text and len(combined_text) > 2 and not combined_text.startswith("/"):
+                    # 清理消息内容
+                    cleaned_text = combined_text.replace('“', '"').replace('”', '"')
+                    cleaned_text = cleaned_text.replace('‘', "'").replace('’', "'")
+                    cleaned_text = cleaned_text.replace('\n', ' ').replace('\r', ' ')
+                    cleaned_text = cleaned_text.replace('\t', ' ')
+                    cleaned_text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', cleaned_text)
+                    
+                    logger.debug(f"build_prompt 清理后的文本: '{cleaned_text}'")
+                    
+                    text_messages.append({
+                        "sender": nickname,
+                        "time": msg_time,
+                        "content": cleaned_text
+                    })
+                else:
+                    logger.debug(f"build_prompt 跳过文本: '{combined_text}' (长度不足或以/开头)")
+            except Exception as e:
+                logger.error(f"build_prompt 处理第 {i+1} 条消息时出错: {e}", exc_info=True)
+                continue
+        
+        logger.debug(f"build_prompt 提取到 {len(text_messages)} 条文本消息")
+        
+        if not text_messages:
+            logger.warning("build_prompt 没有提取到有效的文本消息，返回空prompt")
+            return ""
+        
+        logger.debug(f"build_prompt 第一条文本消息: {text_messages[0] if text_messages else '无'}")
+        
+        # 构建消息文本
+        messages_text = "\n".join([
+            f"[{msg['time']}] {msg['sender']}: {msg['content']}"
+            for msg in text_messages
+        ])
+        
+        max_topics = self.get_max_count()
+        
+        logger.debug(f"build_prompt 准备构建prompt，max_topics={max_topics}")
+        logger.debug(f"build_prompt messages_text 长度: {len(messages_text)}")
+        
+        prompt = f"""
+你是一个帮我进行群聊信息总结的助手，生成总结内容时，你需要严格遵守下面的几个准则：
+请分析接下来提供的群聊记录，提取出最多{max_topics}个主要话题。
+
+对于每个话题，请提供：
+1. 话题名称（突出主题内容，尽量简明扼要）
+2. 主要参与者（最多5人）
+3. 话题详细描述（包含关键信息和结论）
+
+注意：
+- 对于比较有价值的点，稍微用一两句话详细讲讲，比如不要生成 "Nolan 和 SOV 讨论了 galgame 中关于性符号的衍生情况" 这种宽泛的内容，而是生成更加具体的讨论内容，让其他人只看这个消息就能知道讨论中有价值的，有营养的信息。
+- 对于其中的部分信息，你需要特意提到主题施加的主体是谁，是哪个群友做了什么事情，而不要直接生成和群友没有关系的语句。
+- 对于每一条总结，尽量讲清楚前因后果，以及话题的结论，是什么，为什么，怎么做，如果用户没有讲到细节，则可以不用这么做。
+
+群聊记录：
+{messages_text}
+
+重要：必须返回标准JSON格式，严格遵守以下规则：
+1. 只使用英文双引号 " 不要使用中文引号 " "
+2. 字符串内容中的引号必须转义为 \"
+3. 多个对象之间用逗号分隔
+4. 数组元素之间用逗号分隔
+5. 不要在JSON外添加任何文字说明
+6. 描述内容避免使用特殊符号，用普通文字表达
+
+请严格按照以下JSON格式返回，确保可以被标准JSON解析器解析：
+[
+  {{
+    "topic": "话题名称",
+    "contributors": ["用户1", "用户2"],
+    "detail": "话题描述内容"
+  }},
+  {{
+    "topic": "另一个话题",
+    "contributors": ["用户3", "用户4"],
+    "detail": "另一个话题的描述"
+  }}
+]
+
+注意：返回的内容必须是纯JSON，不要包含markdown代码块标记或其他格式
+"""
+        logger.debug(f"build_prompt 构建的prompt长度: {len(prompt)}")
+        logger.debug(f"build_prompt prompt前100字符: {prompt[:100]}...")
+        return prompt
+    
+    def extract_with_regex(self, result_text: str, max_topics: int) -> List[Dict]:
+        """
+        使用正则表达式提取话题信息
+        
+        Args:
+            result_text: LLM响应文本
+            max_topics: 最大话题数量
+            
+        Returns:
+            话题数据列表
+        """
+        return extract_topics_with_regex(result_text, max_topics)
+    
+    def create_data_objects(self, topics_data: List[Dict]) -> List[SummaryTopic]:
+        """
+        创建话题对象列表
+        
+        Args:
+            topics_data: 原始话题数据列表
+            
+        Returns:
+            SummaryTopic对象列表
+        """
+        logger.debug(f"create_data_objects 开始处理，输入数据数量: {len(topics_data) if topics_data else 0}")
+        logger.debug(f"输入数据类型: {type(topics_data)}")
+        
+        try:
+            topics = []
+            max_topics = self.get_max_count()
+            
+            logger.debug(f"处理前 {max_topics} 条话题数据")
+            
+            for i, topic_data in enumerate(topics_data[:max_topics]):
+                logger.debug(f"处理第 {i+1} 条话题数据，类型: {type(topic_data)}")
+                
+                # 确保topic_data是字典类型，避免'str' object has no attribute 'get'错误
+                if not isinstance(topic_data, dict):
+                    logger.warning(f"跳过非字典类型的话题数据: {type(topic_data)} - {topic_data}")
+                    continue
+                    
+                try:
+                    # 确保数据格式正确
+                    topic_name = topic_data.get("topic", "").strip()
+                    contributors = topic_data.get("contributors", [])
+                    detail = topic_data.get("detail", "").strip()
+                    
+                    logger.debug(f"话题数据 - 名称: {topic_name}, 参与者: {contributors}, 详情: {detail[:50]}...")
+                    
+                    # 验证必要字段
+                    if not topic_name or not detail:
+                        logger.warning(f"话题数据格式不完整，跳过: {topic_data}")
+                        continue
+                    
+                    # 确保参与者列表有效
+                    if not contributors or not isinstance(contributors, list):
+                        contributors = ["群友"]
+                    else:
+                        # 清理参与者名称
+                        contributors = [str(c).strip() for c in contributors if c and str(c).strip()] or ["群友"]
+                    
+                    topics.append(SummaryTopic(
+                        topic=topic_name,
+                        contributors=contributors[:5],  # 最多5个参与者
+                        detail=detail
+                    ))
+                except Exception as e:
+                    logger.error(f"处理第 {i+1} 条话题数据时出错: {e}", exc_info=True)
+                    continue
+            
+            logger.debug(f"create_data_objects 完成，创建了 {len(topics)} 个话题对象")
+            return topics
+            
+        except Exception as e:
+            logger.error(f"创建话题对象失败: {e}", exc_info=True)
+            return []
+    
+    def extract_text_messages(self, messages: List[Dict]) -> List[Dict]:
+        """
+        从群聊消息中提取文本消息
+        
+        Args:
+            messages: 群聊消息列表
+            
+        Returns:
+            提取的文本消息列表
+        """
+        logger.debug(f"extract_text_messages 开始处理，输入消息数量: {len(messages) if messages else 0}")
+        logger.debug(f"extract_text_messages 输入消息类型: {type(messages)}")
+        
+        if not messages:
+            logger.warning("extract_text_messages 收到空消息列表")
+            return []
+        
+        text_messages = []
+        
+        for i, msg in enumerate(messages):
+            logger.debug(f"处理第 {i+1} 条消息，类型: {type(msg)}")
+            # 确保msg是字典类型，避免'str' object has no attribute 'get'错误
+            if not isinstance(msg, dict):
+                logger.warning(f"跳过非字典类型的消息: {type(msg)} - {msg}")
+                continue
+                
+            try:
+                sender = msg.get("sender", {})
+                # 确保sender是字典类型，避免'str' object has no attribute 'get'错误
+                if not isinstance(sender, dict):
+                    logger.warning(f"extract_text_messages 跳过sender非字典类型的消息: {type(sender)} - {sender}")
+                    continue
+                    
+                nickname = sender.get("nickname", "") or sender.get("card", "")
+                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
+                
+                for content in msg.get("message", []):
+                    if content.get("type") == "text":
+                        text = content.get("data", {}).get("text", "").strip()
+                        if text and len(text) > 2 and not text.startswith("/"):
+                            # 清理消息内容
+                            text = text.replace('""', '"').replace('""', '"')
+                            text = text.replace(''', "'").replace(''', "'")
+                            text = text.replace('\n', ' ').replace('\r', ' ')
+                            text = text.replace('\t', ' ')
+                            text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
+                            text_messages.append({
+                                "sender": nickname,
+                                "time": msg_time,
+                                "content": text.strip()
+                            })
+            except Exception as e:
+                logger.error(f"处理第 {i+1} 条消息时出错: {e}", exc_info=True)
+                continue
+        
+        logger.debug(f"extract_text_messages 完成，提取到 {len(text_messages)} 条文本消息")
+        if text_messages:
+            logger.debug(f"extract_text_messages 第一条文本消息: {text_messages[0]}")
+        return text_messages
+    
+    async def analyze_topics(self, messages: List[Dict], umo: str = None) -> Tuple[List[SummaryTopic], TokenUsage]:
+        """
+        分析群聊话题
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (话题列表, Token使用统计)
+        """
+        try:
+            logger.debug(f"analyze_topics 开始处理，消息数量: {len(messages) if messages else 0}")
+            logger.debug(f"消息类型: {type(messages)}")
+            if messages:
+                logger.debug(f"第一条消息类型: {type(messages[0]) if messages else '无'}")
+                logger.debug(f"第一条消息内容: {messages[0] if messages else '无'}")
+            
+            # 检查是否有有效的文本消息
+            text_messages = self.extract_text_messages(messages)
+            logger.debug(f"提取到 {len(text_messages)} 条文本消息")
+            
+            if not text_messages:
+                logger.info("没有有效的文本消息，返回空结果")
+                return [], TokenUsage()
+            
+            logger.info(f"开始分析 {len(text_messages)} 条文本消息中的话题")
+            logger.debug(f"文本消息类型: {type(text_messages)}")
+            if text_messages:
+                logger.debug(f"第一条文本消息类型: {type(text_messages[0])}")
+                logger.debug(f"第一条文本消息内容: {text_messages[0]}")
+            
+            # 直接传入原始消息，让 build_prompt 方法处理
+            return await self.analyze(messages, umo)
+            
+        except Exception as e:
+            logger.error(f"话题分析失败: {e}", exc_info=True)
+            return [], TokenUsage()

src/analysis/analyzers/user_title_analyzer.py

@@ -0,0 +1,222 @@
+"""
+用户称号分析模块
+专门处理用户称号和MBTI类型分析
+"""
+
+from typing import List, Dict, Tuple
+from astrbot.api import logger
+from ...models.data_models import UserTitle, TokenUsage
+from .base_analyzer import BaseAnalyzer
+from ..utils.json_utils import extract_user_titles_with_regex
+
+
+class UserTitleAnalyzer(BaseAnalyzer):
+    """
+    用户称号分析器
+    专门处理用户称号分配和MBTI类型分析
+    """
+    
+    def get_data_type(self) -> str:
+        """获取数据类型标识"""
+        return "用户称号"
+    
+    def get_max_count(self) -> int:
+        """获取最大用户称号数量"""
+        return self.config_manager.get_max_user_titles()
+    
+    def get_max_tokens(self) -> int:
+        """获取最大token数"""
+        return 1500
+    
+    def get_temperature(self) -> float:
+        """获取温度参数"""
+        return 0.5
+    
+    def build_prompt(self, user_data: Dict) -> str:
+        """
+        构建用户称号分析提示词
+        
+        Args:
+            user_data: 用户数据字典，包含用户统计信息
+            
+        Returns:
+            提示词字符串
+        """
+        user_summaries = user_data.get("user_summaries", [])
+        
+        if not user_summaries:
+            return ""
+        
+        # 构建用户数据文本
+        users_text = "\n".join([
+            f"- {user['name']} (QQ:{user['qq']}): "
+            f"发言{user['message_count']}条, 平均{user['avg_chars']}字, "
+            f"表情比例{user['emoji_ratio']}, 夜间发言比例{user['night_ratio']}, "
+            f"回复比例{user['reply_ratio']}"
+            for user in user_summaries
+        ])
+        
+        prompt = f"""
+请为以下群友分配合适的称号和MBTI类型。每个人只能有一个称号，每个称号只能给一个人。
+
+可选称号：
+- 龙王: 发言频繁但内容轻松的人
+- 技术专家: 经常讨论技术话题的人
+- 夜猫子: 经常在深夜发言的人
+- 表情包军火库: 经常发表情的人
+- 沉默终结者: 经常开启话题的人
+- 评论家: 平均发言长度很长的人
+- 阳角: 在群里很有影响力的人
+- 互动达人: 经常回复别人的人
+- ... (你可以自行进行拓展添加)
+
+用户数据：
+{users_text}
+
+请以JSON格式返回，格式如下：
+[
+  {{
+    "name": "用户名",
+    "qq": 123456789,
+    "title": "称号",
+    "mbti": "MBTI类型",
+    "reason": "获得此称号的原因"
+  }}
+]
+"""
+        return prompt
+    
+    def extract_with_regex(self, result_text: str, max_count: int) -> List[Dict]:
+        """
+        使用正则表达式提取用户称号信息
+        
+        Args:
+            result_text: LLM响应文本
+            max_count: 最大提取数量
+            
+        Returns:
+            用户称号数据列表
+        """
+        return extract_user_titles_with_regex(result_text, max_count)
+    
+    def create_data_objects(self, titles_data: List[Dict]) -> List[UserTitle]:
+        """
+        创建用户称号对象列表
+        
+        Args:
+            titles_data: 原始用户称号数据列表
+            
+        Returns:
+            UserTitle对象列表
+        """
+        try:
+            titles = []
+            max_titles = self.get_max_count()
+            
+            for title_data in titles_data[:max_titles]:
+                # 确保数据格式正确
+                name = title_data.get("name", "").strip()
+                qq = title_data.get("qq")
+                title = title_data.get("title", "").strip()
+                mbti = title_data.get("mbti", "").strip()
+                reason = title_data.get("reason", "").strip()
+                
+                # 验证必要字段
+                if not name or not title or not mbti or not reason:
+                    logger.warning(f"用户称号数据格式不完整，跳过: {title_data}")
+                    continue
+                
+                # 验证QQ号格式
+                try:
+                    qq = int(qq)
+                except (ValueError, TypeError):
+                    logger.warning(f"QQ号格式无效，跳过: {qq}")
+                    continue
+                
+                titles.append(UserTitle(
+                    name=name,
+                    qq=qq,
+                    title=title,
+                    mbti=mbti,
+                    reason=reason
+                ))
+            
+            return titles
+            
+        except Exception as e:
+            logger.error(f"创建用户称号对象失败: {e}")
+            return []
+    
+    def prepare_user_data(self, messages: List[Dict], user_analysis: Dict) -> Dict:
+        """
+        准备用户数据
+        
+        Args:
+            messages: 群聊消息列表
+            user_analysis: 用户分析统计
+            
+        Returns:
+            准备好的用户数据字典
+        """
+        try:
+            user_summaries = []
+            
+            for user_id, stats in user_analysis.items():
+                if stats["message_count"] < 5:  # 过滤活跃度太低的用户
+                    continue
+                
+                # 分析用户特征
+                night_messages = sum(stats["hours"][h] for h in range(6))
+                day_messages = stats["message_count"] - night_messages
+                avg_chars = stats["char_count"] / stats["message_count"] if stats["message_count"] > 0 else 0
+                
+                user_summaries.append({
+                    "name": stats["nickname"],
+                    "qq": int(user_id),
+                    "message_count": stats["message_count"],
+                    "avg_chars": round(avg_chars, 1),
+                    "emoji_ratio": round(stats["emoji_count"] / stats["message_count"], 2),
+                    "night_ratio": round(night_messages / stats["message_count"], 2),
+                    "reply_ratio": round(stats["reply_count"] / stats["message_count"], 2)
+                })
+            
+            if not user_summaries:
+                return {"user_summaries": []}
+            
+            # 按消息数量排序，取前N名
+            max_user_titles = self.get_max_count()
+            user_summaries.sort(key=lambda x: x["message_count"], reverse=True)
+            user_summaries = user_summaries[:max_user_titles]
+            
+            return {"user_summaries": user_summaries}
+            
+        except Exception as e:
+            logger.error(f"准备用户数据失败: {e}")
+            return {"user_summaries": []}
+    
+    async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict, umo: str = None) -> Tuple[List[UserTitle], TokenUsage]:
+        """
+        分析用户称号
+        
+        Args:
+            messages: 群聊消息列表
+            user_analysis: 用户分析统计
+            umo: 模型唯一标识符
+            
+        Returns:
+            (用户称号列表, Token使用统计)
+        """
+        try:
+            # 准备用户数据
+            user_data = self.prepare_user_data(messages, user_analysis)
+            
+            if not user_data["user_summaries"]:
+                logger.info("没有符合条件的用户，返回空结果")
+                return [], TokenUsage()
+            
+            logger.info(f"开始分析 {len(user_data['user_summaries'])} 个用户的称号")
+            return await self.analyze(user_data, umo)
+            
+        except Exception as e:
+            logger.error(f"用户称号分析失败: {e}")
+            return [], TokenUsage()

src/analysis/llm_analyzer.py

@@ -1,580 +1,130 @@
 """
 LLM分析器模块
-负责使用LLM进行话题分析、用户称号分析和金句分析
+负责协调各个分析器进行话题分析、用户称号分析和金句分析
 """
 
-import json
-import re
-from datetime import datetime
-import asyncio
 from typing import List, Dict, Tuple
 from astrbot.api import logger
-from ...src.models.data_models import SummaryTopic, UserTitle, GoldenQuote, TokenUsage
+from ..models.data_models import SummaryTopic, UserTitle, GoldenQuote, TokenUsage
+from .analyzers.topic_analyzer import TopicAnalyzer
+from .analyzers.user_title_analyzer import UserTitleAnalyzer
+from .analyzers.golden_quote_analyzer import GoldenQuoteAnalyzer
+from .utils.llm_utils import call_provider_with_retry
+from .utils.json_utils import fix_json
+from .utils.json_utils import extract_topics_with_regex, extract_user_titles_with_regex, extract_golden_quotes_with_regex
 
 
 class LLMAnalyzer:
-    """LLM分析器"""
-
+    """
+    LLM分析器
+    作为统一入口，协调各个专门的分析器进行不同类型的分析
+    保持向后兼容性，提供原有的接口
+    """
+    
     def __init__(self, context, config_manager):
+        """
+        初始化LLM分析器
+        
+        Args:
+            context: AstrBot上下文对象
+            config_manager: 配置管理器
+        """
         self.context = context
         self.config_manager = config_manager
-
-    async def _call_provider_with_retry(self, provider, prompt: str, max_tokens: int, temperature: float, umo: str = None):
-        """
-        调用LLM提供者，带超时、重试与退避。支持自定义服务商。
-
-        Args:
-            provider: LLM服务商实例或None。
-            prompt (str): 输入的提示语。
-            max_tokens (int): 最大生成token数。
-            temperature (float): 采样温度。
-            umo (str, optional): 指定使用的模型唯一标识符（Unique Model Object），
-                用于选择特定的LLM服务商或模型。格式通常为字符串，例如 "gpt-3.5-turbo"。
-                如果为None，则使用默认模型。
-
-        Returns:
-            LLM生成的结果。
-        """
-        timeout = self.config_manager.get_llm_timeout()
-        retries = self.config_manager.get_llm_retries()
-        backoff = self.config_manager.get_llm_backoff()
-
-        # 获取自定义服务商参数
-        custom_api_key = self.config_manager.get_custom_api_key()
-        custom_api_base = self.config_manager.get_custom_api_base_url()
-        custom_model = self.config_manager.get_custom_model_name()
-
-        last_exc = None
-        for attempt in range(1, retries + 1):
-            try:
-                if custom_api_key and custom_api_base and custom_model:
-                    logger.info(f"使用自定义LLM提供商: {custom_api_base} model={custom_model}")
-                    import aiohttp
-                    async with aiohttp.ClientSession() as session:
-                        headers = {
-                            "Authorization": f"Bearer {custom_api_key}",
-                            "Content-Type": "application/json"
-                        }
-                        payload = {
-                            "model": custom_model,
-                            "messages": [{"role": "user", "content": prompt}],
-                            "max_tokens": max_tokens,
-                            "temperature": temperature
-                        }
-                        aio_timeout = aiohttp.ClientTimeout(total=timeout)
-                        async with session.post(custom_api_base, json=payload, headers=headers, timeout=aio_timeout) as resp:
-                            if resp.status != 200:
-                                error_text = await resp.text()
-                                logger.error(f"自定义LLM服务商请求失败: HTTP {resp.status}, 内容: {error_text}")
-                            try:
-                                response_json = await resp.json()
-                            except Exception as json_err:
-                                error_text = await resp.text()
-                                logger.error(f"自定义LLM服务商响应JSON解析失败: {json_err}, 内容: {error_text}")
-                                return None
-                            # 兼容 OpenAI 格式，安全访问嵌套字段
-                            content = None
-                            try:
-                                choices = response_json.get("choices")
-                                if choices and isinstance(choices, list) and len(choices) > 0:
-                                    message = choices[0].get("message")
-                                    if message and isinstance(message, dict):
-                                        content = message.get("content")
-                                if content is None:
-                                    logger.error(f"自定义LLM响应格式异常: {response_json}")
-                                    return None
-                            except Exception as key_err:
-                                logger.error(f"自定义LLM响应结构解析失败: {key_err}, 响应内容: {response_json}")
-                                return None
-                            # 构造一个兼容原有逻辑的对象
-                            class CustomResponse:
-                                completion_text = content
-                                raw_completion = response_json
-                            return CustomResponse()
-                else:
-                    # 确保使用当前指定的模型
-                    if provider is None:
-                        provider = self.context.get_using_provider(umo=umo)
-                        provider_id = 'unknown'
-                        if provider:
-                            try:
-                                meta = provider.meta()
-                                provider_id = meta.id
-                            except Exception as e:
-                                logger.debug(f"获取提供商ID失败: {e}")
-                        logger.info(f"获取到的 provider ID: {provider_id}")
-                        if not provider or provider_id == 'unknown':
-                            logger.warning(f"获取的提供商不正确 (Provider ID: {provider_id})")
-
-                    
-                    logger.info(f"使用LLM provider: {provider}")
-                    if not provider:
-                        logger.error("provider 为空，无法调用 text_chat，直接返回 None")
-                        return None
-                    coro = provider.text_chat(prompt=prompt, max_tokens=max_tokens, temperature=temperature)
-                    return await asyncio.wait_for(coro, timeout=timeout)
-            except asyncio.TimeoutError as e:
-                last_exc = e
-                logger.warning(f"LLM请求超时: 第{attempt}次, timeout={timeout}s")
-            except Exception as e:
-                last_exc = e
-                logger.warning(f"LLM请求失败: 第{attempt}次, 错误: {last_exc}")
-            # 若非最后一次，等待退避后重试
-            if attempt < retries:
-                await asyncio.sleep(backoff * attempt)
-
-        # 最终仍失败，记录错误并返回 None 由调用方处理降级，避免抛出异常
-        logger.error(f"LLM请求全部重试失败: {last_exc}")
-        return None
-
+        
+        # 初始化各个专门的分析器
+        self.topic_analyzer = TopicAnalyzer(context, config_manager)
+        self.user_title_analyzer = UserTitleAnalyzer(context, config_manager)
+        self.golden_quote_analyzer = GoldenQuoteAnalyzer(context, config_manager)
+    
     async def analyze_topics(self, messages: List[Dict], umo: str = None) -> Tuple[List[SummaryTopic], TokenUsage]:
-        """使用LLM分析话题"""
+        """
+        使用LLM分析话题
+        保持原有接口，委托给专门的TopicAnalyzer处理
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (话题列表, Token使用统计)
+        """
         try:
-            # 提取文本消息
-            text_messages = []
-            for msg in messages:
-                sender = msg.get("sender", {})
-                nickname = sender.get("nickname", "") or sender.get("card", "")
-                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
-
-                for content in msg.get("message", []):
-                    if content.get("type") == "text":
-                        text = content.get("data", {}).get("text", "").strip()
-                        if text and len(text) > 2 and not text.startswith(("/")):
-                            text_messages.append({
-                                "sender": nickname,
-                                "time": msg_time,
-                                "content": text
-                            })
-
-            if not text_messages:
-                return [], TokenUsage()
-
-            # 构建LLM提示词，清理消息内容
-            def clean_message_content(content):
-                """清理消息内容，移除可能影响JSON解析的字符"""
-                # 替换中文引号
-                content = content.replace('"', '"').replace('"', '"')
-                content = content.replace(''', "'").replace(''', "'")
-                # 移除或替换其他特殊字符
-                content = content.replace('\n', ' ').replace('\r', ' ')
-                content = content.replace('\t', ' ')
-                # 移除可能的控制字符
-                content = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', content)
-                return content.strip()
-
-            messages_text = "\n".join([
-                f"[{msg['time']}] {msg['sender']}: {clean_message_content(msg['content'])}"
-                for msg in text_messages
-            ])
-
-            max_topics = self.config_manager.get_max_topics()
-            prompt = f"""
-你是一个帮我进行群聊信息总结的助手，生成总结内容时，你需要严格遵守下面的几个准则：
-请分析接下来提供的群聊记录，提取出最多{max_topics}个主要话题。
-
-对于每个话题，请提供：
-1. 话题名称（突出主题内容，尽量简明扼要）
-2. 主要参与者（最多5人）
-3. 话题详细描述（包含关键信息和结论）
-
-注意：
-- 对于比较有价值的点，稍微用一两句话详细讲讲，比如不要生成 "Nolan 和 SOV 讨论了 galgame 中关于性符号的衍生情况" 这种宽泛的内容，而是生成更加具体的讨论内容，让其他人只看这个消息就能知道讨论中有价值的，有营养的信息。
-- 对于其中的部分信息，你需要特意提到主题施加的主体是谁，是哪个群友做了什么事情，而不要直接生成和群友没有关系的语句。
-- 对于每一条总结，尽量讲清楚前因后果，以及话题的结论，是什么，为什么，怎么做，如果用户没有讲到细节，则可以不用这么做。
-
-群聊记录：
-{messages_text}
-
-重要：必须返回标准JSON格式，严格遵守以下规则：
-1. 只使用英文双引号 " 不要使用中文引号 " "
-2. 字符串内容中的引号必须转义为 \"
-3. 多个对象之间用逗号分隔
-4. 数组元素之间用逗号分隔
-5. 不要在JSON外添加任何文字说明
-6. 描述内容避免使用特殊符号，用普通文字表达
-
-请严格按照以下JSON格式返回，确保可以被标准JSON解析器解析：
-[
-  {{
-    "topic": "话题名称",
-    "contributors": ["用户1", "用户2"],
-    "detail": "话题描述内容"
-  }},
-  {{
-    "topic": "另一个话题",
-    "contributors": ["用户3", "用户4"],
-    "detail": "另一个话题的描述"
-  }}
-]
-
-注意：返回的内容必须是纯JSON，不要包含markdown代码块标记或其他格式
-"""
-
-            # 调用LLM
-            response = await self._call_provider_with_retry(None, prompt, max_tokens=10000, temperature=0.6, umo=umo)
-            if response is None:
-                logger.error("话题分析调用LLM失败: provider返回None（重试失败）")
-                return [], TokenUsage()
-
-            # 提取token使用统计
-            token_usage = TokenUsage()
-            # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
-            usage = None
-            if getattr(response, 'raw_completion', None) is not None:
-                usage = getattr(response.raw_completion, 'usage', None)
-                if usage:
-                    token_usage.prompt_tokens = getattr(usage, 'prompt_tokens', 0) or 0
-                    token_usage.completion_tokens = getattr(usage, 'completion_tokens', 0) or 0
-                    token_usage.total_tokens = getattr(usage, 'total_tokens', 0) or 0
-
-            # 解析响应
-            if hasattr(response, 'completion_text'):
-                result_text = response.completion_text
-            else:
-                result_text = str(response)
-
-            # 尝试解析JSON
-            try:
-                # 提取JSON部分
-                json_match = re.search(r'\[.*?\]', result_text, re.DOTALL)
-                if json_match:
-                    json_text = json_match.group()
-                    logger.debug(f"话题分析JSON原文: {json_text[:500]}...")
-
-                    # 强化JSON清理和修复
-                    json_text = self._fix_json(json_text)
-                    logger.debug(f"修复后的JSON: {json_text[:300]}...")
-
-                    topics_data = json.loads(json_text)
-                    topics = [SummaryTopic(**topic) for topic in topics_data[:max_topics]]
-                    logger.info(f"话题分析成功，解析到 {len(topics)} 个话题")
-                    return topics, token_usage
-                else:
-                    logger.warning(f"话题分析响应中未找到JSON格式，响应内容: {result_text[:200]}...")
-            except json.JSONDecodeError as e:
-                logger.warning(f"话题分析JSON解析失败: {e}")
-                logger.debug(f"修复后的JSON: {json_text if 'json_text' in locals() else 'N/A'}")
-                logger.debug(f"原始响应: {result_text}")
-
-                # 如果JSON解析失败，尝试用正则表达式提取话题信息
-                topics = self._extract_topics_with_regex(result_text, max_topics)
-                if topics:
-                    logger.info(f"正则表达式提取成功，获得 {len(topics)} 个话题，话题分析 warning 可忽略")
-                    return topics, token_usage
-                else:
-                    # 最后的降级方案
-                    logger.info("正则表达式提取失败，使用默认话题...")
-                    return [SummaryTopic(
-                        topic="群聊讨论",
-                        contributors=["群友"],
-                        detail="今日群聊内容丰富，涵盖多个话题"
-                    )], token_usage
-
-            return [], token_usage
-
+            logger.info("开始话题分析")
+            return await self.topic_analyzer.analyze_topics(messages, umo)
         except Exception as e:
             logger.error(f"话题分析失败: {e}")
             return [], TokenUsage()
-
-    def _fix_json(self, text: str) -> str:
-        """修复JSON格式问题"""
-        # 移除markdown代码块标记
-        text = re.sub(r'```json\s*', '', text)
-        text = re.sub(r'```\s*$', '', text)
-
-        # 基础清理
-        text = text.replace('\n', ' ').replace('\r', ' ')
-        text = re.sub(r'\s+', ' ', text)
-
-        # 替换中文引号为英文引号
-        text = text.replace('"', '"').replace('"', '"')
-        text = text.replace(''', "'").replace(''', "'")
-
-        # 处理字符串内容中的特殊字符
-        # 转义字符串内的双引号
-        def escape_quotes_in_strings(match):
-            content = match.group(1)
-            # 转义内部的双引号
-            content = content.replace('"', '\\"')
-            return f'"{content}"'
-
-        # 先处理字段值中的引号
-        text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text)
-
-        # 修复截断的JSON
-        if not text.endswith(']'):
-            last_complete = text.rfind('}')
-            if last_complete > 0:
-                text = text[:last_complete + 1] + ']'
-
-        # 修复常见的JSON格式问题
-        # 1. 修复缺失的逗号
-        text = re.sub(r'}\s*{', '}, {', text)
-
-        # 2. 确保字段名有引号
-        text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', text)
-
-        # 3. 移除多余的逗号
-        text = re.sub(r',\s*}', '}', text)
-        text = re.sub(r',\s*]', ']', text)
-
-        return text
-
-    def _extract_topics_with_regex(self, result_text: str, max_topics: int) -> List[SummaryTopic]:
-        """使用正则表达式提取话题信息"""
-        try:
-            topics = []
-
-            # 更强的正则表达式提取话题信息，处理转义字符
-            # 匹配每个完整的话题对象
-            topic_pattern = r'\{\s*"topic":\s*"([^"]+)"\s*,\s*"contributors":\s*\[([^\]]+)\]\s*,\s*"detail":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
-            matches = re.findall(topic_pattern, result_text, re.DOTALL)
-
-            if not matches:
-                # 尝试更宽松的匹配
-                topic_pattern = r'"topic":\s*"([^"]+)"[^}]*"contributors":\s*\[([^\]]+)\][^}]*"detail":\s*"([^"]*(?:\\.[^"]*)*)"'
-                matches = re.findall(topic_pattern, result_text, re.DOTALL)
-
-            for match in matches[:max_topics]:
-                topic_name = match[0].strip()
-                contributors_str = match[1].strip()
-                detail = match[2].strip()
-
-                # 清理detail中的转义字符
-                detail = detail.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
-
-                # 解析参与者列表
-                contributors = []
-                for contrib in re.findall(r'"([^"]+)"', contributors_str):
-                    contributors.append(contrib.strip())
-
-                if not contributors:
-                    contributors = ["群友"]
-
-                topics.append(SummaryTopic(
-                    topic=topic_name,
-                    contributors=contributors[:5],  # 最多5个参与者
-                    detail=detail
-                ))
-
-            return topics
-        except Exception as e:
-            logger.error(f"正则表达式提取失败: {e}")
-            return []
-
+    
     async def analyze_user_titles(self, messages: List[Dict], user_analysis: Dict, umo: str = None) -> Tuple[List[UserTitle], TokenUsage]:
-        """使用LLM分析用户称号"""
+        """
+        使用LLM分析用户称号
+        保持原有接口，委托给专门的UserTitleAnalyzer处理
+        
+        Args:
+            messages: 群聊消息列表
+            user_analysis: 用户分析统计
+            umo: 模型唯一标识符
+            
+        Returns:
+            (用户称号列表, Token使用统计)
+        """
         try:
-            # 准备用户数据
-            user_summaries = []
-            for user_id, stats in user_analysis.items():
-                if stats["message_count"] < 5:  # 过滤活跃度太低的用户
-                    continue
-
-                # 分析用户特征
-                night_messages = sum(stats["hours"][h] for h in range(0, 6))
-                day_messages = stats["message_count"] - night_messages
-                avg_chars = stats["char_count"] / stats["message_count"] if stats["message_count"] > 0 else 0
-
-                user_summaries.append({
-                    "name": stats["nickname"],
-                    "qq": int(user_id),
-                    "message_count": stats["message_count"],
-                    "avg_chars": round(avg_chars, 1),
-                    "emoji_ratio": round(stats["emoji_count"] / stats["message_count"], 2),
-                    "night_ratio": round(night_messages / stats["message_count"], 2),
-                    "reply_ratio": round(stats["reply_count"] / stats["message_count"], 2)
-                })
-
-            if not user_summaries:
-                return [], TokenUsage()
-
-            # 按消息数量排序，取前N名
-            max_user_titles = self.config_manager.get_max_user_titles()
-            user_summaries.sort(key=lambda x: x["message_count"], reverse=True)
-            user_summaries = user_summaries[:max_user_titles]
-
-            # 构建LLM提示词
-            users_text = "\n".join([
-                f"- {user['name']} (QQ:{user['qq']}): "
-                f"发言{user['message_count']}条, 平均{user['avg_chars']}字, "
-                f"表情比例{user['emoji_ratio']}, 夜间发言比例{user['night_ratio']}, "
-                f"回复比例{user['reply_ratio']}"
-                for user in user_summaries
-            ])
-
-            prompt = f"""
-请为以下群友分配合适的称号和MBTI类型。每个人只能有一个称号，每个称号只能给一个人。
-
-可选称号：
-- 龙王: 发言频繁但内容轻松的人
-- 技术专家: 经常讨论技术话题的人
-- 夜猫子: 经常在深夜发言的人
-- 表情包军火库: 经常发表情的人
-- 沉默终结者: 经常开启话题的人
-- 评论家: 平均发言长度很长的人
-- 阳角: 在群里很有影响力的人
-- 互动达人: 经常回复别人的人
-- ... (你可以自行进行拓展添加)
-
-用户数据：
-{users_text}
-
-请以JSON格式返回，格式如下：
-[
-  {{
-    "name": "用户名",
-    "qq": 123456789,
-    "title": "称号",
-    "mbti": "MBTI类型",
-    "reason": "获得此称号的原因"
-  }}
-]
-"""
-
-            # 调用LLM
-            response = await self._call_provider_with_retry(None, prompt, max_tokens=1500, temperature=0.5, umo=umo)
-            if response is None:
-                logger.error("用户称号分析调用LLM失败: provider返回None（重试失败）")
-                return [], TokenUsage()
-
-            # 提取token使用统计
-            token_usage = TokenUsage()
-            # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
-            usage = None
-            if getattr(response, 'raw_completion', None) is not None:
-                usage = getattr(response.raw_completion, 'usage', None)
-                if usage:
-                    token_usage.prompt_tokens = getattr(usage, 'prompt_tokens', 0) or 0
-                    token_usage.completion_tokens = getattr(usage, 'completion_tokens', 0) or 0
-                    token_usage.total_tokens = getattr(usage, 'total_tokens', 0) or 0
-
-            # 解析响应
-            if hasattr(response, 'completion_text'):
-                result_text = response.completion_text
-            else:
-                result_text = str(response)
-
-            # debug日志：打印原始响应
-            logger.debug(f"用户称号分析原始响应: {result_text[:500]}...")
-
-            # 尝试解析JSON
-            try:
-                json_match = re.search(r'\[.*\]', result_text, re.DOTALL)
-                if json_match:
-                    logger.debug(f"用户称号分析JSON原文: {json_match.group()[:500]}...")
-                    titles_data = json.loads(json_match.group())
-                    return [UserTitle(**title) for title in titles_data], token_usage
-            except Exception as e:
-                logger.warning(f"用户称号分析JSON解析失败: {e}")
-                logger.warning(f"原始响应: {result_text}")
-
-            return [], token_usage
-
+            logger.info("开始用户称号分析")
+            return await self.user_title_analyzer.analyze_user_titles(messages, user_analysis, umo)
         except Exception as e:
             logger.error(f"用户称号分析失败: {e}")
             return [], TokenUsage()
-
+    
     async def analyze_golden_quotes(self, messages: List[Dict], umo: str = None) -> Tuple[List[GoldenQuote], TokenUsage]:
-        """使用LLM分析群聊金句"""
+        """
+        使用LLM分析群聊金句
+        保持原有接口，委托给专门的GoldenQuoteAnalyzer处理
+        
+        Args:
+            messages: 群聊消息列表
+            umo: 模型唯一标识符
+            
+        Returns:
+            (金句列表, Token使用统计)
+        """
         try:
-            # 提取有趣的文本消息
-            interesting_messages = []
-            for msg in messages:
-                sender = msg.get("sender", {})
-                nickname = sender.get("nickname", "") or sender.get("card", "")
-                msg_time = datetime.fromtimestamp(msg.get("time", 0)).strftime("%H:%M")
-
-                for content in msg.get("message", []):
-                    if content.get("type") == "text":
-                        text = content.get("data", {}).get("text", "").strip()
-                        # 过滤长度适中、可能有趣的消息
-                        if 5 <= len(text) <= 100 and not text.startswith(("http", "www", "/")):
-                            interesting_messages.append({
-                                "sender": nickname,
-                                "time": msg_time,
-                                "content": text
-                            })
-
-            if not interesting_messages:
-                return [], TokenUsage()
-
-            # 构建LLM提示词
-            messages_text = "\n".join([
-                f"[{msg['time']}] {msg['sender']}: {msg['content']}"
-                for msg in interesting_messages
-            ])
-
-            # 计算金句数量
-            max_golden_quotes = self.config_manager.get_max_golden_quotes()
-
-            prompt = f"""
-请从以下群聊记录中挑选出{max_golden_quotes}句最具冲击力、最令人惊叹的"金句"。这些金句需满足：
-- 核心标准：**逆天的神人发言**，即具备颠覆常识的脑洞、逻辑跳脱的表达或强烈反差感的原创内容
-- 典型特征：包含某些争议话题元素、夸张类比、反常规结论、一本正经的"胡说八道"或突破语境的清奇思路，并且具备一定的冲击力，让人印象深刻。
-
-对于每个金句，请提供：
-1. 原文内容（完整保留发言细节）
-2. 发言人昵称
-3. 选择理由（具体说明其"逆天"之处，如逻辑颠覆点/脑洞角度/反差感/争议话题元素）
-
-此外，我将对你进行严格约束：
-- 优先筛选 **逆天指数最高** 的内容：发情、性压抑话题 > 争议话题 > 元素级 > 颠覆认知级 > 逻辑跳脱级 > 趣味调侃级，剔除单纯玩梗或网络热词堆砌的普通发言
-- 重点标记包含极端类比、反常识论证或无厘头结论的内容，并且包含一定的争议话题元素。
-
-群聊记录：
-{messages_text}
-
-请以JSON格式返回，格式如下：
-[
-  {{
-    "content": "金句原文",
-    "sender": "发言人昵称",
-    "reason": "选择这句话的理由（需明确说明逆天特质）"
-  }}
-]
-"""
-
-            # 调用LLM
-            response = await self._call_provider_with_retry(None, prompt, max_tokens=1500, temperature=0.7, umo=umo)
-            if response is None:
-                logger.error("金句分析调用LLM失败: provider返回None（重试失败）")
-                return [], TokenUsage()
-
-            # 提取token使用统计
-            token_usage = TokenUsage()
-            # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
-            usage = None
-            if getattr(response, 'raw_completion', None) is not None:
-                usage = getattr(response.raw_completion, 'usage', None)
-                if usage:
-                    token_usage.prompt_tokens = getattr(usage, 'prompt_tokens', 0) or 0
-                    token_usage.completion_tokens = getattr(usage, 'completion_tokens', 0) or 0
-                    token_usage.total_tokens = getattr(usage, 'total_tokens', 0) or 0
-
-            # 解析响应
-            if hasattr(response, 'completion_text'):
-                result_text = response.completion_text
-            else:
-                result_text = str(response)
-
-            # debug日志：打印原始响应
-            logger.debug(f"金句分析原始响应: {result_text[:500]}...")
-
-            # 尝试解析JSON
-            try:
-                json_match = re.search(r'\[.*\]', result_text, re.DOTALL)
-                if json_match:
-                    logger.debug(f"金句分析JSON原文: {json_match.group()[:500]}...")
-                    quotes_data = json.loads(json_match.group())
-                    return [GoldenQuote(**quote) for quote in quotes_data[:max_golden_quotes]], token_usage
-            except Exception as e:
-                logger.warning(f"金句分析JSON解析失败: {e}")
-                logger.warning(f"原始响应: {result_text}")
-
-            return [], token_usage
-
+            logger.info("开始金句分析")
+            return await self.golden_quote_analyzer.analyze_golden_quotes(messages, umo)
         except Exception as e:
             logger.error(f"金句分析失败: {e}")
-            return [], TokenUsage()
+            return [], TokenUsage()
+    
+    # 向后兼容的方法，保持原有调用方式
+    async def _call_provider_with_retry(self, provider, prompt: str, max_tokens: int, 
+                                      temperature: float, umo: str = None):
+        """
+        向后兼容的LLM调用方法
+        现在委托给llm_utils模块处理
+        
+        Args:
+            provider: LLM服务商实例或None
+            prompt: 输入的提示语
+            max_tokens: 最大生成token数
+            temperature: 采样温度
+            umo: 指定使用的模型唯一标识符
+            
+        Returns:
+            LLM生成的结果
+        """
+        return await call_provider_with_retry(self.context, self.config_manager, 
+                                            prompt, max_tokens, temperature, umo)
+    
+    def _fix_json(self, text: str) -> str:
+        """
+        向后兼容的JSON修复方法
+        现在委托给json_utils模块处理
+        
+        Args:
+            text: 需要修复的JSON文本
+            
+        Returns:
+            修复后的JSON文本
+        """
+        return fix_json(text)
+    

src/analysis/utils/__init__.py

@@ -0,0 +1,32 @@
+"""
+分析工具模块
+包含JSON处理和LLM API请求处理工具
+"""
+
+from .json_utils import (
+    fix_json,
+    parse_json_response,
+    extract_topics_with_regex,
+    extract_user_titles_with_regex,
+    extract_golden_quotes_with_regex
+)
+
+from .llm_utils import (
+    call_provider_with_retry,
+    extract_token_usage,
+    extract_response_text
+)
+
+__all__ = [
+    # JSON处理工具
+    'fix_json',
+    'parse_json_response',
+    'extract_topics_with_regex',
+    'extract_user_titles_with_regex',
+    'extract_golden_quotes_with_regex',
+    
+    # LLM工具
+    'call_provider_with_retry',
+    'extract_token_usage',
+    'extract_response_text'
+]

src/analysis/utils/json_utils.py

@@ -0,0 +1,267 @@
+
+"""
+JSON处理工具模块
+提供JSON解析、修复和正则提取功能
+"""
+
+import json
+import re
+from typing import List, Dict, Tuple, Any, Optional
+from astrbot.api import logger
+
+
+def fix_json(text: str) -> str:
+    """
+    修复JSON格式问题，包括中文符号替换
+    
+    Args:
+        text: 需要修复的JSON文本
+        
+    Returns:
+        修复后的JSON文本
+    """
+    try:
+        # 1. 移除markdown代码块标记
+        text = re.sub(r'```json\s*', '', text)
+        text = re.sub(r'```\s*$', '', text)
+        
+        # 2. 基础清理
+        text = text.replace('\n', ' ').replace('\r', ' ')
+        text = re.sub(r'\s+', ' ', text)
+        
+        # 3. 替换中文符号为英文符号（修复）
+        # 中文引号 -> 英文引号
+        text = text.replace('“', '"').replace('”', '"')
+        text = text.replace('‘', "'").replace('’', "'")
+        # 中文逗号 -> 英文逗号
+        text = text.replace('，', ',')
+        # 中文冒号 -> 英文冒号
+        text = text.replace('：', ':')
+        # 中文括号 -> 英文括号
+        text = text.replace('（', '(').replace('）', ')')
+        text = text.replace('【', '[').replace('】', ']')
+        
+        # 4. 处理字符串内容中的特殊字符
+        # 转义字符串内的双引号
+        def escape_quotes_in_strings(match):
+            content = match.group(1)
+            # 转义内部的双引号
+            content = content.replace('"', '\\"')
+            return f'"{content}"'
+        
+        # 先处理字段值中的引号
+        text = re.sub(r'"([^"]*(?:"[^"]*)*)"', escape_quotes_in_strings, text)
+        
+        # 5. 修复截断的JSON
+        if not text.endswith(']'):
+            last_complete = text.rfind('}')
+            if last_complete > 0:
+                text = text[:last_complete + 1] + ']'
+        
+        # 6. 修复常见的JSON格式问题
+        # 1. 修复缺失的逗号
+        text = re.sub(r'}\s*{', '}, {', text)
+        
+        # 2. 确保字段名有引号（仅在对象开始或逗号后，避免破坏字符串值）
+        def quote_field_names(match):
+            prefix = match.group(1)
+            key = match.group(2)
+            return f'{prefix}"{key}":'
+        
+        # 只在 { 或 , 后面匹配字段名，避免在字符串值中误匹配
+        text = re.sub(r'([{,]\s*)([a-zA-Z_][a-zA-Z0-9_]*)\s*:', quote_field_names, text)
+        
+        # 3. 移除多余的逗号
+        text = re.sub(r',\s*}', '}', text)
+        text = re.sub(r',\s*]', ']', text)
+        
+        return text.strip()
+        
+    except Exception as e:
+        logger.error(f"JSON修复失败: {e}")
+        return text
+
+
+def parse_json_response(result_text: str, data_type: str) -> Tuple[bool, Optional[List[Dict]], Optional[str]]:
+    """
+    统一的JSON解析方法
+    
+    Args:
+        result_text: LLM返回的原始文本
+        data_type: 数据类型 ('topics' | 'user_titles' | 'golden_quotes')
+    
+    Returns:
+        (成功标志, 解析后的数据列表, 错误消息)
+    """
+    try:
+        # 1. 提取JSON部分
+        json_match = re.search(r'\[.*?\]', result_text, re.DOTALL)
+        if not json_match:
+            error_msg = f"{data_type}响应中未找到JSON格式"
+            logger.warning(error_msg)
+            return False, None, error_msg
+        
+        json_text = json_match.group()
+        logger.debug(f"{data_type}分析JSON原文: {json_text[:500]}...")
+        
+        # 2. 修复JSON
+        json_text = fix_json(json_text)
+        logger.debug(f"{data_type}修复后的JSON: {json_text[:300]}...")
+        
+        # 3. 解析JSON
+        data = json.loads(json_text)
+        logger.info(f"{data_type}分析成功，解析到 {len(data)} 条数据")
+        return True, data, None
+        
+    except json.JSONDecodeError as e:
+        error_msg = f"{data_type}JSON解析失败: {e}"
+        logger.warning(error_msg)
+        logger.debug(f"修复后的JSON: {json_text if 'json_text' in locals() else 'N/A'}")
+        return False, None, error_msg
+    except Exception as e:
+        error_msg = f"{data_type}解析异常: {e}"
+        logger.error(error_msg)
+        return False, None, error_msg
+
+
+def extract_topics_with_regex(result_text: str, max_topics: int) -> List[Dict]:
+    """
+    使用正则表达式提取话题信息
+    
+    Args:
+        result_text: 需要提取的文本
+        max_topics: 最大话题数量
+        
+    Returns:
+        话题数据列表
+    """
+    try:
+        # 更强的正则表达式提取话题信息，处理转义字符
+        # 匹配每个完整的话题对象
+        topic_pattern = r'\{\s*"topic":\s*"([^"]+)"\s*,\s*"contributors":\s*\[([^\]]+)\]\s*,\s*"detail":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
+        matches = re.findall(topic_pattern, result_text, re.DOTALL)
+        
+        if not matches:
+            # 尝试更宽松的匹配
+            topic_pattern = r'"topic":\s*"([^"]+)"[^}]*"contributors":\s*\[([^\]]+)\][^}]*"detail":\s*"([^"]*(?:\\.[^"]*)*)"'
+            matches = re.findall(topic_pattern, result_text, re.DOTALL)
+        
+        topics = []
+        for match in matches[:max_topics]:
+            topic_name = match[0].strip()
+            contributors_str = match[1].strip()
+            detail = match[2].strip()
+            
+            # 清理detail中的转义字符
+            detail = detail.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            
+            # 解析参与者列表
+            contributors = [contrib.strip() for contrib in re.findall(r'"([^"]+)"', contributors_str)] or ["群友"]
+            
+            topics.append({
+                "topic": topic_name,
+                "contributors": contributors[:5],  # 最多5个参与者
+                "detail": detail
+            })
+        
+        logger.info(f"话题正则表达式提取成功，提取到 {len(topics)} 条有效话题内容")
+        return topics
+        
+    except Exception as e:
+        logger.error(f"话题正则表达式提取失败: {e}")
+        return []
+
+
+def extract_user_titles_with_regex(result_text: str, max_count: int) -> List[Dict]:
+    """
+    使用正则表达式提取用户称号信息
+    
+    Args:
+        result_text: 需要提取的文本
+        max_count: 最大提取数量
+        
+    Returns:
+        用户称号数据列表
+    """
+    try:
+        titles = []
+        
+        # 正则模式：匹配完整的用户称号对象
+        pattern = r'\{\s*"name":\s*"([^"]+)"\s*,\s*"qq":\s*(\d+)\s*,\s*"title":\s*"([^"]+)"\s*,\s*"mbti":\s*"([^"]+)"\s*,\s*"reason":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
+        matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        if not matches:
+            # 尝试更宽松的匹配（字段顺序可变）
+            pattern = r'"name":\s*"([^"]+)"[^}]*"qq":\s*(\d+)[^}]*"title":\s*"([^"]+)"[^}]*"mbti":\s*"([^"]+)"[^}]*"reason":\s*"([^"]*(?:\\.[^"]*)*)"'
+            matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        for match in matches[:max_count]:
+            name = match[0].strip()
+            qq = int(match[1])
+            title = match[2].strip()
+            mbti = match[3].strip()
+            reason = match[4].strip()
+            
+            # 清理转义字符
+            reason = reason.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            
+            titles.append({
+                "name": name,
+                "qq": qq,
+                "title": title,
+                "mbti": mbti,
+                "reason": reason
+            })
+        
+        logger.info(f"用户称号正则表达式提取成功，提取到 {len(titles)} 条有效用户称号")
+        return titles
+        
+    except Exception as e:
+        logger.error(f"用户称号正则表达式提取失败: {e}")
+        return []
+
+
+def extract_golden_quotes_with_regex(result_text: str, max_count: int) -> List[Dict]:
+    """
+    使用正则表达式提取金句信息
+    
+    Args:
+        result_text: 需要提取的文本
+        max_count: 最大提取数量
+        
+    Returns:
+        金句数据列表
+    """
+    try:
+        quotes = []
+        
+        # 正则模式：匹配完整的金句对象
+        pattern = r'\{\s*"content":\s*"([^"]*(?:\\.[^"]*)*)"\s*,\s*"sender":\s*"([^"]+)"\s*,\s*"reason":\s*"([^"]*(?:\\.[^"]*)*)"\s*\}'
+        matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        if not matches:
+            # 尝试更宽松的匹配（字段顺序可变）
+            pattern = r'"content":\s*"([^"]*(?:\\.[^"]*)*)"[^}]*"sender":\s*"([^"]+)"[^}]*"reason":\s*"([^"]*(?:\\.[^"]*)*)"'
+            matches = re.findall(pattern, result_text, re.DOTALL)
+        
+        for match in matches[:max_count]:
+            content = match[0].strip()
+            sender = match[1].strip()
+            reason = match[2].strip()
+            
+            # 清理转义字符
+            content = content.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            reason = reason.replace('\\"', '"').replace('\\n', ' ').replace('\\t', ' ')
+            
+            quotes.append({
+                "content": content,
+                "sender": sender,
+                "reason": reason
+            })
+        
+        logger.info(f"金句正则表达式提取成功，提取到 {len(quotes)} 条有效金句")
+        return quotes
+        
+    except Exception as e:
+        logger.error(f"金句正则表达式提取失败: {e}")
+        return []

src/analysis/utils/llm_utils.py

@@ -0,0 +1,189 @@
+"""
+LLM API请求处理工具模块
+提供LLM调用和token统计功能
+"""
+
+import asyncio
+from typing import Optional, Any
+from astrbot.api import logger
+import aiohttp
+
+
+async def call_provider_with_retry(context, config_manager, prompt: str, max_tokens: int, 
+                                 temperature: float, umo: str = None) -> Optional[Any]:
+    """
+    调用LLM提供者，带超时、重试与退避。支持自定义服务商。
+    
+    Args:
+        context: AstrBot上下文对象
+        config_manager: 配置管理器
+        prompt: 输入的提示语
+        max_tokens: 最大生成token数
+        temperature: 采样温度
+        umo: 指定使用的模型唯一标识符
+        
+    Returns:
+        LLM生成的结果，失败时返回None
+    """
+    timeout = config_manager.get_llm_timeout()
+    retries = config_manager.get_llm_retries()
+    backoff = config_manager.get_llm_backoff()
+    
+    # 获取自定义服务商参数
+    custom_api_key = config_manager.get_custom_api_key()
+    custom_api_base = config_manager.get_custom_api_base_url()
+    custom_model = config_manager.get_custom_model_name()
+    
+    last_exc = None
+    for attempt in range(1, retries + 1):
+        try:
+            if custom_api_key and custom_api_base and custom_model:
+                logger.info(f"使用自定义LLM提供商: {custom_api_base} model={custom_model}")
+                logger.debug(f"自定义LLM提供商 prompt 长度: {len(prompt) if prompt else 0}")
+                logger.debug(f"自定义LLM提供商 prompt 前100字符: {prompt[:100] if prompt else 'None'}...")
+                
+                # 检查 prompt 是否为空
+                if not prompt or not prompt.strip():
+                    logger.error("自定义LLM提供商: prompt 为空或只包含空白字符，无法发送请求")
+                    return None
+                
+                async with aiohttp.ClientSession() as session:
+                    headers = {
+                        "Authorization": f"Bearer {custom_api_key}",
+                        "Content-Type": "application/json"
+                    }
+                    payload = {
+                        "model": custom_model,
+                        "messages": [{"role": "user", "content": prompt}],
+                        "max_tokens": max_tokens,
+                        "temperature": temperature
+                    }
+                    aio_timeout = aiohttp.ClientTimeout(total=timeout)
+                    async with session.post(custom_api_base, json=payload, headers=headers, timeout=aio_timeout) as resp:
+                        if resp.status != 200:
+                            error_text = await resp.text()
+                            logger.error(f"自定义LLM服务商请求失败: HTTP {resp.status}, 内容: {error_text}")
+                        try:
+                            response_json = await resp.json()
+                        except Exception as json_err:
+                            error_text = await resp.text()
+                            logger.error(f"自定义LLM服务商响应JSON解析失败: {json_err}, 内容: {error_text}")
+                            return None
+                        # 兼容 OpenAI 格式，安全访问嵌套字段
+                        content = None
+                        try:
+                            choices = response_json.get("choices")
+                            if choices and isinstance(choices, list) and len(choices) > 0:
+                                message = choices[0].get("message")
+                                if message and isinstance(message, dict):
+                                    content = message.get("content")
+                            if content is None:
+                                logger.error(f"自定义LLM响应格式异常: {response_json}")
+                                return None
+                        except Exception as key_err:
+                            logger.error(f"自定义LLM响应结构解析失败: {key_err}, 响应内容: {response_json}")
+                            return None
+                        # 构造一个兼容原有逻辑的对象
+                        class CustomResponse:
+                            completion_text = content
+                            raw_completion = response_json
+                        return CustomResponse()
+            else:
+                # 确保使用当前指定的模型
+                provider = context.get_using_provider(umo=umo)
+                provider_id = 'unknown'
+                if provider:
+                    try:
+                        meta = provider.meta()
+                        provider_id = meta.id
+                    except Exception as e:
+                        logger.debug(f"获取提供商ID失败: {e}")
+                logger.info(f"获取到的 provider ID: {provider_id}")
+                if not provider or provider_id == 'unknown':
+                    logger.warning(f"获取的提供商不正确 (Provider ID: {provider_id})")
+                
+                logger.info(f"使用LLM provider: {provider}")
+                if not provider:
+                    logger.error("provider 为空，无法调用 text_chat，直接返回 None")
+                    return None
+                
+                logger.debug(f"LLM provider prompt 长度: {len(prompt) if prompt else 0}")
+                logger.debug(f"LLM provider prompt 前100字符: {prompt[:100] if prompt else 'None'}...")
+                
+                # 检查 prompt 是否为空
+                if not prompt or not prompt.strip():
+                    logger.error("LLM provider: prompt 为空或只包含空白字符，无法调用 text_chat")
+                    return None
+                
+                coro = provider.text_chat(prompt=prompt, max_tokens=max_tokens, temperature=temperature)
+                return await asyncio.wait_for(coro, timeout=timeout)
+        except asyncio.TimeoutError as e:
+            last_exc = e
+            logger.warning(f"LLM请求超时: 第{attempt}次, timeout={timeout}s")
+        except Exception as e:
+            last_exc = e
+            logger.warning(f"LLM请求失败: 第{attempt}次, 错误: {last_exc}")
+        # 若非最后一次，等待退避后重试
+        if attempt < retries:
+            await asyncio.sleep(backoff * attempt)
+    
+    # 最终仍失败，记录错误并返回 None 由调用方处理降级，避免抛出异常
+    logger.error(f"LLM请求全部重试失败: {last_exc}")
+    return None
+
+
+def extract_token_usage(response) -> Optional[dict]:
+    """
+    从LLM响应中提取token使用统计
+    
+    Args:
+        response: LLM响应对象
+        
+    Returns:
+        Token使用统计字典，包含prompt_tokens, completion_tokens, total_tokens
+    """
+    try:
+        token_usage = {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+        
+        # 安全地提取 usage，避免 response.raw_completion.usage 为 None 导致的 AttributeError
+        usage = None
+        if getattr(response, 'raw_completion', None) is not None:
+            usage = getattr(response.raw_completion, 'usage', None)
+            if usage:
+                token_usage["prompt_tokens"] = getattr(usage, 'prompt_tokens', 0) or 0
+                token_usage["completion_tokens"] = getattr(usage, 'completion_tokens', 0) or 0
+                token_usage["total_tokens"] = getattr(usage, 'total_tokens', 0) or 0
+        
+        return token_usage
+        
+    except Exception as e:
+        logger.error(f"提取token使用统计失败: {e}")
+        return {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+
+
+def extract_response_text(response) -> str:
+    """
+    从LLM响应中提取文本内容
+    
+    Args:
+        response: LLM响应对象
+        
+    Returns:
+        响应文本内容
+    """
+    try:
+        if hasattr(response, 'completion_text'):
+            return response.completion_text
+        else:
+            return str(response)
+    except Exception as e:
+        logger.error(f"提取响应文本失败: {e}")
+        return ""

src/reports/generators.py

@@ -11,6 +11,7 @@ from pathlib import Path
 from astrbot.api import logger
 from .templates import HTMLTemplates
 from ..visualization.activity_charts import ActivityVisualizer
+import asyncio
 
 
 class ReportGenerator:
@@ -297,71 +298,204 @@ class ReportGenerator:
             # 尝试启动浏览器，如果 Chromium 不存在会自动下载
             logger.info("启动浏览器进行 PDF 转换")
 
-            # 配置浏览器启动参数，避免 Chromium 下载问题
+            # 配置浏览器启动参数，解决Docker环境中的沙盒问题
             launch_options = {
                 'headless': True,
                 'args': [
-                    '--no-sandbox',
-                    '--disable-setuid-sandbox',
-                    '--disable-dev-shm-usage',
-                    '--disable-gpu',
+                    '--no-sandbox',  # Docker环境必需 - 禁用沙盒
+                    '--disable-setuid-sandbox',  # Docker环境必需 - 禁用setuid沙盒
+                    '--disable-dev-shm-usage',  # 避免共享内存问题
+                    '--disable-gpu',  # 禁用GPU加速
                     '--no-first-run',
                     '--disable-extensions',
-                    '--disable-default-apps'
+                    '--disable-default-apps',
+                    '--disable-background-timer-throttling',
+                    '--disable-backgrounding-occluded-windows',
+                    '--disable-renderer-backgrounding',
+                    '--disable-features=TranslateUI',
+                    '--disable-ipc-flooding-protection',
+                    '--disable-background-networking',
+                    '--enable-features=NetworkService,NetworkServiceInProcess',
+                    '--force-color-profile=srgb',
+                    '--metrics-recording-only',
+                    '--disable-breakpad',
+                    '--disable-component-extensions-with-background-pages',
+                    '--disable-features=Translate,BackForwardCache,AcceptCHFrame,AvoidUnnecessaryBeforeUnloadCheckSync',
+                    '--enable-automation',
+                    '--password-store=basic',
+                    '--use-mock-keychain',
+                    '--export-tagged-pdf',
+                    '--disable-web-security',
+                    '--disable-features=VizDisplayCompositor',
+                    '--disable-blink-features=AutomationControlled',  # 隐藏自动化特征
                 ]
             }
 
-            # 如果是 Windows 系统，尝试使用系统 Chrome
+            # 检测系统 Chrome/Chromium 路径
+            chrome_paths = []
+            
             if sys.platform.startswith('win'):
-                # 常见的 Chrome 安装路径
+                # Windows 系统 Chrome 安装路径
+                username = os.environ.get('USERNAME', '')
                 chrome_paths = [
                     r"C:\Program Files\Google\Chrome\Application\chrome.exe",
                     r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
-                    r"C:\Users\{}\AppData\Local\Google\Chrome\Application\chrome.exe".format(os.environ.get('USERNAME', '')),
+                    rf"C:\Users\{username}\AppData\Local\Google\Chrome\Application\chrome.exe",
+                    r"C:\Program Files\Chromium\Application\chrome.exe",
+                ]
+            elif sys.platform.startswith('linux'):
+                # Linux 系统 Chrome/Chromium 路径
+                chrome_paths = [
+                    '/usr/bin/google-chrome',
+                    '/usr/bin/google-chrome-stable',
+                    '/usr/bin/chromium',
+                    '/usr/bin/chromium-browser',
+                    '/snap/bin/chromium',
+                    '/usr/bin/chromium-freeworld',
+                ]
+            elif sys.platform.startswith('darwin'):
+                # macOS 系统 Chrome 路径
+                chrome_paths = [
+                    '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',
+                    '/Applications/Chromium.app/Contents/MacOS/Chromium',
                 ]
 
-                for chrome_path in chrome_paths:
-                    if Path(chrome_path).exists():
-                        launch_options['executablePath'] = chrome_path
-                        logger.info(f"使用系统 Chrome: {chrome_path}")
-                        break
+            # 查找可用的浏览器
+            found_browser = False
+            for chrome_path in chrome_paths:
+                if Path(chrome_path).exists():
+                    launch_options['executablePath'] = chrome_path
+                    logger.info(f"使用系统浏览器: {chrome_path}")
+                    found_browser = True
+                    break
+            
+            if not found_browser:
+                logger.info("未找到系统浏览器，将使用 pyppeteer 默认下载的 Chromium")
+                # 先尝试确保 Chromium 已下载
+                try:
+                    from pyppeteer import connection, browser, launcher
+                    launcher_instance = launcher.Launcher(
+                        headless=True,
+                        args=['--no-sandbox', '--disable-setuid-sandbox', '--disable-dev-shm-usage']
+                    )
+                    await launcher_instance._get_chromium_revision()
+                    await launcher_instance._download_chromium()
+                    chromium_path = pyppeteer.executablePath()
+                    launch_options['executablePath'] = chromium_path
+                    logger.info(f"使用 pyppeteer 下载的 Chromium: {chromium_path}")
+                except Exception as pre_download_err:
+                    logger.warning(f"预下载 Chromium 失败，继续尝试直接启动: {pre_download_err}")
 
-            browser = await launch(**launch_options)
-            page = await browser.newPage()
-
-            # 设置页面内容 (pyppeteer 1.0.2 版本的 API)
-            await page.setContent(html_content)
-            # 等待页面加载完成
+            # 尝试启动浏览器
             try:
-                await page.waitForSelector('body', {'timeout': 10000})
-            except Exception:
-                # 如果等待失败，继续执行（可能页面已经加载完成）
-                pass
+                logger.info("正在启动浏览器...")
+                browser = await launch(**launch_options)
+                logger.info("浏览器启动成功")
+            except Exception as e:
+                logger.error(f"浏览器启动失败: {e}", exc_info=True)
+                return False
 
-            # 导出 PDF
-            await page.pdf({
-                'path': output_path,
-                'format': 'A4',
-                'printBackground': True,
-                'margin': {
-                    'top': '10mm',
-                    'right': '10mm',
-                    'bottom': '10mm',
-                    'left': '10mm'
-                },
-                'scale': 0.8
-            })
+            try:
+                # 创建新页面，设置更合理的超时时间
+                page = await browser.newPage()
+                
+                # 设置页面视口，减少内存占用
+                await page.setViewport({
+                    'width': 1024,
+                    'height': 768,
+                    'deviceScaleFactor': 1,
+                    'isMobile': False,
+                    'hasTouch': False,
+                    'isLandscape': False
+                })
 
-            await browser.close()
-            logger.info(f"PDF 生成成功: {output_path}")
-            return True
+                # 设置页面内容，使用更安全的加载方式
+                logger.info("开始设置页面内容...")
+                await page.setContent(html_content, {'waitUntil': 'domcontentloaded', 'timeout': 30000})
+                
+                # 等待页面基本加载完成，但不要太长时间
+                try:
+                    await page.waitForSelector('body', {'timeout': 5000})
+                    logger.info("页面基本加载完成")
+                except Exception:
+                    logger.warning("等待页面加载超时，继续执行")
+                
+                # 减少等待时间，避免内存累积
+                await asyncio.sleep(1)
+
+                # 导出 PDF，使用更保守的设置
+                logger.info("开始生成PDF...")
+                pdf_options = {
+                    'path': output_path,
+                    'format': 'A4',
+                    'printBackground': True,
+                    'margin': {
+                        'top': '10mm',
+                        'right': '10mm',
+                        'bottom': '10mm',
+                        'left': '10mm'
+                    },
+                    'scale': 0.8,
+                    'displayHeaderFooter': False,
+                    'preferCSSPageSize': True,
+                    'timeout': 60000  # 增加PDF生成超时时间到60秒
+                }
+                
+                await page.pdf(pdf_options)
+                logger.info(f"PDF 生成成功: {output_path}")
+                return True
+
+            except Exception as e:
+                logger.error(f"PDF生成过程中出错: {e}")
+                return False
+                
+            finally:
+                # 确保浏览器被正确关闭
+                if browser:
+                    try:
+                        logger.info("正在关闭浏览器...")
+                        # 先关闭所有页面
+                        pages = await browser.pages()
+                        for page in pages:
+                            try:
+                                await page.close()
+                            except:
+                                pass
+                        
+                        # 等待一小段时间让资源释放
+                        await asyncio.sleep(0.5)
+                        
+                        # 关闭浏览器
+                        await browser.close()
+                        logger.info("浏览器已关闭")
+                    except Exception as e:
+                        logger.warning(f"关闭浏览器时出错: {e}")
+                        # 强制清理
+                        try:
+                            await browser.disconnect()
+                        except:
+                            pass
 
         except Exception as e:
             error_msg = str(e)
             if "Chromium downloadable not found" in error_msg:
-                logger.error("Chromium 下载失败，建议安装 pyppeteer2 或使用系统 Chrome")
+                logger.error("Chromium 下载失败，建议安装系统 Chrome/Chromium")
+                logger.info("💡 Linux 系统建议: sudo apt-get install chromium-browser 或 sudo yum install chromium")
             elif "No usable sandbox" in error_msg:
                 logger.error("沙盒权限问题，已尝试禁用沙盒")
+            elif "Connection refused" in error_msg or "connect" in error_msg.lower():
+                logger.error("浏览器连接失败，请检查系统资源或尝试重启")
+            elif "executablePath" in error_msg and "not found" in error_msg:
+                logger.error("未找到系统浏览器，请安装 Chrome 或 Chromium")
+                logger.info("💡 安装建议: sudo apt-get install chromium-browser (Ubuntu/Debian) 或 sudo yum install chromium (CentOS/RHEL)")
+            elif "Browser closed unexpectedly" in error_msg:
+                logger.error("浏览器意外关闭，可能是由于内存不足或系统资源限制")
+                logger.info("💡 建议: 检查系统内存，或重启 AstrBot 后重试")
+                logger.info("💡 如果问题持续，可以尝试以下解决方案:")
+                logger.info("   1. 增加系统交换空间")
+                logger.info("   2. 使用更简单的浏览器启动参数")
+                logger.info("   3. 考虑使用其他 PDF 生成方案")
             else:
                 logger.error(f"HTML 转 PDF 失败: {e}")
+                logger.info("💡 可以尝试使用 /安装PDF 命令重新安装依赖，或检查系统日志获取更多信息")
             return False

src/utils/pdf_utils.py

@@ -47,6 +47,187 @@ class PDFInstaller:
             logger.error(f"安装 pyppeteer 时出错: {e}")
             return f"❌ 安装过程中出错: {str(e)}"
 
+    @staticmethod
+    async def install_system_deps():
+        """通过 pyppeteer 自动安装 Chromium"""
+        try:
+            logger.info("正在通过 pyppeteer 自动安装 Chromium...")
+            
+            # 直接通过 pyppeteer 下载 Chromium
+            success = await PDFInstaller._download_chromium_via_pyppeteer()
+            
+            if success:
+                return """✅ Chromium 自动安装成功！
+
+系统依赖已自动配置完成。
+现在可以使用 PDF 功能了。"""
+            else:
+                return """⚠️ 通过 pyppeteer 自动安装 Chromium 失败
+
+请尝试以下方法：
+1. 确保网络连接正常
+2. 检查是否有防火墙或代理限制
+3. 手动运行：path/to/your/actual/sys/executable/python -c "import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())"
+4. 或者手动安装 Chrome/Chromium 浏览器
+
+安装完成后，重启 AstrBot"""
+
+        except Exception as e:
+            logger.error(f"通过 pyppeteer 安装 Chromium 时出错: {e}")
+            return f"❌ 通过 pyppeteer 安装 Chromium 时出错: {str(e)}"
+
+    @staticmethod
+    async def _download_chromium_via_pyppeteer():
+        """通过 pyppeteer 自动下载 Chromium"""
+        try:
+            logger.info("通过 pyppeteer 自动下载 Chromium...")
+            
+            # 导入 pyppeteer 并尝试下载
+            try:
+                import pyppeteer
+                from pyppeteer import launch
+                from pyppeteer.errors import BrowserError
+                
+                # 尝试直接下载 Chromium 而不启动浏览器
+                logger.info("尝试直接下载 Chromium...")
+                try:
+                    # 使用 pyppeteer 的内部下载方法
+                    from pyppeteer.connection import Connection
+                    from pyppeteer.browser import Browser
+                    from pyppeteer.launcher import Launcher
+                    
+                    # 创建 Launcher 实例但不启动浏览器
+                    launcher = Launcher(
+                        headless=True,
+                        args=['--no-sandbox', '--disable-setuid-sandbox']
+                    )
+                    
+                    # 只下载 Chromium
+                    await launcher._get_chromium_revision()
+                    await launcher._download_chromium()
+                    
+                    logger.info("Chromium 下载完成")
+                    return True
+                    
+                except Exception as download_error:
+                    logger.warning(f"直接下载 Chromium 失败，尝试启动浏览器: {download_error}")
+                
+                # 根据操作系统设置不同的参数
+                import platform
+                system = platform.system().lower()
+                
+                if system == "linux":
+                    # Linux 环境下需要更多参数来避免权限问题
+                    browser_args = [
+                        '--no-sandbox',
+                        '--disable-setuid-sandbox',
+                        '--disable-dev-shm-usage',
+                        '--disable-accelerated-2d-canvas',
+                        '--no-first-run',
+                        '--no-zygote',
+                        '--disable-gpu',
+                        '--disable-background-timer-throttling',
+                        '--disable-backgrounding-occluded-windows',
+                        '--disable-renderer-backgrounding',
+                        '--disable-features=TranslateUI',
+                        '--disable-ipc-flooding-protection'
+                    ]
+                else:
+                    # Windows/macOS 环境下的标准参数
+                    browser_args = [
+                        '--no-sandbox',
+                        '--disable-setuid-sandbox',
+                        '--disable-gpu'
+                    ]
+                
+                # 尝试启动浏览器，这会触发自动下载
+                logger.info("启动 pyppeteer 浏览器以触发 Chromium 自动下载...")
+                browser = await launch(
+                    headless=True,
+                    args=browser_args,
+                    ignoreHTTPSErrors=True,
+                    dumpio=True  # 输出浏览器日志用于调试
+                )
+                
+                # 获取 Chromium 路径
+                chromium_path = pyppeteer.executablePath()
+                logger.info(f"Chromium 自动下载完成，路径: {chromium_path}")
+                
+                await browser.close()
+                return True
+                
+            except BrowserError as e:
+                logger.error(f"浏览器错误: {e}", exc_info=True)
+                
+                # 备用方法：使用命令行触发下载
+                try:
+                    logger.info("尝试使用命令行触发 Chromium 自动下载...")
+                    
+                    # 根据操作系统设置不同的命令
+                    import platform
+                    system = platform.system().lower()
+                    
+                    if system == "linux":
+                        cmd = [
+                            sys.executable, "-c",
+                            """
+import pyppeteer
+import asyncio
+import platform
+
+async def download_chrome():
+    try:
+        browser = await pyppeteer.launch(
+            headless=True,
+            args=[
+                '--no-sandbox',
+                '--disable-setuid-sandbox',
+                '--disable-dev-shm-usage',
+                '--disable-gpu',
+                '--single-process'
+            ]
+        )
+        await browser.close()
+        print("Chromium 下载成功")
+    except Exception as e:
+        print(f"下载失败: {e}")
+        raise
+
+asyncio.run(download_chrome())
+                            """
+                        ]
+                    else:
+                        cmd = [
+                            sys.executable, "-c",
+                            "import pyppeteer; import asyncio; asyncio.run(pyppeteer.launch())"
+                        ]
+                    
+                    process = await asyncio.create_subprocess_exec(
+                        *cmd,
+                        stdout=asyncio.subprocess.PIPE,
+                        stderr=asyncio.subprocess.PIPE
+                    )
+                    
+                    stdout, stderr = await process.communicate()
+                    
+                    if process.returncode == 0:
+                        logger.info("成功通过命令行触发 Chromium 自动下载")
+                        return True
+                    else:
+                        logger.error(f"命令行触发自动下载失败: {stderr.decode()}")
+                        
+                        # 最后的备用方案：手动下载
+                        logger.info("尝试手动下载 Chromium...")
+                        return False
+                        
+                except Exception as e2:
+                    logger.error(f"命令行触发自动下载也失败: {e2}")
+                    return False
+                    
+        except Exception as e:
+            logger.error(f"通过 pyppeteer 自动下载 Chromium 时出错: {e}", exc_info=True)
+            return False
+
     @staticmethod
     def get_pdf_status(config_manager) -> str:
         """获取PDF功能状态"""