外部文章质量判断

src/domains/content_decode/__init__.py

@@ -1,3 +1,3 @@
 """
 供给文章解构模块
-"""
+"""

src/domains/content_quality/_mapper.py

@@ -188,7 +188,10 @@ class QualityMapper(ContentQualityConst):
         )
 
     async def get_avg_view_count_60d(
-        self, gh_id: str, item_index: int, publish_time,
+        self,
+        gh_id: str,
+        item_index: int,
+        publish_time,
     ) -> Optional[Dict]:
         query = f"""
             SELECT AVG(show_view_count) AS avg_count, COUNT(*) AS sample_count
@@ -203,8 +206,12 @@ class QualityMapper(ContentQualityConst):
         return await self.pool.fetch_one(
             query=query,
             params=(
-                gh_id, item_index, publish_time,
-                self.AVG_WINDOW_DAYS, publish_time, self.MASS_SEND_TYPE,
+                gh_id,
+                item_index,
+                publish_time,
+                self.AVG_WINDOW_DAYS,
+                publish_time,
+                self.MASS_SEND_TYPE,
             ),
         )
 
@@ -230,8 +237,13 @@ class QualityMapper(ContentQualityConst):
         return await self.pool.save(
             query=query,
             params=(
-                avg_view_count_60d, view_ratio, like_ratio,
-                sample_count_60d, status, item_id, self.Status.PROCESSING,
+                avg_view_count_60d,
+                view_ratio,
+                like_ratio,
+                sample_count_60d,
+                status,
+                item_id,
+                self.Status.PROCESSING,
             ),
         )
 
@@ -241,7 +253,9 @@ class QualityMapper(ContentQualityConst):
 
     async def seed_titles(self, *, limit: int = 500) -> int:
         """relation → 新标题 INSERT title 表"""
-        rows = await self.fetch_relation_titles_not_in_table(self.TITLE_TABLE, limit=limit)
+        rows = await self.fetch_relation_titles_not_in_table(
+            self.TITLE_TABLE, limit=limit
+        )
         if not rows:
             return 0
         query = f"""
@@ -315,7 +329,8 @@ class QualityMapper(ContentQualityConst):
 
     async def seed_categories(self, *, limit: int = 500) -> int:
         rows = await self.fetch_relation_titles_not_in_table(
-            self.CATEGORY_TABLE, limit=limit,
+            self.CATEGORY_TABLE,
+            limit=limit,
         )
         if not rows:
             return 0
@@ -365,7 +380,12 @@ class QualityMapper(ContentQualityConst):
             """
             n = await self.pool.save(
                 query=query,
-                params=(u["category"], self.Status.DONE, u["id"], self.Status.PROCESSING),
+                params=(
+                    u["category"],
+                    self.Status.DONE,
+                    u["id"],
+                    self.Status.PROCESSING,
+                ),
             )
             total += n
         return total

src/domains/content_quality/article_quality_category.py

@@ -16,7 +16,6 @@ logger = logging.getLogger(__name__)
 
 
 class ArticleQualityCategory(ContentQualityConst):
-
     def __init__(self, pool: MysqlManager):
         self.mapper = QualityMapper(pool)
         self._evaluated = 0
@@ -41,11 +40,13 @@ class ArticleQualityCategory(ContentQualityConst):
             batch_ids = [r["id"] for r in batch]
             titles = [{"id": r["id"], "title": r["title"]} for r in batch]
             prompt = build_category_prompt(titles)
-            batches.append({
-                "batch_ids": batch_ids,
-                "titles": titles,
-                "prompt": prompt,
-            })
+            batches.append(
+                {
+                    "batch_ids": batch_ids,
+                    "titles": titles,
+                    "prompt": prompt,
+                }
+            )
 
         # 先一次性标记 PROCESSING，避免并发 CAS 冲突
         for b in batches:
