解构服务优化

app/domains/decode_task/__init__.py

@@ -1 +0,0 @@
-from .ad_platform_articles_decode import AdPlatformArticlesDecodeTask

app/domains/decode_task/ad_platform_articles_decode/__init__.py

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

app/domains/decode_task/ad_platform_articles_decode/_const.py

@@ -1,29 +0,0 @@
-class AdPlatformArticlesDecodeConst:
-    # 任务状态
-    INIT_STATUS = 0
-    PROCESSING_STATUS = 1
-    SUCCESS_STATUS = 2
-    FAILED_STATUS = 99
-
-    # 解构结果状态
-    PENDING = 0
-    RUNNING = 1
-    SUCCESS = 2
-    FAILED = 3
-
-    # 业务场景
-    POINT_PICK = 0
-    CREATE = 1
-    MAKE = 2
-
-    # 内容类型
-    LONG_ARTICLE = 1
-    PICTURE_TEXT = 2
-    VIDEO = 3
-
-    # 返回 code
-    SUCCESS_CODE = 0
-
-    # 获取详情状态
-    FETCH_DETAIL_SUCCESS = 2
-    TASK_BATCH = 100

app/domains/decode_task/ad_platform_articles_decode/_mapper.py

@@ -1,130 +0,0 @@
-from typing import List, Dict
-
-from app.core.database import DatabaseManager
-
-from ._const import AdPlatformArticlesDecodeConst
-
-
-class AdPlatformArticlesDecodeMapper(AdPlatformArticlesDecodeConst):
-    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 INTO long_articles_decode_tasks (task_id, wx_sn, remark)
-            VALUES (%s, %s, %s)
-        """
-        return await self.pool.async_save(query=query, params=(task_id, wx_sn, remark))
-
-    # 更新解构任务状态
-    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 update_article_decode_status(
-        self, id_: int, ori_status: int, new_status: int
-    ) -> int:
-        query = """
-            UPDATE ad_platform_accounts_daily_detail
-            SET decode_status = %s
-            WHERE id = %s AND decode_status = %s;
-        """
-        return await self.pool.async_save(
-            query=query, params=(new_status, 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.SUCCESS_STATUS,
-                remark,
-                result,
-                task_id,
-                self.PROCESSING_STATUS,
-            ),
-        )
-
-    # 获取待解构文章
-    async def fetch_decode_articles(self) -> List[Dict]:
-        query = """
-            SELECT id, account_name, gh_id, article_title, article_cover,
-                   article_text, article_images, wx_sn
-            FROM ad_platform_accounts_daily_detail WHERE fetch_status = %s AND decode_status = %s
-            LIMIT %s;
-        """
-        return await self.pool.async_fetch(
-            query=query, params=(self.SUCCESS_STATUS, self.INIT_STATUS, self.TASK_BATCH)
-        )
-
-    # 获取待拉取结果的解构任务（status=INIT，尚未拿到解构结果）
-    async def fetch_decoding_tasks(self) -> List[Dict]:
-        query = """
-            SELECT task_id FROM long_articles_decode_tasks WHERE status = %s LIMIT %s;
-        """
-        return await self.pool.async_fetch(
-            query=query, params=(self.INIT_STATUS, 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.INIT_STATUS, self.SUCCESS_STATUS)
-        )
-
-    # 修改解析状态（用于加锁与状态流转）
-    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", ""),
-            ),
-        )

app/domains/decode_task/ad_platform_articles_decode/_util.py

@@ -1,89 +0,0 @@
-import json
-from typing import Dict, List
-
-from app.infra.internal import DecodeServer
-
-from ._const import AdPlatformArticlesDecodeConst
-
-
-class AdPlatformArticlesDecodeUtil(AdPlatformArticlesDecodeConst):
-    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):
-        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.POINT_PICK,
-            "content_type": self.LONG_ARTICLE,
-            "content": {
-                "channel_content_id": article.get("wx_sn", ""),
-                "video_url": "",
-                "images": self.format_images(article.get("article_images") or ""),
-                "body_text": article.get("article_text", ""),
-                "title": article.get("article_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/ad_platform_articles_decode/entrance.py

@@ -1,321 +0,0 @@
-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 AdPlatformArticlesDecodeConst
-from ._mapper import AdPlatformArticlesDecodeMapper
-from ._util import AdPlatformArticlesDecodeUtil
-
-
-class AdPlatformArticlesDecodeTask(AdPlatformArticlesDecodeConst):
-    def __init__(self, pool: DatabaseManager, log_service: LogService):
-        self.pool = pool
-        self.log_service = log_service
-        self.mapper = AdPlatformArticlesDecodeMapper(self.pool)
-        self.tool = AdPlatformArticlesDecodeUtil()
-
-    async def create_single_decode_task(self, article: Dict):
-        # Acquire Lock
-        article_id = article["id"]
-        acquire_lock = await self.mapper.update_article_decode_status(
-            article_id, self.INIT_STATUS, self.PROCESSING_STATUS
-        )
-        if not acquire_lock:
-            await self.log_service.log(
-                contents={
-                    "article_id": article_id,
-                    "task": "create_decode_task",
-                    "status": "skip",
-                    "message": "acquire lock failed",
-                }
-            )
-            return
-
-        # 与解构系统交互，创建解构任务
-        response = await self.tool.create_decode_task(article)
-        response_code = response.get("code")
-        if response_code != self.SUCCESS_CODE:
-            # 解构任务创建失败
-            await self.mapper.update_article_decode_status(
-                article_id, self.PROCESSING_STATUS, self.FAILED_STATUS
-            )
-            await self.log_service.log(
-                contents={
-                    "article_id": article_id,
-                    "task": "create_decode_task",
-                    "status": "fail",
-                    "data": response,
-                }
-            )
-            return
-
-        task_id = response.get("data", {}).get("task_id") or response.get(
-            "data", {}
-        ).get("taskId")
-        if not task_id:
-            # 解构任务创建失败
-            await self.mapper.update_article_decode_status(
-                article_id, self.PROCESSING_STATUS, self.FAILED_STATUS
-            )
-            await self.log_service.log(
-                contents={
-                    "article_id": article_id,
-                    "task": "create_decode_task",
-                    "status": "fail",
-                    "data": response,
-                }
-            )
-            return
-
-        # 创建 decode 任务成功
-        await self.log_service.log(
-            contents={
-                "article_id": article_id,
-                "task": "create_decode_task",
-                "status": "success",
-                "data": response,
-            }
-        )
-
-        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:
-            # 记录解构任务失败
-            await self.mapper.update_article_decode_status(
-                article_id, self.PROCESSING_STATUS, self.FAILED_STATUS
-            )
-            await self.log_service.log(
-                contents={
-                    "article_id": article_id,
-                    "task": "record_decode_task",
-                    "status": "fail",
-                    "message": "创建 decode 记录失败",
-                    "data": response,
-                }
-            )
-            return
-
-        # 记录创建成功
-        await self.mapper.update_article_decode_status(
-            article_id, self.PROCESSING_STATUS, self.SUCCESS_STATUS
-        )
-
-    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.INIT_STATUS, self.PROCESSING_STATUS
-        )
-        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.PROCESSING_STATUS,
-                new_status=self.INIT_STATUS,
-                remark="获取解构结果失败，服务异常，已回滚状态",
-            )
-            return
-
-        # 请求成功
-        response_code = response.get("code")
-        if response_code != self.SUCCESS_CODE:
-            # 解构任务获取失败
-            await self.mapper.update_decode_task_status(
-                task_id=task_id,
-                ori_status=self.PROCESSING_STATUS,
-                new_status=self.FAILED_STATUS,
-                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.PROCESSING_STATUS,
-                new_status=self.FAILED_STATUS,
-                remark=f"请求解构接口TaskId异常：{json.dumps(response, ensure_ascii=False)}",
-            )
-            return
-
-        status = response_data.get("status")
-        match status:
-            case self.PENDING:
-                await self.mapper.update_decode_task_status(
-                    task_id=task_id,
-                    ori_status=self.PROCESSING_STATUS,
-                    new_status=self.INIT_STATUS,
-                    remark=f"解构任务状态为PENDING，继续轮询",
-                )
-
-            case self.RUNNING:
-                await self.mapper.update_decode_task_status(
-                    task_id=task_id,
-                    ori_status=self.PROCESSING_STATUS,
-                    new_status=self.INIT_STATUS,
-                    remark=f"解构任务状态为RUNNING，继续轮询",
-                )
-
-            case self.SUCCESS:
-                await self.mapper.set_decode_result(
-                    task_id=task_id,
-                    result=json.dumps(response_data, ensure_ascii=False),
-                )
-
-            case self.FAILED:
-                await self.mapper.update_decode_task_status(
-                    task_id=task_id,
-                    ori_status=self.PROCESSING_STATUS,
-                    new_status=self.FAILED_STATUS,
-                    remark=f"解构任务状态为FAILED，标记为失败",
-                )
-
-            case _:
-                await self.mapper.update_decode_task_status(
-                    task_id=task_id,
-                    ori_status=self.PROCESSING_STATUS,
-                    new_status=self.INIT_STATUS,
-                    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.INIT_STATUS, self.PROCESSING_STATUS
-        )
-        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.PROCESSING_STATUS,
-                self.FAILED_STATUS,
-            )
-            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.PROCESSING_STATUS,
-                self.FAILED_STATUS,
-            )
-            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.PROCESSING_STATUS,
-                self.FAILED_STATUS,
-            )
-            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.PROCESSING_STATUS,
-            self.SUCCESS_STATUS,
-        )
-
-    async def create_tasks(self):
-        article_list = await self.mapper.fetch_decode_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, 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 decoding_tasks:
-            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__ = ["AdPlatformArticlesDecodeTask"]

app/domains/decode_task/inner_articles_decode/__init__.py

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

app/domains/decode_task/inner_articles_decode/entrance.py

@@ -1,260 +0,0 @@
-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, 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/llm_tasks/aigc_decode_task/__init__.py

@@ -0,0 +1,13 @@
+from .create_decode_tasks import (
+    CreateAdPlatformArticlesDecodeTask,
+    CreateInnerArticlesDecodeTask,
+)
+from .fetch_decode_results import FetchDecodeResults
+from .extract_decode_task_detail import ExtractDecodeTaskDetail
+
+__all__ = [
+    "CreateAdPlatformArticlesDecodeTask",
+    "CreateInnerArticlesDecodeTask",
+    "FetchDecodeResults",
+    "ExtractDecodeTaskDetail",
+]

app/domains/decode_task/inner_articles_decode/_const.py → app/domains/llm_tasks/aigc_decode_task/_const.py

@@ -1,4 +1,4 @@
-class InnerArticlesDecodeConst:
+class DecodeTaskConst:
     TASK_BATCH = 100
 
     class TaskStatus:
@@ -42,3 +42,6 @@ class InnerArticlesDecodeConst:
 
     class RequestDecode:
         SUCCESS = 0
+
+
+__all__ = ["DecodeTaskConst"]

app/domains/decode_task/inner_articles_decode/_mapper.py → app/domains/llm_tasks/aigc_decode_task/_mapper.py

@@ -2,10 +2,10 @@ from typing import List, Dict
 
 from app.core.database import DatabaseManager
 
-from ._const import InnerArticlesDecodeConst
+from ._const import DecodeTaskConst
 
 
-class InnerArticlesDecodeMapper(InnerArticlesDecodeConst):
+class ArticlesDecodeTaskMapper(DecodeTaskConst):
     def __init__(self, pool: DatabaseManager):
         self.pool = pool
 
@@ -14,12 +14,10 @@ class InnerArticlesDecodeMapper(InnerArticlesDecodeConst):
         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)
+            INSERT INTO long_articles_decode_tasks (task_id, wx_sn, remark)
+            VALUES (%s, %s, %s)
         """
-        return await self.pool.async_save(
-            query=query, params=(task_id, wx_sn, remark, self.SourceType.INNER)
-        )
+        return await self.pool.async_save(query=query, params=(task_id, wx_sn, remark))
 
     # 更新解构任务状态
     async def update_decode_task_status(
@@ -54,24 +52,13 @@ class InnerArticlesDecodeMapper(InnerArticlesDecodeConst):
             ),
         )
 
-    # 获取内部文章生成信息
-    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;
+            SELECT task_id FROM long_articles_decode_tasks WHERE status = %s LIMIT %s;
         """
         return await self.pool.async_fetch(
-            query=query, params=(self.TaskStatus.INIT, self.SourceType.INNER, self.TASK_BATCH)
+            query=query, params=(self.TaskStatus.INIT, self.TASK_BATCH)
         )
 
     # 获取待解析的任务（获取处理成功的任务）
@@ -117,6 +104,42 @@ class InnerArticlesDecodeMapper(InnerArticlesDecodeConst):
             ),
         )
 
+
+class AdPlatformArticlesDecodeTaskMapper(ArticlesDecodeTaskMapper):
+    def __init__(self, pool: DatabaseManager):
+        super().__init__(pool)
+
+    # 修改文章解构状态
+    async def update_article_decode_status(
+        self, id_: int, ori_status: int, new_status: int
+    ) -> int:
+        query = """
+            UPDATE ad_platform_accounts_daily_detail
+            SET decode_status = %s
+            WHERE id = %s AND decode_status = %s;
+        """
+        return await self.pool.async_save(
+            query=query, params=(new_status, id_, ori_status)
+        )
+
+    # 获取待解构文章
+    async def fetch_decode_articles(self) -> List[Dict]:
+        query = """
+            SELECT id, account_name, gh_id, article_title, article_cover,
+                   article_text, article_images, wx_sn
+            FROM ad_platform_accounts_daily_detail WHERE fetch_status = %s AND decode_status = %s
+            LIMIT %s;
+        """
+        return await self.pool.async_fetch(
+            query=query,
+            params=(self.TaskStatus.SUCCESS, self.TaskStatus.INIT, self.TASK_BATCH),
+        )
+
+
+class InnerArticlesDecodeTaskMapper(ArticlesDecodeTaskMapper):
+    def __init__(self, pool: DatabaseManager):
+        super().__init__(pool)
+
     # 获取内部文章
     async def fetch_inner_articles(self):
         query = """
@@ -136,3 +159,21 @@ class InnerArticlesDecodeMapper(InnerArticlesDecodeConst):
 
         """
         return await self.pool.async_fetch(query=query)
+
+    # 获取内部文章生成信息
+    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,)
+        )
+
+
+__all__ = [
+    "ArticlesDecodeTaskMapper",
+    "AdPlatformArticlesDecodeTaskMapper",
+    "InnerArticlesDecodeTaskMapper",
+]

app/domains/decode_task/inner_articles_decode/_util.py → app/domains/llm_tasks/aigc_decode_task/_utils.py

@@ -3,51 +3,12 @@ from typing import Dict, List
 
 from app.infra.internal import DecodeServer
 
-from ._const import InnerArticlesDecodeConst
+from ._const import DecodeTaskConst
 
 
-class InnerArticlesDecodeUtil(InnerArticlesDecodeConst):
+class DecodeTaskUtil(DecodeTaskConst):
     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,
@@ -55,9 +16,9 @@ class InnerArticlesDecodeUtil(InnerArticlesDecodeConst):
             "content": {
                 "channel_content_id": article.get("wx_sn", ""),
                 "video_url": "",
-                "images": article.get("images", []),
+                "images": article.get("article_images"),
                 "body_text": article.get("article_text", ""),
-                "title": article.get("title", ""),
+                "title": article.get("article_title", ""),
                 "channel_account_id": article.get("gh_id", ""),
                 "channel_account_name": article.get("account_name", ""),
             },
@@ -100,3 +61,59 @@ class InnerArticlesDecodeUtil(InnerArticlesDecodeConst):
             "key_point": _join_points(keypoint_list, "关键点"),
             "topic": topic_text,
         }
+
+    async def fetch_decode_result(self, task_id: str):
+        return await self.decode_server.fetch_result(task_id)
+
+
+class AdPlatformArticlesDecodeUtils(DecodeTaskUtil):
+    @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):
+        images = self.format_images(article.get("article_images") or "")
+        article["article_images"] = images
+        request_body = self.prepare_extract_body(article)
+        return await self.decode_server.create_decode_task(request_body)
+
+
+class InnerArticlesDecodeUtils(DecodeTaskUtil):
+    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["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)
+
+
+__all__ = [
+    "AdPlatformArticlesDecodeUtils",
+    "InnerArticlesDecodeUtils",
+    "DecodeTaskUtil",
+]

app/domains/llm_tasks/aigc_decode_task/create_decode_tasks.py

@@ -0,0 +1,174 @@
+from typing import Dict
+from tqdm import tqdm
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+
+from ._const import DecodeTaskConst
+from ._mapper import AdPlatformArticlesDecodeTaskMapper, InnerArticlesDecodeTaskMapper
+from ._utils import AdPlatformArticlesDecodeUtils, InnerArticlesDecodeUtils
+
+
+class CreateAdPlatformArticlesDecodeTask(DecodeTaskConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = AdPlatformArticlesDecodeTaskMapper(self.pool)
+        self.tool = AdPlatformArticlesDecodeUtils()
+
+    async def create_single_decode_task(self, article: Dict):
+        # Acquire Lock
+        article_id = article["id"]
+        acquire_lock = await self.mapper.update_article_decode_status(
+            article_id, self.TaskStatus.INIT, self.TaskStatus.PROCESSING
+        )
+        if not acquire_lock:
+            await self.log_service.log(
+                contents={
+                    "article_id": article_id,
+                    "task": "create_decode_task",
+                    "status": "skip",
+                    "message": "acquire lock failed",
+                }
+            )
+            return
+
+        # 与解构系统交互，创建解构任务
+        response = await self.tool.create_decode_task(article)
+        response_code = response.get("code")
+        if response_code != self.RequestDecode.SUCCESS:
+            # 解构任务创建失败
+            await self.mapper.update_article_decode_status(
+                article_id, self.TaskStatus.PROCESSING, self.TaskStatus.FAILED
+            )
+            await self.log_service.log(
+                contents={
+                    "article_id": article_id,
+                    "task": "create_decode_task",
+                    "status": "fail",
+                    "data": response,
+                }
+            )
+            return
+
+        task_id = response.get("data", {}).get("task_id") or response.get(
+            "data", {}
+        ).get("taskId")
+        if not task_id:
+            # 解构任务创建失败
+            await self.mapper.update_article_decode_status(
+                article_id, self.TaskStatus.PROCESSING, self.TaskStatus.FAILED
+            )
+            await self.log_service.log(
+                contents={
+                    "article_id": article_id,
+                    "task": "create_decode_task",
+                    "status": "fail",
+                    "data": response,
+                }
+            )
+            return
+
+        # 创建 decode 任务成功
+        await self.log_service.log(
+            contents={
+                "article_id": article_id,
+                "task": "create_decode_task",
+                "status": "success",
+                "data": response,
+            }
+        )
+
+        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:
+            # 记录解构任务失败
+            await self.mapper.update_article_decode_status(
+                article_id, self.TaskStatus.PROCESSING, self.TaskStatus.FAILED
+            )
+            await self.log_service.log(
+                contents={
+                    "article_id": article_id,
+                    "task": "record_decode_task",
+                    "status": "fail",
+                    "message": "创建 decode 记录失败",
+                    "data": response,
+                }
+            )
+            return
+
+        # 记录创建成功
+        await self.mapper.update_article_decode_status(
+            article_id, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+        )
+
+    async def create_tasks(self):
+        article_list = await self.mapper.fetch_decode_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, desc="Creating decode tasks"):
+            await self.create_single_decode_task(article)
+
+    async def deal(self):
+        await self.create_tasks()
+
+
+class CreateInnerArticlesDecodeTask(DecodeTaskConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = InnerArticlesDecodeTaskMapper(self.pool)
+        self.tool = InnerArticlesDecodeUtils()
+
+    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 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, desc="Creating decode tasks"):
+            await self.create_single_decode_task(article)
+
+    async def deal(self):
+        await self.create_tasks()
+
+
+__all__ = ["CreateAdPlatformArticlesDecodeTask", "CreateInnerArticlesDecodeTask"]

app/domains/llm_tasks/aigc_decode_task/extract_decode_task_detail.py

@@ -0,0 +1,106 @@
+import json
+
+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 DecodeTaskConst
+from ._mapper import ArticlesDecodeTaskMapper
+from ._utils import DecodeTaskUtil
+
+
+class ExtractDecodeTaskDetail(DecodeTaskConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = ArticlesDecodeTaskMapper(self.pool)
+        self.tool = DecodeTaskUtil()
+
+    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 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):
+        await self.extract_task()
+
+
+__all__ = ["ExtractDecodeTaskDetail"]

app/domains/llm_tasks/aigc_decode_task/fetch_decode_results.py

@@ -0,0 +1,127 @@
+import json
+from typing import Dict
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+
+from ._const import DecodeTaskConst
+from ._mapper import ArticlesDecodeTaskMapper
+from ._utils import DecodeTaskUtil
+
+
+class FetchDecodeResults(DecodeTaskConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = ArticlesDecodeTaskMapper(self.pool)
+        self.tool = DecodeTaskUtil()
+
+    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 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 decoding_tasks:
+            await self.fetch_single_task(task)
+
+    async def deal(self):
+        await self.fetch_results()
+
+
+__all__ = ["FetchDecodeResults"]

app/infra/internal/__init__.py

@@ -2,7 +2,7 @@
 from .piaoquan import change_video_audit_status
 from .piaoquan import publish_video_to_piaoquan
 from .piaoquan import fetch_piaoquan_video_list_detail
-from .piaoquan import DecodeServer
+from .piaoquan_decode_server import DecodeServer
 
 # aigc system api
 from .aigc_system import delete_illegal_gzh_articles

app/infra/internal/piaoquan.py

@@ -75,66 +75,4 @@ async def publish_video_to_piaoquan(oss_path: str, uid: str, title: str) -> Dict
     async with AsyncHttpClient() as client:
         response = await client.post(url, data=payload, headers=headers)
 
-    return response
-
-
-class DecodeServer:
-    base_url: str = "http://supply-content-deconstruction-api.piaoquantv.com"
-
-    # 创建解构任务
-    async def create_decode_task(self, data: Dict) -> Dict:
-        """
-        scene: 业务场景：0 选题; 1 创作; 2 制作;
-        content_type: 内容类型： 1 长文; 2 图文; 3 视频;
-        content:
-            channel_content_id:
-            video_url:
-            images: [ image_url_1, image_url_2, ...]
-            body_text: 长文内容
-            title: 文章标题
-            channel_account_id: 作者 id
-            channel_account_name: 作者名称
-        output_type: {
-
-        }
-        """
-        url = f"{self.base_url}/api/v1/content/tasks/decode"
-        headers = {
-            "User-Agent": "PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0",
-            "Content-Type": "application/json",
-        }
-        async with AsyncHttpClient() as client:
-            response = await client.post(url, json=data, headers=headers)
-
-        return response
-
-    # 获取解构结果
-    async def fetch_result(self, task_id: str) -> Dict:
-        """
-        INPUT: TaskId
-        OUTPUT: Dict
-        {
-          code: 0 | 404(task not exist),
-          msg: 'ok',
-          data: {
-              "taskId": "123",
-              "status": 3, 0 PENDING, 1 RUNNING, 2 SUCCESS, 3 FAILED
-              "result": None | JSON,
-              "reason": "" | 失败原因,
-              "url": {
-                  "pointUrl": "", 选题点结构结果页地址（仅解构任务有
-                  "weightUrl": "", 权重页地址（解构聚类都有）
-                  "patternUrl": "" 选题点聚类结果页地址（仅聚类任务有）
-              }
-          },
-        }
-        """
-        url = f"{self.base_url}/api/v1/content/tasks/{task_id}"
-        headers = {
-            "User-Agent": "PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0",
-            "Content-Type": "application/json",
-        }
-        async with AsyncHttpClient() as client:
-            response = await client.get(url, headers=headers)
-
-        return response
+    return response

app/infra/internal/piaoquan_decode_server.py

@@ -0,0 +1,65 @@
+from typing import Optional, Dict, List
+
+from app.infra.shared import AsyncHttpClient
+
+
+class DecodeServer:
+    base_url: str = "http://supply-content-deconstruction-api.piaoquantv.com"
+
+    # 创建解构任务
+    async def create_decode_task(self, data: Dict) -> Dict:
+        """
+        scene: 业务场景：0 选题; 1 创作; 2 制作;
+        content_type: 内容类型： 1 长文; 2 图文; 3 视频;
+        content:
+            channel_content_id:
+            video_url:
+            images: [ image_url_1, image_url_2, ...]
+            body_text: 长文内容
+            title: 文章标题
+            channel_account_id: 作者 id
+            channel_account_name: 作者名称
+        output_type: {
+
+        }
+        """
+        url = f"{self.base_url}/api/v1/content/tasks/decode"
+        headers = {
+            "User-Agent": "PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0",
+            "Content-Type": "application/json",
+        }
+        async with AsyncHttpClient() as client:
+            response = await client.post(url, json=data, headers=headers)
+
+        return response
+
+    # 获取解构结果
+    async def fetch_result(self, task_id: str) -> Dict:
+        """
+        INPUT: TaskId
+        OUTPUT: Dict
+        {
+          code: 0 | 404(task not exist),
+          msg: 'ok',
+          data: {
+              "taskId": "123",
+              "status": 3, 0 PENDING, 1 RUNNING, 2 SUCCESS, 3 FAILED
+              "result": None | JSON,
+              "reason": "" | 失败原因,
+              "url": {
+                  "pointUrl": "", 选题点结构结果页地址（仅解构任务有
+                  "weightUrl": "", 权重页地址（解构聚类都有）
+                  "patternUrl": "" 选题点聚类结果页地址（仅聚类任务有）
+              }
+          },
+        }
+        """
+        url = f"{self.base_url}/api/v1/content/tasks/{task_id}"
+        headers = {
+            "User-Agent": "PQSpeed/486 CFNetwork/1410.1 Darwin/22.6.0",
+            "Content-Type": "application/json",
+        }
+        async with AsyncHttpClient() as client:
+            response = await client.get(url, headers=headers)
+
+        return response

app/jobs/task_handler.py

@@ -31,8 +31,10 @@ from app.domains.data_recycle_tasks import (
     UpdateOutsideRootSourceIdAndUpdateTimeTask,
 )
 
-from app.domains.decode_task import AdPlatformArticlesDecodeTask
-
+from app.domains.llm_tasks.aigc_decode_task import CreateAdPlatformArticlesDecodeTask
+from app.domains.llm_tasks.aigc_decode_task import CreateInnerArticlesDecodeTask
+from app.domains.llm_tasks.aigc_decode_task import FetchDecodeResults
+from app.domains.llm_tasks.aigc_decode_task import ExtractDecodeTaskDetail
 from app.domains.llm_tasks import TitleRewrite
 from app.domains.llm_tasks import ArticlePoolCategoryGeneration
 from app.domains.llm_tasks import CandidateAccountQualityScoreRecognizer
@@ -476,28 +478,28 @@ class TaskHandler:
     @register("create_ad_platform_accounts_decode_task")
     async def _create_ad_platform_accounts_decode_task(self) -> int:
         """创建解构任务"""
-        task = AdPlatformArticlesDecodeTask(
+        task = CreateAdPlatformArticlesDecodeTask(
             pool=self.db_client, log_service=self.log_client
         )
-        await task.deal(task_name="create_tasks")
+        await task.deal()
         return TaskStatus.SUCCESS
 
-    @register("fetch_ad_platform_accounts_decode_result")
-    async def _fetch_ad_platform_accounts_decode_result(self) -> int:
+    @register("fetch_decode_result")
+    async def _fetch_decode_result(self) -> int:
         """获取解构任务结果"""
-        task = AdPlatformArticlesDecodeTask(
+        task = FetchDecodeResults(
             pool=self.db_client, log_service=self.log_client
         )
-        await task.deal(task_name="fetch_results")
+        await task.deal()
         return TaskStatus.SUCCESS
 
-    @register("extract_ad_platform_accounts_decode_result")
+    @register("extract_decode_result")
     async def _extract_ad_platform_accounts_decode_result(self) -> int:
         """提取解构任务结果"""
-        task = AdPlatformArticlesDecodeTask(
+        task = ExtractDecodeTaskDetail(
             pool=self.db_client, log_service=self.log_client
         )
-        await task.deal(task_name="extract")
+        await task.deal()
         return TaskStatus.SUCCESS