解构内部优质文章

app/api/v1/utils/schemas.py

@@ -47,4 +47,6 @@ class LongArticlesMcpRequest(BaseRequest):
     page: int = Field(default=1, ge=1, description="页码，从 1 开始")
     page_size: int = Field(default=20, ge=1, le=100, description="每页条数")
     sort_by: Optional[str] = Field(default=None, description="排序字段")
-    sort_order: Optional[Literal["asc", "desc"]] = Field(default=None, description="排序方向: asc / desc")
+    sort_order: Optional[Literal["asc", "desc"]] = Field(
+        default=None, description="排序方向: asc / desc"
+    )

app/domains/data_recycle_tasks/recycle_daily_publish_articles.py

@@ -53,7 +53,7 @@ class Const:
         "gh_341e08f852f5",
         "gh_0c89e11f8bf3",
         "gh_9eef14ad6c16",
-        "gh_c5cdf60d9ab4"
+        "gh_c5cdf60d9ab4",
     ]
 
     # NOT USED SERVER ACCOUNT

app/domains/data_recycle_tasks/recycle_mini_program_info/_const.py

@@ -1,5 +1,4 @@
 class RecycleMiniProgramInfoConst:
-
     class TaskSatus:
         INIT = 0
         PROCESSING = 1
@@ -17,4 +16,4 @@ class RecycleMiniProgramInfoConst:
         REQUEST_FAIL = -1
         DELETED = -2
         UNKNOWN = -3
-        ILLEGAL = -4
+        ILLEGAL = -4

app/domains/data_recycle_tasks/recycle_mini_program_info/_mapper.py

@@ -18,15 +18,16 @@ class RecycleMiniProgramInfoMapper(RecycleMiniProgramInfoConst):
                     [
                         self.PublishTimestampStatus.DEFAULT,
                         self.PublishTimestampStatus.REQUEST_FAIL,
-                        self.PublishTimestampStatus.REQUEST_FAIL
+                        self.PublishTimestampStatus.REQUEST_FAIL,
                     ]
                 ),
-            )
+            ),
         )
 
     # 更新文章发文时间戳 &&  root_source_id_list
