init

src/domains/demand_search_account/_const.py

@@ -14,12 +14,12 @@ class DemandSearchAccountConst:
         SUCCESS = 2
         FAIL = 99
 
-    ACCOUNT_DEAL_LIMIT = 100       # 单次取待抓账号条数
-    ACCOUNT_INTERVAL_SEC = 1       # API 调用间隔（秒）
+    ACCOUNT_DEAL_LIMIT = 100  # 单次取待抓账号条数
+    ACCOUNT_INTERVAL_SEC = 1  # API 调用间隔（秒）
 
-    ARTICLE_DEAL_LIMIT = 10        # 单次处理账号数
+    ARTICLE_DEAL_LIMIT = 10  # 单次处理账号数
     ARTICLE_PAGE_INTERVAL_SEC = 2  # 翻页间隔（秒）
-    ARTICLE_BATCH_INSERT = 200     # 文章批量写入每批条数
+    ARTICLE_BATCH_INSERT = 200  # 文章批量写入每批条数
     ONE_YEAR_SECONDS = 365 * 24 * 3600
 
 

src/domains/demand_search_account/_mapper.py

@@ -29,7 +29,8 @@ class AccountMapper(DemandSearchAccountConst):
             LIMIT %s
         """
         return await self.pool.fetch(
-            query=query, params=(self.AccountSaveStatus.INIT, limit),
+            query=query,
+            params=(self.AccountSaveStatus.INIT, limit),
         )
 
     async def fetch_relation_ids_by_biz(self, biz: str) -> List[int]:
@@ -44,7 +45,11 @@ class AccountMapper(DemandSearchAccountConst):
         """取某个 biz 下未完成（INIT/PROCESSING）的 relation id，异常恢复用"""
         rows = await self.pool.fetch(
             query=f"SELECT id FROM {self.RELATION_TABLE} WHERE biz = %s AND save_account_status IN (%s, %s)",
-            params=(biz, self.AccountSaveStatus.INIT, self.AccountSaveStatus.PROCESSING),
+            params=(
+                biz,
+                self.AccountSaveStatus.INIT,
+                self.AccountSaveStatus.PROCESSING,
+            ),
         )
         return [r["id"] for r in rows]
 
@@ -52,7 +57,9 @@ class AccountMapper(DemandSearchAccountConst):
     # relation 表 — save_account_status 状态更新
     # ═══════════════════════════════════════════════════════════════
 
-    async def _mark_batch(self, target_status: int, ids: List[int], from_statuses: List[int]) -> int:
+    async def _mark_batch(
+        self, target_status: int, ids: List[int], from_statuses: List[int]
+    ) -> int:
         """通用：将 ids 从 from_statuses 任一改为 target_status"""
         if not ids:
             return 0
@@ -70,20 +77,28 @@ class AccountMapper(DemandSearchAccountConst):
 
     async def relation_mark_processing(self, ids: List[int]) -> int:
         """INIT → PROCESSING，CAS 防重复"""
-        return await self._mark_batch(self.AccountSaveStatus.PROCESSING, ids, [self.AccountSaveStatus.INIT])
+        return await self._mark_batch(
+            self.AccountSaveStatus.PROCESSING, ids, [self.AccountSaveStatus.INIT]
+        )
 
     async def relation_mark_success(self, ids: List[int]) -> int:
         """PROCESSING → SUCCESS"""
-        return await self._mark_batch(self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.PROCESSING])
+        return await self._mark_batch(
+            self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.PROCESSING]
+        )
 
     async def relation_mark_success_direct(self, ids: List[int]) -> int:
         """INIT → SUCCESS，跳过 PROCESSING（查重命中时用）"""
-        return await self._mark_batch(self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.INIT])
+        return await self._mark_batch(
+            self.AccountSaveStatus.SUCCESS, ids, [self.AccountSaveStatus.INIT]
+        )
 
     async def relation_mark_failed(self, ids: List[int]) -> int:
         """INIT/PROCESSING → FAIL，异常恢复用"""
         return await self._mark_batch(
-            self.AccountSaveStatus.FAIL, ids, [self.AccountSaveStatus.INIT, self.AccountSaveStatus.PROCESSING],
+            self.AccountSaveStatus.FAIL,
+            ids,
+            [self.AccountSaveStatus.INIT, self.AccountSaveStatus.PROCESSING],
         )
 
     # ═══════════════════════════════════════════════════════════════
@@ -106,8 +121,14 @@ class AccountMapper(DemandSearchAccountConst):
         """
         return await self.pool.save(
             query=query,
-            params=(account["channel_account_id"], account["gh_id"], account["account_name"],
-                    account["biz"], account["biz_info"], account["description"]),
+            params=(
+                account["channel_account_id"],
+                account["gh_id"],
+                account["account_name"],
+                account["biz"],
+                account["biz_info"],
+                account["description"],
+            ),
         )
 
     async def list_accounts_for_article_crawl(self) -> List[Dict]:
