content-find-agent-new/content_agent/integrations/aigc_platform.py

"""AIGC 平台接口:把 CFA 入池视频推进 AIGC 生成计划。

3 个接口(文档见 `tech_documents/AIGC 接口调用说明（create_crawler_plan_by_{douyin,kuaishou}_content_id 链路）.md`):
1. 建爬取计划 `/aigc/crawler/plan/save`(传一批平台内容 ID,抖音 aweme_id / 快手 photo_id,返回 crawler_plan_id)
2. 查生成计划详情 `/aigc/produce/plan/detail`
3. 存生成计划 `/aigc/produce/plan/save`(把 crawler_plan_id 作为输入源增量追加)

本模块只做纯接口调用,不含 dry-run / 安全闸等编排逻辑(那些在 scripts/push_runs_to_aigc.py)。
"""

from __future__ import annotations

from typing import Any

import httpx

# 各平台建爬取计划的固定参数(crawlerMode=5 按内容 ID 抓取)。
# 抖音文档 §2.2:channel=2、filterMatchMode=2;快手文档 §2.2:channel=7、filterMatchMode=1。
_PLATFORM_CRAWLER_PARAMS: dict[str, dict[str, Any]] = {
    "douyin": {
        "channel": 2,
        "contentModal": 4,
        "crawlerComment": 0,
        "crawlerMode": 5,
        "filterAccountMatchMode": 2,
        "filterContentMatchMode": 2,
        "frequencyType": 2,
        "planType": 2,
        "searchModeValues": [],
        "srtExtractFlag": 1,
        "videoKeyFrameType": 1,
        "voiceExtractFlag": 1,
    },
    "kuaishou": {
        "channel": 7,
        "contentModal": 4,
        "crawlerComment": 0,
        "crawlerMode": 5,
        "filterAccountMatchMode": 1,
        "filterContentMatchMode": 1,
        "frequencyType": 2,
        "planType": 2,
        "searchModeValues": [],
        "srtExtractFlag": 1,
        "videoKeyFrameType": 1,
        "voiceExtractFlag": 1,
    },
}

# 绑进生成计划时,匹配输入源组 + 新输入源的固定字段(channel 随平台:抖音 2 / 快手 7,文档 4.2)。
_PLATFORM_INPUT_MATCH: dict[str, dict[str, Any]] = {
    "douyin": {"inputSourceModal": 4, "inputSourceChannel": 2, "contentType": 1},
    "kuaishou": {"inputSourceModal": 4, "inputSourceChannel": 7, "contentType": 1},
}


class AigcPlatformError(RuntimeError):
    pass


class AigcPlatformClient:
    def __init__(
        self,
        *,
        base_url: str,
        token: str,
        paths: dict[str, str],
        timeout: float = 30.0,
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._token = token
        self._paths = paths
        self._timeout = timeout

    @classmethod
    def from_env(cls, env: dict[str, str]) -> "AigcPlatformClient":
        token = env.get("CONTENTFIND_API_AIGC_TOKEN") or env.get("AIGC_TOKEN") or ""
        if not token:
            raise AigcPlatformError("缺 CONTENTFIND_API_AIGC_TOKEN")
        return cls(
            base_url=env.get("CONTENTFIND_API_AIGC_BASE_URL") or "https://aigc-api.aiddit.com",
            token=token,
            paths={
                "crawler_plan_save": env.get("CONTENTFIND_AIGC_CRAWLER_PLAN_SAVE_PATH") or "/aigc/crawler/plan/save",
                "produce_plan_detail": env.get("CONTENTFIND_AIGC_PRODUCE_PLAN_DETAIL_PATH") or "/aigc/produce/plan/detail",
                "produce_plan_save": env.get("CONTENTFIND_AIGC_PRODUCE_PLAN_SAVE_PATH") or "/aigc/produce/plan/save",
            },
        )

    def _post(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
        body = {"baseInfo": {"token": self._token}, "params": params}
        response = httpx.post(
            f"{self._base_url}{path}",
            json=body,
            headers={"Content-Type": "application/json"},
            timeout=self._timeout,
        )
        response.raise_for_status()
        data = response.json()
        if not isinstance(data, dict):
            raise AigcPlatformError(f"{path}: 非法响应 {data!r}")
        if data.get("code") not in (0, "0"):
            raise AigcPlatformError(f"{path}: 业务失败 code={data.get('code')} msg={data.get('msg')}")
        return data.get("data") or {}

    def create_crawler_plan(self, content_ids: list[str], *, platform: str, name: str) -> str:
        """接口一:按平台建爬取计划(抖音/快手,按内容 ID 抓取),返回 crawler_plan_id(响应 data.id)。"""
        crawler_params = _PLATFORM_CRAWLER_PARAMS.get(platform)
        if crawler_params is None:
            raise AigcPlatformError(f"不支持的平台:{platform}")
        params = {**crawler_params, "inputModeValues": list(content_ids), "name": name}
        data = self._post(self._paths["crawler_plan_save"], params)
        plan_id = str(data.get("id") or "")
        if not plan_id:
            raise AigcPlatformError("接口一未返回 data.id")
        return plan_id

    def get_produce_plan(self, produce_plan_id: str) -> dict[str, Any]:
        """接口二:查生成计划完整详情(含 inputSourceGroups)。"""
        data = self._post(self._paths["produce_plan_detail"], {"id": produce_plan_id})
        if not data:
            raise AigcPlatformError(f"接口二空 data,produce_plan_id={produce_plan_id}")
        return data

    def bind_crawler_to_produce(
        self,
        *,
        crawler_plan_id: str,
        produce_plan_id: str,
        label: str,
        platform: str,
    ) -> None:
        """接口二+三:把 crawler_plan_id 作为新输入源,增量追加进生成计划并整体保存。

        文档 4.4 要求:把接口二返回的完整 data 原样带上,只增量追加,避免字段缺失保存失败。
        """
        input_match = _PLATFORM_INPUT_MATCH.get(platform)
        if input_match is None:
            raise AigcPlatformError(f"不支持的平台:{platform}")
        plan = self.get_produce_plan(produce_plan_id)
        group = _match_input_group(plan.get("inputSourceGroups") or [], input_match)
        if group is None:
            raise AigcPlatformError(f"生成计划 {produce_plan_id} 找不到匹配的输入源组")
        group.setdefault("inputSources", []).append(
            {
                **input_match,
                "inputSourceType": 2,
                "inputSourceValue": crawler_plan_id,
                "inputSourceLabel": label,
            }
        )
        self._post(self._paths["produce_plan_save"], plan)


def _match_input_group(
    groups: list[dict[str, Any]], input_match: dict[str, Any]
) -> dict[str, Any] | None:
    """找含匹配输入源(modal/channel/contentType,channel 随平台)的那一组(文档 4.2 step2);
    没匹配但只有一组时退而用它;否则返回 None(不瞎猜)。"""
    for group in groups:
        for source in group.get("inputSources") or []:
            if all(source.get(key) == value for key, value in input_match.items()):
                return group
    return groups[0] if len(groups) == 1 else None