-    async def set_publish_detail_info(self, wx_sn,
-                                      publish_timestamp, root_source_id_list):
+    async def set_publish_detail_info(
+        self, wx_sn, publish_timestamp, root_source_id_list
+    ):
         query = """
             UPDATE official_articles_v2 
             SET publish_timestamp = %s, root_source_id_list = %s
@@ -35,11 +36,5 @@ class RecycleMiniProgramInfoMapper(RecycleMiniProgramInfoConst):
         await self.pool.async_save(
             query=query,
             db_name="piaoquan_crawler",
-            params=(
-                publish_timestamp,
-                root_source_id_list,
-                wx_sn
-            ),
+            params=(publish_timestamp, root_source_id_list, wx_sn),
         )
-
-

app/domains/decode_task/inner_articles_decode/__init__.py

@@ -0,0 +1 @@
+from .entrance import InnerArticlesDecodeTask

app/domains/decode_task/inner_articles_decode/_const.py

@@ -0,0 +1,44 @@
+class InnerArticlesDecodeConst:
+    TASK_BATCH = 100
+
+    class TaskStatus:
+        # 任务状态
+        INIT = 0
+        PROCESSING = 1
+        SUCCESS = 2
+        FAILED = 99
+
+    class ExtractStatus(TaskStatus): ...
+
+    class DecodeStatus:
+        # 解构结果状态
+        PENDING = 0
+        RUNNING = 1
+        SUCCESS = 2
+        FAILED = 3
+
+    class BusinessScene:
+        # 业务场景
+        POINT_PICK = 0
+        CREATE = 1
+        MAKE = 2
+
+    class ContentType:
+        # 内容类型
+        LONG_ARTICLE = 1
+        PICTURE_TEXT = 2
+        VIDEO = 3
+
+    class SourceType:
+        AD_PLATFORM = 1
+        INNER = 2
+
+    class ProduceModuleType:
+        COVER = 1
+        IMAGE = 2
+        TITLE = 3
+        CONTENT = 4
+        SUMMARY = 18
+
+    class RequestDecode:
+        SUCCESS = 0

app/domains/decode_task/inner_articles_decode/_mapper.py

@@ -0,0 +1,133 @@
+from typing import List, Dict
+
+from app.core.database import DatabaseManager
+
+from ._const import InnerArticlesDecodeConst
+
+
+class InnerArticlesDecodeMapper(InnerArticlesDecodeConst):
+    def __init__(self, pool: DatabaseManager):
+        self.pool = pool
+
+    # 存储解构任务
+    async def record_decode_task(
+        self, task_id: str, wx_sn: str, remark: str = None
+    ) -> int:
+        query = """
+            INSERT IGNORE INTO long_articles_decode_tasks (task_id, wx_sn, remark, source)
+            VALUES (%s, %s, %s, %s)
+        """
+        return await self.pool.async_save(
+            query=query, params=(task_id, wx_sn, remark, self.SourceType.INNER)
+        )
+
+    # 更新解构任务状态
+    async def update_decode_task_status(
+        self, task_id: str, ori_status: int, new_status: int, remark: str = None
+    ) -> int:
+        query = """
+            UPDATE long_articles_decode_tasks
+            SET status = %s, remark = %s
+            WHERE task_id = %s AND status = %s;
+        """
+        return await self.pool.async_save(
+            query=query, params=(new_status, remark, task_id, ori_status)
+        )
+
+    # 设置解构结果
+    async def set_decode_result(
+        self, task_id: str, result: str, remark: str = None
+    ) -> int:
+        query = """
+            UPDATE long_articles_decode_tasks
+            SET status = %s, remark = %s, result = %s
+            WHERE task_id = %s AND status = %s;
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                self.TaskStatus.SUCCESS,
+                remark,
+                result,
+                task_id,
+                self.TaskStatus.PROCESSING,
+            ),
+        )
+
+    # 获取内部文章生成信息
+    async def fetch_inner_articles_produce_detail(self, source_id) -> List[Dict]:
+        query = """
+            SELECT produce_module_type, output
+            FROM produce_plan_module_output WHERE plan_exe_id = %s
+            AND produce_module_type in (1,2,3,4,18); 
+        """
+        return await self.pool.async_fetch(
+            query=query, db_name="aigc", params=(source_id,)
+        )
+
+    # 获取待拉取结果的解构任务（status=INIT，尚未拿到解构结果）
+    async def fetch_decoding_tasks(self) -> List[Dict]:
+        query = """
+            SELECT task_id FROM long_articles_decode_tasks WHERE status = %s AND source = %s LIMIT %s;
+        """
+        return await self.pool.async_fetch(
+            query=query, params=(self.TaskStatus.INIT, self.SourceType.INNER, self.TASK_BATCH)
+        )
+
+    # 获取待解析的任务（获取处理成功的任务）
+    async def fetch_extract_tasks(self):
+        query = """
+            SELECT id, result FROM long_articles_decode_tasks
+            WHERE extract_status = %s AND status = %s;
+        """
+        return await self.pool.async_fetch(
+            query=query, params=(self.ExtractStatus.INIT, self.TaskStatus.SUCCESS)
+        )
+
+    # 修改解析状态（用于加锁与状态流转）
+    async def update_extract_status(self, task_id, ori_status, new_status):
+        query = """
+            UPDATE long_articles_decode_tasks
+            SET extract_status = %s WHERE extract_status = %s AND id = %s;
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                new_status,
+                ori_status,
+                task_id,
+            ),
+        )
+
+    # 记录解析结果明细到 long_articles_decode_task_detail
+    async def record_extract_detail(self, decode_task_id: int, detail: Dict) -> int:
+        query = """
+            INSERT INTO long_articles_decode_task_detail
+                (decode_task_id, inspiration, purpose, key_point, topic)
+            VALUES (%s, %s, %s, %s, %s);
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                decode_task_id,
+                detail.get("inspiration", ""),
+                detail.get("purpose", ""),
+                detail.get("key_point", ""),
+                detail.get("topic", ""),
+            ),
+        )
+
+    # 获取内部文章
+    async def fetch_inner_articles(self):
+        query = """
+            SELECT source_id, title, wx_sn
+            FROM datastat_sort_strategy
+            WHERE date_str >= '20260101'
+                AND account_type != '服务号'
+                AND position = 1
+                AND source_id is not null
+            GROUP BY source_id
+            HAVING sum(view_count) / sum(avg_view_count) >= 1.2
+            AND min(read_rate) >= 0.2;
+        """
+        return await self.pool.async_fetch(query=query)

app/domains/decode_task/inner_articles_decode/_util.py

@@ -0,0 +1,102 @@
+import json
+from typing import Dict, List
+
+from app.infra.internal import DecodeServer
+
+from ._const import InnerArticlesDecodeConst
+
+
+class InnerArticlesDecodeUtil(InnerArticlesDecodeConst):
+    decode_server = DecodeServer()
+
+    @staticmethod
+    def format_images(images: str) -> List[str]:
+        """
+        格式化图片字符串，空/非法 JSON 返回空列表。
+        """
+        if not images or not images.strip():
+            return []
+        try:
+            image_list = json.loads(images)
+        except (json.JSONDecodeError, TypeError):
+            return []
+        if not isinstance(image_list, list):
+            return []
+        return [
+            i.get("image_url")
+            for i in image_list
+            if isinstance(i, dict) and i.get("image_url")
+        ]
+
+    async def create_decode_task(self, article: Dict, article_produce_info: List[Dict]):
+        images = [
+            i["output"]
+            for i in article_produce_info
+            if i["produce_module_type"]
+            in (self.ProduceModuleType.COVER, self.ProduceModuleType.IMAGE)
+        ]
+        article["images"] = images
+        text = [
+            i["output"]
+            for i in article_produce_info
+            if i["produce_module_type"] == self.ProduceModuleType.CONTENT
+        ]
+        article["article_text"] = "\n".join(text)
+        request_body = self.prepare_extract_body(article)
+        return await self.decode_server.create_decode_task(request_body)
+
+    async def fetch_decode_result(self, task_id: str):
+        return await self.decode_server.fetch_result(task_id)
+
+    def prepare_extract_body(self, article: Dict) -> Dict:
+        return {
+            "scene": self.BusinessScene.POINT_PICK,
+            "content_type": self.ContentType.LONG_ARTICLE,
+            "content": {
+                "channel_content_id": article.get("wx_sn", ""),
+                "video_url": "",
+                "images": article.get("images", []),
+                "body_text": article.get("article_text", ""),
+                "title": article.get("title", ""),
+                "channel_account_id": article.get("gh_id", ""),
+                "channel_account_name": article.get("account_name", ""),
+            },
+        }
+
+    @staticmethod
+    def extract_decode_result(result: Dict) -> Dict:
+        """
+        从结构的结果中，解析出灵感点、目的点、关键点；
+        """
+        final_result = result.get("final_normalization_rebuild")
+        if not final_result:
+            return {"error": "解构结果中无 final_normalization_rebuild 信息"}
+        # 灵感点
+        inspiration_list = final_result.get("inspiration_final_result", {}).get(
+            "最终灵感点列表", []
+        )
+        # 目的
+        purpose_list = final_result.get("purpose_final_result", {}).get(
+            "最终目的点列表", []
+        )
+        # 关键点
+        keypoint_list = final_result.get("keypoint_final", {}).get("最终关键点列表", [])
+
+        topic_fusion = final_result.get("topic_fusion_result", {})
+        # 选题
+        topic_text = (
+            topic_fusion.get("最终选题", {}).get("选题", "")
+            if isinstance(topic_fusion.get("最终选题"), dict)
+            else ""
+        )
+
+        def _join_points(items: list, key: str) -> str:
+            parts = [str(p[key]) for p in items if isinstance(p, dict) and p.get(key)]
+            return ",".join(parts)
+
+        return {
+            "inspiration": _join_points(inspiration_list, "灵感点"),
+            "purpose": _join_points(purpose_list, "目的点"),
+            "key_point": _join_points(keypoint_list, "关键点"),
+            "topic": topic_text,
+        }

app/domains/decode_task/inner_articles_decode/entrance.py

@@ -0,0 +1,260 @@
+import json
+from typing import Dict
+from tqdm import tqdm
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+
+from app.infra.shared import run_tasks_with_asyncio_task_group
+
+from ._const import InnerArticlesDecodeConst
+from ._mapper import InnerArticlesDecodeMapper
+from ._util import InnerArticlesDecodeUtil
+
+
+class InnerArticlesDecodeTask(InnerArticlesDecodeConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = InnerArticlesDecodeMapper(self.pool)
+        self.tool = InnerArticlesDecodeUtil()
+
+    async def create_single_decode_task(self, article: Dict):
+        # Acquire Lock
+        source_id = article["source_id"]
+        article_produce_info = await self.mapper.fetch_inner_articles_produce_detail(
+            source_id
+        )
+
+        # 与解构系统交互，创建解构任务
+        response = await self.tool.create_decode_task(article, article_produce_info)
+        response_code = response.get("code")
+        if response_code != self.RequestDecode.SUCCESS:
+            return
+
+        task_id = response.get("data", {}).get("task_id") or response.get(
+            "data", {}
+        ).get("taskId")
+        if not task_id:
+            return
+
+        wx_sn = article["wx_sn"]
+        remark = f"task_id: {task_id}-创建解构任务"
+        record_row = await self.mapper.record_decode_task(task_id, wx_sn, remark)
+        if not record_row:
+            return
+
+
+    async def fetch_single_task(self, task: Dict):
+        task_id = task["task_id"]
+
+        # acquire lock
+        acquire_lock = await self.mapper.update_decode_task_status(
+            task_id, self.TaskStatus.INIT, self.TaskStatus.PROCESSING
+        )
+        if not acquire_lock:
+            return
+
+        response = await self.tool.fetch_decode_result(task_id)
+        if not response:
+            await self.mapper.update_decode_task_status(
+                task_id=task_id,
+                ori_status=self.TaskStatus.PROCESSING,
+                new_status=self.TaskStatus.INIT,
+                remark="获取解构结果失败，服务异常，已回滚状态",
+            )
+            return
+
+        # 请求成功
+        response_code = response.get("code")
+        if response_code != self.RequestDecode.SUCCESS:
+            # 解构任务获取失败
+            await self.mapper.update_decode_task_status(
+                task_id=task_id,
+                ori_status=self.TaskStatus.PROCESSING,
+                new_status=self.TaskStatus.FAILED,
+                remark=f"请求解构接口返回异常，标记为失败：{json.dumps(response, ensure_ascii=False)}",
+            )
+            return
+
+        response_data = response.get("data", {})
+        response_task_id = response_data.get("taskId") or response_data.get("task_id")
+        if task_id != response_task_id:
+            # 解构任务获取失败
+            await self.mapper.update_decode_task_status(
+                task_id=task_id,
+                ori_status=self.TaskStatus.PROCESSING,
+                new_status=self.TaskStatus.FAILED,
+                remark=f"请求解构接口TaskId异常：{json.dumps(response, ensure_ascii=False)}",
+            )
+            return
+
+        status = response_data.get("status")
+        match status:
+            case self.DecodeStatus.PENDING:
+                await self.mapper.update_decode_task_status(
+                    task_id=task_id,
+                    ori_status=self.TaskStatus.PROCESSING,
+                    new_status=self.TaskStatus.INIT,
+                    remark=f"解构任务状态为PENDING，继续轮询",
+                )
+
+            case self.DecodeStatus.RUNNING:
+                await self.mapper.update_decode_task_status(
+                    task_id=task_id,
+                    ori_status=self.TaskStatus.PROCESSING,
+                    new_status=self.TaskStatus.INIT,
+                    remark=f"解构任务状态为RUNNING，继续轮询",
+                )
+
+            case self.DecodeStatus.SUCCESS:
+                await self.mapper.set_decode_result(
+                    task_id=task_id,
+                    result=json.dumps(response_data, ensure_ascii=False),
+                )
+
+            case self.DecodeStatus.FAILED:
+                await self.mapper.update_decode_task_status(
+                    task_id=task_id,
+                    ori_status=self.TaskStatus.PROCESSING,
+                    new_status=self.TaskStatus.FAILED,
+                    remark=f"解构任务状态为FAILED，标记为失败",
+                )
+
+            case _:
+                await self.mapper.update_decode_task_status(
+                    task_id=task_id,
+                    ori_status=self.TaskStatus.PROCESSING,
+                    new_status=self.TaskStatus.INIT,
+                    remark=f"解构任务状态未知(status={status})，回滚待重试：{json.dumps(response_data, ensure_ascii=False)}",
+                )
+                await self.log_service.log(
+                    contents={
+                        "task": "fetch_single_task",
+                        "task_id": task_id,
+                        "status": "unknown",
+                        "message": f"unexpected decode status: {status}",
+                        "data": response_data,
+                    }
+                )
+
+    async def extract_single_result(self, task):
+        task_id = task["id"]
+
+        # acquire lock by extract_status
+        acquire_lock = await self.mapper.update_extract_status(
+            task_id, self.ExtractStatus.INIT, self.ExtractStatus.PROCESSING
+        )
+        if not acquire_lock:
+            return
+
+        try:
+            result = json.loads(task["result"])["result"]
+        except (TypeError, KeyError, json.JSONDecodeError) as e:
+            await self.mapper.update_extract_status(
+                task_id,
+                self.ExtractStatus.PROCESSING,
+                self.ExtractStatus.FAILED,
+            )
+            await self.log_service.log(
+                contents={
+                    "task": "extract_single_result",
+                    "task_id": task_id,
+                    "status": "fail",
+                    "message": f"parse decode result error: {e}",
+                    "raw": task.get("result"),
+                }
+            )
+            return
+
+        detail = self.tool.extract_decode_result(result)
+        # 如果工具返回错误信息，直接标记为失败
+        if detail.get("error"):
+            await self.mapper.update_extract_status(
+                task_id,
+                self.ExtractStatus.PROCESSING,
+                self.ExtractStatus.FAILED,
+            )
+            await self.log_service.log(
+                contents={
+                    "task": "extract_single_result",
+                    "task_id": task_id,
+                    "status": "fail",
+                    "message": detail["error"],
+                }
+            )
+            return
+
+        # 写入明细表
+        saved = await self.mapper.record_extract_detail(task_id, detail)
+        if not saved:
+            await self.mapper.update_extract_status(
+                task_id,
+                self.ExtractStatus.PROCESSING,
+                self.ExtractStatus.FAILED,
+            )
+            await self.log_service.log(
+                contents={
+                    "task": "extract_single_result",
+                    "task_id": task_id,
+                    "status": "fail",
+                    "message": "insert long_articles_decode_task_detail failed",
+                    "detail": detail,
+                }
+            )
+            return
+
+        # 写入成功，更新状态为成功
+        await self.mapper.update_extract_status(
+            task_id,
+            self.ExtractStatus.PROCESSING,
+            self.ExtractStatus.SUCCESS,
+        )
+
+    async def create_tasks(self):
+        article_list = await self.mapper.fetch_inner_articles()
+        if not article_list:
+            await self.log_service.log(
+                contents={
+                    "task": "create_tasks",
+                    "message": "No more articles to decode",
+                }
+            )
+            return
+
+        for article in tqdm(article_list[1:], desc="Creating decode tasks"):
+            await self.create_single_decode_task(article)
+
+    async def fetch_results(self):
+        decoding_tasks = await self.mapper.fetch_decoding_tasks()
+        if not decoding_tasks:
+            await self.log_service.log(
+                contents={"task": "fetch_results", "message": "No more tasks to fetch"}
+            )
+            return
+
+        for task in tqdm(decoding_tasks, desc="Fetching decode results"):
+            await self.fetch_single_task(task)
+
+    async def extract_task(self):
+        tasks = await self.mapper.fetch_extract_tasks()
+        await run_tasks_with_asyncio_task_group(
+            task_list=tasks,
+            handler=self.extract_single_result,
+            description="批量解析结构结果",
+            unit="task",
+        )
+
+    async def deal(self, task_name):
+        match task_name:
+            case "create_tasks":
+                await self.create_tasks()
+
+            # case "fetch_results":
+            #     await self.fetch_results()
+            #
+            # case "extract":
+            #     await self.extract_task()
+
+
+__all__ = ["InnerArticlesDecodeTask"]

app/domains/mcp/_const.py

@@ -3,13 +3,16 @@ import math
 
 class LongArticlesMcpConst:
     """MCP 配置层：排序字段、SQL 片段、分页默认值等。"""
+
     # 分页配置
     DEFAULT_PAGE = 1
     DEFAULT_PAGE_SIZE = 20
     MAX_PAGE_SIZE = 100
 
     @staticmethod
-    def normalize_pagination(page: int | None, page_size: int | None) -> tuple[int, int]:
+    def normalize_pagination(
+        page: int | None, page_size: int | None
+    ) -> tuple[int, int]:
         page = page or LongArticlesMcpConst.DEFAULT_PAGE
         page = max(page, 1)
 

app/domains/mcp/_handler_map.py

@@ -4,7 +4,9 @@ from app.core.database import DatabaseManager
 from app.core.observability import LogService
 
 
-HandlerType = Callable[[DatabaseManager, LogService, Dict[str, Any] | None], Awaitable[Any]]
+HandlerType = Callable[
+    [DatabaseManager, LogService, Dict[str, Any] | None], Awaitable[Any]
+]
 
 
 async def _get_decode_response_wrapper(

app/domains/monitor_tasks/__init__.py

@@ -24,5 +24,5 @@ __all__ = [
     "AutoReplyCardsMonitor",
     "CooperateAccountsMonitorTask",
     "FwhGroupPublishMonitor",
-    "AdPlatformAccountsMonitorTask"
+    "AdPlatformAccountsMonitorTask",
 ]

app/domains/monitor_tasks/ad_platform_accounts_monitor/_mapper.py

@@ -55,9 +55,7 @@ class AdPlatformAccountsMonitorMapper(AdPlatformAccountsMonitorTaskConst):
         query = """
             UPDATE ad_platform_accounts SET need_extract = %s WHERE gh_id = %s;
         """
-        return await self.pool.async_save(
-            query=query, params=(target_status, gh_id)
-        )
+        return await self.pool.async_save(query=query, params=(target_status, gh_id))
 
     # ============================文章操作===================================
     # 修改文章 fetch 状态

app/domains/monitor_tasks/ad_platform_accounts_monitor/entrance.py

@@ -111,9 +111,13 @@ class AdPlatformAccountsMonitorTask(AdPlatformAccountsMonitorTaskConst):
         if publish_time_detail:
             max_publish_timestamp = publish_time_detail[0]["publish_timestamp"]
             if max_publish_timestamp is None:
-                max_publish_timestamp = self.tool.get_now_timestamp() - self.ACCOUNT_CRAWL_DURATION
+                max_publish_timestamp = (
+                    self.tool.get_now_timestamp() - self.ACCOUNT_CRAWL_DURATION
+                )
         else:
-            max_publish_timestamp = self.tool.get_now_timestamp() - self.ACCOUNT_CRAWL_DURATION
+            max_publish_timestamp = (
+                self.tool.get_now_timestamp() - self.ACCOUNT_CRAWL_DURATION
+            )
 
         cursor = None
         while True:
@@ -190,9 +194,7 @@ class AdPlatformAccountsMonitorTask(AdPlatformAccountsMonitorTaskConst):
             )
 
         else:
-            read_list = [
-                (i.get("read_cnt") or self.INT_ZERO) for i in head_articles
-            ]
+            read_list = [(i.get("read_cnt") or self.INT_ZERO) for i in head_articles]
             read_median = self.tool.get_median(read_list)
             remark = f"{execute_time}--计算阅读头条阅读中位数"
             await self.mapper.update_account_read_median(
@@ -203,9 +205,13 @@ class AdPlatformAccountsMonitorTask(AdPlatformAccountsMonitorTaskConst):
             )
 
             # 更新账号的解构状态
-            read_median_rate = read_median / fans_level if fans_level else self.ZERO_FLOAT
+            read_median_rate = (
+                read_median / fans_level if fans_level else self.ZERO_FLOAT
+            )
             if read_median_rate >= self.READ_MEDIAN_RATE_THRESHOLD:
-                await self.mapper.update_account_extract_status(gh_id, self.NEED_EXTRACT)
+                await self.mapper.update_account_extract_status(
+                    gh_id, self.NEED_EXTRACT
+                )
 
             # 更新文章的阅读中位数倍数
             for article in head_articles:
@@ -213,7 +219,9 @@ class AdPlatformAccountsMonitorTask(AdPlatformAccountsMonitorTaskConst):
                 read_cnt = article.get("read_cnt") or self.INT_ZERO
 
                 if read_cnt > self.INT_ZERO:
-                    multiplier = read_cnt / read_median if read_median else self.ZERO_FLOAT
+                    multiplier = (
+                        read_cnt / read_median if read_median else self.ZERO_FLOAT
+                    )
                 else:
                     multiplier = self.ZERO_FLOAT
                 article_remark = f"{execute_time}--更新文章阅读中位数倍数"

app/domains/monitor_tasks/kimi_balance.py

@@ -31,4 +31,4 @@ async def check_kimi_balance() -> Dict:
             title="kimi余额接口处理失败，数据结构异常",
             detail={"error": str(e), "error_msg": error_stack},
         )