@@ -121,7 +142,11 @@ class AccountMapper(DemandSearchAccountConst):
         return await self.pool.fetch(query=query)
 
     async def update_crawl_status(
-        self, gh_id: str, *, cursor: str = "", deep_done: int = 0,
+        self,
+        gh_id: str,
+        *,
+        cursor: str = "",
+        deep_done: int = 0,
     ) -> int:
         """更新账号的爬取游标和深翻状态"""
         query = f"""
@@ -153,13 +178,28 @@ class AccountMapper(DemandSearchAccountConst):
         """
         params = [
             (
-                a["gh_id"], a["app_msg_id"], a["item_index"], a["title"],
-                a["content_url"], a["wx_sn"],
-                a["digest"], a["is_original"], a["item_show_type"],
-                a["source_url"], a["cover_img_url"], a["show_desc"],
-                a["show_view_count"], a["show_like_count"], a["show_pay_count"], a["show_zs_count"],
-                a["type"], a["send_time"], a["create_time"], a["update_time"],
-                a["date_time"], json.dumps(a["base_info"], ensure_ascii=False),
+                a["gh_id"],
+                a["app_msg_id"],
+                a["item_index"],
+                a["title"],
+                a["content_url"],
+                a["wx_sn"],
+                a["digest"],
+                a["is_original"],
+                a["item_show_type"],
+                a["source_url"],
+                a["cover_img_url"],
+                a["show_desc"],
+                a["show_view_count"],
+                a["show_like_count"],
+                a["show_pay_count"],
+                a["show_zs_count"],
+                a["type"],
+                a["send_time"],
+                a["create_time"],
+                a["update_time"],
+                a["date_time"],
+                json.dumps(a["base_info"], ensure_ascii=False),
             )
             for a in articles
         ]

src/domains/demand_search_account/_utils.py

@@ -22,6 +22,7 @@ def parse_account_response(api_data: Dict, biz: str = "") -> Dict:
 # ShowDesc 解析：阅读/赞/付费/赞赏 → 数字
 # ═══════════════════════════════════════════════════════════════
 