@@ -62,7 +63,10 @@ class ArticleQualityCategory(ContentQualityConst):
 
         logger.info(
             "品类识别完成: seeded=%d, batches=%d, evaluated=%d, errors=%d",
-            seeded, len(batches), self._evaluated, len(result["errors"]),
+            seeded,
+            len(batches),
+            self._evaluated,
+            len(result["errors"]),
         )
         return {"seeded": seeded, "evaluated": self._evaluated}
 
@@ -87,7 +91,9 @@ class ArticleQualityCategory(ContentQualityConst):
 
     async def _call_llm(self, prompt: str) -> str | None:
         loop = asyncio.get_running_loop()
-        return await loop.run_in_executor(None, fetch_deepseek_completion, prompt, "text")
+        return await loop.run_in_executor(
+            None, fetch_deepseek_completion, prompt, "text"
+        )
 
 
 __all__ = ["ArticleQualityCategory"]

src/domains/content_quality/article_quality_quantify.py

@@ -13,7 +13,6 @@ logger = logging.getLogger(__name__)
 
 
 class ArticleQualityQuantify(ContentQualityConst):
-
     def __init__(self, pool: MysqlManager):
         self.mapper = QualityMapper(pool)
 
@@ -24,7 +23,13 @@ class ArticleQualityQuantify(ContentQualityConst):
         account_ids = await self.mapper.fetch_crawled_account_ids()
         if not account_ids:
             logger.info("无已完成深翻的账号")
-            return {"validated": 0, "skipped": 0, "inserted": 0, "computed": 0, "errors": 0}
+            return {
+                "validated": 0,
+                "skipped": 0,
+                "inserted": 0,
+                "computed": 0,
+                "errors": 0,
+            }
         logger.info("已深翻账号数: %d", len(account_ids))
 
         total_validated = 0
