外部文章质量判断

src/domains/demand_enqueue_task/enqueue_demands.py

@@ -1,16 +1,13 @@
 """需求入队服务 —— 将 FetchDemands 产出的队列条目批量写入 MySQL"""
 
 import logging
-from typing import Dict, List, TYPE_CHECKING
+from typing import Dict, List
 
 from src.infra.database.mysql.manager import MysqlManager
 
 from ._const import DemandConst
 from ._mapper import DemandQueueMapper
 
-if TYPE_CHECKING:
-    from src.handlers._logging import TaskLogger
-
 logger = logging.getLogger(__name__)
 
 
@@ -23,10 +20,9 @@ class EnqueueDemands(DemandConst):
         n = await enqueuer.deal(items)
     """
 
-    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
+    def __init__(self, pool: MysqlManager):
         self.pool = pool
         self.mapper = DemandQueueMapper(self.pool)
-        self._tlog = task_logger
 
     # ═══════════════════════════════════════════════════════════════
     # 入口

src/handlers/_logging.py

@@ -1,4 +1,4 @@
-"""handler 共享日志工具 —— LogService 注入 + TaskLogger + trace_id 上下文
+"""handler 共享日志工具 —— TaskLogger + 便捷工厂
 
 镜像 _utils.py 的 get_db() 模式：app.py 启动时注入 LogService，handler 通过
 get_task_logger() 获得 TaskLogger 实例，自动管理 trace_id 和计时。
@@ -11,7 +11,6 @@ trace_id 传递链路:
 
 from __future__ import annotations
 
-import contextvars
 import logging
 import time
 from typing import TYPE_CHECKING
@@ -23,6 +22,10 @@ from src.infra.observability import (
     LogRecord,
     TaskEvent,
 )
+from src.infra.observability.context import (
+    _current_trace_id,
+    get_log_service,
+)
 from src.infra.shared.tools import generate_task_trace_id
 
 if TYPE_CHECKING:
@@ -30,35 +33,6 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-# ═══════════════════════════════════════════════════════════════
-# trace_id 上下文变量 —— 跨 handler/service 隐式穿透
-# ═══════════════════════════════════════════════════════════════
-
-_current_trace_id: contextvars.ContextVar[str] = contextvars.ContextVar(
-    "task_trace_id", default=""
-)
-
-# ═══════════════════════════════════════════════════════════════
-# LogService 注入槽 —— app.py 启动时注入
-# ═══════════════════════════════════════════════════════════════
-
-_log_service_ref: "LogService | None" = None
-
-
-def set_log_service(ls: "LogService") -> None:
-    """启动时注入 LogService，与 set_db() 对称"""
-    global _log_service_ref
-    _log_service_ref = ls
-
-
-def get_log_service() -> "LogService":
-    """运行时获取已注入的 LogService"""
-    if _log_service_ref is None:
-        raise RuntimeError(
-            "LogService 尚未注入，请确认 app.py 中已调用 set_log_service()"
-        )
-    return _log_service_ref
-
 
 # ═══════════════════════════════════════════════════════════════
 # TaskLogger —— 任务级日志工具

src/handlers/demand_enqueue.py

@@ -35,12 +35,13 @@ async def demand_enqueue(param: str) -> dict:
         pg = get_db().pg_manager("pg")
         fetcher = FetchDemands(OdpsConfig(), pg, task_logger=tlog)
         items = await fetcher.deal(dt)
-        await tlog.phase_end("fetch", item_count=len(items), dt=dt)
     except Exception as e:
         logger.exception("ODPS 拉取失败, dt=%s", dt)
         await tlog.task_failed(e, dt=dt)
         return {"code": 500, "msg": "ODPS 拉取异常"}
 
+    await tlog.phase_end("fetch", item_count=len(items), dt=dt)
+
     if not items:
         logger.info("无队列项生成, dt=%s", dt)
         await tlog.task_completed(message="无队列项", dt=dt, items=0)
@@ -50,15 +51,16 @@ async def demand_enqueue(param: str) -> dict:
     try:
         await tlog.phase_start("enqueue", dt=dt)
         mysql = get_db().mysql_manager("long_articles")
-        enqueuer = EnqueueDemands(mysql, task_logger=tlog)
+        enqueuer = EnqueueDemands(mysql)
         n = await enqueuer.deal(items)
         accounts = len(set(it["account"] for it in items))
-        await tlog.phase_end("enqueue", enqueued=n, accounts=accounts, dt=dt)
     except Exception as e:
         logger.exception("MySQL 入队失败, dt=%s", dt)
         await tlog.task_failed(e, dt=dt)
         return {"code": 500, "msg": "入队异常"}
 
+    await tlog.phase_end("enqueue", enqueued=n, accounts=accounts, dt=dt)
+
     logger.info(
         "demandEnqueue done: enqueued %d items (%d accounts) for dt=%s",
         n,

src/infra/observability/__init__.py

@@ -1,4 +1,4 @@
-"""可观测性设施层 —— 结构化日志模型、SLS 索引定义、日志服务"""
+"""可观测性设施层 —— 结构化日志模型、SLS 索引定义、日志服务、追踪上下文"""
 
 from src.infra.observability.log_schema import (
     LogCategory,
@@ -9,6 +9,11 @@ from src.infra.observability.log_schema import (
     SLS_INDEX_CONFIG,
 )
 from src.infra.observability.logging import LogService
+from src.infra.observability.context import (
+    _current_trace_id,
+    set_log_service,
+    get_log_service,
+)
 
 __all__ = [
     "LogCategory",
@@ -18,4 +23,7 @@ __all__ = [
     "LogRecord",
     "SLS_INDEX_CONFIG",
     "LogService",
+    "_current_trace_id",
+    "set_log_service",
+    "get_log_service",
 ]

src/infra/observability/context.py

@@ -0,0 +1,45 @@
+"""可观测性上下文 —— trace_id ContextVar + LogService 模块级注入
+
+本模块位于 infra/observability，处于依赖链底层：
+  - handlers 依赖 infra → 可安全 import
+  - xxl_executor 本身在 infra → 同级 import，不违反对称
+
+避免 core/ 新建依赖 infra 的方向——core 当前不依赖 infra，保持干净。
+"""
+
+from __future__ import annotations
+
+import contextvars
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from src.infra.observability.logging import LogService
+
+# ═══════════════════════════════════════════════════════════════
+# trace_id 上下文变量 —— xxl_executor 写入，TaskLogger 读取
+# ═══════════════════════════════════════════════════════════════
+
+_current_trace_id: contextvars.ContextVar[str] = contextvars.ContextVar(
+    "task_trace_id", default=""
+)
+
+# ═══════════════════════════════════════════════════════════════
+# LogService 注入槽 —— app.py 启动时注入
+# ═══════════════════════════════════════════════════════════════
+
+_log_service_ref: "LogService | None" = None
+
+
+def set_log_service(ls: "LogService") -> None:
+    """启动时注入 LogService，与 set_db() 对称"""
+    global _log_service_ref
+    _log_service_ref = ls
+
+
+def get_log_service() -> "LogService":
+    """运行时获取已注入的 LogService，未注入则抛 RuntimeError"""
+    if _log_service_ref is None:
+        raise RuntimeError(
+            "LogService 尚未注入，请确认 app.py 中已调用 set_log_service()"
+        )
+    return _log_service_ref

src/infra/xxl_jobs/xxl_executor.py

@@ -16,7 +16,7 @@ import aiohttp
 
 from src.config.xxljob import XxlJobConfig
 from src.infra.shared.tools import generate_task_trace_id
-from src.handlers._logging import _current_trace_id, get_log_service
+from src.infra.observability.context import _current_trace_id, get_log_service
 from src.infra.observability import (
     LogCategory,
     LogLevel,

src/server/app.py

@@ -13,7 +13,7 @@ from src.infra.xxl_jobs import XxlJobExecutor
 from src.config.logging_config import setup_logging
 from src.handlers import discover_and_register
 from src.handlers._utils import set_db
-from src.handlers._logging import set_log_service
+from src.infra.observability.context import set_log_service
 from src.server.xxl_endpoints import set_xxl_executor
 
 logger = logging.getLogger(__name__)