+
 def show_desc_to_sta(show_desc: str) -> Dict[str, int]:
     """解析 ShowDesc 中的阅读量、点赞量等
 
@@ -61,17 +62,30 @@ def show_desc_to_sta(show_desc: str) -> Dict[str, int]:
             return "show_unknown"
         raw = raw.strip().lower()
         mapping = {
-            "阅读": "show_view_count", "看过": "show_view_count", "观看": "show_view_count",
-            "reads": "show_view_count", "views": "show_view_count", "view": "show_view_count",
-            "赞": "show_like_count", "点赞": "show_like_count",
-            "likes": "show_like_count", "like": "show_like_count",
-            "付费": "show_pay_count", "payments": "show_pay_count", "paid": "show_pay_count",
+            "阅读": "show_view_count",
+            "看过": "show_view_count",
+            "观看": "show_view_count",
+            "reads": "show_view_count",
+            "views": "show_view_count",
+            "view": "show_view_count",
+            "赞": "show_like_count",
+            "点赞": "show_like_count",
+            "likes": "show_like_count",
+            "like": "show_like_count",
+            "付费": "show_pay_count",
+            "payments": "show_pay_count",
+            "paid": "show_pay_count",
             "赞赏": "show_zs_count",
         }
         return mapping.get(raw, "show_unknown")
 
     if not show_desc:
-        return {"show_view_count": 0, "show_like_count": 0, "show_pay_count": 0, "show_zs_count": 0}
+        return {
+            "show_view_count": 0,
+            "show_like_count": 0,
+            "show_pay_count": 0,
+            "show_zs_count": 0,
+        }
 
     show_desc = show_desc.replace("+", "")
     sta: Dict[str, int] = {}
@@ -106,10 +120,11 @@ def show_desc_to_sta(show_desc: str) -> Dict[str, int]:
 # 文章列表响应解析
 # ═══════════════════════════════════════════════════════════════
 
+
 def parse_article_list_response(data_list: List[Dict], gh_id: str) -> List[Dict]:
     """get_article_list_from_account 返回的 data.data[] → articles 表 INSERT dict 列表"""
     rows = []
-    for group in (data_list or []):
+    for group in data_list or []:
         app_msg = group.get("AppMsg") or {}
         base_info = app_msg.get("BaseInfo") or {}
         detail_list = app_msg.get("DetailInfo") or []
@@ -120,30 +135,32 @@ def parse_article_list_response(data_list: List[Dict], gh_id: str) -> List[Dict]
             show_desc = detail.get("ShowDesc", "") or ""
             counts = show_desc_to_sta(show_desc)
 
-            rows.append({
-                "gh_id": gh_id,
-                "app_msg_id": base_info.get("AppMsgId", 0),
-                "item_index": detail.get("ItemIndex", 0),
-                "title": detail.get("Title", "") or "",
-                "content_url": url,
-                "wx_sn": extract_wx_sn(url) or "",
-                "digest": detail.get("Digest", "") or "",
-                "is_original": 1 if detail.get("IsOriginal") else 0,
-                "item_show_type": detail.get("ItemShowType", 0) or 0,
-                "source_url": detail.get("SourceUrl", "") or "",
-                "cover_img_url": detail.get("CoverImgUrl", "") or "",
-                "show_desc": show_desc,
-                "show_view_count": counts["show_view_count"],
-                "show_like_count": counts["show_like_count"],
-                "show_pay_count": counts["show_pay_count"],
-                "show_zs_count": counts["show_zs_count"],
-                "type": base_info.get("Type", 0) or 0,
-                "send_time": detail.get("send_time", 0) or 0,
-                "create_time": base_info.get("CreateTime", 0) or 0,
-                "update_time": base_info.get("UpdateTime", 0) or 0,
-                "date_time": top_base.get("DateTime", 0) or 0,
-                "base_info": top_base,
-            })
+            rows.append(
+                {
+                    "gh_id": gh_id,
+                    "app_msg_id": base_info.get("AppMsgId", 0),
+                    "item_index": detail.get("ItemIndex", 0),
+                    "title": detail.get("Title", "") or "",
+                    "content_url": url,
+                    "wx_sn": extract_wx_sn(url) or "",
+                    "digest": detail.get("Digest", "") or "",
+                    "is_original": 1 if detail.get("IsOriginal") else 0,
+                    "item_show_type": detail.get("ItemShowType", 0) or 0,
+                    "source_url": detail.get("SourceUrl", "") or "",
+                    "cover_img_url": detail.get("CoverImgUrl", "") or "",
+                    "show_desc": show_desc,
+                    "show_view_count": counts["show_view_count"],
+                    "show_like_count": counts["show_like_count"],
+                    "show_pay_count": counts["show_pay_count"],
+                    "show_zs_count": counts["show_zs_count"],
+                    "type": base_info.get("Type", 0) or 0,
+                    "send_time": detail.get("send_time", 0) or 0,
+                    "create_time": base_info.get("CreateTime", 0) or 0,
+                    "update_time": base_info.get("UpdateTime", 0) or 0,
+                    "date_time": top_base.get("DateTime", 0) or 0,
+                    "base_info": top_base,
+                }
+            )
     return rows
 
 

src/domains/demand_search_account/account_article_fetch.py

@@ -67,14 +67,22 @@ class AccountArticleFetch(DemandSearchAccountConst):
             prev_count = await self.mapper.count_articles_by_gh_id(gh_id)
             logger.info(
                 "gh_id=%s 抓取完成: 本批写入 %d 条, 累计 %d 条",
-                gh_id, n, prev_count,
+                gh_id,
+                n,
+                prev_count,
             )
 
         logger.info(
             "文章抓取完成: 账号 %d 个, 写入 %d 条, 异常 %d",
-            len(accounts), total_articles, errors,
+            len(accounts),
+            total_articles,
+            errors,
         )
-        return {"accounts": len(accounts), "articles_written": total_articles, "errors": errors}
+        return {
+            "accounts": len(accounts),
+            "articles_written": total_articles,
+            "errors": errors,
+        }
 
     # ═══════════════════════════════════════════════════════════════
     # 抓取模式
@@ -87,7 +95,9 @@ class AccountArticleFetch(DemandSearchAccountConst):
         cutoff = int(time.time()) - self.ONE_YEAR_SECONDS
 
         while True:
-            articles, next_cursor, has_more, oldest_send_time = await self._fetch_page(gh_id, cursor)
+            articles, next_cursor, has_more, oldest_send_time = await self._fetch_page(
+                gh_id, cursor
+            )
 
             if articles:
                 # 筛选一年内的文章，过滤一年前的
@@ -101,7 +111,9 @@ class AccountArticleFetch(DemandSearchAccountConst):
                 if old_count > 0 or (oldest_send_time and oldest_send_time < cutoff):
                     logger.info(
                         "gh_id=%s 深翻到达一年前边界: 本页 %d 条中 %d 条超一年, 停止",
-                        gh_id, len(articles), old_count,
+                        gh_id,
+                        len(articles),
+                        old_count,
                     )
                     break
 
@@ -122,14 +134,18 @@ class AccountArticleFetch(DemandSearchAccountConst):
         return n
 
     async def _fetch_page(
-        self, gh_id: str, cursor: str | None,
+        self,
+        gh_id: str,
+        cursor: str | None,
     ) -> tuple[List[dict], str | None, bool, int | None]:
         """调用 API 获取一页文章，返回 (articles, next_cursor, has_more, oldest_send_time)"""
         resp = await get_article_list_from_account(gh_id, index=cursor)
         if not resp or resp.get("code") != 0:
             logger.warning(
                 "get_article_list_from_account 返回异常: gh_id=%s, cursor=%s, code=%s",
-                gh_id, cursor, resp.get("code") if resp else "None",
+                gh_id,
+                cursor,
+                resp.get("code") if resp else "None",
             )
             return [], None, False, None
 
@@ -143,7 +159,9 @@ class AccountArticleFetch(DemandSearchAccountConst):
 
         articles = parse_article_list_response(data_list, gh_id)
         # 本页最老的 send_time
-        oldest = min((a["send_time"] for a in articles if a["send_time"] > 0), default=None)
+        oldest = min(
+            (a["send_time"] for a in articles if a["send_time"] > 0), default=None
+        )
         return articles, next_cursor, has_more, oldest
 
 

src/domains/demand_search_article/_mapper.py

@@ -203,7 +203,12 @@ class ArticleSearchRelationMapper(DemandSearchArticleConst):
         """
         return await self.pool.save(
             query=query,
-            params=(self.SearchStatus.SUCCESS, channel_content_id, item_id, self.SearchStatus.PROCESSING),
+            params=(
+                self.SearchStatus.SUCCESS,
+                channel_content_id,
+                item_id,
+                self.SearchStatus.PROCESSING,
+            ),
         )
 
     async def mark_failed(self, ids: List[int]) -> int:

src/handlers/account_article_fetch.py

@@ -10,7 +10,10 @@ import logging
 from typing import TYPE_CHECKING
 
 from src.infra.xxl_jobs import xxl_job
-from src.domains.demand_search_account import AccountArticleFetch, DemandSearchAccountConst
+from src.domains.demand_search_account import (
+    AccountArticleFetch,
+    DemandSearchAccountConst,
+)
 
 if TYPE_CHECKING:
     from src.infra.database import DatabaseManager
@@ -40,7 +43,9 @@ async def account_article_fetch(param: str) -> dict:
 
     logger.info(
         "accountArticleFetch done: accounts=%d, articles_written=%d, errors=%d",
-        result["accounts"], result["articles_written"], result["errors"],
+        result["accounts"],
+        result["articles_written"],
+        result["errors"],
     )
     return {
         "code": 200,

src/handlers/article_fetch_detail.py

@@ -10,7 +10,10 @@ import logging
 from typing import TYPE_CHECKING
 
 from src.infra.xxl_jobs import xxl_job
-from src.domains.demand_search_article import ArticleFetchDetail, DemandSearchArticleConst
+from src.domains.demand_search_article import (
+    ArticleFetchDetail,
+    DemandSearchArticleConst,
+)
 
 if TYPE_CHECKING:
     from src.infra.database import DatabaseManager
@@ -40,7 +43,9 @@ async def article_fetch_detail(param: str) -> dict:
 
     logger.info(
         "articleFetchDetail done: pending=%d, success=%d, failed=%d",
-        result["pending"], result["success"], result["failed"],
+        result["pending"],
+        result["success"],
+        result["failed"],
     )
     return {
         "code": 200,

src/handlers/article_search.py

@@ -11,7 +11,10 @@ import logging
 from typing import TYPE_CHECKING
 
 from src.infra.xxl_jobs import xxl_job
-from src.domains.demand_search_article import DemandSearchArticle, DemandSearchArticleConst
+from src.domains.demand_search_article import (
+    DemandSearchArticle,
+    DemandSearchArticleConst,
+)
 
 if TYPE_CHECKING:
     from src.infra.database import DatabaseManager
@@ -51,7 +54,10 @@ async def article_search(param: str) -> dict:
 
     logger.info(
         "articleSearch done: dt=%s, queued=%d, processed=%d, written=%d",
-        dt, result["queued"], result["processed"], result["results_written"],
+        dt,
+        result["queued"],
+        result["processed"],
+        result["results_written"],
     )
     return {
         "code": 200,

src/handlers/demand_enqueue.py

@@ -67,6 +67,8 @@ async def demand_enqueue(param: str) -> dict:
     accounts = len(set(it["account"] for it in items))
     logger.info(
         "demandEnqueue done: enqueued %d items (%d accounts) for dt=%s",
-        n, accounts, dt,
+        n,
+        accounts,
+        dt,
     )
     return {"code": 200, "msg": f"enqueued={n}, accounts={accounts}"}