Agent/examples/auto_put_ad_mini/tools/ai_generated_material.py

"""AI generated creative image assets for module B.

This module is intentionally isolated from the historical material recall path.
External side effects happen only when an account explicitly uses
material_source=ai_generated.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import mimetypes
import os
import re
import uuid
from dataclasses import dataclass
from email.utils import formatdate
from io import BytesIO
from pathlib import Path
from typing import Iterable, Optional
from urllib.parse import quote, urlparse

import httpx
from PIL import Image

from tools.material_recall import Material
from tools.video_feature_query import VideoElementFeature, read_cached_video_element_features
from tools.video_recall import LandingVideo

logger = logging.getLogger(__name__)

OPENROUTER_CHAT_COMPLETIONS_URL = os.getenv(
    "OPENROUTER_CHAT_COMPLETIONS_URL",
    "https://openrouter.ai/api/v1/chat/completions",
)
OPENROUTER_IMAGES_URL = os.getenv(
    "OPENROUTER_IMAGES_URL",
    "https://openrouter.ai/api/v1/images",
)
OPENROUTER_IMAGE_MODEL = os.getenv(
    "OPENROUTER_IMAGE_MODEL",
    "google/gemini-3-pro-image",
)
OPENROUTER_TEXT_MODEL = os.getenv(
    "OPENROUTER_TEXT_MODEL",
    "google/gemini-2.5-flash",
)
AI_IMAGE_OSS_PREFIX = os.getenv("AI_IMAGE_OSS_PREFIX", "auto_put_tencent/image").strip("/")
AI_IMAGE_ASPECT_RATIO = os.getenv("AI_IMAGE_ASPECT_RATIO", "16:9")
AI_IMAGE_RESOLUTION = os.getenv("AI_IMAGE_RESOLUTION", "1K")
AI_IMAGE_OUTPUT_FORMAT = os.getenv("AI_IMAGE_OUTPUT_FORMAT", "jpeg")
AI_IMAGE_TARGET_WIDTH = int(os.getenv("AI_IMAGE_TARGET_WIDTH", "1280"))
AI_IMAGE_TARGET_HEIGHT = int(os.getenv("AI_IMAGE_TARGET_HEIGHT", "720"))
DEFAULT_AI_IMAGE_PROMPT_TEMPLATE_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "ai_generated_material.md"
)
DEFAULT_AI_SANITIZE_PROMPT_TEMPLATE_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "ai_sanitize_video_description.md"
)
DEFAULT_AI_IMAGE_OSS_BUCKET = "art-pubbucket"
DEFAULT_AI_IMAGE_PUBLIC_BASE_URL = "https://rescdn.yishihui.com"
FORBIDDEN_SANITIZED_DESCRIPTION_TERMS = (
    "公众号", "私信", "评论区", "扫码", "二维码", "加群", "关注",
    "回复", "关键词", "转发", "推广", "引导", "点击",
    "下载", "按钮", "弹窗", "底部固定", "领取", "能领", "赶紧",
    "通知", "已办成", "不错过", "社交分享", "分享给", "转发分享",
    "优惠券", "卡券", "领券", "下单", "购买", "价格", "订单",
    "商品售卖", "课程售卖", "App下载", "小程序推广",
    "真人相册", "私人相册", "小视频", "小短片", "寂寞", "性感",
    "风情", "舞女", "小姐", "软妹", "学妹", "女郎", "娇妇", "俏妹",
)


CREATE_AI_MATERIAL_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS ai_generated_material (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
    adgroup_id BIGINT DEFAULT NULL COMMENT '广告ID',
    crowd_package VARCHAR(200) NOT NULL COMMENT '投放人群包名称',
    landing_video_id BIGINT NOT NULL COMMENT '承接视频ID',
    landing_title VARCHAR(500) DEFAULT NULL COMMENT '承接视频标题',
    landing_category VARCHAR(200) DEFAULT NULL COMMENT '承接视频品类',
    prompt_type VARCHAR(50) NOT NULL COMMENT '生成角度:topic',
    prompt_text MEDIUMTEXT NOT NULL COMMENT '最终生成prompt',
    prompt_feature_json MEDIUMTEXT DEFAULT NULL COMMENT 'prompt使用的视频特征JSON',
    model VARCHAR(200) NOT NULL COMMENT 'OpenRouter模型',
    oss_object_key VARCHAR(500) DEFAULT NULL COMMENT 'OSS对象key',
    oss_url VARCHAR(1000) DEFAULT NULL COMMENT 'OSS公网URL',
    status VARCHAR(50) NOT NULL DEFAULT 'generated' COMMENT 'generated/prepared/approved/rejected/hold/skip/posted_ok/post_failed/error',
    approval_status VARCHAR(50) DEFAULT NULL COMMENT '人工审批状态',
    tencent_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
    dynamic_creative_id BIGINT DEFAULT NULL COMMENT '腾讯动态创意ID',
    error TEXT DEFAULT NULL COMMENT '生成/上传/创建错误',
    raw_response MEDIUMTEXT DEFAULT NULL COMMENT '模型原始响应摘要',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    KEY idx_account_ad_landing_status (account_id, adgroup_id, landing_video_id, status),
    KEY idx_crowd_created (crowd_package, created_at),
    KEY idx_status_created (status, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='AI生成创意图片素材'
"""


@dataclass(frozen=True)
class GenerationPrompt:
    prompt_type: str
    prompt_text: str
    feature_hits: list[dict]


@dataclass(frozen=True)
class GeneratedMaterialAsset:
    id: int
    account_id: int
    adgroup_id: Optional[int]
    crowd_package: str
    landing_video_id: int
    prompt_type: str
    prompt_text: str
    model: str
    oss_url: str
    oss_object_key: str
    feature_hits: list[dict]

    @property
    def material_id(self) -> str:
        return f"ai:{self.id}"

    def to_material(self) -> Material:
        return Material(
            material_id=self.material_id,
            score=1.0,
            title=f"AI生成素材-{self.prompt_type}",
            cover=self.oss_url,
            cost=None,
            ctr=None,
            cvr=None,
            roi=None,
            impressions=None,
            quality_score=None,
            recall_strategy=f"AI生成-{self.prompt_type}",
            recall_query_text=self.prompt_text[:500],
            recall_config_code="AI_GENERATED_IMAGE",
            recall_element_dimension="AI生成素材",
            recall_point_type=self.prompt_type,
            recall_standard_element="",
            recall_hit_queries=self.feature_hits,
            raw={
                "ai_generated_material_id": self.id,
                "oss_url": self.oss_url,
                "prompt_type": self.prompt_type,
                "prompt_text": self.prompt_text,
                "model": self.model,
            },
        )


def _feature_attr(feature, name: str, default=""):
    if isinstance(feature, dict):
        return feature.get(name, default)
    return getattr(feature, name, default)


def _feature_to_hit(feature) -> dict:
    return {
        "element_dimension": str(_feature_attr(feature, "element_dimension") or ""),
        "point_type": str(_feature_attr(feature, "point_type") or ""),
        "standard_element": str(_feature_attr(feature, "standard_element") or ""),
        "contribution_score": float(_feature_attr(feature, "contribution_score", 0) or 0),
        "dt": str(_feature_attr(feature, "dt") or ""),
    }


def _top_feature(features: Iterable, *, dimension: str, point_type: str = "") -> Optional[dict]:
    candidates = []
    for feature in features:
        hit = _feature_to_hit(feature)
        if hit["element_dimension"] != dimension:
            continue
        if point_type and hit["point_type"] != point_type:
            continue
        if not hit["standard_element"] or hit["standard_element"] == "-":
            continue
        candidates.append(hit)
    if not candidates:
        return None
    return sorted(candidates, key=lambda x: x["contribution_score"], reverse=True)[0]


def _load_prompt_template() -> str:
    raw_path = os.getenv("AI_IMAGE_PROMPT_TEMPLATE_PATH", "").strip()
    path = Path(raw_path).expanduser() if raw_path else DEFAULT_AI_IMAGE_PROMPT_TEMPLATE_PATH
    return path.read_text(encoding="utf-8")


def _load_sanitize_prompt_messages(raw_description: str) -> list[dict]:
    raw_path = os.getenv("AI_SANITIZE_PROMPT_TEMPLATE_PATH", "").strip()
    path = Path(raw_path).expanduser() if raw_path else DEFAULT_AI_SANITIZE_PROMPT_TEMPLATE_PATH
    template = path.read_text(encoding="utf-8")
    if "【system】" not in template or "【user】" not in template:
        raise RuntimeError(f"清洗prompt模板缺少【system】/【user】标记:{path}")
    system_part, user_part = template.split("【user】", 1)
    system_text = system_part.replace("【system】", "", 1).strip()
    user_text = user_part.replace("{{raw_description}}", raw_description).strip()
    if not system_text or not user_text:
        raise RuntimeError(f"清洗prompt模板为空:{path}")
    return [
        {"role": "system", "content": system_text},
        {"role": "user", "content": user_text},
    ]


def _render_prompt(video_description: str) -> str:
    return (
        _load_prompt_template()
        .replace("{{video_description}}", video_description)
        .replace("{{aspect_ratio}}", AI_IMAGE_ASPECT_RATIO)
    )


def _extract_chat_completion_text(data: dict) -> str:
    choices = data.get("choices") or []
    if not choices:
        raise RuntimeError("OpenRouter 清洗响应缺 choices")
    message = choices[0].get("message") or {}
    content = message.get("content")
    if isinstance(content, str):
        return content.strip()
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, dict) and isinstance(item.get("text"), str):
                parts.append(item["text"])
        return "\n".join(parts).strip()
    return ""


def sanitize_video_description(raw_description: str, model: str = OPENROUTER_TEXT_MODEL) -> str:
    """Rewrite ODPS topic into a clean visual description for image generation."""
    raw = (raw_description or "").strip()
    if not raw:
        raise RuntimeError("缺少视频解构选题,无法清洗生成描述")

    body = {
        "model": model,
        "temperature": 0.2,
        "max_tokens": 180,
        "messages": _load_sanitize_prompt_messages(raw),
    }
    headers = {
        "authorization": f"Bearer {_openrouter_api_key()}",
        "content-type": "application/json",
        "accept": "application/json",
    }
    resp = httpx.post(OPENROUTER_CHAT_COMPLETIONS_URL, json=body, headers=headers, timeout=60)
    resp.raise_for_status()
    cleaned = _extract_chat_completion_text(resp.json())
    cleaned = cleaned.strip().strip("`").strip().strip("“”\"'").replace("\n", "")
    if not cleaned:
        raise RuntimeError("OpenRouter 清洗后描述为空")
    hit_terms = [term for term in FORBIDDEN_SANITIZED_DESCRIPTION_TERMS if term in cleaned]
    if hit_terms:
        raise RuntimeError(f"OpenRouter 清洗后仍包含互动/营销词:{','.join(hit_terms)}")
    return cleaned


def build_generation_prompts(
    *,
    video_id: int,
    title: str,
    category: str,
    features: Iterable[VideoElementFeature],
    sanitized_description: str = "",
) -> list[GenerationPrompt]:
    """Build one generation prompt from the video's ODPS topic."""
    feature_list = list(features or [])
    topic = _top_feature(feature_list, dimension="解构选题")
    topic_text = (topic or {}).get("standard_element")
    if not topic_text:
        return []
    video_description = sanitized_description.strip() or topic_text
    topic_hit = dict(topic)
    topic_hit["original_standard_element"] = topic_text
    topic_hit["video_description"] = video_description
    return [
        GenerationPrompt(
            prompt_type="topic",
            prompt_text=_render_prompt(video_description),
            feature_hits=[topic_hit],
        )
    ]


def ensure_ai_material_table() -> None:
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(CREATE_AI_MATERIAL_TABLE_SQL)
        conn.commit()
    finally:
        conn.close()


def _openrouter_api_key() -> str:
    key = os.getenv("OPEN_ROUTER_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not key:
        raise RuntimeError("缺少 OPENROUTER_API_KEY/OPEN_ROUTER_API_KEY,无法生成AI素材")
    return key


def _to_jpeg_bytes(image_bytes: bytes) -> bytes:
    """Normalize model output to RGB JPEG for Tencent material upload."""
    with Image.open(BytesIO(image_bytes)) as img:
        if img.mode not in ("RGB", "L"):
            background = Image.new("RGB", img.size, (255, 255, 255))
            if img.mode in ("RGBA", "LA"):
                alpha = img.getchannel("A")
                background.paste(img.convert("RGB"), mask=alpha)
            else:
                background.paste(img.convert("RGB"))
            out_img = background
        else:
            out_img = img.convert("RGB")
        target_ratio = AI_IMAGE_TARGET_WIDTH / AI_IMAGE_TARGET_HEIGHT
        image_ratio = out_img.width / out_img.height
        if image_ratio > target_ratio:
            new_width = int(out_img.height * target_ratio)
            left = (out_img.width - new_width) // 2
            out_img = out_img.crop((left, 0, left + new_width, out_img.height))
        elif image_ratio < target_ratio:
            new_height = int(out_img.width / target_ratio)
            top = (out_img.height - new_height) // 2
            out_img = out_img.crop((0, top, out_img.width, top + new_height))
        out_img = out_img.resize((AI_IMAGE_TARGET_WIDTH, AI_IMAGE_TARGET_HEIGHT), Image.Resampling.LANCZOS)
        out = BytesIO()
        out_img.save(out, format="JPEG", quality=92, optimize=True)
        return out.getvalue()


def _extract_image_ref_from_openrouter(data: dict) -> str:
    choices = data.get("choices") or []
    if not choices:
        raise RuntimeError("OpenRouter 响应缺 choices")
    msg = choices[0].get("message") or {}
    images = msg.get("images") or []
    for image in images:
        image_url = image.get("image_url") if isinstance(image, dict) else None
        if isinstance(image_url, dict) and image_url.get("url"):
            return image_url["url"]
        if isinstance(image_url, str):
            return image_url

    content = msg.get("content")
    if isinstance(content, list):
        for part in content:
            if not isinstance(part, dict):
                continue
            image_url = part.get("image_url")
            if isinstance(image_url, dict) and image_url.get("url"):
                return image_url["url"]
            if part.get("type") in ("image_url", "output_image") and part.get("url"):
                return part["url"]
    if isinstance(content, str):
        match = re.search(r"!\[[^\]]*]\(([^)]+)\)", content)
        if match:
            return match.group(1)
        data_match = re.search(r"(data:image/[^;\s]+;base64,[A-Za-z0-9+/=]+)", content)
        if data_match:
            return data_match.group(1)
        url_match = re.search(r"https?://\S+", content)
        if url_match:
            return url_match.group(0).rstrip(").,，。")
    raise RuntimeError("OpenRouter 响应未包含可识别图片URL/base64")


def generate_image_bytes(prompt: str, model: str = OPENROUTER_IMAGE_MODEL) -> tuple[bytes, str, dict]:
    """Generate one image through OpenRouter Images API.

    The Images API lets us pass aspect_ratio directly. We still normalize the
    returned raster to JPEG in code because not every model supports
    output_format.
    """
    body = {
        "model": model,
        "prompt": prompt,
        "aspect_ratio": AI_IMAGE_ASPECT_RATIO,
        "resolution": AI_IMAGE_RESOLUTION,
        "n": 1,
    }
    headers = {
        "authorization": f"Bearer {_openrouter_api_key()}",
        "content-type": "application/json",
        "accept": "application/json",
    }
    resp = httpx.post(OPENROUTER_IMAGES_URL, json=body, headers=headers, timeout=120)
    resp.raise_for_status()
    data = resp.json()
    images = data.get("data") or []
    if not images or not images[0].get("b64_json"):
        raise RuntimeError("OpenRouter Images API 响应缺少 data[0].b64_json")
    image_bytes = base64.b64decode(images[0]["b64_json"])
    if AI_IMAGE_OUTPUT_FORMAT.lower() in ("jpg", "jpeg"):
        return _to_jpeg_bytes(image_bytes), "image/jpeg", data
    media_type = images[0].get("media_type") or "image/png"
    return image_bytes, media_type, data


def _public_oss_url(public_base_url: str, object_key: str) -> str:
    return f"{public_base_url.rstrip('/')}/{quote(object_key, safe='/')}"


def _oss_env() -> tuple[str, str, str, str, str, str]:
    endpoint = os.getenv("ALIYUN_OSS_ENDPOINT", "").strip()
    bucket = (
        os.getenv("ALIYUN_OSS_BUCKET", "").strip()
        or os.getenv("AI_IMAGE_OSS_BUCKET", "").strip()
        or DEFAULT_AI_IMAGE_OSS_BUCKET
    )
    access_key_id = os.getenv("ALIYUN_OSS_ACCESS_KEY_ID", "").strip()
    access_key_secret = os.getenv("ALIYUN_OSS_ACCESS_KEY_SECRET", "").strip()
    public_base_url = (
        os.getenv("AI_IMAGE_PUBLIC_BASE_URL", "").strip()
        or DEFAULT_AI_IMAGE_PUBLIC_BASE_URL
    )
    object_prefix = os.getenv("AI_IMAGE_OSS_PREFIX", AI_IMAGE_OSS_PREFIX).strip("/")
    missing = [
        name for name, value in [
            ("ALIYUN_OSS_ENDPOINT", endpoint),
            ("ALIYUN_OSS_BUCKET", bucket),
            ("ALIYUN_OSS_ACCESS_KEY_ID", access_key_id),
            ("ALIYUN_OSS_ACCESS_KEY_SECRET", access_key_secret),
        ]
        if not value
    ]
    if missing:
        raise RuntimeError(f"缺少 OSS 配置:{','.join(missing)}")
    return endpoint, bucket, access_key_id, access_key_secret, public_base_url.rstrip("/"), object_prefix


def _normalize_oss_endpoint(endpoint: str, bucket: str) -> str:
    endpoint = endpoint.rstrip("/")
    if not endpoint.startswith(("http://", "https://")):
        endpoint = "https://" + endpoint
    parsed = urlparse(endpoint)
    host = parsed.netloc
    if not host.startswith(f"{bucket}."):
        host = f"{bucket}.{host}"
    return f"{parsed.scheme}://{host}"


def upload_image_to_oss(image_bytes: bytes, content_type: str, object_key: str) -> str:
    endpoint, bucket, access_key_id, access_key_secret, public_base_url, _ = _oss_env()
    content_md5 = base64.b64encode(hashlib.md5(image_bytes).digest()).decode("ascii")
    date = formatdate(usegmt=True)
    canonical_resource = f"/{bucket}/{object_key}"
    string_to_sign = f"PUT\n{content_md5}\n{content_type}\n{date}\n{canonical_resource}"
    signature = base64.b64encode(
        hmac.new(
            access_key_secret.encode("utf-8"),
            string_to_sign.encode("utf-8"),
            hashlib.sha1,
        ).digest()
    ).decode("ascii")
    upload_base = _normalize_oss_endpoint(endpoint, bucket)
    url = f"{upload_base}/{quote(object_key, safe='/')}"
    resp = httpx.put(
        url,
        content=image_bytes,
        headers={
            "Date": date,
            "Content-Type": content_type,
            "Content-MD5": content_md5,
            "Authorization": f"OSS {access_key_id}:{signature}",
        },
        timeout=60,
    )
    resp.raise_for_status()
    return _public_oss_url(public_base_url, object_key)


def build_ai_image_object_key(
    *,
    account_id: int | str,
    landing_video_id: int | str,
    prompt_type: str,
    extension: str,
    debug: bool = False,
) -> str:
    object_prefix = os.getenv("AI_IMAGE_OSS_PREFIX", AI_IMAGE_OSS_PREFIX).strip("/")
    ext = extension if extension.startswith(".") else f".{extension}"
    if debug:
        scope = "debug"
    else:
        scope = f"account_{account_id}/video_{landing_video_id}"
    return f"{object_prefix}/{scope}/{prompt_type}_{uuid.uuid4().hex}{ext}"


def _insert_generated_material(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
    prompt: GenerationPrompt,
    model: str,
    object_key: str,
    oss_url: str,
    raw_response: dict,
) -> GeneratedMaterialAsset:
    ensure_ai_material_table()
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO ai_generated_material
                    (account_id, adgroup_id, crowd_package, landing_video_id,
                     landing_title, landing_category, prompt_type, prompt_text,
                     prompt_feature_json, model, oss_object_key, oss_url, raw_response)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    account_id,
                    adgroup_id,
                    crowd_package,
                    landing.video_id,
                    landing.title,
                    landing.category,
                    prompt.prompt_type,
                    prompt.prompt_text,
                    json.dumps(prompt.feature_hits, ensure_ascii=False),
                    model,
                    object_key,
                    oss_url,
                    json.dumps(raw_response, ensure_ascii=False, default=str)[:16000000],
                ),
            )
            asset_id = int(cur.lastrowid)
        conn.commit()
    finally:
        conn.close()
    return GeneratedMaterialAsset(
        id=asset_id,
        account_id=account_id,
        adgroup_id=adgroup_id,
        crowd_package=crowd_package,
        landing_video_id=landing.video_id,
        prompt_type=prompt.prompt_type,
        prompt_text=prompt.prompt_text,
        model=model,
        oss_url=oss_url,
        oss_object_key=object_key,
        feature_hits=prompt.feature_hits,
    )


