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