From b4abee55d42deeb03e36f0a5a2e2b37b7b997c44 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Thu, 12 Mar 2026 14:51:59 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20TraceContext=20?= =?UTF-8?q?=E5=85=BC=E5=AE=B9=E6=80=A7=E4=B8=8E=E6=9C=8D=E5=8A=A1=E8=BF=94?= =?UTF-8?q?=E5=9B=9E=E7=BB=93=E6=9E=9C=E7=BC=BA=E5=A4=B1=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E7=9A=84=E5=B4=A9=E6=BA=83=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【链路追踪系统】 - 在 TraceContext 中补全 generate()、set() 和 get() 兼容性接口，确保旧代码无缝迁移。 - 重新实现 TraceLogFilter 日志过滤器，恢复日志流中的 TraceID 显示。 - 统一全插件 import 路径至 src/shared/trace_context.py，并删除冗余的旧文件。 【应用服务层】 - 修复 KeyError: 'group_id'：在 execute_daily_analysis 和 execute_incremental_final_report 的返回结果中补全 group_id 和 platform_id。 - 完善 execute_incremental_analysis 的返回字段，保持 API 返回格式的一致性。 【基础设施与工具】 - 在 AutoScheduler 中增加对适配器获取失败的防御性检查。 - 修复 uto_scheduler.py 和 dispatcher.py 中的静态类型检查报错。 --- main.py | 2 +- .../services/analysis_application_service.py | 2 + src/infrastructure/reporting/dispatcher.py | 2 +- .../scheduler/auto_scheduler.py | 2 +- src/shared/trace_context.py | 47 ++++++++++ src/utils/trace_context.py | 94 ------------------- 6 files changed, 52 insertions(+), 97 deletions(-) delete mode 100644 src/utils/trace_context.py diff --git a/main.py b/main.py index d409734..77f5ff1 100644 --- a/main.py +++ b/main.py @@ -45,9 +45,9 @@ from .src.infrastructure.platform.template_preview import ( from .src.infrastructure.reporting.generators import ReportGenerator from .src.infrastructure.scheduler.auto_scheduler import AutoScheduler from .src.infrastructure.scheduler.retry import RetryManager +from .src.shared.trace_context import TraceContext, TraceLogFilter from .src.utils.logger import logger from .src.utils.pdf_utils import PDFInstaller -from .src.utils.trace_context import TraceContext, TraceLogFilter class GroupDailyAnalysis(Star): diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index 004e656..94a19fa 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -462,6 +462,8 @@ class AnalysisApplicationService: "success": True, "batch_summary": batch.get_summary(), "messages_count": len(unified_messages), + "group_id": group_id, + "platform_id": getattr(adapter, "platform_id", platform_id), } async def execute_incremental_final_report( diff --git a/src/infrastructure/reporting/dispatcher.py b/src/infrastructure/reporting/dispatcher.py index a1142c3..f65ca6b 100644 --- a/src/infrastructure/reporting/dispatcher.py +++ b/src/infrastructure/reporting/dispatcher.py @@ -5,8 +5,8 @@ from collections.abc import Callable from datetime import datetime from typing import Any +from ...shared.trace_context import TraceContext from ...utils.logger import logger -from ...utils.trace_context import TraceContext class ReportDispatcher: diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index 10f4fa1..c65f733 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -10,8 +10,8 @@ from typing import Any from apscheduler.triggers.cron import CronTrigger from ...application.services.analysis_application_service import DuplicateGroupTaskError +from ...shared.trace_context import TraceContext from ...utils.logger import logger -from ...utils.trace_context import TraceContext from ..messaging.message_sender import MessageSender from ..platform.factory import PlatformAdapterFactory from ..reporting.dispatcher import ReportDispatcher diff --git a/src/shared/trace_context.py b/src/shared/trace_context.py index 3d15a8c..727ef90 100644 --- a/src/shared/trace_context.py +++ b/src/shared/trace_context.py @@ -5,6 +5,7 @@ """ import functools +import logging import uuid from contextvars import ContextVar, Token from dataclasses import dataclass, field @@ -144,6 +145,52 @@ class TraceContext: new_ctx._token = _current_trace.set(new_ctx) return new_ctx + @staticmethod + def generate(prefix: str = "") -> str: + """ + [兼容性接口] 生成一个带前缀的唯一追踪 ID。 + + Args: + prefix (str): 前缀，如 'manual_12345' + + Returns: + str: 格式为 'prefix-uuid' 的字符串 + """ + uid = str(uuid.uuid4())[:8] + return f"{prefix}-{uid}" if prefix else uid + + @classmethod + def set(cls, trace_id: str) -> None: + """ + [兼容性接口] 直接设置当前上下文的 TraceID。 + 这会创建一个新的 TraceContext 实例并将其推入 ContextVar。 + + Args: + trace_id (str): 要设置的追踪 ID 字符串 + """ + ctx = cls(trace_id=trace_id) + # 注意：此处不手动存储 Token，依靠异步任务结束时 ContextVar 的自动清理。 + _current_trace.set(ctx) + + @classmethod + def get(cls) -> str: + """ + [兼容性接口] 获取当前活跃的追踪 ID 字符串。 + """ + return get_trace_id() + + +class TraceLogFilter(logging.Filter): + """ + 日志过滤器：自动将当前的 TraceID 注入每一条日志记录中。 + + 配合日志格式化字符串 `[%(trace_id)s]` 使用。 + """ + + def filter(self, record: logging.LogRecord) -> bool: + record.trace_id = get_trace_id() + return True + def get_trace_id() -> str: """ diff --git a/src/utils/trace_context.py b/src/utils/trace_context.py deleted file mode 100644 index 6a58d62..0000000 --- a/src/utils/trace_context.py +++ /dev/null @@ -1,94 +0,0 @@ -import contextvars -import logging -import time -import uuid -from typing import Any - -# 定义 ContextVar -_trace_id_ctx = contextvars.ContextVar("trace_id", default="") - - -class TraceContext: - """ - 链路追踪：追踪上下文管理者 - - 利用 `contextvars` 在异步任务流中传递全局唯一的 `trace_id`， - 实现对单一请求/分析任务的全流程日志记录追踪。 - """ - - @staticmethod - 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。 - - Returns: - str: 当前任务的追踪 ID，若无则返回空字符串 - """ - return _trace_id_ctx.get() - - @staticmethod - def generate(prefix: str = "") -> str: - """ - 构建生成一个新的高辨识度 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: - return f"{prefix}-{timestamp}-{unique_id}" - return f"{timestamp}-{unique_id}" - - @staticmethod - def clear() -> None: - """ - 重置/清除当前上下文的 TraceID 记录。 - """ - _trace_id_ctx.set("") - - -class TraceLogFilter(logging.Filter): - """ - 日志治理：TraceID 注入过滤器 - - 该过滤器被挂载到日志系统后，会自动从流水上下文中提取 `trace_id` - 并注入到每一条日志记录中，便于日后通过 ID 检索完整的任务执行链路。 - """ - - def filter(self, record: logging.LogRecord) -> bool: - """ - 拦截日志记录进行 TraceID 动态修饰。 - - Args: - record (logging.LogRecord): 日志记录对象 - - Returns: - bool: 始终返回 True (仅修改不拦截) - """ - trace_id = _trace_id_ctx.get() - if trace_id: - # 同时注入属性和修饰消息文本，保证在简易日志格式下也能直接可见 - record.trace_id = trace_id - record.msg = f"[{trace_id}] {record.msg}" - else: - record.trace_id = "" - return True