def _rows_to_assets(rows: list[dict]) -> list[GeneratedMaterialAsset]:
    out = []
    for row in rows:
        try:
            feature_hits = json.loads(row.get("prompt_feature_json") or "[]")
        except Exception:
            feature_hits = []
        out.append(GeneratedMaterialAsset(
            id=int(row["id"]),
            account_id=int(row["account_id"]),
            adgroup_id=int(row["adgroup_id"]) if row.get("adgroup_id") else None,
            crowd_package=str(row.get("crowd_package") or ""),
            landing_video_id=int(row["landing_video_id"]),
            prompt_type=str(row.get("prompt_type") or ""),
            prompt_text=str(row.get("prompt_text") or ""),
            model=str(row.get("model") or ""),
            oss_url=str(row.get("oss_url") or ""),
            oss_object_key=str(row.get("oss_object_key") or ""),
            feature_hits=feature_hits,
        ))
    return out


def load_available_generated_assets(
    account_id: int,
    adgroup_id: int,
    landing_video_id: int,
) -> list[GeneratedMaterialAsset]:
    ensure_ai_material_table()
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT *
                FROM ai_generated_material
                WHERE account_id=%s
                  AND adgroup_id=%s
                  AND landing_video_id=%s
                  AND status='generated'
                  AND oss_url IS NOT NULL
                  AND oss_url <> ''
                ORDER BY id ASC
                """,
                (account_id, adgroup_id, landing_video_id),
            )
            rows = cur.fetchall() or []
    finally:
        conn.close()
    return _rows_to_assets(rows)


def generate_assets_for_landing(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
    model: str = OPENROUTER_IMAGE_MODEL,
) -> list[GeneratedMaterialAsset]:
    db_features = read_cached_video_element_features([landing.video_id]).get(landing.video_id) or []
    if not db_features:
        logger.info(
            "[ai_generated_material] landing=%d no cached DB features for generation",
            landing.video_id,
        )
        return []
    topic = _top_feature(db_features, dimension="解构选题")
    topic_text = (topic or {}).get("standard_element")
    if not topic_text:
        logger.info(
            "[ai_generated_material] landing=%d no topic feature for generation",
            landing.video_id,
        )
        return []
    sanitized_description = sanitize_video_description(topic_text)
    logger.info(
        "[ai_generated_material] landing=%d sanitized description=%r",
        landing.video_id, sanitized_description,
    )
    prompts = build_generation_prompts(
        video_id=landing.video_id,
        title=landing.title,
        category=landing.category,
        features=db_features,
        sanitized_description=sanitized_description,
    )
    assets: list[GeneratedMaterialAsset] = []
    for prompt in prompts:
        image_bytes, content_type, raw_response = generate_image_bytes(prompt.prompt_text, model)
        ext = mimetypes.guess_extension(content_type) or ".jpg"
        if ext == ".jpe":
            ext = ".jpg"
        object_key = build_ai_image_object_key(
            account_id=account_id,
            landing_video_id=landing.video_id,
            prompt_type=prompt.prompt_type,
            extension=ext,
        )
        oss_url = upload_image_to_oss(image_bytes, content_type, object_key)
        asset = _insert_generated_material(
            account_id=account_id,
            adgroup_id=adgroup_id,
            crowd_package=crowd_package,
            landing=landing,
            prompt=prompt,
            model=model,
            object_key=object_key,
            oss_url=oss_url,
            raw_response=raw_response,
        )
        assets.append(asset)
        logger.info(
            "[ai_generated_material] generated account=%d adgroup=%d landing=%d asset=%d type=%s url=%s",
            account_id, adgroup_id, landing.video_id, asset.id, prompt.prompt_type, oss_url,
        )
    return assets


def get_or_generate_assets_for_landing(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
) -> list[GeneratedMaterialAsset]:
    existing = load_available_generated_assets(account_id, adgroup_id, landing.video_id)
    if existing:
        return existing
    return generate_assets_for_landing(
        account_id=account_id,
        adgroup_id=adgroup_id,
        crowd_package=crowd_package,
        landing=landing,
    )


def update_generated_material_status(
    record: dict,
    status: str,
    *,
    dynamic_creative_id: int | str | None = None,
    tencent_image_id: str = "",
    error: str = "",
) -> None:
    asset_id = record.get("_ai_generated_material_id")
    if not asset_id:
        return
    ensure_ai_material_table()
    from db.connection import get_connection

    normalized_status = {
        "approve": "approved",
        "reject": "rejected",
    }.get(status, status)
    approval_status = {
        "approve": "approved",
        "reject": "rejected",
        "hold": "hold",
        "skip": "skip",
    }.get(status, normalized_status)
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                UPDATE ai_generated_material
                SET status=%s,
                    approval_status=%s,
                    dynamic_creative_id=COALESCE(%s, dynamic_creative_id),
                    tencent_image_id=COALESCE(NULLIF(%s, ''), tencent_image_id),
                    error=COALESCE(NULLIF(%s, ''), error),
                    updated_at=CURRENT_TIMESTAMP
                WHERE id=%s
                """,
                (
                    normalized_status,
                    approval_status,
                    int(dynamic_creative_id) if dynamic_creative_id else None,
                    tencent_image_id,
                    error[:2000] if error else "",
                    int(asset_id),
                ),
            )
        conn.commit()
    finally:
        conn.close()