-        return {"code": 99, "data": error_stack}
+        return {"code": 99, "data": error_stack}

app/infra/external/odps_service.py

@@ -11,8 +11,7 @@ class OdpsService:
 
         def _execute():
             instance = self.odps_client.execute_sql(
-                query,
-                hints={"odps.sql.submit.mode": "script"}
+                query, hints={"odps.sql.submit.mode": "script"}
             )
             instance.wait_for_success()
 

app/infra/internal/long_articles.py

@@ -32,8 +32,4 @@ async def insert_crawler_plan(pool: DatabaseManager, data_tuple: tuple):
         INSERT INTO article_crawler_plan (crawler_plan_id, name, create_timestamp)
         VALUES (%s, %s, %s);
     """
-    return await pool.async_save(
-        query=query, params=data_tuple
-    )
-
-
+    return await pool.async_save(query=query, params=data_tuple)

app/infra/internal/piaoquan.py

@@ -79,7 +79,6 @@ async def publish_video_to_piaoquan(oss_path: str, uid: str, title: str) -> Dict
 
 
 class DecodeServer:
-
     base_url: str = "http://supply-content-deconstruction-api.piaoquantv.com"
 
     # 创建解构任务
@@ -139,5 +138,3 @@ class DecodeServer:
             response = await client.get(url, headers=headers)
 
         return response
-
-

app/jobs/task_handler.py

@@ -337,9 +337,7 @@ class TaskHandler:
     @register("ad_platform_article_publish")
     async def _ad_platform_article_publish(self) -> int:
         """文章池冷启动"""
-        task = AdPlatformArticlePublishTask(
-            self.db_client, self.log_client
-        )
+        task = AdPlatformArticlePublishTask(self.db_client, self.log_client)
         await task.deal()
         return TaskStatus.SUCCESS