fix(docs): 完善注释

src/utils/helpers.py

@@ -4,6 +4,7 @@
 """
 
 import asyncio
+from typing import Any
 
 from ..analysis.llm_analyzer import LLMAnalyzer
 from ..analysis.statistics import UserAnalyzer
@@ -13,9 +14,32 @@ from .logger import logger
 
 
 class MessageAnalyzer:
-    """消息分析器 - 整合所有分析功能"""
+    """
+    业务逻辑：消息分析整合器
 
-    def __init__(self, context, config_manager, bot_manager=None):
+    该类作为一个门面（Facade），将消息存储、统计计算、LLM 智能分析以及用户画像分析
+    等多个底层组件整合在一起，提供统一的消息分析流程接口。
+
+    Attributes:
+        context (Any): AstrBot 上下文环境
+        config_manager (Any): 配置管理者实例
+        bot_manager (Any, optional): 机器人多实例管理者
+        message_handler (MessageHandler): 负责消息过滤和基础统计
+        llm_analyzer (LLMAnalyzer): 负责调用大模型进行语义分析
+        user_analyzer (UserAnalyzer): 负责用户活跃度及角色分析
+    """
+
+    def __init__(
+        self, context: Any, config_manager: Any, bot_manager: Any | None = None
+    ):
+        """
+        初始化消息分析器。
+
+        Args:
+            context (Any): AstrBot 核心上下文
+            config_manager (Any): 插件配置管理器
+            bot_manager (Any, optional): 多平台机器人管理器实例
+        """
         self.context = context
         self.config_manager = config_manager
         self.bot_manager = bot_manager
@@ -23,71 +47,91 @@ class MessageAnalyzer:
         self.llm_analyzer = LLMAnalyzer(context, config_manager)
         self.user_analyzer = UserAnalyzer(config_manager)
 
-    def _extract_bot_self_id_from_instance(self, bot_instance):
-        """从bot实例中提取ID（单个）"""
+    def _extract_bot_self_id_from_instance(self, bot_instance: Any) -> str | None:
+        """
+        内部方法：从不同平台的机器人实例中探测其自身 ID。
+
+        Args:
+            bot_instance (Any): 宿主机器人实例 (如 OneBot, Discord 实例)
+
+        Returns:
+            str | None: 探测到的用户 ID 或 None
+        """
         if hasattr(bot_instance, "self_id") and bot_instance.self_id:
             return str(bot_instance.self_id)
-        elif hasattr(bot_instance, "qq") and bot_instance.qq:
-            return str(bot_instance.qq)
         elif hasattr(bot_instance, "user_id") and bot_instance.user_id:
             return str(bot_instance.user_id)
         return None
 
-    def _extract_bot_qq_id_from_instance(self, bot_instance):
-        """从bot实例中提取QQ号（已弃用）"""
-        return self._extract_bot_self_id_from_instance(bot_instance)
+    async def set_bot_instance(
+        self, bot_instance: Any, platform_id: str | None = None
+    ) -> None:
+        """
+        向分析组件注入当前活跃的机器人实例。
 
-    async def set_bot_instance(self, bot_instance, platform_id=None):
-        """设置bot实例（保持向后兼容）"""
+        Args:
+            bot_instance (Any): 活跃的机器人 SDK 实例
+            platform_id (str, optional): 平台标识符，用于多实例路由
+        """
         if self.bot_manager:
             self.bot_manager.set_bot_instance(bot_instance, platform_id)
         else:
-            # 从bot实例提取ID并设置为列表
+            # 降级逻辑：仅设置单个默认 ID
             bot_self_id = self._extract_bot_self_id_from_instance(bot_instance)
             if bot_self_id:
-                # 将单个ID转换为列表，保持统一处理
                 await self.message_handler.set_bot_self_ids([bot_self_id])
 
     async def analyze_messages(
-        self, messages: list[dict], group_id: str, unified_msg_origin: str = None
-    ) -> dict:
-        """完整的消息分析流程"""
+        self, messages: list[dict], group_id: str, unified_msg_origin: str | None = None
+    ) -> dict | None:
+        """
+        执行完整的群消息流水化分析。
+
+        包含：消息预处理 -> 词频统计 -> 活跃用户识别 -> LLM 摘要/金句提取。
+
+        Args:
+            messages (list[dict]): 待处理的原始或统一格式消息字典列表
+            group_id (str): 群组 ID，用于上下文标识
+            unified_msg_origin (str, optional): 统一消息来源标识
+
+        Returns:
+            dict | None: 包含 statistics, topics, user_titles, user_analysis 的字典，失败返回 None
+        """
         try:
-            # 基础统计
+            # 1. 基础消息统计 (耗时操作，放入线程池避免阻塞事件循环)
             statistics = await asyncio.to_thread(
                 self.message_handler.calculate_statistics, messages
             )
 
-            # 用户分析
+            # 2. 用户维度分析 (等级、发言习惯等)
             user_analysis = await asyncio.to_thread(
                 self.user_analyzer.analyze_users, messages
             )
 
-            # 获取活跃用户列表 - 使用get_top_users方法,limit从配置中读取
+            # 3. 筛选分析范围：提取 Top N 活跃用户用于深度称号分析
             max_user_titles = self.config_manager.get_max_user_titles()
             top_users = self.user_analyzer.get_top_users(
                 user_analysis, limit=max_user_titles
             )
             logger.info(
-                f"获取到 {len(top_users)} 个活跃用户用于称号分析(配置上限: {max_user_titles})"
+                f"已为称号分析筛选出 {len(top_users)} 名活跃用户 (最大限制: {max_user_titles})"
             )
 
-            # LLM分析 - 使用并发方式
+            # 4. LLM 语义分析阶段
             topics = []
             user_titles = []
             golden_quotes = []
             total_token_usage = TokenUsage()
 
-            # 检查各个分析功能是否启用
+            # 检查开关设置
             topic_enabled = self.config_manager.get_topic_analysis_enabled()
             user_title_enabled = self.config_manager.get_user_title_analysis_enabled()
             golden_quote_enabled = (
                 self.config_manager.get_golden_quote_analysis_enabled()
             )
 
-            # 如果三个分析都启用，使用并发执行
+            # 策略：如果多项功能均开启，则通过 LLMAnalyzer 并发调用，显著降低分析总时长
             if topic_enabled and user_title_enabled and golden_quote_enabled:
-                # 并发执行所有三个分析任务，传入活跃用户列表
                 (
                     topics,
                     user_titles,
@@ -97,7 +141,7 @@ class MessageAnalyzer:
                     messages, user_analysis, umo=unified_msg_origin, top_users=top_users
                 )
             else:
-                # 如果只启用部分分析，则按需执行
+                # 串行降级路径：根据开关按需串行调用 (适用于 Token 敏感或单项测试)
                 if topic_enabled:
                     topics, topic_tokens = await self.llm_analyzer.analyze_topics(
                         messages, umo=unified_msg_origin
@@ -109,7 +153,6 @@ class MessageAnalyzer:
                     total_token_usage.total_tokens += topic_tokens.total_tokens
 
                 if user_title_enabled:
-                    # 传入活跃用户列表
                     (
                         user_titles,
                         title_tokens,
@@ -138,7 +181,7 @@ class MessageAnalyzer:
                     )
                     total_token_usage.total_tokens += quote_tokens.total_tokens
 
-            # 更新统计数据
+            # 5. 回填分析结果并组装返回字典
             statistics.golden_quotes = golden_quotes
             statistics.token_usage = total_token_usage
 
@@ -150,5 +193,5 @@ class MessageAnalyzer:
             }
 
         except Exception as e:
-            logger.error(f"消息分析失败: {e}")
+            logger.error(f"消息分析流水线执行失败: {e}")
             return None

src/utils/logger.py

@@ -1,15 +1,28 @@
 import logging
+from typing import Any
 
 from astrbot.api import logger as astrbot_logger
 
 
 class PluginLoggerAdapter(logging.LoggerAdapter):
     """
-    插件日志适配器
-    自动为日志添加 [QQ群分析] 前缀，方便区分
+    日志适配器：插件级统一日志装饰器
+
+    自动向所有通过该实例输出的日志信息前缀添加 `[QQ群分析]` 标签，
+    以便用户在 AstrBot 混合日志流中快速定位属于本插件的输出。
     """
 
-    def process(self, msg, kwargs):
+    def process(self, msg: str, kwargs: Any) -> tuple[str, Any]:
+        """
+        加工日志消息，注入插件专有前缀。
+
+        Args:
+            msg (str): 原始日志消息
+            kwargs (Any): 额外的日志参数映射
+
+        Returns:
+            tuple[str, Any]: (格式化后的消息, 参数)
+        """
         return f"[QQ群分析] {msg}", kwargs
 
 

src/utils/pdf_utils.py

@@ -6,18 +6,26 @@ PDF工具模块
 import asyncio
 import sys
 from concurrent.futures import ThreadPoolExecutor
+from typing import Any
 
 from .logger import logger
 
 
 class PDFInstaller:
-    """PDF功能安装器"""
+    """
+    工具组件：PDF 渲染引擎 (Playwright) 安装器
 
-    # 类级别的线程池，用于异步下载任务
+    该组件负责管理 Playwright 及其对应浏览器内核 (Chromium) 的安装生命周期。
+    由于内核下载耗时较长且受网络波动影响，采用非阻塞的后台任务模式执行。
+    """
+
+    # 类级别的线程池，专用于隔离耗时的 IO/Shell 操作
     _executor = ThreadPoolExecutor(
         max_workers=1, thread_name_prefix="playwright_install"
     )
-    _install_status = {
+
+    # 静态安装状态追踪
+    _install_status: dict[str, Any] = {
         "in_progress": False,
         "completed": False,
         "failed": False,
@@ -25,13 +33,26 @@ class PDFInstaller:
     }
 
     @staticmethod
-    async def install_playwright(config_manager):
-        """安装 Playwright 依赖"""
-        try:
-            logger.info("开始安装 Playwright...")
+    async def install_playwright(config_manager: Any) -> str:
+        """
+        异步入口：安装 Playwright 环境。
 
-            # 1. 安装 pip 包
-            logger.info("正在运行 pip install playwright...")
+        流程：
+        1. 调用 pip 安装 `playwright` Python 包。
+        2. 验证自定义浏览器路径配置。
+        3. 若无自定义路径，则触发浏览器内核安装。
+
+        Args:
+            config_manager (Any): 配置管理实例，用于读取/设置安装状态。
+
+        Returns:
+            str: 安装阶段提示信息
+        """
+        try:
+            logger.info("正在初始化 Playwright 安装流程...")
+
+            # 1. 下载并安装库文件
+            logger.info("第一步：正在运行 pip install playwright...")
             process = await asyncio.create_subprocess_exec(
                 sys.executable,
                 "-m",
@@ -45,52 +66,58 @@ class PDFInstaller:
             stdout, stderr = await process.communicate()
 
             if process.returncode != 0:
-                error_msg = stderr.decode()
-                logger.error(f"playwright pip 安装失败: {error_msg}")
+                error_msg = stderr.decode().strip()
+                logger.error(f"Playwright 库安装失败: {error_msg}")
                 return f"❌ pip install playwright 失败: {error_msg}"
 
-            logger.info("pip 包安装成功，检查是否需要安装浏览器内核...")
+            logger.info("第一步完成。正在检查浏览器内核...")
 
-            # 2. 检查自定义路径
+            # 2. 检查自定义路径：若用户已手动提供内核，则跳过下载步骤
             from pathlib import Path
 
             custom_path = config_manager.get_browser_path()
             if custom_path and Path(custom_path).exists():
-                logger.info(
-                    f"检测到自定义浏览器路径: {custom_path}，将跳过 Chromium 内核安装。"
-                )
-                return f"✅ Playwright 包安装成功。检测到自定义浏览器路径 `{custom_path}`，已跳过浏览器内核安装。您可以现在尝试生成 PDF。"
+                logger.info(f"检测到自定义浏览器路径: {custom_path}。跳过内核下载。")
+                return f"✅ Playwright 库已就绪。已检测到自定义浏览器 `{custom_path}`，无需额外安装内核。您可以直接开始生成 PDF。"
 
-            # 3. 安装浏览器内核
+            # 3. 部署浏览器内核
             return await PDFInstaller.install_system_deps()
 
         except Exception as e:
-            logger.error(f"安装 playwright 时出错: {e}")
+            logger.error(f"Playwright 设置过程中出错: {e}")
             return f"❌ 安装过程中出错: {str(e)}"
 
     @staticmethod
-    async def install_system_deps():
-        """安装系统依赖 (运行 playwright install chromium)"""
+    async def install_system_deps() -> str:
+        """
+        触发浏览器内核的后台异步安装流程。
+
+        该方法检查防重入状态，并立即返回任务启动信息，不会阻塞主线程。
+
+        Returns:
+            str: 任务排队状态提示
+        """
         try:
-            # 检查是否已经在安装中
             if PDFInstaller._install_status["in_progress"]:
-                return "⏳ 浏览器内核正在后台安装中，请稍候..."
+                return "⏳ 浏览器内核正在后台部署中，请稍后检查日志或状态。"
 
-            PDFInstaller._install_status["in_progress"] = True
-            PDFInstaller._install_status["completed"] = False
-            PDFInstaller._install_status["failed"] = False
-            PDFInstaller._install_status["error_message"] = None
+            PDFInstaller._install_status.update(
+                {
+                    "in_progress": True,
+                    "completed": False,
+                    "failed": False,
+                    "error_message": None,
+                }
+            )
 
-            logger.info("启动后台任务安装 Chromium...")
+            logger.info("正在启动后台线程以部署 Chromium 内核...")
             asyncio.create_task(PDFInstaller._background_playwright_install())
 
-            return """🚀 浏览器内核安装任务已启动
-
-正在运行 `playwright install chromium`...
-这可能需要几分钟时间，取决于网络速度。
-安装过程不会阻塞 Bot 的正常运行。
-下载完成后平台日志会显示安装完成的日志。
-"""
+            return (
+                "🚀 浏览器内核安装任务已成功在后台启动。\n\n"
+                "程序正在执行 `playwright install chromium`，由于体积较大，通常需花费 2-5 分钟。\n"
+                "此过程不会影响机器人正常响应。安装完成后，系统日志将进行通知。"
+            )
 
         except Exception as e:
             PDFInstaller._install_status["in_progress"] = False
@@ -98,13 +125,14 @@ class PDFInstaller:
             return f"❌ 启动安装任务失败: {e}"
 
     @staticmethod
-    async def _background_playwright_install():
-        """后台运行 playwright install"""
+    async def _background_playwright_install() -> None:
+        """
+        底层宿主任务：驱动系统 shell 执行浏览器二进制文件部署。
+        """
         try:
-            logger.info("开始运行 playwright install chromium...")
+            logger.info("正在执行二进制文件：playwright install chromium")
 
-            # 使用 shell 命令确保能找到 path 中的 playwright
-            # 或者使用 python -m playwright install chromium
+            # 通过当前 Python 解释器环境调用子模块，确保环境隔离
             process = await asyncio.create_subprocess_exec(
                 sys.executable,
                 "-m",
@@ -115,48 +143,50 @@ class PDFInstaller:
                 stderr=asyncio.subprocess.PIPE,
             )
 
-            # 等待完成，设置较长的超时
             stdout, stderr = await process.communicate()
 
             if process.returncode == 0:
                 PDFInstaller._install_status["completed"] = True
-                logger.info(f"✅ Playwright Chromium 安装成功: {stdout.decode()}")
+                logger.info("✅ Chromium 内核安装成功。")
 
-                # 尝试安装系统依赖 (Linux only，通常不需要 root 无法执行，但尝试一下无妨或者提示用户)
+                # Linux 特殊处理：提示用户补充系统依赖
                 if sys.platform.startswith("linux"):
-                    logger.info("正在尝试安装系统依赖 (install-deps)...")
-                    # 无需 await 阻塞太久，这步通常需要 sudo，可能会失败，仅做尝试或提示
-                    # 真正的系统依赖安装通常由 Dockerfile 或用户手动完成
-                    # 这里我们仅记录日志建议
                     logger.info(
-                        "💡 如果 Linux 下仍无法生成 PDF，请尝试运行: sudo playwright install-deps"
+                        "提示：在 Linux 上，如果 PDF 生成仍然失败，请尝试运行 'sudo playwright install-deps'。"
                     )
-
             else:
                 PDFInstaller._install_status["failed"] = True
-                PDFInstaller._install_status["error_message"] = stderr.decode()
-                logger.error(f"❌ Playwright Chromium 安装失败: {stderr.decode()}")
+                PDFInstaller._install_status["error_message"] = stderr.decode().strip()
+                logger.error(f"❌ Chromium 安装二进制文件执行失败: {stderr.decode()}")
 
         except Exception as e:
-            PDFInstaller._install_status["failed"] = True
-            PDFInstaller._install_status["error_message"] = str(e)
-            logger.error(f"Playwright 安装后台任务出错: {e}")
+            PDFInstaller._install_status.update(
+                {"failed": True, "error_message": str(e)}
+            )
+            logger.error(f"Playwright 后台任务遇到异常: {e}")
         finally:
             PDFInstaller._install_status["in_progress"] = False
 
     @staticmethod
-    def get_pdf_status(config_manager) -> str:
-        """获取PDF功能状态"""
-        if config_manager.playwright_available:
-            version = config_manager.playwright_version or "未知版本"
+    def get_pdf_status(config_manager: Any) -> str:
+        """
+        查询当前系统的 PDF 功能可用性状态描述。
 
-            status = f"✅ PDF 功能可用 (playwright {version})"
+        Args:
+            config_manager (Any): 配置管理器，用于读取核心探测开关。
+
+        Returns:
+            str: 用户友好的状态文本
+        """
+        if config_manager.playwright_available:
+            version = config_manager.playwright_version or "Unknown"
+            status = f"✅ PDF 功能可用 (核心版本: {version})"
 
             if PDFInstaller._install_status["in_progress"]:
-                status += "\n⏳ 正在后台安装浏览器内核..."
+                status += "\n⏳ 警告：浏览器内核仍在后台下载/部署中..."
             elif PDFInstaller._install_status["failed"]:
-                status += f"\n❌ 上次浏览器安装失败: {PDFInstaller._install_status.get('error_message', '未知错误')}"
+                status += f"\n⚠️ 上次内核安装异常: {PDFInstaller._install_status.get('error_message')}"
 
             return status
         else:
-            return "❌ PDF 功能不可用 - 请输入 /安装PDF 进行安装"
+            return "❌ PDF 渲染核心未安装 - 请发送管理员指令 `/安装PDF`。"

src/utils/resilience.py

@@ -6,7 +6,15 @@ from .logger import logger
 
 class CircuitBreaker:
     """
-    简单的熔断器实现 (Simple Circuit Breaker)
+    韧性设计：熔断器 (Circuit Breaker)
+
+    用于监控外部服务（如 LLM API）的调用状态。当错误率达到阈值时，自动开启熔断，
+    拦截对故障服务的进一步请求，保护系统不被连锁故障拖累，直到服务窗口恢复。
+
+    States:
+        CLOSED: 正常工作状态，允许请求
+        OPEN: 熔断状态，拒绝请求
+        HALF_OPEN: 尝试恢复状态，允许少量测试请求
     """
 
     STATE_CLOSED = "CLOSED"
@@ -19,16 +27,24 @@ class CircuitBreaker:
         recovery_timeout: int = 60,
         name: str = "default",
     ):
+        """
+        初始化熔断器。
+
+        Args:
+            failure_threshold (int): 连续失败触发熔断的次数上限
+            recovery_timeout (int): 熔断开启后尝试恢复之前的冷却时间（秒）
+            name (str): 熔断器标识符（用于日志区分）
+        """
         self.name = name
         self.failure_threshold = failure_threshold
         self.recovery_timeout = recovery_timeout
 
         self.failure_count = 0
         self.state = self.STATE_CLOSED
-        self.last_failure_time = 0
+        self.last_failure_time = 0.0
 
-    def record_failure(self):
-        """记录一次失败"""
+    def record_failure(self) -> None:
+        """记录一次调用失败，并根据阈值决定是否切换到 OPEN 状态。"""
         self.failure_count += 1
         if (
             self.state == self.STATE_CLOSED
@@ -36,66 +52,91 @@ class CircuitBreaker:
         ):
             self._open_circuit()
         elif self.state == self.STATE_HALF_OPEN:
-            # 在半开状态下，一次失败直接重新打开熔断器
+            # 半开状态下任何一次失败都将立即导致熔断重开
             self._open_circuit()
 
-    def record_success(self):
-        """记录一次成功"""
+    def record_success(self) -> None:
+        """记录一次调用成功，并尝试重置或关闭熔断器。"""
         if self.state == self.STATE_HALF_OPEN:
             self._close_circuit()
         elif self.state == self.STATE_CLOSED:
-            # 成功则重置失败计数 (可选，这里选择连续失败才熔断)
+            # 正常状态下的成功重置累积计数值
             self.failure_count = 0
 
     def allow_request(self) -> bool:
-        """是否允许请求"""
+        """
+        判断是否允许本次服务请求。
+
+        Returns:
+            bool: True 为允许，False 为拦截
+        """
         if self.state == self.STATE_OPEN:
+            # 检查冷却时间是否已过，过则进入试探性的半开状态
             if time.time() - self.last_failure_time > self.recovery_timeout:
                 self._half_open_circuit()
                 return True
             return False
         return True
 
-    def _open_circuit(self):
+    def _open_circuit(self) -> None:
+        """动作：开启熔断"""
         self.state = self.STATE_OPEN
         self.last_failure_time = time.time()
         logger.warning(
-            f"CircuitBreaker[{self.name}] 熔断器已打开! 暂停请求 {self.recovery_timeout} 秒。"
+            f"熔断器 CircuitBreaker[{self.name}] 已激活！将拦截请求 {self.recovery_timeout} 秒。"
         )
 
-    def _close_circuit(self):
+    def _close_circuit(self) -> None:
+        """动作：关闭熔断，恢复常态"""
         self.state = self.STATE_CLOSED
         self.failure_count = 0
-        logger.info(f"CircuitBreaker[{self.name}] 熔断器已关闭，服务恢复。")
+        logger.info(f"熔断器 CircuitBreaker[{self.name}] 已恢复至关闭 (CLOSED) 状态。")
 
-    def _half_open_circuit(self):
+    def _half_open_circuit(self) -> None:
+        """动作：进入半开状态"""
         self.state = self.STATE_HALF_OPEN
-        logger.info(f"CircuitBreaker[{self.name}] 进入半开状态，尝试恢复...")
+        logger.info(
+            f"熔断器 CircuitBreaker[{self.name}] 进入半开 (HALF_OPEN) 测试模式。"
+        )
 
 
 class GlobalRateLimiter:
     """
-    全局限流器 (Global Rate Limiter)
-    使用 asyncio.Semaphore 控制并发数
+    韧性设计：全局并发动态限流器
+
+    基于单例模式管理 asyncio.Semaphore，确保在插件内的异步任务
+    不会超过设定的最大并发限制（如保护 LLM 账单或避免 API 拥塞）。
     """
 
-    _instance = None
-    _semaphore = None
+    _instance: "GlobalRateLimiter | None" = None
+    _semaphore: asyncio.Semaphore | None = None
 
     @classmethod
-    def get_instance(cls, max_concurrency: int = 3):
+    def get_instance(cls, max_concurrency: int = 3) -> "GlobalRateLimiter":
+        """
+        获取或创建限流器单例。
+
+        Args:
+            max_concurrency (int): 允许的最大并发行数
+
+        Returns:
+            GlobalRateLimiter: 唯一实例
+        """
         if cls._instance is None:
             cls._instance = cls()
             cls._semaphore = asyncio.Semaphore(max_concurrency)
         return cls._instance
 
     @property
-    def semaphore(self):
+    def semaphore(self) -> asyncio.Semaphore:
+        """返回核心的异步信号量对象。"""
         if self._semaphore is None:
-            # Fallback if accessed before get_instance called with arg
+            # 兜底：若直接通过属性访问则初始化默认值
             self._semaphore = asyncio.Semaphore(3)
         return self._semaphore
 
 
-# 默认全局限流实例
-global_llm_rate_limiter = GlobalRateLimiter.get_instance(max_concurrency=3).semaphore
+# 导出默认实例：用于 LLM 调用的全局限流
+global_llm_rate_limiter: asyncio.Semaphore = GlobalRateLimiter.get_instance(
+    max_concurrency=3
+).semaphore

src/utils/trace_context.py

@@ -2,6 +2,7 @@ import contextvars
 import logging
 import time
 import uuid
+from typing import Any
 
 # 定义 ContextVar
 _trace_id_ctx = contextvars.ContextVar("trace_id", default="")
@@ -9,22 +10,48 @@ _trace_id_ctx = contextvars.ContextVar("trace_id", default="")
 
 class TraceContext:
     """
-    链路追踪上下文管理器
+    链路追踪：追踪上下文管理者
+
+    利用 `contextvars` 在异步任务流中传递全局唯一的 `trace_id`，
+    实现对单一请求/分析任务的全流程日志记录追踪。
     """
 
     @staticmethod
-    def set(trace_id: str):
-        """设置当前上下文的 TraceID"""
+    def set(trace_id: str) -> Any:
+        """
+        设置当前异步上下文的 TraceID。
+
+        Args:
+            trace_id (str): 追踪 ID 字符串
+
+        Returns:
+            Token: contextvars 令牌，用于后续重置
+        """
         return _trace_id_ctx.set(trace_id)
 
     @staticmethod
     def get() -> str:
-        """获取当前上下文的 TraceID"""
+        """
+        获取当前异步上下文中的 TraceID。
+
+        Returns:
+            str: 当前任务的追踪 ID，若无则返回空字符串
+        """
         return _trace_id_ctx.get()
 
     @staticmethod
     def generate(prefix: str = "") -> str:
-        """生成一个新的 TraceID (Prefix + Timestamp + UUID前8位)"""
+        """
+        构建生成一个新的高辨识度 TraceID。
+
+        格式：[prefix-]时间戳-UUID前8位
+
+        Args:
+            prefix (str, optional): ID 前缀 (如 'ANALYSIS')
+
+        Returns:
+            str: 生成的追踪 ID
+        """
         timestamp = int(time.time())
         unique_id = str(uuid.uuid4())[:8]
         if prefix:
@@ -32,20 +59,34 @@ class TraceContext:
         return f"{timestamp}-{unique_id}"
 
     @staticmethod
-    def clear():
-        """清除当前上下文的 TraceID"""
+    def clear() -> None:
+        """
+        重置/清除当前上下文的 TraceID 记录。
+        """
         _trace_id_ctx.set("")
 
 
 class TraceLogFilter(logging.Filter):
     """
-    日志过滤器，自动注入 TraceID
+    日志治理：TraceID 注入过滤器
+
+    该过滤器被挂载到日志系统后，会自动从流水上下文中提取 `trace_id`
+    并注入到每一条日志记录中，便于日后通过 ID 检索完整的任务执行链路。
     """
 
-    def filter(self, record):
+    def filter(self, record: logging.LogRecord) -> bool:
+        """
+        拦截日志记录进行 TraceID 动态修饰。
+
+        Args:
+            record (logging.LogRecord): 日志记录对象
+
+        Returns:
+            bool: 始终返回 True (仅修改不拦截)
+        """
         trace_id = _trace_id_ctx.get()
         if trace_id:
-            # 将 trace_id 注入到 record 中，同时也修改 msg 以便在不支持自定义 format 的 logger 中也能看到
+            # 同时注入属性和修饰消息文本，保证在简易日志格式下也能直接可见
             record.trace_id = trace_id
             record.msg = f"[{trace_id}] {record.msg}"
         else: