LongArticleTaskServer/app/domains/decode_task/inner_articles_decode/_util.py

import json
from typing import Dict, List

from app.infra.internal import DecodeServer

from ._const import InnerArticlesDecodeConst


class InnerArticlesDecodeUtil(InnerArticlesDecodeConst):
    decode_server = DecodeServer()

    @staticmethod
    def format_images(images: str) -> List[str]:
        """
        格式化图片字符串，空/非法 JSON 返回空列表。
        """
        if not images or not images.strip():
            return []
        try:
            image_list = json.loads(images)
        except (json.JSONDecodeError, TypeError):
            return []
        if not isinstance(image_list, list):
            return []
        return [
            i.get("image_url")
            for i in image_list
            if isinstance(i, dict) and i.get("image_url")
        ]

    async def create_decode_task(self, article: Dict, article_produce_info: List[Dict]):
        images = [
            i["output"]
            for i in article_produce_info
            if i["produce_module_type"]
            in (self.ProduceModuleType.COVER, self.ProduceModuleType.IMAGE)
        ]
        article["images"] = images
        text = [
            i["output"]
            for i in article_produce_info
            if i["produce_module_type"] == self.ProduceModuleType.CONTENT
        ]
        article["article_text"] = "\n".join(text)
        request_body = self.prepare_extract_body(article)
        return await self.decode_server.create_decode_task(request_body)

    async def fetch_decode_result(self, task_id: str):
        return await self.decode_server.fetch_result(task_id)

    def prepare_extract_body(self, article: Dict) -> Dict:
        return {
            "scene": self.BusinessScene.POINT_PICK,
            "content_type": self.ContentType.LONG_ARTICLE,
            "content": {
                "channel_content_id": article.get("wx_sn", ""),
                "video_url": "",
                "images": article.get("images", []),
                "body_text": article.get("article_text", ""),
                "title": article.get("title", ""),
                "channel_account_id": article.get("gh_id", ""),
                "channel_account_name": article.get("account_name", ""),
            },
        }

    @staticmethod
    def extract_decode_result(result: Dict) -> Dict:
        """
        从结构的结果中，解析出灵感点、目的点、关键点；
        """
        final_result = result.get("final_normalization_rebuild")
        if not final_result:
            return {"error": "解构结果中无 final_normalization_rebuild 信息"}
        # 灵感点
        inspiration_list = final_result.get("inspiration_final_result", {}).get(
            "最终灵感点列表", []
        )
        # 目的
        purpose_list = final_result.get("purpose_final_result", {}).get(
            "最终目的点列表", []
        )
        # 关键点
        keypoint_list = final_result.get("keypoint_final", {}).get("最终关键点列表", [])

        topic_fusion = final_result.get("topic_fusion_result", {})
        # 选题
        topic_text = (
            topic_fusion.get("最终选题", {}).get("选题", "")
            if isinstance(topic_fusion.get("最终选题"), dict)
            else ""
        )

        def _join_points(items: list, key: str) -> str:
            parts = [str(p[key]) for p in items if isinstance(p, dict) and p.get(key)]
            return ",".join(parts)

        return {
            "inspiration": _join_points(inspiration_list, "灵感点"),
            "purpose": _join_points(purpose_list, "目的点"),
            "key_point": _join_points(keypoint_list, "关键点"),
            "topic": topic_text,
        }