卡片解构

app/domains/llm_tasks/decode_article/_const.py

@@ -41,6 +41,8 @@ class DecodeArticleConst:
         WECHAT = 5  # 微信公众号
         TOUTIAO = 6  # 头条号
         PIAOQUAN = 10  # 票圈
+        GROWTH_MATERIAL = 14  # 长文素材
+        GROWTH_AUTO_REPLY_CARD = 15  # 长文自动回复卡片
 
     class ProduceModuleType:
         COVER = 1  # 封面

app/domains/llm_tasks/decode_cards/__init__.py

@@ -0,0 +1,7 @@
+from .create_decode_tasks import CreateCardsDecodeTask
+from .fetch_decode_results import FetchCardDecodeResults
+
+__all__ = [
+    "CreateCardsDecodeTask",
+    "FetchCardDecodeResults",
+]

app/domains/llm_tasks/decode_cards/_const.py

@@ -0,0 +1,41 @@
+class DecodeCardConst:
+    TASK_BATCH = 500
+    SUBMIT_BATCH = 50
+
+    # auto_reply_top_cards_daily.channel → AIGC config_id
+    CHANNEL_CONFIG_MAP = {
+        "公众号合作-即转-稳定": 67,
+        "公众号投流-稳定": 68,
+    }
+
+    class TaskStatus:
+        INIT = 0
+        PROCESSING = 1
+        SUCCESS = 2
+        FAILED = 99
+
+    class SubmitStatus:
+        SUCCESS = "SUCCESS"
+        PENDING = "PENDING"
+        FAILED = "FAILED"
+
+    class QueryStatus:
+        SUCCESS = "SUCCESS"
+        PENDING = "PENDING"
+        RUNNING = "RUNNING"
+        FAILED = "FAILED"
+
+    class SourceType:
+        PARTNER_CARD = 4  # 合作方即转卡片
+
+    class TaskChannel:
+        PARTNER_CARD = 4  # long_articles_decode_tasks.channel: 合作方即转卡片
+
+    class ContentModal:
+        PICTURE_TEXT = 2  # 图文
+
+    class Channel:
+        GROWTH_AUTO_REPLY_CARD = 15  # 长文自动回复卡片（AIGC API post channel）
+
+
+__all__ = ["DecodeCardConst"]

app/domains/llm_tasks/decode_cards/_mapper.py

@@ -0,0 +1,180 @@
+from typing import Dict, List
+
+from app.core.database import DatabaseManager
+
+from ._const import DecodeCardConst
+
+TABLE_SOURCE = "auto_reply_top_cards_daily"
+TABLE_TASK = "long_articles_decode_tasks"
+
+
+class CardDecodeTaskMapper(DecodeCardConst):
+    """卡片解构 Mapper — 操作 auto_reply_top_cards_daily 与 long_articles_decode_tasks"""
+
+    def __init__(self, pool: DatabaseManager):
+        self.pool = pool
+
+    # ——— auto_reply_top_cards_daily ———
+
+    async def fetch_cards(self) -> List[Dict]:
+        """获取待解构卡片：status=INIT"""
+        query = f"""
+            SELECT id, channel, share_cover, share_title, card_cover_id
+            FROM {TABLE_SOURCE}
+            WHERE status = %s
+            LIMIT %s
+        """
+        return await self.pool.async_fetch(
+            query=query, params=(self.TaskStatus.INIT, self.TASK_BATCH)
+        )
+
+    async def update_card_status(
+        self, id_: int, ori_status: int, new_status: int
+    ) -> int:
+        """更新卡片解构状态（乐观锁）"""
+        query = f"""
+            UPDATE {TABLE_SOURCE}
+            SET status = %s
+            WHERE id = %s AND status = %s
+        """
+        return await self.pool.async_save(
+            query=query, params=(new_status, id_, ori_status)
+        )
+
+    async def set_card_cover_id(self, id_: int, cover_id: str) -> int:
+        """回填 card_cover_id"""
+        query = f"""
+            UPDATE {TABLE_SOURCE}
+            SET card_cover_id = %s
+            WHERE id = %s
+        """
+        return await self.pool.async_save(query=query, params=(cover_id, id_))
+
+    # ——— long_articles_decode_tasks ———
+
+    async def insert_decode_task(
+        self,
+        source_id: str,
+        config_id: int,
+        payload: str,
+        remark: str = None,
+        status: int = None,
+    ) -> int:
+        if status is not None:
+            query = f"""
+                INSERT IGNORE INTO {TABLE_TASK}
+                    (source_id, config_id, source, channel, payload, remark, status)
+                VALUES (%s, %s, %s, %s, %s, %s, %s)
+            """
+            params = (
+                source_id, config_id, self.SourceType.PARTNER_CARD,
+                self.TaskChannel.PARTNER_CARD, payload, remark, status,
+            )
+        else:
+            query = f"""
+                INSERT IGNORE INTO {TABLE_TASK}
+                    (source_id, config_id, source, channel, payload, remark)
+                VALUES (%s, %s, %s, %s, %s, %s)
+            """
+            params = (
+                source_id, config_id, self.SourceType.PARTNER_CARD,
+                self.TaskChannel.PARTNER_CARD, payload, remark,
+            )
+        return await self.pool.async_save(query=query, params=params)
+
+    async def set_decode_result(
+        self,
+        source_id: str,
+        config_id: int,
+        result: str,
+        remark: str = None,
+    ) -> int:
+        query = f"""
+            UPDATE {TABLE_TASK}
+            SET status = %s, result = %s, remark = %s
+            WHERE source_id = %s AND status IN (%s, %s) AND config_id = %s
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                self.TaskStatus.SUCCESS,
+                result,
+                remark,
+                source_id,
+                self.TaskStatus.INIT,
+                self.TaskStatus.PROCESSING,
+                config_id,
+            ),
+        )
+
+    async def fetch_pending_tasks(self) -> List[Dict]:
+        query = f"""
+            SELECT source_id, config_id
+            FROM {TABLE_TASK}
+            WHERE status IN (%s, %s) AND source = %s
+            ORDER BY config_id
+            LIMIT %s
+        """
+        return await self.pool.async_fetch(
+            query=query,
+            params=(
+                self.TaskStatus.INIT,
+                self.TaskStatus.PROCESSING,
+                self.SourceType.PARTNER_CARD,
+                self.TASK_BATCH,
+            ),
+        )
+
+    async def update_task_status_by_source_id(
+        self,
+        source_id: str,
+        config_id: int,
+        new_status: int,
+        remark: str = None,
+    ) -> int:
+        query = f"""
+            UPDATE {TABLE_TASK}
+            SET status = %s, remark = %s
+            WHERE source_id = %s AND status IN (%s, %s) AND config_id = %s
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                new_status, remark, source_id,
+                self.TaskStatus.INIT, self.TaskStatus.PROCESSING,
+                config_id,
+            ),
+        )
+
+    async def fetch_existing_source_ids(
+        self, source_ids: List[str], config_id: int
+    ) -> set:
+        """批量查询已有任务记录的 source_id，用于去重跳过"""
+        if not source_ids:
+            return set()
+        placeholders = ",".join(["%s"] * len(source_ids))
+        query = f"""
+            SELECT source_id FROM {TABLE_TASK}
+            WHERE source_id IN ({placeholders})
+              AND config_id = %s
+              AND source = %s
+              AND status IN (%s, %s, %s, %s)
+        """
+        return {
+            r["source_id"]
+            for r in await self.pool.async_fetch(
+                query=query,
+                params=(
+                    *source_ids,
+                    config_id,
+                    self.SourceType.PARTNER_CARD,
+                    self.TaskStatus.INIT,
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.SUCCESS,
+                    self.TaskStatus.FAILED,
+                ),
+            )
+        }
+
+
+__all__ = ["CardDecodeTaskMapper"]

app/domains/llm_tasks/decode_cards/_utils.py

@@ -0,0 +1,119 @@
+import hashlib
+import random
+import time
+from typing import Dict, List
+
+import requests
+
+from app.infra.internal.aigc_decode_server import AigcDecodeServer
+
+from ._const import DecodeCardConst
+
+# 封面下载重试配置
+COVER_DOWNLOAD_MAX_RETRIES = 3
+COVER_DOWNLOAD_BACKOFF_BASE = 1.0  # 基础退避秒数
+COVER_DOWNLOAD_BACKOFF_MAX = 10.0  # 最大退避秒数
+
+
+class CardDecodeUtils(DecodeCardConst):
+    decode_server = AigcDecodeServer()
+
+    async def submit_decode_batch(
+        self, posts: List[Dict], *, config_id: int, skip_completed: bool = False
+    ) -> Dict[str, Dict]:
+        """分批提交卡片解构任务，返回 {content_id: {status, errorMessage}}"""
+        result = {}
+        for i in range(0, len(posts), self.SUBMIT_BATCH):
+            batch = posts[i : i + self.SUBMIT_BATCH]
+            response = await self.decode_server.submit_decode(
+                config_id=config_id, posts=batch, skip_completed=skip_completed
+            )
+            if response.get("code") == 0:
+                for item in response.get("data", []):
+                    result[item["channelContentId"]] = item
+            else:
+                for post in batch:
+                    cid = post["channelContentId"]
+                    result[cid] = {
+                        "channelContentId": cid,
+                        "status": "FAILED",
+                        "errorMessage": f"batch submit failed: {response}",
+                    }
+        return result
+
+    async def query_decode_results_batch(
+        self, content_ids: List[str], *, config_id: int
+    ) -> Dict[str, Dict]:
+        """分批查询卡片解构结果"""
+        result = {}
+        for i in range(0, len(content_ids), self.SUBMIT_BATCH):
+            batch = content_ids[i : i + self.SUBMIT_BATCH]
+            response = await self.decode_server.query_decode_results(
+                config_id=config_id, channel_content_ids=batch
+            )
+            if response.get("code") == 0:
+                for item in response.get("data", []):
+                    result[item["channelContentId"]] = item
+            else:
+                for cid in batch:
+                    result[cid] = {
+                        "channelContentId": cid,
+                        "status": "API_ERROR",
+                        "errorMessage": f"query API failed: {response}",
+                    }
+        return result
+
+    COVER_CDN_PREFIX = "https://rescdn.yishihui.com/"
+
+    @staticmethod
+    def _normalize_cover_url(url: str) -> str:
+        """不以 https: 开头的封面 URL 拼接 CDN 前缀"""
+        if not url.startswith("https:"):
+            return f"{CardDecodeUtils.COVER_CDN_PREFIX}{url}"
+        return url
+
+    @staticmethod
+    def compute_cover_md5(image_url: str) -> str:
+        """下载封面图片并计算 MD5，带重试 + 指数退避 + 抖动，应对高并发 CDN 限流"""
+        url = CardDecodeUtils._normalize_cover_url(image_url)
+        last_exc = None
+        for attempt in range(COVER_DOWNLOAD_MAX_RETRIES):
+            try:
+                resp = requests.get(url, timeout=30)
+                resp.raise_for_status()
+                return hashlib.md5(resp.content).hexdigest()
+            except requests.RequestException as e:
+                last_exc = e
+                if attempt < COVER_DOWNLOAD_MAX_RETRIES - 1:
+                    delay = min(
+                        COVER_DOWNLOAD_BACKOFF_BASE * (2 ** attempt),
+                        COVER_DOWNLOAD_BACKOFF_MAX,
+                    )
+                    jitter = random.uniform(0, delay * 0.5)
+                    time.sleep(delay + jitter)
+        raise last_exc
+
+    @staticmethod
+    def prepare_posts(cards: List[Dict]) -> List[Dict]:
+        """将卡片数据转换为 AIGC 解构 API 所需的 post 格式"""
+        posts = []
+        for card in cards:
+            images = []
+            cover = card.get("share_cover")
+            if cover:
+                images.append(CardDecodeUtils._normalize_cover_url(cover))
+            posts.append(
+                {
+                    "channelContentId": str(card["card_cover_id"]),
+                    "title": card.get("share_title", ""),
+                    "bodyText": "",
+                    "images": images,
+                    "video": None,
+                    "contentModal": DecodeCardConst.ContentModal.PICTURE_TEXT,
+                    "channel": DecodeCardConst.Channel.GROWTH_AUTO_REPLY_CARD,
+                }
+            )
+        return posts
+
+
+__all__ = ["CardDecodeUtils"]

app/domains/llm_tasks/decode_cards/create_decode_tasks.py

@@ -0,0 +1,356 @@
+import json
+from collections import defaultdict
+from typing import Dict, List
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+
+from ._const import DecodeCardConst
+from ._mapper import CardDecodeTaskMapper
+from ._utils import CardDecodeUtils
+
+
+class CreateCardsDecodeTask(DecodeCardConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = CardDecodeTaskMapper(self.pool)
+        self.tool = CardDecodeUtils()
+
+    async def _acquire_cards(self) -> List[Dict]:
+        """获取待解构卡片并加锁（status INIT → PROCESSING）"""
+        cards = await self.mapper.fetch_cards()
+        locked = []
+        for card in cards:
+            card_id = card["id"]
+            acquired = await self.mapper.update_card_status(
+                card_id, self.TaskStatus.INIT, self.TaskStatus.PROCESSING
+            )
+            if acquired:
+                locked.append(card)
+            else:
+                await self.log_service.log(
+                    contents={
+                        "card_id": card_id,
+                        "task": "create_card_decode_task",
+                        "status": "skip",
+                        "message": "acquire lock failed",
+                    }
+                )
+        return locked
+
+    async def _ensure_cover_ids(self, cards: List[Dict]):
+        """为缺少 card_cover_id 的卡片下载封面并计算 MD5"""
+        for card in cards:
+            if card.get("card_cover_id"):
+                continue
+            if not card.get("share_cover"):
+                continue  # 空封面留给 _submit_and_record 标记 FAILED
+            try:
+                cover_id = self.tool.compute_cover_md5(card["share_cover"])
+                await self.mapper.set_card_cover_id(card["id"], cover_id)
+                card["card_cover_id"] = cover_id
+                await self.log_service.log(
+                    contents={
+                        "card_id": card["id"],
+                        "task": "create_card_decode_task",
+                        "status": "info",
+                        "message": f"computed cover_id={cover_id}",
+                    }
+                )
+            except Exception as e:
+                await self.mapper.update_card_status(
+                    card["id"],
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.FAILED,
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card["id"],
+                        "share_cover": card.get("share_cover"),
+                        "task": "create_card_decode_task",
+                        "status": "fail",
+                        "message": f"compute cover md5 failed, marked as FAILED: {e}",
+                    }
+                )
+
+    @staticmethod
+    def _group_cards_by_channel(cards: List[Dict]) -> Dict[int, List[Dict]]:
+        """按 auto_reply_top_cards_daily.channel 分组，映射到对应的 config_id"""
+        grouped = defaultdict(list)
+        for card in cards:
+            source_channel = card.get("channel", "")
+            config_id = DecodeCardConst.CHANNEL_CONFIG_MAP.get(source_channel)
+            if config_id:
+                grouped[config_id].append(card)
+        return dict(grouped)
+
+    async def _submit_and_record_for_config(
+        self, cards: List[Dict], config_id: int
+    ):
+        """对同一 config_id 的卡片执行提交与落库"""
+        if not cards:
+            return
+
+        # 过滤已有任务记录的卡片（按 card_cover_id 去重）
+        cards_with_cid = [c for c in cards if c.get("card_cover_id")]
+        if not cards_with_cid:
+            return
+
+        # 跨批次去重：查 DB 已有任务
+        all_source_ids = [str(c["card_cover_id"]) for c in cards_with_cid]
+        existing = await self.mapper.fetch_existing_source_ids(
+            all_source_ids, config_id
+        )
+
+        # 同批次去重：相同 card_cover_id 只保留第一条
+        seen = set()
+        deduped_cards = []
+        intra_dups = []
+        for c in cards_with_cid:
+            cid = str(c["card_cover_id"])
+            if cid in existing or cid in seen:
+                intra_dups.append(c)
+            else:
+                seen.add(cid)
+                deduped_cards.append(c)
+
+        if intra_dups:
+            await self.log_service.log(
+                contents={
+                    "task": "create_card_decode_task",
+                    "config_id": config_id,
+                    "message": f"Skipped {len(intra_dups)} duplicate cards (already submitted or same-batch)",
+                }
+            )
+            for card in intra_dups:
+                await self.mapper.update_card_status(
+                    card["id"],
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.SUCCESS,
+                )
+
+        new_cards = deduped_cards
+
+        if not new_cards:
+            return
+
+        posts = self.tool.prepare_posts(new_cards)
+        submit_results = await self.tool.submit_decode_batch(
+            posts, config_id=config_id
+        )
+        posts_by_cid = {p["channelContentId"]: p for p in posts}
+
+        for card in new_cards:
+            source_id = str(card["card_cover_id"])
+            card_id = card["id"]
+            result = submit_results.get(source_id)
+
+            if not result:
+                await self.mapper.update_card_status(
+                    card_id, self.TaskStatus.PROCESSING, self.TaskStatus.INIT
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card_id,
+                        "source_id": source_id,
+                        "config_id": config_id,
+                        "task": "create_card_decode_task",
+                        "status": "fail",
+                        "message": "no response for source_id, rolled back to INIT",
+                    }
+                )
+                continue
+
+            status = result.get("status")
+            if status == self.SubmitStatus.FAILED:
+                await self.mapper.update_card_status(
+                    card_id, self.TaskStatus.PROCESSING, self.TaskStatus.INIT
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card_id,
+                        "source_id": source_id,
+                        "config_id": config_id,
+                        "task": "create_card_decode_task",
+                        "status": "fail",
+                        "data": result,
+                    }
+                )
+                continue
+
+            if status == self.SubmitStatus.SUCCESS:
+                query_results = await self.tool.query_decode_results_batch(
+                    [source_id], config_id=config_id
+                )
+                result_data = query_results.get(source_id)
+                if (
+                    result_data
+                    and result_data.get("status") == self.QueryStatus.SUCCESS
+                ):
+                    data_content = result_data.get("dataContent") or "{}"
+                    html = result_data.get("html")
+                    await self.mapper.insert_decode_task(
+                        source_id=source_id,
+                        config_id=config_id,
+                        payload=json.dumps(
+                            posts_by_cid.get(source_id, {}), ensure_ascii=False
+                        ),
+                        remark="提交时已有解构结果，直接落库",
+                    )
+                    await self.mapper.set_decode_result(
+                        source_id=source_id,
+                        config_id=config_id,
+                        result=json.dumps(
+                            {"dataContent": data_content, "html": html},
+                            ensure_ascii=False,
+                        ),
+                        remark="提交时已返回 SUCCESS，结果已落库",
+                    )
+                    await self.mapper.update_card_status(
+                        card_id, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+                    )
+                    await self.log_service.log(
+                        contents={
+                            "card_id": card_id,
+                            "source_id": source_id,
+                            "config_id": config_id,
+                            "task": "create_card_decode_task",
+                            "status": "success",
+                            "message": "decode result already available on submit",
+                        }
+                    )
+                else:
+                    await self.mapper.insert_decode_task(
+                        source_id=source_id,
+                        config_id=config_id,
+                        payload=json.dumps(
+                            posts_by_cid.get(source_id, {}), ensure_ascii=False
+                        ),
+                        remark="提交返回SUCCESS，查询未果，等待轮询",
+                        status=self.TaskStatus.PROCESSING,
+                    )
+                    await self.mapper.update_card_status(
+                        card_id, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+                    )
+                    await self.log_service.log(
+                        contents={
+                            "card_id": card_id,
+                            "source_id": source_id,
+                            "config_id": config_id,
+                            "task": "create_card_decode_task",
+                            "status": "pending",
+                            "message": "submit SUCCESS but query not ready, inserted for polling",
+                        }
+                    )
+            elif status == self.SubmitStatus.PENDING:
+                await self.mapper.insert_decode_task(
+                    source_id=source_id,
+                    config_id=config_id,
+                    payload=json.dumps(
+                        posts_by_cid.get(source_id, {}), ensure_ascii=False
+                    ),
+                    remark="卡片解构任务已提交，等待轮询",
+                    status=self.TaskStatus.PROCESSING,
+                )
+                await self.mapper.update_card_status(
+                    card_id, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card_id,
+                        "source_id": source_id,
+                        "config_id": config_id,
+                        "task": "create_card_decode_task",
+                        "status": "pending",
+                        "message": "task submitted, waiting for polling",
+                    }
+                )
+            else:
+                await self.mapper.update_card_status(
+                    card_id, self.TaskStatus.PROCESSING, self.TaskStatus.INIT
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card_id,
+                        "source_id": source_id,
+                        "config_id": config_id,
+                        "task": "create_card_decode_task",
+                        "status": "fail",
+                        "message": f"unexpected submit status: {status}, rolled back to INIT",
+                        "data": result,
+                    }
+                )
+
+    async def _submit_and_record(self, cards: List[Dict]):
+        if not cards:
+            return
+
+        # 无 share_cover 的卡片直接标记失败
+        valid_cards = []
+        for card in cards:
+            if not card.get("share_cover"):
+                await self.mapper.update_card_status(
+                    card["id"],
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.FAILED,
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card["id"],
+                        "task": "create_card_decode_task",
+                        "status": "fail",
+                        "message": "share_cover is empty, marked as FAILED",
+                    }
+                )
+            else:
+                valid_cards.append(card)
+
+        if not valid_cards:
+            return
+
+        grouped = self._group_cards_by_channel(valid_cards)
+        for config_id, group_cards in grouped.items():
+            await self._submit_and_record_for_config(group_cards, config_id)
+
+        # 处理不在映射表中的卡片（回滚状态）
+        mapped_ids = {id(c) for g in grouped.values() for c in g}
+        for card in valid_cards:
+            if id(card) not in mapped_ids:
+                await self.mapper.update_card_status(
+                    card["id"],
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.INIT,
+                )
+                await self.log_service.log(
+                    contents={
+                        "card_id": card["id"],
+                        "channel": card.get("channel"),
+                        "task": "create_card_decode_task",
+                        "status": "fail",
+                        "message": "unknown channel, rolled back to INIT",
+                    }
+                )
+
+    async def deal(self):
+        cards = await self._acquire_cards()
+        if not cards:
+            await self.log_service.log(
+                contents={
+                    "task": "create_card_decode_task",
+                    "message": "No more cards to decode",
+                }
+            )
+            return
+
+        await self._ensure_cover_ids(cards)
+        await self._submit_and_record(cards)
+        await self.log_service.log(
+            contents={
+                "task": "create_card_decode_task",
+                "message": f"Processed {len(cards)} cards",
+            }
+        )
+
+
+__all__ = ["CreateCardsDecodeTask"]

app/domains/llm_tasks/decode_cards/fetch_decode_results.py

@@ -0,0 +1,129 @@
+import json
+from collections import defaultdict
+from typing import List, Dict
+
+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 DecodeCardConst
+from ._mapper import CardDecodeTaskMapper
+from ._utils import CardDecodeUtils
+
+
+class FetchCardDecodeResults(DecodeCardConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = CardDecodeTaskMapper(self.pool)
+        self.tool = CardDecodeUtils()
+
+    @staticmethod
+    def _group_tasks_by_config(tasks: List[Dict]) -> Dict[int, List[Dict]]:
+        grouped = defaultdict(list)
+        for task in tasks:
+            grouped[task["config_id"]].append(task)
+        return dict(grouped)
+
+    async def _process_batch(self, tasks: List[Dict], config_id: int):
+        source_ids = [t["source_id"] for t in tasks]
+        results = await self.tool.query_decode_results_batch(
+            source_ids, config_id=config_id
+        )
+
+        for task in tasks:
+            source_id = task["source_id"]
+            result = results.get(source_id)
+
+            if not result:
+                await self.mapper.update_task_status_by_source_id(
+                    source_id=source_id,
+                    config_id=config_id,
+                    new_status=self.TaskStatus.FAILED,
+                    remark="卡片解构任务在结果查询中未返回",
+                )
+                await self.log_service.log(
+                    contents={
+                        "task": "fetch_card_decode_results",
+                        "source_id": source_id,
+                        "config_id": config_id,
+                        "status": "fail",
+                        "message": "source_id not in query response",
+                    }
+                )
+                continue
+
+            status = result.get("status")
+            if status == "API_ERROR":
+                continue
+            elif status == self.QueryStatus.SUCCESS:
+                data_content = result.get("dataContent") or "{}"
+                html = result.get("html")
+                await self.mapper.set_decode_result(
+                    source_id=source_id,
+                    config_id=config_id,
+                    result=json.dumps(
+                        {"dataContent": data_content, "html": html},
+                        ensure_ascii=False,
+                    ),
+                    remark="卡片解构结果获取成功",
+                )
+            elif status in (self.QueryStatus.PENDING, self.QueryStatus.RUNNING):
+                pass
+            elif status == self.QueryStatus.FAILED:
+                await self.mapper.update_task_status_by_source_id(
+                    source_id=source_id,
+                    config_id=config_id,
+                    new_status=self.TaskStatus.FAILED,
+                    remark=f"卡片解构任务失败: {result.get('errorMessage', '')}",
+                )
+            else:
+                await self.log_service.log(
+                    contents={
+                        "task": "fetch_card_decode_results",
+                        "source_id": source_id,
+                        "config_id": config_id,
+                        "status": "unknown",
+                        "message": f"unexpected query status: {status}",
+                        "data": result,
+                    }
+                )
+
+    async def deal(self):
+        pending_tasks = await self.mapper.fetch_pending_tasks()
+        if not pending_tasks:
+            await self.log_service.log(
+                contents={
+                    "task": "fetch_card_decode_results",
+                    "message": "No more card tasks to fetch",
+                }
+            )
+            return
+
+        grouped = self._group_tasks_by_config(pending_tasks)
+        for config_id, tasks in grouped.items():
+            batches = [
+                tasks[i : i + self.SUBMIT_BATCH]
+                for i in range(0, len(tasks), self.SUBMIT_BATCH)
+            ]
+            await run_tasks_with_asyncio_task_group(
+                task_list=[
+                    {"batch": b, "config_id": config_id} for b in batches
+                ],
+                handler=lambda item, cid=config_id: self._process_batch(
+                    item["batch"], cid
+                ),
+                description="批量查询卡片解构结果",
+                unit="batch",
+            )
+
+        await self.log_service.log(
+            contents={
+                "task": "fetch_card_decode_results",
+                "message": f"Processed {len(pending_tasks)} pending card tasks across {len(grouped)} configs",
+            }
+        )
+
+
+__all__ = ["FetchCardDecodeResults"]

app/domains/llm_tasks/decode_material/__init__.py

@@ -0,0 +1,7 @@
+from .create_decode_tasks import CreateMaterialsDecodeTask
+from .fetch_decode_results import FetchMaterialDecodeResults
+
+__all__ = [
+    "CreateMaterialsDecodeTask",
+    "FetchMaterialDecodeResults",
+]

app/domains/llm_tasks/decode_material/_const.py

@@ -0,0 +1,36 @@
+class DecodeMaterialConst:
+    CONFIG_ID = 69
+    TASK_BATCH = 100
+    SUBMIT_BATCH = 50
+
+    class TaskStatus:
+        INIT = 0
+        PROCESSING = 1
+        SUCCESS = 2
+        FAILED = 99
+
+    class SubmitStatus:
+        SUCCESS = "SUCCESS"
+        PENDING = "PENDING"
+        FAILED = "FAILED"
+
+    class QueryStatus:
+        SUCCESS = "SUCCESS"
+        PENDING = "PENDING"
+        RUNNING = "RUNNING"
+        FAILED = "FAILED"
+
+    class SourceType:
+        MATERIAL = 5  # 素材
+
+    class TaskChannel:
+        MATERIAL = 3  # long_articles_decode_tasks.channel: 素材
+
+    class ContentModal:
+        PICTURE_TEXT = 2  # 图文
+
+    class Channel:
+        GROWTH_MATERIAL = 14  # 长文素材（AIGC API post channel）
+
+
+__all__ = ["DecodeMaterialConst"]

app/domains/llm_tasks/decode_material/_mapper.py

@@ -0,0 +1,166 @@
+from typing import Dict, List
+
+from app.core.database import DatabaseManager
+
+from ._const import DecodeMaterialConst
+
+TABLE_SOURCE = "growth_daily_material"
+TABLE_TASK = "long_articles_decode_tasks"
+
+
+class MaterialDecodeTaskMapper(DecodeMaterialConst):
+    """素材解构 Mapper — 操作 growth_daily_material 与 long_articles_decode_tasks"""
+
+    def __init__(self, pool: DatabaseManager):
+        self.pool = pool
+
+    # ——— growth_daily_material ———
+
+    async def fetch_materials(self) -> List[Dict]:
+        """获取待解构素材：status=INIT"""
+        query = f"""
+            SELECT id, material_id, material_title, material_cover
+            FROM {TABLE_SOURCE}
+            WHERE status = %s
+            LIMIT %s
+        """
+        return await self.pool.async_fetch(
+            query=query, params=(self.TaskStatus.INIT, self.TASK_BATCH)
+        )
+
+    async def update_material_status(
+        self, id_: int, ori_status: int, new_status: int
+    ) -> int:
+        """更新素材解构状态（乐观锁）"""
+        query = f"""
+            UPDATE {TABLE_SOURCE}
+            SET status = %s
+            WHERE id = %s AND status = %s
+        """
+        return await self.pool.async_save(
+            query=query, params=(new_status, id_, ori_status)
+        )
+
+    # ——— long_articles_decode_tasks ———
+
+    async def insert_decode_task(
+        self,
+        source_id: str,
+        payload: str,
+        remark: str = None,
+        status: int = None,
+    ) -> int:
+        if status is not None:
+            query = f"""
+                INSERT IGNORE INTO {TABLE_TASK}
+                    (source_id, config_id, source, channel, payload, remark, status)
+                VALUES (%s, %s, %s, %s, %s, %s, %s)
+            """
+            params = (
+                source_id, self.CONFIG_ID, self.SourceType.MATERIAL,
+                self.TaskChannel.MATERIAL, payload, remark, status,
+            )
+        else:
+            query = f"""
+                INSERT IGNORE INTO {TABLE_TASK}
+                    (source_id, config_id, source, channel, payload, remark)
+                VALUES (%s, %s, %s, %s, %s, %s)
+            """
+            params = (
+                source_id, self.CONFIG_ID, self.SourceType.MATERIAL,
+                self.TaskChannel.MATERIAL, payload, remark,
+            )
+        return await self.pool.async_save(query=query, params=params)
+
+    async def set_decode_result(
+        self,
+        source_id: str,
+        result: str,
+        remark: str = None,
+    ) -> int:
+        query = f"""
+            UPDATE {TABLE_TASK}
+            SET status = %s, result = %s, remark = %s
+            WHERE source_id = %s AND status IN (%s, %s) AND config_id = %s
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                self.TaskStatus.SUCCESS,
+                result,
+                remark,
+                source_id,
+                self.TaskStatus.INIT,
+                self.TaskStatus.PROCESSING,
+                self.CONFIG_ID,
+            ),
+        )
+
+    async def fetch_pending_tasks(self) -> List[Dict]:
+        query = f"""
+            SELECT source_id
+            FROM {TABLE_TASK}
+            WHERE status IN (%s, %s) AND source = %s AND config_id = %s
+            LIMIT %s
+        """
+        return await self.pool.async_fetch(
+            query=query,
+            params=(
+                self.TaskStatus.INIT,
+                self.TaskStatus.PROCESSING,
+                self.SourceType.MATERIAL,
+                self.CONFIG_ID,
+                self.TASK_BATCH,
+            ),
+        )
+
+    async def update_task_status_by_source_id(
+        self,
+        source_id: str,
+        new_status: int,
+        remark: str = None,
+    ) -> int:
+        query = f"""
+            UPDATE {TABLE_TASK}
+            SET status = %s, remark = %s
+            WHERE source_id = %s AND status IN (%s, %s) AND config_id = %s
+        """
+        return await self.pool.async_save(
+            query=query,
+            params=(
+                new_status, remark, source_id,
+                self.TaskStatus.INIT, self.TaskStatus.PROCESSING,
+                self.CONFIG_ID,
+            ),
+        )
+
+    async def fetch_existing_source_ids(self, source_ids: List[str]) -> set:
+        """批量查询已有任务记录的 source_id，用于去重跳过"""
+        if not source_ids:
+            return set()
+        placeholders = ",".join(["%s"] * len(source_ids))
+        query = f"""
+            SELECT source_id FROM {TABLE_TASK}
+            WHERE source_id IN ({placeholders})
+              AND config_id = %s
+              AND source = %s
+              AND status IN (%s, %s, %s, %s)
+        """
+        return {
+            r["source_id"]
+            for r in await self.pool.async_fetch(
+                query=query,
+                params=(
+                    *source_ids,
+                    self.CONFIG_ID,
+                    self.SourceType.MATERIAL,
+                    self.TaskStatus.INIT,
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.SUCCESS,
+                    self.TaskStatus.FAILED,
+                ),
+            )
+        }
+
+
+__all__ = ["MaterialDecodeTaskMapper"]

app/domains/llm_tasks/decode_material/_utils.py

@@ -0,0 +1,79 @@
+from typing import Dict, List
+
+from app.infra.internal.aigc_decode_server import AigcDecodeServer
+
+from ._const import DecodeMaterialConst
+
+
+class MaterialDecodeUtils(DecodeMaterialConst):
+    decode_server = AigcDecodeServer()
+
+    async def submit_decode_batch(
+        self, posts: List[Dict], *, skip_completed: bool = False
+    ) -> Dict[str, Dict]:
+        """分批提交素材解构任务"""
+        result = {}
+        for i in range(0, len(posts), self.SUBMIT_BATCH):
+            batch = posts[i : i + self.SUBMIT_BATCH]
+            response = await self.decode_server.submit_decode(
+                config_id=self.CONFIG_ID, posts=batch, skip_completed=skip_completed
+            )
+            if response.get("code") == 0:
+                for item in response.get("data", []):
+                    result[item["channelContentId"]] = item
+            else:
+                for post in batch:
+                    cid = post["channelContentId"]
+                    result[cid] = {
+                        "channelContentId": cid,
+                        "status": "FAILED",
+                        "errorMessage": f"batch submit failed: {response}",
+                    }
+        return result
+
+    async def query_decode_results_batch(
+        self, content_ids: List[str]
+    ) -> Dict[str, Dict]:
+        """分批查询素材解构结果"""
+        result = {}
+        for i in range(0, len(content_ids), self.SUBMIT_BATCH):
+            batch = content_ids[i : i + self.SUBMIT_BATCH]
+            response = await self.decode_server.query_decode_results(
+                config_id=self.CONFIG_ID, channel_content_ids=batch
+            )
+            if response.get("code") == 0:
+                for item in response.get("data", []):
+                    result[item["channelContentId"]] = item
+            else:
+                for cid in batch:
+                    result[cid] = {
+                        "channelContentId": cid,
+                        "status": "API_ERROR",
+                        "errorMessage": f"query API failed: {response}",
+                    }
+        return result
+
+    @staticmethod
+    def prepare_posts(materials: List[Dict]) -> List[Dict]:
+        """将素材数据转换为 AIGC 解构 API 所需的 post 格式"""
+        posts = []
+        for m in materials:
+            images = []
+            cover = m.get("material_cover")
+            if cover:
+                images.append(cover)
+            posts.append(
+                {
+                    "channelContentId": str(m["material_id"]),
+                    "title": m.get("material_title", ""),
+                    "bodyText": "",
+                    "images": images,
+                    "video": None,
+                    "contentModal": DecodeMaterialConst.ContentModal.PICTURE_TEXT,
+                    "channel": DecodeMaterialConst.Channel.GROWTH_MATERIAL,
+                }
+            )
+        return posts
+
+
+__all__ = ["MaterialDecodeUtils"]

app/domains/llm_tasks/decode_material/create_decode_tasks.py

@@ -0,0 +1,254 @@
+import json
+from typing import Dict, List
+
+from app.core.database import DatabaseManager
+from app.core.observability import LogService
+
+from ._const import DecodeMaterialConst
+from ._mapper import MaterialDecodeTaskMapper
+from ._utils import MaterialDecodeUtils
+
+
+class CreateMaterialsDecodeTask(DecodeMaterialConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = MaterialDecodeTaskMapper(self.pool)
+        self.tool = MaterialDecodeUtils()
+
+    async def _acquire_materials(self) -> List[Dict]:
+        """获取待解构素材并加锁（status INIT → PROCESSING）"""
+        materials = await self.mapper.fetch_materials()
+        locked = []
+        for m in materials:
+            mid = m["id"]
+            acquired = await self.mapper.update_material_status(
+                mid, self.TaskStatus.INIT, self.TaskStatus.PROCESSING
+            )
+            if acquired:
+                locked.append(m)
+            else:
+                await self.log_service.log(
+                    contents={
+                        "material_id": mid,
+                        "task": "create_material_decode_task",
+                        "status": "skip",
+                        "message": "acquire lock failed",
+                    }
+                )
+        return locked
+
+    async def _submit_and_record(self, materials: List[Dict]):
+        if not materials:
+            return
+
+        # 无 material_cover 的素材直接标记失败
+        valid = []
+        for m in materials:
+            if not m.get("material_cover"):
+                await self.mapper.update_material_status(
+                    m["id"],
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.FAILED,
+                )
+                await self.log_service.log(
+                    contents={
+                        "material_id": m["id"],
+                        "task": "create_material_decode_task",
+                        "status": "fail",
+                        "message": "material_cover is empty, marked as FAILED",
+                    }
+                )
+            else:
+                valid.append(m)
+
+        if not valid:
+            return
+
+        # 跨批次去重：查 DB 已有任务
+        all_source_ids = [str(m["material_id"]) for m in valid]
+        existing = await self.mapper.fetch_existing_source_ids(all_source_ids)
+
+        # 同批次去重：相同 material_id 只保留第一条
+        seen = set()
+        deduped = []
+        dups = []
+        for m in valid:
+            mid = str(m["material_id"])
+            if mid in existing or mid in seen:
+                dups.append(m)
+            else:
+                seen.add(mid)
+                deduped.append(m)
+
+        if dups:
+            await self.log_service.log(
+                contents={
+                    "task": "create_material_decode_task",
+                    "message": f"Skipped {len(dups)} duplicate materials (already submitted or same-batch)",
+                }
+            )
+            for m in dups:
+                await self.mapper.update_material_status(
+                    m["id"],
+                    self.TaskStatus.PROCESSING,
+                    self.TaskStatus.SUCCESS,
+                )
+
+        if not deduped:
+            return
+
+        posts = self.tool.prepare_posts(deduped)
+        submit_results = await self.tool.submit_decode_batch(posts)
+        posts_by_cid = {p["channelContentId"]: p for p in posts}
+
+        for m in deduped:
+            source_id = str(m["material_id"])
+            mid = m["id"]
+            result = submit_results.get(source_id)
+
+            if not result:
+                await self.mapper.update_material_status(
+                    mid, self.TaskStatus.PROCESSING, self.TaskStatus.INIT
+                )
+                await self.log_service.log(
+                    contents={
+                        "material_id": mid,
+                        "source_id": source_id,
+                        "task": "create_material_decode_task",
+                        "status": "fail",
+                        "message": "no response for source_id, rolled back to INIT",
+                    }
+                )
+                continue
+
+            status = result.get("status")
+            if status == self.SubmitStatus.FAILED:
+                await self.mapper.update_material_status(
+                    mid, self.TaskStatus.PROCESSING, self.TaskStatus.INIT
+                )
+                await self.log_service.log(
+                    contents={
+                        "material_id": mid,
+                        "source_id": source_id,
+                        "task": "create_material_decode_task",
+                        "status": "fail",
+                        "data": result,
+                    }
+                )
+                continue
+
+            if status == self.SubmitStatus.SUCCESS:
+                query_results = await self.tool.query_decode_results_batch([source_id])
+                result_data = query_results.get(source_id)
+                if (
+                    result_data
+                    and result_data.get("status") == self.QueryStatus.SUCCESS
+                ):
+                    data_content = result_data.get("dataContent") or "{}"
+                    html = result_data.get("html")
+                    await self.mapper.insert_decode_task(
+                        source_id=source_id,
+                        payload=json.dumps(
+                            posts_by_cid.get(source_id, {}), ensure_ascii=False
+                        ),
+                        remark="提交时已有解构结果，直接落库",
+                    )
+                    await self.mapper.set_decode_result(
+                        source_id=source_id,
+                        result=json.dumps(
+                            {"dataContent": data_content, "html": html},
+                            ensure_ascii=False,
+                        ),
+                        remark="提交时已返回 SUCCESS，结果已落库",
+                    )
+                    await self.mapper.update_material_status(
+                        mid, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+                    )
+                    await self.log_service.log(
+                        contents={
+                            "material_id": mid,
+                            "source_id": source_id,
+                            "task": "create_material_decode_task",
+                            "status": "success",
+                            "message": "decode result already available on submit",
+                        }
+                    )
+                else:
+                    await self.mapper.insert_decode_task(
+                        source_id=source_id,
+                        payload=json.dumps(
+                            posts_by_cid.get(source_id, {}), ensure_ascii=False
+                        ),
+                        remark="提交返回SUCCESS，查询未果，等待轮询",
+                        status=self.TaskStatus.PROCESSING,
+                    )
+                    await self.mapper.update_material_status(
+                        mid, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+                    )
+                    await self.log_service.log(
+                        contents={
+                            "material_id": mid,
+                            "source_id": source_id,
+                            "task": "create_material_decode_task",
+                            "status": "pending",
+                            "message": "submit SUCCESS but query not ready, inserted for polling",
+                        }
+                    )
+            elif status == self.SubmitStatus.PENDING:
+                await self.mapper.insert_decode_task(
+                    source_id=source_id,
+                    payload=json.dumps(
+                        posts_by_cid.get(source_id, {}), ensure_ascii=False
+                    ),
+                    remark="素材解构任务已提交，等待轮询",
+                    status=self.TaskStatus.PROCESSING,
+                )
+                await self.mapper.update_material_status(
+                    mid, self.TaskStatus.PROCESSING, self.TaskStatus.SUCCESS
+                )
+                await self.log_service.log(
+                    contents={
+                        "material_id": mid,
+                        "source_id": source_id,
+                        "task": "create_material_decode_task",
+                        "status": "pending",
+                        "message": "task submitted, waiting for polling",
+                    }
+                )
+            else:
+                await self.mapper.update_material_status(
+                    mid, self.TaskStatus.PROCESSING, self.TaskStatus.INIT
+                )
+                await self.log_service.log(
+                    contents={
+                        "material_id": mid,
+                        "source_id": source_id,
+                        "task": "create_material_decode_task",
+                        "status": "fail",
+                        "message": f"unexpected submit status: {status}, rolled back to INIT",
+                        "data": result,
+                    }
+                )
+
+    async def deal(self):
+        materials = await self._acquire_materials()
+        if not materials:
+            await self.log_service.log(
+                contents={
+                    "task": "create_material_decode_task",
+                    "message": "No more materials to decode",
+                }
+            )
+            return
+
+        await self._submit_and_record(materials)
+        await self.log_service.log(
+            contents={
+                "task": "create_material_decode_task",
+                "message": f"Processed {len(materials)} materials",
+            }
+        )
+
+
+__all__ = ["CreateMaterialsDecodeTask"]

app/domains/llm_tasks/decode_material/fetch_decode_results.py

@@ -0,0 +1,108 @@
+import json
+from typing import List, Dict
+
+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 DecodeMaterialConst
+from ._mapper import MaterialDecodeTaskMapper
+from ._utils import MaterialDecodeUtils
+
+
+class FetchMaterialDecodeResults(DecodeMaterialConst):
+    def __init__(self, pool: DatabaseManager, log_service: LogService):
+        self.pool = pool
+        self.log_service = log_service
+        self.mapper = MaterialDecodeTaskMapper(self.pool)
+        self.tool = MaterialDecodeUtils()
+
+    async def _process_batch(self, tasks: List[Dict]):
+        source_ids = [t["source_id"] for t in tasks]
+        results = await self.tool.query_decode_results_batch(source_ids)
+
+        for task in tasks:
+            source_id = task["source_id"]
+            result = results.get(source_id)
+
+            if not result:
+                await self.mapper.update_task_status_by_source_id(
+                    source_id=source_id,
+                    new_status=self.TaskStatus.FAILED,
+                    remark="素材解构任务在结果查询中未返回",
+                )
+                await self.log_service.log(
+                    contents={
+                        "task": "fetch_material_decode_results",
+                        "source_id": source_id,
+                        "status": "fail",
+                        "message": "source_id not in query response",
+                    }
+                )
+                continue
+
+            status = result.get("status")
+            if status == "API_ERROR":
+                continue
+            elif status == self.QueryStatus.SUCCESS:
+                data_content = result.get("dataContent") or "{}"
+                html = result.get("html")
+                await self.mapper.set_decode_result(
+                    source_id=source_id,
+                    result=json.dumps(
+                        {"dataContent": data_content, "html": html},
+                        ensure_ascii=False,
+                    ),
+                    remark="素材解构结果获取成功",
+                )
+            elif status in (self.QueryStatus.PENDING, self.QueryStatus.RUNNING):
+                pass
+            elif status == self.QueryStatus.FAILED:
+                await self.mapper.update_task_status_by_source_id(
+                    source_id=source_id,
+                    new_status=self.TaskStatus.FAILED,
+                    remark=f"素材解构任务失败: {result.get('errorMessage', '')}",
+                )
+            else:
+                await self.log_service.log(
+                    contents={
+                        "task": "fetch_material_decode_results",
+                        "source_id": source_id,
+                        "status": "unknown",
+                        "message": f"unexpected query status: {status}",
+                        "data": result,
+                    }
+                )
+
+    async def deal(self):
+        pending_tasks = await self.mapper.fetch_pending_tasks()
+        if not pending_tasks:
+            await self.log_service.log(
+                contents={
+                    "task": "fetch_material_decode_results",
+                    "message": "No more material tasks to fetch",
+                }
+            )
+            return
+
+        batches = [
+            pending_tasks[i : i + self.SUBMIT_BATCH]
+            for i in range(0, len(pending_tasks), self.SUBMIT_BATCH)
+        ]
+        await run_tasks_with_asyncio_task_group(
+            task_list=batches,
+            handler=self._process_batch,
+            description="批量查询素材解构结果",
+            unit="batch",
+        )
+
+        await self.log_service.log(
+            contents={
+                "task": "fetch_material_decode_results",
+                "message": f"Processed {len(pending_tasks)} pending material tasks in {len(batches)} batches",
+            }
+        )
+
+
+__all__ = ["FetchMaterialDecodeResults"]

app/jobs/domains/llm_task.py

@@ -2,6 +2,10 @@ from app.domains.llm_tasks.decode_article import CreateAdPlatformArticlesDecodeT
 from app.domains.llm_tasks.decode_article import CreateInnerArticlesDecodeTask
 from app.domains.llm_tasks.decode_article import FetchDecodeResults
 from app.domains.llm_tasks.decode_article import ExtractDecodeTaskDetail
+from app.domains.llm_tasks.decode_cards import CreateCardsDecodeTask
+from app.domains.llm_tasks.decode_cards import FetchCardDecodeResults
+from app.domains.llm_tasks.decode_material import CreateMaterialsDecodeTask
+from app.domains.llm_tasks.decode_material import FetchMaterialDecodeResults
 from app.domains.llm_tasks import TitleRewrite
 from app.domains.llm_tasks import ArticlePoolCategoryGeneration
 from app.domains.llm_tasks import CandidateAccountQualityScoreRecognizer
@@ -13,6 +17,10 @@ __all__ = [
     "CreateInnerArticlesDecodeTask",
     "FetchDecodeResults",
     "ExtractDecodeTaskDetail",
+    "CreateCardsDecodeTask",
+    "FetchCardDecodeResults",
+    "CreateMaterialsDecodeTask",
+    "FetchMaterialDecodeResults",
     "TitleRewrite",
     "ArticlePoolCategoryGeneration",
     "CandidateAccountQualityScoreRecognizer",

app/jobs/task_config.py

@@ -134,6 +134,24 @@ TASK_CONFIGS = {
         max_concurrent=3,
         retry_times=2,
     ),
+    # 卡片解构任务
+    "create_cards_decode_task": TaskConfig(
+        timeout=3600,
+        max_concurrent=2,
+    ),
+    "fetch_cards_decode_result": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
+    # 素材解构任务
+    "create_material_decode_task": TaskConfig(
+        timeout=1800,
+        max_concurrent=2,
+    ),
+    "fetch_material_decode_result": TaskConfig(
+        timeout=1800,
+        max_concurrent=3,
+    ),
     # 统计分析类任务
     "update_account_read_rate_avg": TaskConfig(
         timeout=1800,

app/jobs/task_handler.py

@@ -478,6 +478,46 @@ class TaskHandler:
         await task.deal()
         return TaskStatus.SUCCESS
 
+    # ====================== 卡片解构任务 ======================
+
+    @register("create_cards_decode_task")
+    async def _create_cards_decode_task(self) -> int:
+        """创建卡片解构任务"""
+        task = CreateCardsDecodeTask(
+            pool=self.db_client, log_service=self.log_client
+        )
+        await task.deal()
+        return TaskStatus.SUCCESS
+
+    @register("fetch_cards_decode_result")
+    async def _fetch_cards_decode_result(self) -> int:
+        """获取卡片解构结果"""
+        task = FetchCardDecodeResults(
+            pool=self.db_client, log_service=self.log_client
+        )
+        await task.deal()
+        return TaskStatus.SUCCESS
+
+    # ====================== 素材解构任务 ======================
+
+    @register("create_material_decode_task")
+    async def _create_material_decode_task(self) -> int:
+        """创建素材解构任务"""
+        task = CreateMaterialsDecodeTask(
+            pool=self.db_client, log_service=self.log_client
+        )
+        await task.deal()
+        return TaskStatus.SUCCESS
+
+    @register("fetch_material_decode_result")
+    async def _fetch_material_decode_result(self) -> int:
+        """获取素材解构结果"""
+        task = FetchMaterialDecodeResults(
+            pool=self.db_client, log_service=self.log_client
+        )
+        await task.deal()
+        return TaskStatus.SUCCESS
+
     # ====================== Recommend Tasks=====================
     @register("i2i_recommend_data_sync")
     async def _i2i_recommend_data_sync_handler(self) -> int: