Merge branch 'feature_20260707_MARK_add_aliyun_log' of Server/long_articles_agentic_supply into master

src/config/logging_config.py

@@ -52,6 +52,14 @@ LOGGING_CONFIG: dict = {
             "formatter": "default",
             "encoding": "utf-8",
         },
+        "handler_quality": {
+            "class": "logging.handlers.RotatingFileHandler",
+            "filename": _log("handler_quality"),
+            "maxBytes": 10 * 1024 * 1024,
+            "backupCount": 3,
+            "formatter": "default",
+            "encoding": "utf-8",
+        },
         "domain_supply": {
             "class": "logging.handlers.RotatingFileHandler",
             "filename": _log("domain_content_supply"),
@@ -68,6 +76,14 @@ LOGGING_CONFIG: dict = {
             "formatter": "default",
             "encoding": "utf-8",
         },
+        "domain_quality": {
+            "class": "logging.handlers.RotatingFileHandler",
+            "filename": _log("domain_content_quality"),
+            "maxBytes": 10 * 1024 * 1024,
+            "backupCount": 3,
+            "formatter": "default",
+            "encoding": "utf-8",
+        },
         "infra": {
             "class": "logging.handlers.RotatingFileHandler",
             "filename": _log("infra"),
@@ -122,6 +138,21 @@ LOGGING_CONFIG: dict = {
             "level": "INFO",
             "propagate": False,
         },
+        "src.handlers.article_quality_category": {
+            "handlers": ["handler_quality", "console"],
+            "level": "INFO",
+            "propagate": False,
+        },
+        "src.handlers.article_quality_quantify": {
+            "handlers": ["handler_quality", "console"],
+            "level": "INFO",
+            "propagate": False,
+        },
+        "src.handlers.article_quality_title": {
+            "handlers": ["handler_quality", "console"],
+            "level": "INFO",
+            "propagate": False,
+        },
         # domain
         "src.domains.content_supply": {
             "handlers": ["domain_supply", "console"],
@@ -133,6 +164,11 @@ LOGGING_CONFIG: dict = {
             "level": "INFO",
             "propagate": False,
         },
+        "src.domains.content_quality": {
+            "handlers": ["domain_quality", "console"],
+            "level": "INFO",
+            "propagate": False,
+        },
         # infra
         "src.infra": {
             "handlers": ["infra", "console"],

src/domains/content_quality/_const.py

@@ -44,7 +44,8 @@ class ContentQualityConst:
     # LLM 参数
     # ═══════════════════════════════════════════════════════════════
 
-    LLM_BATCH_SIZE = 20
+    LLM_BATCH_SIZE = 20  # 品类识别：ID 显式绑定，20 条安全
+    TITLE_BATCH_SIZE = 10  # 标题评分：对齐精度要求高，控制在 10 条/批
     LLM_INTERVAL_SEC = 2
 
     # ═══════════════════════════════════════════════════════════════
@@ -69,8 +70,13 @@ class ContentQualityConst:
 7.阵营对立：7分
 8.军事危机暗示：5分
 
-基于以上的爆款标题特征分评分标准，如果下列标题符合某个特征的、那么就加上这个特征对应的权重，标题最终的得分是各个特征分的总和。请直接输出最终的分数，不用给任何理由和描述、解释。
-待评分的标题为：
+基于以上的爆款标题特征分评分标准，如果下列标题符合某个特征的、那么就加上这个特征对应的权重，标题最终的得分是各个特征分的总和。如果标题完全不符合任何特征，则分数为0。
+
+输入是一个 LIST，每个元素是一个元组，元组的第一个元素是标题的 ID，第二个元素是标题的文本。
+输出格式为 JSON 对象，key 为标题 id（数字），value 为评分（0-100 的整数）。
+仅输出 JSON，不要 markdown 包裹，不要任何其他内容。
+
+输入的 LIST 是：
 {title_list}"""
 
     PROMPT_TITLE_CATEGORY = """

src/domains/content_quality/_utils.py

@@ -1,7 +1,5 @@
 """内容质量评估工具函数 —— Prompt 构建 + LLM 响应解析"""
 
-import hashlib
-import re
 import logging
 from typing import Dict, List
 
@@ -12,20 +10,16 @@ from ._const import ContentQualityConst
 logger = logging.getLogger(__name__)
 
 
-def md5(text: str) -> str:
-    return hashlib.md5(text.encode("utf-8")).hexdigest()
-
-
 # ═══════════════════════════════════════════════════════════════
 # Prompt 构建
 # ═══════════════════════════════════════════════════════════════
 
 
 def build_title_quality_prompt(titles: List[Dict]) -> str:
-    """构造标题质量评分 prompt，每行一个标题"""
-    lines = [t["title"] for t in titles]
+    """构造标题质量评分 prompt，格式: [(id, title), ...] 元组列表"""
+    tuples = [(t["id"], t["title"]) for t in titles]
     return ContentQualityConst.PROMPT_TITLE_QUALITY.format(
-        title_list="\n".join(lines),
+        title_list=str(tuples),
     )
 
 
@@ -46,7 +40,10 @@ def parse_title_quality_response(
     raw: str | None,
     batch_titles: List[Dict],
 ) -> List[Dict]:
-    """解析 prompt1 返回值：LLM 按行输出纯数字分数，按顺序对应输入标题
+    """解析标题质量评分返回值：{id: score} JSON 对象，key 为标题 id
+
+    与品类识别 prompt 对齐——ID 显式绑定，消除位置依赖导致的分数错位。
+    如果某条标题在 LLM 响应中缺失 key 或值非法，则静默跳过（不影响同批次其他标题）。
 
     返回: [{"id": 1, "score": 85}, ...]
     """
@@ -55,18 +52,21 @@ def parse_title_quality_response(
         logger.warning("LLM 标题质量返回为空")
         return result
 
-    # 从文本中提取所有数字
-    scores = re.findall(r"\d+", raw)
-    if not scores:
-        logger.warning("LLM 标题质量未提取到分数: %s", raw[:200])
+    parsed = safe_json_parse(raw)
+    if not isinstance(parsed, dict):
+        logger.warning("LLM 标题质量返回非 JSON 对象: %s", raw[:200])
         return result
 
-    for i, s in enumerate(scores):
-        if i >= len(batch_titles):
-            break
-        score = int(s)
-        if 0 <= score <= 100:
-            result.append({"id": batch_titles[i]["id"], "score": score})
+    for t in batch_titles:
+        score_val = parsed.get(str(t["id"]))
+        if score_val is None:
+            continue
+        try:
+            score = int(score_val)
+            if 0 <= score <= 100:
+                result.append({"id": t["id"], "score": score})
+        except (ValueError, TypeError):
+            pass
 
     return result
 
@@ -98,7 +98,6 @@ def parse_category_response(
 
 
 __all__ = [
-    "md5",
     "build_title_quality_prompt",
     "build_category_prompt",
     "parse_title_quality_response",

src/domains/content_quality/article_quality_category.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import logging
-from typing import Dict
+from typing import Dict, TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.external.volcengine import fetch_deepseek_completion
@@ -12,26 +12,37 @@ from ._const import ContentQualityConst
 from ._mapper import QualityMapper
 from ._utils import build_category_prompt, parse_category_response
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class ArticleQualityCategory(ContentQualityConst):
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.mapper = QualityMapper(pool)
+        self._tlog = task_logger
         self._evaluated = 0
         self._lock = asyncio.Lock()
 
     async def deal(self, *, limit: int | None = None) -> dict:
         _limit = limit if limit is not None else self.CATEGORY_DEAL_LIMIT
 
+        if self._tlog:
+            await self._tlog.phase_start("category", limit=_limit)
+
         # Step 0: 种子同步
         seeded = await self.mapper.seed_categories(limit=_limit)
         logger.info("品类种子同步: seeded=%d", seeded)
+        if self._tlog:
+            await self._tlog.info("phase.start", phase="category", message=f"种子同步完成: {seeded}", seeded=seeded)
 
         # Step 1: LLM 识别
         rows = await self.mapper.fetch_pending_categories(limit=_limit)
         if not rows:
             logger.info("无待识别品类")
+            if self._tlog:
+                await self._tlog.phase_end("category", seeded=seeded, evaluated=0)
             return {"seeded": seeded, "evaluated": 0}
 
         batches = []
@@ -68,6 +79,14 @@ class ArticleQualityCategory(ContentQualityConst):
             self._evaluated,
             len(result["errors"]),
         )
+        if self._tlog:
+            await self._tlog.phase_end(
+                "category",
+                seeded=seeded,
+                batches=len(batches),
+                evaluated=self._evaluated,
+                errors=len(result["errors"]),
+            )
         return {"seeded": seeded, "evaluated": self._evaluated}
 
     async def _evaluate_one_batch(self, batch: Dict) -> None:

src/domains/content_quality/article_quality_quantify.py

@@ -1,6 +1,7 @@
 """量化评估 Service —— 前置校验 + 入库 + 量化计算"""
 
 import logging
+from typing import TYPE_CHECKING
 
 from tqdm import tqdm
 
@@ -9,20 +10,29 @@ from src.infra.database.mysql.manager import MysqlManager
 from ._const import ContentQualityConst
 from ._mapper import QualityMapper
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class ArticleQualityQuantify(ContentQualityConst):
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.mapper = QualityMapper(pool)
+        self._tlog = task_logger
 
     async def deal(self, *, limit: int | None = None) -> dict:
         _limit = limit if limit is not None else self.QUANTIFY_DEAL_LIMIT
 
+        if self._tlog:
+            await self._tlog.phase_start("quantify_validate", limit=_limit)
+
         # 取已完成深翻的账号，挨个处理
         account_ids = await self.mapper.fetch_crawled_account_ids()
         if not account_ids:
             logger.info("无已完成深翻的账号")
+            if self._tlog:
+                await self._tlog.phase_end("quantify_validate", accounts=0)
             return {
                 "validated": 0,
                 "skipped": 0,
@@ -48,9 +58,26 @@ class ArticleQualityQuantify(ContentQualityConst):
                 pbar.update(1)
                 pbar.set_postfix(inserted=total_inserted, skipped=total_skipped)
 
+        if self._tlog:
+            await self._tlog.phase_end(
+                "quantify_validate",
+                validated=total_validated,
+                skipped=total_skipped,
+                inserted=total_inserted,
+                errors=total_errors,
+            )
+
+        if self._tlog:
+            await self._tlog.phase_start("quantify_compute", limit=_limit)
         c = await self._compute(limit=_limit)
         total_computed = c["computed"]
         total_errors += c["errors"]
+        if self._tlog:
+            await self._tlog.phase_end(
+                "quantify_compute",
+                computed=total_computed,
+                errors=c["errors"],
+            )
 
         logger.info(
             "量化评估完成: validated=%d skipped=%d inserted=%d computed=%d errors=%d",

src/domains/content_quality/article_quality_title.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import logging
-from typing import Dict, List
+from typing import Dict, List, TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.external.volcengine import fetch_deepseek_completion
@@ -12,32 +12,43 @@ from ._const import ContentQualityConst
 from ._mapper import QualityMapper
 from ._utils import build_title_quality_prompt, parse_title_quality_response
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class ArticleQualityTitle(ContentQualityConst):
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.mapper = QualityMapper(pool)
+        self._tlog = task_logger
         self._evaluated = 0
         self._lock = asyncio.Lock()
 
     async def deal(self, *, limit: int | None = None) -> dict:
         _limit = limit if limit is not None else self.TITLE_DEAL_LIMIT
 
+        if self._tlog:
+            await self._tlog.phase_start("title_scoring", limit=_limit)
+
         # Step 0: 种子同步
         seeded = await self.mapper.seed_titles(limit=_limit)
         logger.info("标题种子同步: seeded=%d", seeded)
+        if self._tlog:
+            await self._tlog.info("phase.start", phase="title_scoring", message=f"种子同步完成: {seeded}", seeded=seeded)
 
         # Step 1: LLM 评分
         rows = await self.mapper.fetch_pending_titles(limit=_limit)
         if not rows:
             logger.info("无待评分标题")
+            if self._tlog:
+                await self._tlog.phase_end("title_scoring", seeded=seeded, evaluated=0)
             return {"seeded": seeded, "evaluated": 0}
 
         # 拆批
         batches = []
-        for start in range(0, len(rows), self.LLM_BATCH_SIZE):
-            batch = rows[start : start + self.LLM_BATCH_SIZE]
+        for start in range(0, len(rows), self.TITLE_BATCH_SIZE):
+            batch = rows[start : start + self.TITLE_BATCH_SIZE]
             batch_ids = [r["id"] for r in batch]
             titles = [{"id": r["id"], "title": r["title"]} for r in batch]
             prompt = build_title_quality_prompt(titles)
@@ -74,6 +85,14 @@ class ArticleQualityTitle(ContentQualityConst):
             self._evaluated,
             len(result["errors"]),
         )
+        if self._tlog:
+            await self._tlog.phase_end(
+                "title_scoring",
+                seeded=seeded,
+                batches=len(batches),
+                evaluated=self._evaluated,
+                errors=len(result["errors"]),
+            )
         return {"seeded": seeded, "evaluated": self._evaluated}
 
     async def _evaluate_one_batch(self, batch: Dict) -> None:

src/domains/content_supply/_utils.py

@@ -1,12 +1,12 @@
 """供给域工具函数 —— API 响应解析、数据转换、搜索/账号/文章通用工具"""
 
-import hashlib
 import logging
 import re
 from datetime import datetime
 from typing import Dict, List, Tuple
 from urllib.parse import parse_qs, urlparse
 
+from src.infra.shared.tools import str_to_md5
 from src.infra.spider.wechat.gzh import get_article_list_from_account
 
 logger = logging.getLogger(__name__)
@@ -94,7 +94,7 @@ def parse_search_result(
         "wx_sn": extract_wx_sn(article_url) or "",
         "biz": extract_biz(article_url) or "",
         "title": title,
-        "title_md5": hashlib.md5(title.encode("utf-8")).hexdigest() if title else "",
+        "title_md5": str_to_md5(title) if title else "",
         "cover_url": article.get("cover_url", ""),
         "publish_time": publish_time,
     }

src/domains/content_supply/account_article_fetch.py

@@ -12,6 +12,7 @@
 import asyncio
 import logging
 import time
+from typing import TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.shared.async_tasks import run_tasks_with_async_worker_group
@@ -20,14 +21,18 @@ from ._const import ContentSupplyConst
 from ._mapper import AccountMapper
 from ._utils import fetch_article_page
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class AccountArticleFetch(ContentSupplyConst):
     """从 accounts 取 gh_id → 调文章列表 API → 写入 articles 表"""
 
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.mapper = AccountMapper(pool)
+        self._tlog = task_logger
 
     async def deal(self, *, limit: int = 10) -> dict:
         """处理 N 个账号的文章抓取
@@ -35,9 +40,14 @@ class AccountArticleFetch(ContentSupplyConst):
         Returns:
             {"accounts": int, "articles_written": int, "errors": int}
         """
+        if self._tlog:
+            await self._tlog.phase_start("article_fetch", limit=limit)
+
         accounts = await self.mapper.list_accounts_for_article_crawl(limit=limit)
         if not accounts:
             logger.info("无待抓取文章的账号")
+            if self._tlog:
+                await self._tlog.phase_end("article_fetch", accounts=0)
             return {"accounts": 0, "articles_written": 0, "errors": 0}
 
         total_articles = 0
@@ -95,6 +105,13 @@ class AccountArticleFetch(ContentSupplyConst):
             total_articles,
             errors_count,
         )
+        if self._tlog:
+            await self._tlog.phase_end(
+                "article_fetch",
+                accounts=len(accounts),
+                articles_written=total_articles,
+                errors=errors_count,
+            )
         return {
             "accounts": len(accounts),
             "articles_written": total_articles,

src/domains/content_supply/account_article_update.py

@@ -9,6 +9,7 @@
 
 import asyncio
 import logging
+from typing import TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 
@@ -16,14 +17,18 @@ from ._const import ContentSupplyConst
 from ._mapper import AccountMapper
 from ._utils import fetch_article_page
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class AccountArticleUpdate(ContentSupplyConst):
     """取已完成深翻的账号 → 拉第一页文章 → 按 wx_sn 更新指标"""
 
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.mapper = AccountMapper(pool)
+        self._tlog = task_logger
 
     async def deal(self, *, limit: int = 10) -> dict:
         """处理 N 个账号的文章指标更新
@@ -31,9 +36,14 @@ class AccountArticleUpdate(ContentSupplyConst):
         Returns:
             {"accounts": int, "articles_updated": int, "errors": int}
         """
+        if self._tlog:
+            await self._tlog.phase_start("article_update", limit=limit)
+
         accounts = await self.mapper.list_accounts_for_article_update(limit=limit)
         if not accounts:
             logger.info("无待更新文章指标的账号")
+            if self._tlog:
+                await self._tlog.phase_end("article_update", accounts=0)
             return {"accounts": 0, "articles_updated": 0, "errors": 0}
 
         total_updated = 0
@@ -61,6 +71,13 @@ class AccountArticleUpdate(ContentSupplyConst):
             total_updated,
             errors,
         )
+        if self._tlog:
+            await self._tlog.phase_end(
+                "article_update",
+                accounts=len(accounts),
+                articles_updated=total_updated,
+                errors=errors,
+            )
         return {
             "accounts": len(accounts),
             "articles_updated": total_updated,

src/domains/content_supply/account_fetch.py

@@ -8,6 +8,7 @@
 
 import asyncio
 import logging
+from typing import TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.spider.wechat.gzh import get_account_from_url
@@ -16,14 +17,18 @@ from ._const import ContentSupplyConst
 from ._mapper import AccountMapper
 from ._utils import parse_account_response
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class AccountFetch(ContentSupplyConst):
     """从 relation 取待抓 biz → 查重 → 调账号 API → 写入 accounts 表"""
 
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.mapper = AccountMapper(pool)
+        self._tlog = task_logger
 
     async def deal(self, *, limit: int = 100) -> dict:
         """串行处理待抓账号
@@ -31,9 +36,14 @@ class AccountFetch(ContentSupplyConst):
         Returns:
             {"pending": int, "success": int, "skipped": int, "failed": int}
         """
+        if self._tlog:
+            await self._tlog.phase_start("account_fetch", limit=limit)
+
         rows = await self.mapper.fetch_pending_relations(limit=limit)
         if not rows:
             logger.info("无待抓账号")
+            if self._tlog:
+                await self._tlog.phase_end("account_fetch", pending=0)
             return {"pending": 0, "success": 0, "skipped": 0, "failed": 0}
 
         success = 0
@@ -73,6 +83,8 @@ class AccountFetch(ContentSupplyConst):
             skipped,
             failed,
         )
+        if self._tlog:
+            await self._tlog.phase_end("account_fetch", pending=len(rows), success=success, skipped=skipped, failed=failed)
         return {
             "pending": len(rows),
             "success": success,

src/domains/content_supply/article_fetch_detail.py

@@ -6,7 +6,7 @@
 
 import asyncio
 import logging
-from typing import Dict
+from typing import Dict, TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.spider.wechat.gzh import get_article_detail
@@ -15,15 +15,19 @@ from ._const import ContentSupplyConst
 from ._mapper import ArticleSearchRelationMapper, ArticleDetailMapper
 from ._utils import parse_detail
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
 class ArticleFetchDetail(ContentSupplyConst):
     """从 relation 表取待拉项 → 调详情 API → 写入 detail 表"""
 
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.relation_mapper = ArticleSearchRelationMapper(pool)
         self.detail_mapper = ArticleDetailMapper(pool)
+        self._tlog = task_logger
 
     # ═══════════════════════════════════════════════════════════════
     # 入口
@@ -41,6 +45,9 @@ class ArticleFetchDetail(ContentSupplyConst):
         if limit is None:
             limit = self.DETAIL_DEAL_LIMIT
 
+        if self._tlog:
+            await self._tlog.phase_start("detail_fetch", limit=limit)
+
         # 将低于阈值的低质文章标记为 FAIL，不拉详情
         dropped = await self.relation_mapper.mark_low_quality(limit=limit)
         if dropped:
@@ -49,6 +56,8 @@ class ArticleFetchDetail(ContentSupplyConst):
                 dropped,
                 self.MIN_QUALITY_SCORE,
             )
+            if self._tlog:
+                await self._tlog.info("item.processed", message=f"标记低质文章: {dropped} 条", phase="detail_fetch", dropped=dropped)
 
         items = await self.relation_mapper.fetch_pending(limit=limit)
         if not items:
@@ -83,6 +92,8 @@ class ArticleFetchDetail(ContentSupplyConst):
             success,
             failed,
         )
+        if self._tlog:
+            await self._tlog.phase_end("detail_fetch", pending=len(items), success=success, failed=failed)
         return {
             "pending": len(items),
             "success": success,

src/domains/content_supply/article_search.py

@@ -7,7 +7,7 @@
 
 import asyncio
 import logging
-from typing import Dict
+from typing import Dict, TYPE_CHECKING
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.spider.wechat.gzh import weixin_search
@@ -16,6 +16,9 @@ from ._const import ContentSupplyConst
 from ._mapper import ArticleSearchRelationMapper
 from ._utils import parse_search_result, is_valid_keyword
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 
@@ -25,8 +28,9 @@ class DemandSearchArticle(ContentSupplyConst):
     按 dt 分区读取，experiment_id 只透传不过滤。
     """
 
-    def __init__(self, pool: MysqlManager):
+    def __init__(self, pool: MysqlManager, task_logger: "TaskLogger | None" = None):
         self.relation_mapper = ArticleSearchRelationMapper(pool)
+        self._tlog = task_logger
 
     # ═══════════════════════════════════════════════════════════════
     # 入口
@@ -40,9 +44,14 @@ class DemandSearchArticle(ContentSupplyConst):
         Returns:
             {"queued": int, "processed": int, "results_written": int}
         """
+        if self._tlog:
+            await self._tlog.phase_start("search", dt=dt, limit=limit)
+
         items = await self.relation_mapper.fetch_queue_pending(dt, limit=limit)
         if not items:
             logger.info("无待处理队列项")
+            if self._tlog:
+                await self._tlog.phase_end("search", queued=0, processed=0)
             return {"queued": 0, "processed": 0, "results_written": 0}
 
         total_written = 0
@@ -57,6 +66,15 @@ class DemandSearchArticle(ContentSupplyConst):
                     item["id"],
                     item.get("experiment_id"),
                 )
+                if self._tlog:
+                    await self._tlog.item_processed(
+                        "item.failed",
+                        message=f"搜索失败: queue_id={item['id']}",
+                        phase="search",
+                        successful=False,
+                        queue_id=item["id"],
+                        experiment_id=item.get("experiment_id"),
+                    )
                 await self.relation_mapper.queue_mark_failed(
                     item["id"],
                     reason=str(e),
@@ -86,6 +104,10 @@ class DemandSearchArticle(ContentSupplyConst):
             len(items),
             total_written,
         )
+        if self._tlog:
+            await self._tlog.phase_end(
+                "search", queued=len(items), processed=processed, results_written=total_written
+            )
         return {
             "queued": len(items),
             "processed": processed,
@@ -171,9 +193,13 @@ class DemandSearchArticle(ContentSupplyConst):
         cached = await self.relation_mapper.cache_get(keyword, page)
         if cached:
             logger.info("缓存命中: keyword=%s, page=%d", keyword, page)
+            if self._tlog:
+                await self._tlog.info("cache.hit", message=f"缓存命中: {keyword}", phase="search", keyword=keyword, page=page)
             return cached, True
 
-        result = await weixin_search(keyword, page=cursor)
+        if self._tlog:
+            await self._tlog.info("cache.miss", message=f"缓存未命中, 调API: {keyword}", phase="search", keyword=keyword, page=page)
+        result = await weixin_search(keyword, page=cursor, tlog=self._tlog)
         if result and result.get("code") == 0:
             await self.relation_mapper.cache_set(keyword, page, result)
         return result, False

src/domains/demand_enqueue_task/fetch_demands.py

@@ -9,7 +9,7 @@
 
 import asyncio
 import logging
-from typing import Dict, List, Set
+from typing import Dict, List, Set, TYPE_CHECKING
 
 from tqdm import tqdm
 
@@ -20,6 +20,9 @@ from src.infra.database.postgresql.manager import PgManager
 from ._const import DemandConst
 from ._mapper import VideoDeconstructMapper
 
+if TYPE_CHECKING:
+    from src.handlers._logging import TaskLogger
+
 logger = logging.getLogger(__name__)
 
 # ODPS SELECT 列清单（匹配结果表）
@@ -44,9 +47,10 @@ class FetchDemands(DemandConst):
         items = await fetcher.deal("20260701")
     """
 
-    def __init__(self, config: OdpsConfig, pg: PgManager):
+    def __init__(self, config: OdpsConfig, pg: PgManager, task_logger: "TaskLogger | None" = None):
         self._odps_config = config
         self._deconstruct_mapper = VideoDeconstructMapper(pg)
+        self._tlog = task_logger
 
     # ═══════════════════════════════════════════════════════════════
     # 入口
@@ -58,6 +62,8 @@ class FetchDemands(DemandConst):
         valid_accounts = await self._load_valid_accounts()
         if not valid_accounts:
             logger.warning("白名单为空, dt=%s, 终止", dt)
+            if self._tlog:
+                await self._tlog.warn("item.failed", "白名单为空", phase="fetch", dt=dt)
             return []
 
         # 2. 查询匹配视频（ODPS）
@@ -77,6 +83,15 @@ class FetchDemands(DemandConst):
         # 5. 获取视频解构 + 展开搜索词
         items = await self._explode_search_items(deduped)
         logger.info("搜索词展开: %d 条队列项", len(items))
+        if self._tlog:
+            await self._tlog.info(
+                "phase.end", phase="fetch",
+                message=f"搜索词展开完成: {len(items)} 条",
+                dt=dt,
+                raw_matches=len(rows),
+                deduped=len(deduped),
+                items=len(items),
+            )
         return items
 
     # ═══════════════════════════════════════════════════════════════

src/handlers/_logging.py

@@ -0,0 +1,251 @@
+"""handler 共享日志工具 —— TaskLogger + 便捷工厂
+
+镜像 _utils.py 的 get_db() 模式：app.py 启动时注入 LogService，handler 通过
+get_task_logger() 获得 TaskLogger 实例，自动管理 trace_id 和计时。
+
+trace_id 传递链路:
+  XxlJobExecutor.run() → _current_trace_id.set(...) → ContextVar
+    └─ handler(param) → get_task_logger()  → 自动读取 ContextVar 中的 trace_id
+         └─ service.deal(..., task_logger=tlog)  → 传入 TaskLogger
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import TYPE_CHECKING
+
+from src.infra.observability import (
+    LogCategory,
+    LogLevel,
+    LogStatus,
+    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:
+    from src.infra.observability.logging import LogService
+
+logger = logging.getLogger(__name__)
+
+
+# ═══════════════════════════════════════════════════════════════
+# TaskLogger —— 任务级日志工具
+# ═══════════════════════════════════════════════════════════════
+
+
+class TaskLogger:
+    """任务日志器 —— 封装 LogService 提供业务友好的结构化日志 API
+
+    自动管理 trace_id、计时、事件标准化。每个 handler 实例创建一个。
+
+    使用方式:
+        tlog = get_task_logger("articleSearch")
+        await tlog.task_started(dt="20260701", limit=2000)
+        try:
+            ...
+            await tlog.phase_start("search")
+            ...
+            await tlog.phase_end("search", items=100)
+            await tlog.task_completed(processed=100)
+        except Exception as e:
+            await tlog.task_failed(e)
+    """
+
+    def __init__(
+        self,
+        log_service: "LogService",
+        task_name: str,
+        trace_id: str | None = None,
+    ):
+        self._ls = log_service
+        self._task_name = task_name
+        self.trace_id = trace_id or _current_trace_id.get() or generate_task_trace_id()
+        self._started_at = time.monotonic()
+        self._phase_timers: dict[str, float] = {}
+
+    # ── 生命周期 ──
+
+    async def task_started(self, message: str = "", **extra) -> None:
+        """任务开始 —— 在 handler 入口调用一次"""
+        await self._emit(
+            level=LogLevel.INFO,
+            event=TaskEvent.TASK_STARTED,
+            message=message or f"{self._task_name} 开始执行",
+            status=LogStatus.PENDING,
+            **extra,
+        )
+
+    async def task_completed(self, message: str = "", **extra) -> None:
+        """任务完成 —— handler 正常结束时调用"""
+        await self._emit(
+            level=LogLevel.INFO,
+            event=TaskEvent.TASK_COMPLETED,
+            message=message or f"{self._task_name} 执行完成",
+            status=LogStatus.SUCCESS,
+            **extra,
+        )
+
+    async def task_failed(self, error: Exception | None = None, message: str = "", **extra) -> None:
+        """任务失败 —— handler 异常时调用"""
+        if not message:
+            message = str(error) if error else f"{self._task_name} 执行失败"
+        await self._emit(
+            level=LogLevel.ERROR,
+            event=TaskEvent.TASK_FAILED,
+            message=message,
+            status=LogStatus.FAILED,
+            error_type=type(error).__name__ if error else "",
+            error_message=str(error) if error else "",
+            **extra,
+        )
+
+    # ── 阶段标记 ──
+
+    async def phase_start(self, phase: str, message: str = "", **extra) -> None:
+        """阶段开始 —— 标记管道阶段"""
+        self._phase_timers[phase] = time.monotonic()
+        await self._emit(
+            level=LogLevel.INFO,
+            event=TaskEvent.PHASE_START,
+            phase=phase,
+            message=message or f"阶段开始: {phase}",
+            **extra,
+        )
+
+    async def phase_end(self, phase: str, message: str = "", **extra) -> None:
+        """阶段结束 —— 自动计算阶段耗时"""
+        phase_start = self._phase_timers.pop(phase, self._started_at)
+        phase_duration = int((time.monotonic() - phase_start) * 1000)
+        await self._emit(
+            level=LogLevel.INFO,
+            event=TaskEvent.PHASE_END,
+            phase=phase,
+            message=message or f"阶段结束: {phase}",
+            duration_ms=phase_duration,
+            **extra,
+        )
+
+    # ── 业务事件 ──
+
+    async def info(
+        self,
+        event: str | TaskEvent,
+        message: str = "",
+        phase: str = "",
+        **extra,
+    ) -> None:
+        """通用信息日志"""
+        await self._emit(
+            level=LogLevel.INFO,
+            event=event if isinstance(event, str) else event.value,
+            phase=phase,
+            message=message,
+            **extra,
+        )
+
+    async def warn(
+        self,
+        event: str | TaskEvent,
+        message: str = "",
+        phase: str = "",
+        **extra,
+    ) -> None:
+        """警告日志"""
+        await self._emit(
+            level=LogLevel.WARN,
+            event=event if isinstance(event, str) else event.value,
+            phase=phase,
+            message=message,
+            **extra,
+        )
+
+    async def error(
+        self,
+        event: str | TaskEvent,
+        message: str = "",
+        exc: Exception | None = None,
+        phase: str = "",
+        **extra,
+    ) -> None:
+        """错误日志"""
+        await self._emit(
+            level=LogLevel.ERROR,
+            event=event if isinstance(event, str) else event.value,
+            phase=phase,
+            message=message or (str(exc) if exc else ""),
+            error_type=type(exc).__name__ if exc else "",
+            error_message=str(exc) if exc else "",
+            status=LogStatus.FAILED,
+            **extra,
+        )
+
+    async def item_processed(
+        self,
+        event: str | TaskEvent,
+        message: str = "",
+        phase: str = "",
+        successful: bool = True,
+        **extra,
+    ) -> None:
+        """单条处理结果"""
+        await self._emit(
+            level=LogLevel.INFO,
+            event=event if isinstance(event, str) else event.value,
+            phase=phase,
+            message=message,
+            status=LogStatus.SUCCESS if successful else LogStatus.FAILED,
+            **extra,
+        )
+
+    # ── 内部方法 ──
+
+    async def _emit(
+        self,
+        level: LogLevel,
+        event: str,
+        message: str = "",
+        phase: str = "",
+        status: LogStatus = LogStatus.SUCCESS,
+        duration_ms: int = 0,
+        error_type: str = "",
+        error_message: str = "",
+        **extra,
+    ) -> None:
+        """构造 LogRecord 并推送到 SLS"""
+        total_elapsed = int((time.monotonic() - self._started_at) * 1000)
+        record = LogRecord(
+            level=level,
+            category=LogCategory.TASK,
+            event=event,
+            message=message,
+            trace_id=self.trace_id,
+            task_name=self._task_name,
+            phase=phase,
+            duration_ms=duration_ms or total_elapsed,
+            status=status,
+            error_type=error_type,
+            error_message=error_message,
+            extra=extra,
+        )
+        await self._ls.log(record)
+
+
+# ═══════════════════════════════════════════════════════════════
+# 便捷工厂
+# ═══════════════════════════════════════════════════════════════
+
+
+def get_task_logger(task_name: str, trace_id: str | None = None) -> TaskLogger:
+    """从注入的 LogService 创建 TaskLogger —— handler 入口一行获取
+
+    使用方式:
+        from src.handlers._logging import get_task_logger
+        tlog = get_task_logger("articleSearch")
+    """
+    return TaskLogger(get_log_service(), task_name, trace_id=trace_id)

src/handlers/account_article_fetch.py

@@ -11,6 +11,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_supply import AccountArticleFetch, ContentSupplyConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -18,14 +19,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("accountArticleFetch")
 async def account_article_fetch(param: str) -> dict:
     limit = parse_limit(param, ContentSupplyConst.ARTICLE_DEAL_LIMIT)
-    logger.info("accountArticleFetch started, limit=%d", limit)
+    tlog = get_task_logger("accountArticleFetch")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        fetcher = AccountArticleFetch(mysql)
+        fetcher = AccountArticleFetch(mysql, task_logger=tlog)
         result = await fetcher.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("账号文章抓取失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "账号文章抓取异常"}
 
     logger.info(
@@ -34,6 +37,7 @@ async def account_article_fetch(param: str) -> dict:
         result["articles_written"],
         result["errors"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": f"accounts={result['accounts']}, written={result['articles_written']}, errors={result['errors']}",

src/handlers/account_article_update.py

@@ -12,6 +12,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_supply import AccountArticleUpdate, ContentSupplyConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -19,14 +20,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("accountArticleUpdate")
 async def account_article_update(param: str) -> dict:
     limit = parse_limit(param, ContentSupplyConst.ARTICLE_DEAL_LIMIT)
-    logger.info("accountArticleUpdate started, limit=%d", limit)
+    tlog = get_task_logger("accountArticleUpdate")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        updater = AccountArticleUpdate(mysql)
+        updater = AccountArticleUpdate(mysql, task_logger=tlog)
         result = await updater.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("文章指标更新失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "文章指标更新异常"}
 
     logger.info(
@@ -35,6 +38,7 @@ async def account_article_update(param: str) -> dict:
         result["articles_updated"],
         result["errors"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": f"accounts={result['accounts']}, updated={result['articles_updated']}, errors={result['errors']}",

src/handlers/account_fetch.py

@@ -10,6 +10,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_supply import AccountFetch, ContentSupplyConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -17,14 +18,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("accountFetch")
 async def account_fetch(param: str) -> dict:
     limit = parse_limit(param, ContentSupplyConst.ACCOUNT_DEAL_LIMIT)
-    logger.info("accountFetch started, limit=%d", limit)
+    tlog = get_task_logger("accountFetch")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        fetcher = AccountFetch(mysql)
+        fetcher = AccountFetch(mysql, task_logger=tlog)
         result = await fetcher.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("账号抓取失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "账号抓取异常"}
 
     logger.info(
@@ -34,6 +37,7 @@ async def account_fetch(param: str) -> dict:
         result["skipped"],
         result["failed"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": f"success={result['success']}, skipped={result['skipped']}, failed={result['failed']}",

src/handlers/article_fetch_detail.py

@@ -11,6 +11,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_supply import ArticleFetchDetail, ContentSupplyConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -18,14 +19,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("articleFetchDetail")
 async def article_fetch_detail(param: str) -> dict:
     limit = parse_limit(param, ContentSupplyConst.DETAIL_DEAL_LIMIT)
-    logger.info("articleFetchDetail started, limit=%d", limit)
+    tlog = get_task_logger("articleFetchDetail")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        fetcher = ArticleFetchDetail(mysql)
+        fetcher = ArticleFetchDetail(mysql, task_logger=tlog)
         result = await fetcher.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("详情拉取失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "详情拉取异常"}
 
     logger.info(
@@ -34,6 +37,7 @@ async def article_fetch_detail(param: str) -> dict:
         result["success"],
         result["failed"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": f"success={result['success']}, failed={result['failed']}",

src/handlers/article_quality_category.py

@@ -12,6 +12,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_quality import ArticleQualityCategory, ContentQualityConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -19,14 +20,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("articleQualityCategory")
 async def article_quality_category(param: str) -> dict:
     limit = parse_limit(param, ContentQualityConst.CATEGORY_DEAL_LIMIT)
-    logger.info("articleQualityCategory started, limit=%d", limit)
+    tlog = get_task_logger("articleQualityCategory")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        service = ArticleQualityCategory(mysql)
+        service = ArticleQualityCategory(mysql, task_logger=tlog)
         result = await service.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("品类识别失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "异常"}
 
     logger.info(
@@ -34,6 +37,7 @@ async def article_quality_category(param: str) -> dict:
         result["seeded"],
         result["evaluated"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": f"seeded={result['seeded']} evaluated={result['evaluated']}",

src/handlers/article_quality_quantify.py

@@ -12,6 +12,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_quality import ArticleQualityQuantify, ContentQualityConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -19,14 +20,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("articleQualityQuantify")
 async def article_quality_quantify(param: str) -> dict:
     limit = parse_limit(param, ContentQualityConst.QUANTIFY_DEAL_LIMIT)
-    logger.info("articleQualityQuantify started, limit=%d", limit)
+    tlog = get_task_logger("articleQualityQuantify")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        service = ArticleQualityQuantify(mysql)
+        service = ArticleQualityQuantify(mysql, task_logger=tlog)
         result = await service.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("量化评估失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "异常"}
 
     logger.info(
@@ -37,6 +40,7 @@ async def article_quality_quantify(param: str) -> dict:
         result["computed"],
         result["errors"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": (

src/handlers/article_quality_title.py

@@ -12,6 +12,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_quality import ArticleQualityTitle, ContentQualityConst
 from src.handlers._utils import get_db, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -19,14 +20,16 @@ logger = logging.getLogger(__name__)
 @xxl_job("articleQualityTitle")
 async def article_quality_title(param: str) -> dict:
     limit = parse_limit(param, ContentQualityConst.TITLE_DEAL_LIMIT)
-    logger.info("articleQualityTitle started, limit=%d", limit)
+    tlog = get_task_logger("articleQualityTitle")
+    await tlog.task_started(limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        service = ArticleQualityTitle(mysql)
+        service = ArticleQualityTitle(mysql, task_logger=tlog)
         result = await service.deal(limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("标题评分失败")
+        await tlog.task_failed(e)
         return {"code": 500, "msg": "异常"}
 
     logger.info(
@@ -34,6 +37,7 @@ async def article_quality_title(param: str) -> dict:
         result["seeded"],
         result["evaluated"],
     )
+    await tlog.task_completed(**result)
     return {
         "code": 200,
         "msg": f"seeded={result['seeded']} evaluated={result['evaluated']}",

src/handlers/article_search.py

@@ -12,6 +12,7 @@ import logging
 from src.infra.xxl_jobs import xxl_job
 from src.domains.content_supply import DemandSearchArticle, ContentSupplyConst
 from src.handlers._utils import get_db, parse_params, parse_limit
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -21,14 +22,16 @@ async def article_search(param: str) -> dict:
     p = parse_params(param)
     dt = p.get("dt") or None
     limit = parse_limit(param, ContentSupplyConst.SEARCH_DEAL_LIMIT)
-    logger.info("articleSearch started, dt=%s, limit=%d", dt, limit)
+    tlog = get_task_logger("articleSearch")
+    await tlog.task_started(dt=dt, limit=limit)
 
     try:
         mysql = get_db().mysql_manager("long_articles")
-        searcher = DemandSearchArticle(mysql)
+        searcher = DemandSearchArticle(mysql, task_logger=tlog)
         result = await searcher.deal(dt, limit=limit)
-    except Exception:
+    except Exception as e:
         logger.exception("搜索失败, dt=%s", dt)
+        await tlog.task_failed(e, dt=dt)
         return {"code": 500, "msg": "搜索异常"}
 
     logger.info(
@@ -38,6 +41,7 @@ async def article_search(param: str) -> dict:
         result["processed"],
         result["results_written"],
     )
+    await tlog.task_completed(**result, dt=dt)
     return {
         "code": 200,
         "msg": f"processed={result['processed']}, written={result['results_written']}",

src/handlers/demand_enqueue.py

@@ -14,6 +14,7 @@ from src.config.odps import OdpsConfig
 from src.infra.xxl_jobs import xxl_job
 from src.domains.demand_enqueue_task import FetchDemands, EnqueueDemands
 from src.handlers._utils import get_db, parse_dt
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
@@ -25,35 +26,46 @@ def _default_dt() -> str:
 @xxl_job("demandEnqueue")
 async def demand_enqueue(param: str) -> dict:
     dt = parse_dt(param, _default_dt())
-    logger.info("demandEnqueue started, dt=%s", dt)
+    tlog = get_task_logger("demandEnqueue")
+    await tlog.task_started(dt=dt)
 
     # 1. ODPS 拉取 → PG 查搜索词
     try:
+        await tlog.phase_start("fetch", dt=dt)
         pg = get_db().pg_manager("pg")
-        fetcher = FetchDemands(OdpsConfig(), pg)
+        fetcher = FetchDemands(OdpsConfig(), pg, task_logger=tlog)
         items = await fetcher.deal(dt)
-    except Exception:
+    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)
         return {"code": 200, "msg": "no items"}
 
     # 2. MySQL 入队
     try:
+        await tlog.phase_start("enqueue", dt=dt)
         mysql = get_db().mysql_manager("long_articles")
         enqueuer = EnqueueDemands(mysql)
         n = await enqueuer.deal(items)
-    except Exception:
+        accounts = len(set(it["account"] for it in items))
+    except Exception as e:
         logger.exception("MySQL 入队失败, dt=%s", dt)
+        await tlog.task_failed(e, dt=dt)
         return {"code": 500, "msg": "入队异常"}
 
-    accounts = len(set(it["account"] for it in items))
+    await tlog.phase_end("enqueue", enqueued=n, accounts=accounts, dt=dt)
+
     logger.info(
         "demandEnqueue done: enqueued %d items (%d accounts) for dt=%s",
         n,
         accounts,
         dt,
     )
+    await tlog.task_completed(enqueued=n, accounts=accounts, dt=dt)
     return {"code": 200, "msg": f"enqueued={n}, accounts={accounts}"}

src/handlers/test_hello_world.py

@@ -3,13 +3,18 @@
 import logging
 
 from src.infra.xxl_jobs import xxl_job
+from src.handlers._logging import get_task_logger
 
 logger = logging.getLogger(__name__)
 
 
 @xxl_job("testHelloWorld")
 async def test_hello_world(param: str) -> dict:
-    """测试 handler：打印 hello world"""
+    """测试 handler：验证 SLS 日志全链路 trace_id 穿透"""
+    tlog = get_task_logger("testHelloWorld")
+    await tlog.task_started(param=param)
     logger.info("testHelloWorld triggered, param=%s", param)
-    print(f"hello world! param: {param}")
+    msg = f"hello world! param: {param}"
+    print(msg)
+    await tlog.task_completed(message=msg)
     return {"code": 200, "msg": "hello world printed"}

src/infra/__init__.py

@@ -20,6 +20,7 @@ from src.infra.observability import (
     LogCategory,
     LogLevel,
     LogStatus,
+    TaskEvent,
     LogRecord,
     LogService,
     SLS_INDEX_CONFIG,
@@ -76,6 +77,7 @@ __all__ = [
     "LogCategory",
     "LogLevel",
     "LogStatus",
+    "TaskEvent",
     "LogRecord",
     "LogService",
     "SLS_INDEX_CONFIG",

src/infra/external/volcengine.py

@@ -19,7 +19,13 @@ deepseek_model = config.volcengine.deepseek_v4_pro
 def fetch_deepseek_completion(
     prompt: str,
     output_type: str = "text",
-) -> Optional[Dict | List]:
+) -> Optional[Dict | List | str]:
+    """调用火山引擎 DeepSeek API
+
+    Args:
+        prompt: 提示词
+        output_type: "text" 返回原始文本, "json" 返回解析后的 JSON
+    """
     input_messages = [{"role": "user", "content": prompt}]
     kwargs: dict[str, Any] = {
         "model": deepseek_model,

src/infra/observability/__init__.py

@@ -1,18 +1,29 @@
-"""可观测性设施层 —— 结构化日志模型、SLS 索引定义、日志服务"""
+"""可观测性设施层 —— 结构化日志模型、SLS 索引定义、日志服务、追踪上下文"""
 
 from src.infra.observability.log_schema import (
     LogCategory,
     LogLevel,
     LogStatus,
+    TaskEvent,
     LogRecord,
     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",
     "LogLevel",
+    "LogStatus",
+    "TaskEvent",
     "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/observability/log_schema.py

@@ -115,6 +115,31 @@ class LogStatus(str, Enum):
     PENDING = "pending"
 
 
+class TaskEvent(str, Enum):
+    """任务事件命名标准 —— 统一 event 字段取值，方便 SLS 精确检索"""
+
+    # ── 任务生命周期 ──
+    TASK_RECEIVED = "task.received"  # XXL-JOB 调度到达
+    TASK_STARTED = "task.started"  # handler 开始执行
+    TASK_COMPLETED = "task.completed"  # handler 正常结束
+    TASK_FAILED = "task.failed"  # handler 异常终止
+    TASK_TIMEOUT = "task.timeout"  # handler 超时
+    TASK_CANCELLED = "task.cancelled"  # handler 被取消
+
+    # ── 阶段标记 ──
+    PHASE_START = "phase.start"
+    PHASE_END = "phase.end"
+
+    # ── 业务事件 ──
+    ITEM_PROCESSED = "item.processed"  # 单条处理完成
+    ITEM_FAILED = "item.failed"  # 单条处理失败
+    API_CALL = "api.call"  # 外部 API 调用
+    DB_OPERATION = "db.operation"  # 数据库操作
+    LLM_CALL = "llm.call"  # LLM 推理调用
+    CACHE_HIT = "cache.hit"  # 缓存命中
+    CACHE_MISS = "cache.miss"  # 缓存未命中
+
+
 # ==================== 结构化日志模型 ====================
 
 
@@ -148,6 +173,10 @@ class LogRecord(BaseModel):
 
     # ── 业务标识 ──
     task_name: str = Field(default="", description="任务名，如 content_generation")
+    phase: str = Field(
+        default="",
+        description="当前管道阶段: search/detail/account/article/update/quality/category/title/quantify/enqueue",
+    )
     agent_name: str = Field(default="", description="Agent 名，如 ContentWriter")
     tool_name: str = Field(default="", description="Tool 名，如 WebSearchTool")
 
@@ -179,6 +208,7 @@ class LogRecord(BaseModel):
             ("trace_id", self.trace_id),
             ("span_id", self.span_id),
             ("task_name", self.task_name),
+            ("phase", self.phase),
             ("agent_name", self.agent_name),
             ("tool_name", self.tool_name),
             ("duration_ms", str(self.duration_ms)),
@@ -210,7 +240,9 @@ SLS_INDEX_CONFIG = {
         {"key": "category", "type": "text", "alias": "log_category"},
         {"key": "event", "type": "text"},
         {"key": "trace_id", "type": "text"},
+        {"key": "span_id", "type": "text"},
         {"key": "task_name", "type": "text"},
+        {"key": "phase", "type": "text"},
         {"key": "agent_name", "type": "text"},
         {"key": "tool_name", "type": "text"},
         {"key": "duration_ms", "type": "long", "stats": True},

src/infra/spider/wechat/gzh.py

@@ -31,7 +31,7 @@ HEADERS = {"Content-Type": "application/json"}
 
 @retry(**retry_desc)
 async def get_article_detail(
-    article_link: str, is_count: bool = False, is_cache: bool = True
+    article_link: str, is_count: bool = False, is_cache: bool = True, tlog=None
 ) -> dict | None:
     """获取公众号文章详情
 
@@ -39,6 +39,7 @@ async def get_article_detail(
         article_link: 文章链接
         is_count: 是否仅返回计数
         is_cache: 是否优先使用缓存
+        tlog: 可选的 TaskLogger，用于记录 API 调用事件
     """
     target_url = f"{BASE_URL}/detail"
     payload = json.dumps(
@@ -54,13 +55,15 @@ async def get_article_detail(
             response = await http_client.post(target_url, headers=HEADERS, data=payload)
     except Exception as e:
         logger.error("get_article_detail API请求失败: %s, link=%s", e, article_link)
+        if tlog:
+            await tlog.error("api.call", f"详情API请求失败: {article_link[:80]}", exc=e, phase="detail_fetch")
         return None
     return response
 
 
 @retry(**retry_desc)
 async def get_article_list_from_account(
-    account_id: str, index: str | None = None, is_cache: bool = True
+    account_id: str, index: str | None = None, is_cache: bool = True, tlog=None
 ) -> dict | None:
     """获取公众号文章列表
 
@@ -68,6 +71,7 @@ async def get_article_list_from_account(
         account_id: 公众号 ID
         index: 分页游标
         is_cache: 是否优先使用缓存
+        tlog: 可选的 TaskLogger
     """
     target_url = f"{BASE_URL}/blogger"
     payload = json.dumps(
@@ -88,6 +92,13 @@ async def get_article_list_from_account(
             account_id,
             index,
         )
+        if tlog:
+            await tlog.error(
+                "api.call",
+                f"文章列表API请求失败: account_id={account_id[:40]}",
+                exc=e,
+                phase="article_fetch",
+            )
         return None
     return response
 
@@ -131,12 +142,13 @@ def get_source_account_from_article(article_link: str) -> dict | None:
 
 
 @retry(**retry_desc)
-async def weixin_search(keyword: str, page: str = "1") -> dict | None:
+async def weixin_search(keyword: str, page: str = "1", tlog=None) -> dict | None:
     """微信关键词搜索
 
     Args:
         keyword: 搜索关键词
         page: 分页游标
+        tlog: 可选的 TaskLogger
     """
     url = f"{BASE_URL}/keyword"
     payload = json.dumps({"keyword": keyword, "cursor": page})
@@ -147,16 +159,19 @@ async def weixin_search(keyword: str, page: str = "1") -> dict | None:
         logger.error(
             "weixin_search API请求失败: %s, keyword=%s, page=%s", e, keyword, page
         )
+        if tlog:
+            await tlog.error("api.call", f"搜索API请求失败: {keyword[:40]}", exc=e, phase="search")
         return None
     return response
 
 
 @retry(**retry_desc)
-async def get_account_from_url(article_url) -> dict | None:
-    """微信关键词搜索
+async def get_account_from_url(article_url, tlog=None) -> dict | None:
+    """通过文章链接获取公众号账号信息
 
     Args:
         article_url: 文章链接
+        tlog: 可选的 TaskLogger
     """
     url = f"{BASE_URL}/account_info"
     payload = json.dumps({"content_link": article_url})
@@ -167,5 +182,7 @@ async def get_account_from_url(article_url) -> dict | None:
         logger.error(
             "get_account_from_url API请求失败: %s, article_url=%s", e, article_url
         )
+        if tlog:
+            await tlog.error("api.call", f"账号API请求失败: {article_url[:80]}", exc=e, phase="account_fetch")
         return None
     return response

src/infra/xxl_jobs/xxl_executor.py

@@ -15,6 +15,15 @@ from typing import Any, Dict, Optional
 import aiohttp
 
 from src.config.xxljob import XxlJobConfig
+from src.infra.shared.tools import generate_task_trace_id
+from src.infra.observability.context import _current_trace_id, get_log_service
+from src.infra.observability import (
+    LogCategory,
+    LogLevel,
+    LogStatus,
+    LogRecord,
+    TaskEvent,
+)
 from .xxl_decorator import xxl_registry
 
 logger = logging.getLogger(__name__)
@@ -129,7 +138,7 @@ class XxlJobExecutor:
             except Exception:
                 pass
 
-    async def _callback(self, log_id: int, log_date_time: int, result: dict) -> None:
+    async def _callback(self, log_id: int, log_date_time: int, result: dict) -> bool:
         """向 Admin 回调上报任务执行结果 —— POST /api/callback
 
         对标 Java 版的 TriggerCallbackThread.callback()。
@@ -138,10 +147,13 @@ class XxlJobExecutor:
             log_id: XXL-JOB 调度日志 ID
             log_date_time: 调度时间戳（毫秒）
             result: handler 返回的 {"code": 200, "msg": "..."} 或异常结果
+
+        Returns:
+            True 如果至少向一个 Admin 成功上报，False 全部失败
         """
         if not log_id:
             logger.warning("Skip callback: logId is empty")
-            return
+            return False
 
         body = {
             "logId": log_id,
@@ -159,13 +171,14 @@ class XxlJobExecutor:
                     if resp.status == 200:
                         data = await resp.json()
                         if data.get("code") == 200:
-                            return
+                            return True
                         logger.warning(
                             "Callback to %s rejected: %s", admin_url, data.get("msg")
                         )
             except Exception as e:
                 logger.warning("Callback to %s failed: %s", admin_url, e)
         logger.error("All callback attempts failed for logId=%s", log_id)
+        return False
 
     async def _heartbeat_loop(self) -> None:
         """定期心跳：每隔 registry_interval 秒重新注册（续约）"""
@@ -223,8 +236,13 @@ class XxlJobExecutor:
                 f"Registered: {', '.join(registered) if registered else '(none)'}",
             }
 
+        # 生成 trace_id 并写入 ContextVar，穿透到 handler → service 全链路
+        trace_id = generate_task_trace_id()
+        _current_trace_id.set(trace_id)
+
         async def _execute_and_callback():
             """后台执行 handler 并回调 Admin 上报结果"""
+            _log_task_received(trace_id, handler_name, job_id, param)
             try:
                 if timeout and timeout > 0:
                     result = await asyncio.wait_for(handler(param), timeout=timeout)
@@ -234,17 +252,26 @@ class XxlJobExecutor:
                 logger.error(
                     "Job %s(id=%s) timeout after %ss", handler_name, job_id, timeout
                 )
+                _log_task_timeout(trace_id, handler_name, job_id, timeout)
                 result = {"code": 500, "msg": f"timeout after {timeout}s"}
             except asyncio.CancelledError:
                 logger.info("Job %s(id=%s) cancelled", handler_name, job_id)
+                _log_task_cancelled(trace_id, handler_name, job_id)
                 result = {"code": 500, "msg": "job cancelled"}
             except Exception as e:
                 logger.exception("Job %s(id=%s) failed", handler_name, job_id)
+                _log_task_failed(trace_id, handler_name, job_id, e)
                 result = {"code": 500, "msg": str(e)}
 
             # 回调 Admin 上报执行结果
-            await self._callback(log_id, log_date_time, result)
-            logger.info("Job %s(id=%s) callback done: %s", handler_name, job_id, result)
+            callback_ok = await self._callback(log_id, log_date_time, result)
+            if callback_ok:
+                logger.info(
+                    "Job %s(id=%s) callback succeeded: %s",
+                    handler_name,
+                    job_id,
+                    result,
+                )
 
         task = asyncio.create_task(_execute_and_callback())
         if job_id is not None:
@@ -254,9 +281,10 @@ class XxlJobExecutor:
         task.add_done_callback(lambda t: self._running_jobs.pop(job_id, None))
 
         logger.info(
-            "Job %s(id=%s) dispatched, will callback when done",
+            "Job %s(id=%s) dispatched, trace_id=%s, will callback when done",
             handler_name,
             job_id,
+            trace_id,
         )
         return {"code": 200, "msg": None}
 
@@ -336,3 +364,91 @@ class XxlJobExecutor:
             return ip
         except Exception:
             return "127.0.0.1"
+
+
+# ═══════════════════════════════════════════════════════════════
+# SLS 日志辅助函数 —— XXL-JOB 生命周期事件
+# ═══════════════════════════════════════════════════════════════
+
+
+def _log_task_received(
+    trace_id: str, handler_name: str, job_id: int, param: str
+) -> None:
+    """记录 task.received 到 SLS（非阻塞，fire-and-forget）"""
+    try:
+        ls = get_log_service()
+        record = LogRecord(
+            level=LogLevel.INFO,
+            category=LogCategory.TASK,
+            event=TaskEvent.TASK_RECEIVED,
+            message=f"XXL-JOB 调度到达: {handler_name}",
+            trace_id=trace_id,
+            task_name=handler_name,
+            extra={"job_id": job_id, "param": param},
+        )
+        asyncio.create_task(ls.log(record))
+    except RuntimeError:
+        pass  # LogService 未注入（测试环境），静默跳过
+
+
+def _log_task_timeout(
+    trace_id: str, handler_name: str, job_id: int, timeout: int
+) -> None:
+    """记录 task.timeout"""
+    try:
+        ls = get_log_service()
+        record = LogRecord(
+            level=LogLevel.ERROR,
+            category=LogCategory.TASK,
+            event=TaskEvent.TASK_TIMEOUT,
+            message=f"任务超时: {handler_name} (timeout={timeout}s)",
+            trace_id=trace_id,
+            task_name=handler_name,
+            status=LogStatus.TIMEOUT,
+            extra={"job_id": job_id, "timeout_s": timeout},
+        )
+        asyncio.create_task(ls.log(record))
+    except RuntimeError:
+        pass
+
+
+def _log_task_cancelled(trace_id: str, handler_name: str, job_id: int) -> None:
+    """记录 task.cancelled"""
+    try:
+        ls = get_log_service()
+        record = LogRecord(
+            level=LogLevel.WARN,
+            category=LogCategory.TASK,
+            event=TaskEvent.TASK_CANCELLED,
+            message=f"任务被取消: {handler_name}",
+            trace_id=trace_id,
+            task_name=handler_name,
+            status=LogStatus.CANCELLED,
+            extra={"job_id": job_id},
+        )
+        asyncio.create_task(ls.log(record))
+    except RuntimeError:
+        pass
+
+
+def _log_task_failed(
+    trace_id: str, handler_name: str, job_id: int, error: Exception
+) -> None:
+    """记录 task.failed（handler 未捕获的顶层异常）"""
+    try:
+        ls = get_log_service()
+        record = LogRecord(
+            level=LogLevel.ERROR,
+            category=LogCategory.TASK,
+            event=TaskEvent.TASK_FAILED,
+            message=str(error),
+            trace_id=trace_id,
+            task_name=handler_name,
+            status=LogStatus.FAILED,
+            error_type=type(error).__name__,
+            error_message=str(error),
+            extra={"job_id": job_id},
+        )
+        asyncio.create_task(ls.log(record))
+    except RuntimeError:
+        pass

src/server/app.py

@@ -13,6 +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.infra.observability.context import set_log_service
 from src.server.xxl_endpoints import set_xxl_executor
 
 logger = logging.getLogger(__name__)
@@ -46,6 +47,7 @@ def create_app(global_config: GlobalConfig = None) -> Quart:
 
     set_xxl_executor(xxl_executor)
     set_db(db)
+    set_log_service(log_service)
 
     app = Quart(__name__)
     app = cors(app, allow_origin="*")