@@ -49,7 +54,11 @@ class ArticleQualityQuantify(ContentQualityConst):
 
         logger.info(
             "量化评估完成: validated=%d skipped=%d inserted=%d computed=%d errors=%d",
-            total_validated, total_skipped, total_inserted, total_computed, total_errors,
+            total_validated,
+            total_skipped,
+            total_inserted,
+            total_computed,
+            total_errors,
         )
         return {
             "validated": total_validated,
@@ -61,7 +70,9 @@ class ArticleQualityQuantify(ContentQualityConst):
 
     async def _validate_and_insert(self, account_id: str, *, limit: int) -> dict:
         """处理单个账号的待校验文章"""
-        rows = await self.mapper.fetch_pending_detail_by_account(account_id, limit=limit)
+        rows = await self.mapper.fetch_pending_detail_by_account(
+            account_id, limit=limit
+        )
         if not rows:
             return {"validated": 0, "skipped": 0, "inserted": 0, "errors": 0}
 
@@ -87,7 +98,10 @@ class ArticleQualityQuantify(ContentQualityConst):
                 article_type = pos.get("type", 0)
                 item_index = pos.get("item_index", 0)
 
-                if article_type != self.MASS_SEND_TYPE or item_index not in self.VALID_POSITIONS:
+                if (
+                    article_type != self.MASS_SEND_TYPE
+                    or item_index not in self.VALID_POSITIONS
+                ):
                     await self.mapper.mark_detail_failed(
                         detail_id,
                         f"type={article_type}, pos={item_index}",
@@ -95,11 +109,13 @@ class ArticleQualityQuantify(ContentQualityConst):
                     skipped += 1
                     continue
 
-                await self.mapper.insert_quantile({
-                    "wx_sn": wx_sn,
-                    "view_count": row.get("view_count") or 0,
-                    "like_count": row.get("like_count") or 0,
-                })
+                await self.mapper.insert_quantile(
+                    {
+                        "wx_sn": wx_sn,
+                        "view_count": row.get("view_count") or 0,
+                        "like_count": row.get("like_count") or 0,
+                    }
+                )
                 await self.mapper.mark_detail_done(detail_id)
                 inserted += 1
 
@@ -111,7 +127,12 @@ class ArticleQualityQuantify(ContentQualityConst):
                     pass
                 errors += 1
 
-        return {"validated": len(rows), "skipped": skipped, "inserted": inserted, "errors": errors}
+        return {
+            "validated": len(rows),
+            "skipped": skipped,
+            "inserted": inserted,
+            "errors": errors,
+        }
 
     async def _compute(self, *, limit: int) -> dict:
         rows = await self.mapper.fetch_pending_quantile(limit=limit)
@@ -144,7 +165,9 @@ class ArticleQualityQuantify(ContentQualityConst):
                     item_index = pos.get("item_index", 0)
 
                     avg_row = await self.mapper.get_avg_view_count_60d(
-                        gh_id, item_index, row.get("publish_at"),
+                        gh_id,
+                        item_index,
+                        row.get("publish_at"),
                     )
 
                     avg_count = None
@@ -175,7 +198,8 @@ class ArticleQualityQuantify(ContentQualityConst):
                     logger.exception("量化计算失败 id=%d wx_sn=%s", item_id, wx_sn)
                     try:
                         await self.mapper.update_quantile_result(
-                            item_id, status=self.Status.FAIL,
+                            item_id,
+                            status=self.Status.FAIL,
                         )
                     except Exception:
                         pass

src/domains/content_quality/article_quality_title.py

@@ -16,7 +16,6 @@ logger = logging.getLogger(__name__)
 
 
 class ArticleQualityTitle(ContentQualityConst):
-
     def __init__(self, pool: MysqlManager):
         self.mapper = QualityMapper(pool)
         self._evaluated = 0
@@ -42,11 +41,13 @@ class ArticleQualityTitle(ContentQualityConst):
             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)
-            batches.append({
-                "batch_ids": batch_ids,
-                "titles": titles,
-                "prompt": prompt,
-            })
+            batches.append(
+                {
+                    "batch_ids": batch_ids,
+                    "titles": titles,
+                    "prompt": prompt,
+                }
+            )
 
         # 并发执行（先标记 PROCESSING，避免并发冲突）
         for b in batches:
@@ -68,7 +69,10 @@ class ArticleQualityTitle(ContentQualityConst):
 
         logger.info(
             "标题评分完成: seeded=%d, batches=%d, evaluated=%d, errors=%d",
-            seeded, len(batches), self._evaluated, len(result["errors"]),
+            seeded,
+            len(batches),
+            self._evaluated,
+            len(result["errors"]),
         )
         return {"seeded": seeded, "evaluated": self._evaluated}
 
@@ -94,7 +98,9 @@ class ArticleQualityTitle(ContentQualityConst):
 
     async def _call_llm(self, prompt: str) -> str | None:
         loop = asyncio.get_running_loop()
-        return await loop.run_in_executor(None, fetch_deepseek_completion, prompt, "text")
+        return await loop.run_in_executor(
+            None, fetch_deepseek_completion, prompt, "text"
+        )
 
 
 __all__ = ["ArticleQualityTitle"]

src/domains/content_supply/_const.py

@@ -87,7 +87,7 @@ class ContentSupplyConst:
 
     ARTICLE_DEAL_LIMIT = 10  # 单次处理账号数
     ARTICLE_PAGE_INTERVAL_SEC = 2  # 翻页间隔（秒）
-    ARTICLE_BATCH_INSERT = 200  # 文章批量写入每批条数
+    MAX_DEEP_CRAWL_PAGES = 100  # 深翻单次最大翻页数，防止无限循环
     ONE_YEAR_SECONDS = 365 * 24 * 3600
 
     # ═══════════════════════════════════════════════════════════════

src/domains/content_supply/_mapper.py

@@ -237,7 +237,12 @@ class ArticleSearchRelationMapper(ContentSupplyConst):
         """
         return await self.pool.save(
             query=query,
-            params=(self.SearchStatus.FAIL, self.SearchStatus.INIT, self.MIN_QUALITY_SCORE, limit),
+            params=(
+                self.SearchStatus.FAIL,
+                self.SearchStatus.INIT,
+                self.MIN_QUALITY_SCORE,
+                limit,
+            ),
         )
 
     # ═══════════════════════════════════════════════════════════════
@@ -466,7 +471,7 @@ class AccountMapper(ContentSupplyConst):
             SELECT a.channel_account_id, a.gh_id, a.crawl_cursor, a.crawl_deep_done,
                    MAX(r.quality_score) AS max_quality_score
             FROM {self.ACCOUNT_TABLE} a
-            LEFT JOIN {self.RELATION_TABLE} r ON a.biz = r.biz
+            LEFT JOIN {self.RELATION_TABLE} r ON a.biz COLLATE utf8mb4_unicode_ci = r.biz
             WHERE a.gh_id IS NOT NULL AND a.gh_id != ''
             GROUP BY a.channel_account_id, a.gh_id, a.crawl_cursor, a.crawl_deep_done
             ORDER BY a.crawl_deep_done ASC, max_quality_score DESC, a.gh_id ASC
@@ -538,14 +543,6 @@ class AccountMapper(ContentSupplyConst):
         ]
         return await self.pool.save(query=query, params=params, batch=True)
 
-    async def count_articles_by_gh_id(self, gh_id: str) -> int:
-        """统计某账号已抓文章数"""
-        row = await self.pool.fetch_one(
-            query=f"SELECT COUNT(*) AS cnt FROM {self.ARTICLES_TABLE} WHERE gh_id = %s",
-            params=(gh_id,),
-        )
-        return row["cnt"] if row else 0
-
     # ═══════════════════════════════════════════════════════════════
     # demand_account_articles 表 — 指标更新
     # ═══════════════════════════════════════════════════════════════

src/domains/content_supply/account_article_fetch.py

@@ -3,8 +3,9 @@
 流程:
   1. 取 accounts 表中所有有 gh_id 的账号（crawl_deep_done=0 的优先）
   2. 每个账号:
-     - 0: 首次深翻 → 翻页 → send_time < 一年前停止 → 标记 deep_done=1
-     - 1: 增量更新 → 只取第一页
+     - crawl_deep_done=0: 深翻 → 逐页拉取一年内文章（每页持久化游标，崩溃可断点续翻）
+       → 遇到一年前文章或翻完停止 → 标记 deep_done=1
+     - crawl_deep_done=1: 增量更新 → 只取第一页
   3. 文章 INSERT IGNORE (wx_sn 去重)
 """
 
@@ -13,6 +14,7 @@ import logging
 import time
 
 from src.infra.database.mysql.manager import MysqlManager
+from src.infra.shared.async_tasks import run_tasks_with_async_worker_group
 
 from ._const import ContentSupplyConst
 from ._mapper import AccountMapper
@@ -37,60 +39,85 @@ class AccountArticleFetch(ContentSupplyConst):
         if not accounts:
             logger.info("无待抓取文章的账号")
             return {"accounts": 0, "articles_written": 0, "errors": 0}
+
         total_articles = 0
-        errors = 0
+        lock = asyncio.Lock()
+        first_done = False
+
+        async def _process_one(acct: dict) -> None:
+            nonlocal total_articles, first_done
 
-        for i, acct in enumerate(accounts):
             gh_id = acct["gh_id"]
-            deep_done = acct.get("crawl_deep_done", 0)
-            cursor = acct.get("crawl_cursor") or None
+            try:
+                async with lock:
+                    need_sleep = first_done
+                    first_done = True
 
-            if i > 0:
-                await asyncio.sleep(self.ARTICLE_PAGE_INTERVAL_SEC)
+                if need_sleep:
+                    await asyncio.sleep(self.ARTICLE_PAGE_INTERVAL_SEC)
+
+                deep_done = acct.get("crawl_deep_done", 0)
+                cursor = acct.get("crawl_cursor") or None
 
-            try:
                 if deep_done:
-                    n = await self._incremental_crawl(gh_id, cursor)
+                    n = await self._incremental_crawl(gh_id)
                 else:
-                    n = await self._deep_crawl(gh_id)
-                    await self.mapper.update_crawl_status(gh_id, cursor="", deep_done=1)
+                    # _deep_crawl 内部逐页持久化游标，完成后自动标记 deep_done=1
+                    n = await self._deep_crawl(gh_id, start_cursor=cursor)
+
+                async with lock:
+                    total_articles += n
+
+                logger.info(
+                    "gh_id=%s 抓取完成: 本轮写入 %d 条, 累计 %d 条",
+                    gh_id,
+                    n,
+                    total_articles,
+                )
             except Exception:
                 logger.exception("账号文章抓取异常: gh_id=%s", gh_id)
-                errors += 1
-                continue
-
-            total_articles += n
-            prev_count = await self.mapper.count_articles_by_gh_id(gh_id)
-            logger.info(
-                "gh_id=%s 抓取完成: 本批写入 %d 条, 累计 %d 条",
-                gh_id,
-                n,
-                prev_count,
-            )
+                raise
+
+        result = await run_tasks_with_async_worker_group(
+            accounts,
+            _process_one,
+            description="账号文章抓取",
+            unit="account",
+            max_concurrency=1,
+            fail_fast=False,
+            show_progress=False,
+        )
 
+        errors_count = len(result["errors"])
         logger.info(
             "文章抓取完成: 账号 %d 个, 写入 %d 条, 异常 %d",
             len(accounts),
             total_articles,
-            errors,
+            errors_count,
         )
         return {
             "accounts": len(accounts),
             "articles_written": total_articles,
-            "errors": errors,
+            "errors": errors_count,
         }
 
     # ═══════════════════════════════════════════════════════════════
     # 抓取模式
     # ═══════════════════════════════════════════════════════════════
 
-    async def _deep_crawl(self, gh_id: str) -> int:
-        """首次深翻：逐页拉取，直到遇到一年前的文章"""
+    async def _deep_crawl(self, gh_id: str, *, start_cursor: str | None = None) -> int:
+        """首次深翻：逐页拉取直到遇到一年前的文章，每页持久化游标防崩溃丢进度
+
+        断点续翻：start_cursor 非空时从上次中断位置继续，而非从头开始。
+        send_time=0 的文章保留（时间戳未知，不丢弃）。
+        """
         total = 0
-        cursor = None
+        cursor = start_cursor
         cutoff = int(time.time()) - self.ONE_YEAR_SECONDS
+        pages = 0
 
-        while True:
+        while pages < self.MAX_DEEP_CRAWL_PAGES:
+            pages += 1
             (
                 articles,
                 next_cursor,
@@ -99,14 +126,27 @@ class AccountArticleFetch(ContentSupplyConst):
             ) = await fetch_article_page(gh_id, cursor)
 
             if articles:
-                # 筛选一年内的文章，过滤一年前的
-                fresh = [a for a in articles if a["send_time"] >= cutoff]
+                # 筛选一年内的文章；send_time=0（未知时间戳）保留不丢弃
+                fresh = [
+                    a
+                    for a in articles
+                    if a["send_time"] >= cutoff or a["send_time"] == 0
+                ]
                 old_count = len(articles) - len(fresh)
+                zero_ts = [a["wx_sn"] for a in articles if a["send_time"] == 0]
+                if zero_ts:
+                    logger.warning(
+                        "gh_id=%s 第 %d 页有 %d 条文章 send_time=0 已保留: %s",
+                        gh_id,
+                        pages,
+                        len(zero_ts),
+                        zero_ts,
+                    )
                 if fresh:
                     n = await self.mapper.insert_articles_batch(fresh)
                     total += n
 
-                # 出现一年前的文章 → 停止翻页
+                # 出现一年前的文章 → 停止翻页，标记完成
                 if old_count > 0 or (oldest_send_time and oldest_send_time < cutoff):
                     logger.info(
                         "gh_id=%s 深翻到达一年前边界: 本页 %d 条中 %d 条超一年, 停止",
@@ -114,17 +154,31 @@ class AccountArticleFetch(ContentSupplyConst):
                         len(articles),
                         old_count,
                     )
-                    break
+                    await self.mapper.update_crawl_status(gh_id, cursor="", deep_done=1)
+                    return total
 
             if not has_more or not next_cursor:
+                await self.mapper.update_crawl_status(gh_id, cursor="", deep_done=1)
                 break
 
             cursor = next_cursor
+            # 逐页持久化游标，崩溃后可从断点继续
+            await self.mapper.update_crawl_status(
+                gh_id, cursor=next_cursor or "", deep_done=0
+            )
             await asyncio.sleep(self.ARTICLE_PAGE_INTERVAL_SEC)
 
+        else:
+            # while 正常结束（非 break）→ 达到最大翻页数
+            logger.warning(
+                "gh_id=%s 深翻达到最大翻页数 %d，下次调度从当前游标继续",
+                gh_id,
+                self.MAX_DEEP_CRAWL_PAGES,
+            )
+
         return total
 
-    async def _incremental_crawl(self, gh_id: str, _cursor: str | None = None) -> int:
+    async def _incremental_crawl(self, gh_id: str) -> int:
         """增量更新：只取第一页即停"""
         articles, _, _, _ = await fetch_article_page(gh_id, None)
         if not articles:

src/domains/content_supply/article_fetch_detail.py

@@ -44,7 +44,11 @@ class ArticleFetchDetail(ContentSupplyConst):
         # 将低于阈值的低质文章标记为 FAIL，不拉详情
         dropped = await self.relation_mapper.mark_low_quality(limit=limit)
         if dropped:
-            logger.info("标记低质文章: dropped=%d (quality_score < %d)", dropped, self.MIN_QUALITY_SCORE)
+            logger.info(
+                "标记低质文章: dropped=%d (quality_score < %d)",
+                dropped,
+                self.MIN_QUALITY_SCORE,
+            )
 
         items = await self.relation_mapper.fetch_pending(limit=limit)
         if not items:

src/handlers/article_quality_category.py

@@ -31,7 +31,8 @@ async def article_quality_category(param: str) -> dict:
 
     logger.info(
         "done: seeded=%d evaluated=%d",
-        result["seeded"], result["evaluated"],
+        result["seeded"],
+        result["evaluated"],
     )
     return {
         "code": 200,

src/handlers/article_quality_quantify.py

@@ -31,8 +31,11 @@ async def article_quality_quantify(param: str) -> dict:
 
     logger.info(
         "done: validated=%d skipped=%d inserted=%d computed=%d errors=%d",
-        result["validated"], result["skipped"],
-        result["inserted"], result["computed"], result["errors"],
+        result["validated"],
+        result["skipped"],
+        result["inserted"],
+        result["computed"],
+        result["errors"],
     )
     return {
         "code": 200,

src/handlers/article_quality_title.py

@@ -31,7 +31,8 @@ async def article_quality_title(param: str) -> dict:
 
     logger.info(
         "done: seeded=%d evaluated=%d",
-        result["seeded"], result["evaluated"],
+        result["seeded"],
+        result["evaluated"],
     )
     return {
         "code": 200,

src/infra/xxl_jobs/xxl_executor.py

@@ -153,7 +153,7 @@ class XxlJobExecutor:
             try:
                 async with self._session.post(
                     f"{admin_url}{_PATH_CALLBACK}",
-                    json=body,
+                    json=[body],  # Admin 端接口签名为 List<HandleCallbackParam>，须传数组
                     headers=self._auth_headers(),
                 ) as resp:
                     if resp.status == 200: