Agent/examples/auto_put_ad_mini/tools/ai_generated_material.py

"""模块 B 的 AI 生成创意图片资产。

本模块与历史素材召回链路保持隔离；只有账户明确配置
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 Any, Iterable, Optional, Sequence
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.1-flash-image",
)
OPENROUTER_TEXT_MODEL = os.getenv(
    "OPENROUTER_TEXT_MODEL",
    "google/gemini-3-flash-preview",
)
AI_MATERIAL_REVIEW_MODEL = os.getenv("AI_MATERIAL_REVIEW_MODEL", "google/gemini-3-flash-preview")
AI_IMAGE_USE_PATTERN_SELECTOR = os.getenv("AI_IMAGE_USE_PATTERN_SELECTOR", "1").strip().lower() not in (
    "0", "false", "no", "否",
)
AI_IMAGE_PATTERN_TOP_K = int(os.getenv("AI_IMAGE_PATTERN_TOP_K", "1"))
AI_IMAGE_PATTERN_PLACEMENT = os.getenv("AI_IMAGE_PATTERN_PLACEMENT", "WECHAT_OFFICIAL_ACCOUNTS")
AI_COVER_COPY_REQUIRED = os.getenv("AI_COVER_COPY_REQUIRED", "1").strip().lower() not in (
    "0", "false", "no", "否",
)
AI_COVER_COPY_MAX_TOKENS = int(os.getenv("AI_COVER_COPY_MAX_TOKENS", "900"))
AI_COVER_COPY_TITLE_MAX_LEN = int(os.getenv("AI_COVER_COPY_TITLE_MAX_LEN", "22"))
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_COVER_COPY_PROMPT_TEMPLATE_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "ai_cover_copy.md"
)
DEFAULT_AI_IMAGE_OSS_BUCKET = "art-pubbucket"
DEFAULT_AI_IMAGE_PUBLIC_BASE_URL = "https://rescdn.yishihui.com"


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',
    ai_review_status VARCHAR(50) DEFAULT NULL COMMENT 'AI审核状态:pass/reject/hold/error',
    ai_review_score INT DEFAULT NULL COMMENT 'AI审核评分0-100',
    ai_review_model VARCHAR(200) DEFAULT NULL COMMENT 'AI审核模型',
    ai_review_reason TEXT DEFAULT NULL COMMENT 'AI审核原因摘要',
    ai_review_json MEDIUMTEXT DEFAULT NULL COMMENT 'AI审核原始JSON',
    ai_reviewed_at TIMESTAMP NULL DEFAULT NULL COMMENT 'AI审核时间',
    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生成创意图片素材'
"""

AI_MATERIAL_REVIEW_COLUMNS_SQL = {
    "ai_review_status": "ALTER TABLE ai_generated_material ADD COLUMN ai_review_status VARCHAR(50) DEFAULT NULL COMMENT 'AI审核状态:pass/reject/hold/error' AFTER dynamic_creative_id",
    "ai_review_score": "ALTER TABLE ai_generated_material ADD COLUMN ai_review_score INT DEFAULT NULL COMMENT 'AI审核评分0-100' AFTER ai_review_status",
    "ai_review_model": "ALTER TABLE ai_generated_material ADD COLUMN ai_review_model VARCHAR(200) DEFAULT NULL COMMENT 'AI审核模型' AFTER ai_review_score",
    "ai_review_reason": "ALTER TABLE ai_generated_material ADD COLUMN ai_review_reason TEXT DEFAULT NULL COMMENT 'AI审核原因摘要' AFTER ai_review_model",
    "ai_review_json": "ALTER TABLE ai_generated_material ADD COLUMN ai_review_json MEDIUMTEXT DEFAULT NULL COMMENT 'AI审核原始JSON' AFTER ai_review_reason",
    "ai_reviewed_at": "ALTER TABLE ai_generated_material ADD COLUMN ai_reviewed_at TIMESTAMP NULL DEFAULT NULL COMMENT 'AI审核时间' AFTER ai_review_json",
}


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


@dataclass(frozen=True)
class CoverCopy:
    main_title: str
    hook_angle: str = ""
    hook_point: str = ""
    highlight_terms: list[str] | None = None
    reason: str = ""
    candidates: list[dict[str, Any]] | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "main_title": self.main_title,
            "hook_angle": self.hook_angle,
            "hook_point": self.hook_point,
            "highlight_terms": self.highlight_terms or [],
            "reason": self.reason,
            "candidates": self.candidates or [],
        }


@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 _load_cover_copy_prompt_messages(video_description: str, selection) -> list[dict]:
    raw_path = os.getenv("AI_COVER_COPY_PROMPT_TEMPLATE_PATH", "").strip()
    path = Path(raw_path).expanduser() if raw_path else DEFAULT_AI_COVER_COPY_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}")
    pattern = selection.pattern
    pattern_json = {
        "pattern_key": pattern.pattern_key,
        "pattern_name": pattern.pattern_name,
        "hook_category": pattern.hook_category,
        "visual_template": pattern.visual_template,
        "title_hook_rule": pattern.title_hook_rule,
        "visual_rule": pattern.visual_rule,
        "relevance_rule": pattern.relevance_rule,
        "positive_examples": pattern.positive_examples or [],
        "negative_examples": pattern.negative_examples or [],
    }
    system_part, user_part = template.split("【user】", 1)
    system_text = system_part.replace("【system】", "", 1).strip()
    user_text = (
        user_part
        .replace("{{video_description}}", video_description)
        .replace("{{pattern_json}}", json.dumps(pattern_json, ensure_ascii=False, indent=2))
        .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 _extract_json_object(text: str) -> dict[str, Any]:
    raw = str(text or "").strip()
    if raw.startswith("```"):
        raw = re.sub(r"^```(?:json)?", "", raw).strip()
        raw = re.sub(r"```$", "", raw).strip()
    try:
        parsed = json.loads(raw)
    except json.JSONDecodeError:
        start = raw.find("{")
        if start >= 0:
            try:
                parsed, _ = json.JSONDecoder().raw_decode(raw[start:])
            except json.JSONDecodeError:
                parsed = None
            if isinstance(parsed, dict):
                return parsed
        match = re.search(r"\{.*?\}", raw, flags=re.S)
        if not match:
            raise
        parsed = json.loads(match.group(0))
    if isinstance(parsed, list):
        parsed = next((item for item in parsed if isinstance(item, dict)), None)
    if isinstance(parsed, dict) and "selected" in parsed and isinstance(parsed["selected"], list):
        selected = next((item for item in parsed["selected"] if isinstance(item, dict)), None)
        if selected:
            parsed = selected
    if not isinstance(parsed, dict):
        raise ValueError("模型返回不是JSON object")
    return parsed


def sanitize_video_description(raw_description: str, model: str = OPENROUTER_TEXT_MODEL) -> str:
    """将 ODPS 主题改写为图片生成使用的广告主题种子。"""
    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 清洗后描述为空")
    return cleaned



def _clean_cover_copy_text(value: Any, *, max_len: int) -> str:
    text = str(value or "").strip()
    text = text.strip("`").strip().strip("“”\"'‘’")
    text = re.sub(r"[\r\n\t]+", "", text)
    text = re.sub(r"[，,。.!！?？:：;；、]+", "", text)
    text = re.sub(r"\s+", "", text)
    return text[:max_len]


def _normalize_cover_copy_candidates(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    candidates: list[dict[str, Any]] = []
    for item in value:
        if not isinstance(item, dict):
            continue
        title = _clean_cover_copy_text(item.get("title"), max_len=64)
        if not title:
            continue
        if len(title) > AI_COVER_COPY_TITLE_MAX_LEN:
            continue
        normalized = dict(item)
        normalized["title"] = title
        for key in ("hook_score", "plain_score", "relevance_score"):
            try:
                normalized[key] = int(float(normalized.get(key) or 0))
            except (TypeError, ValueError):
                normalized[key] = 0
        candidates.append(normalized)
    return candidates


def _best_cover_copy_candidate(candidates: list[dict[str, Any]]) -> dict[str, Any] | None:
    if not candidates:
        return None
    return sorted(
        candidates,
        key=lambda item: (
            int(item.get("hook_score") or 0),
            int(item.get("plain_score") or 0),
            int(item.get("relevance_score") or 0),
        ),
        reverse=True,
    )[0]


def _normalize_highlight_terms(value: Any, *, title: str) -> list[str]:
    if not isinstance(value, list):
        return []
    out: list[str] = []
    for item in value:
        term = _clean_cover_copy_text(item, max_len=8)
        if not term:
            continue
        if term not in title:
            continue
        if term in out:
            continue
        out.append(term)
        if len(out) >= 2:
            break
    return out


def generate_cover_copy(video_description: str, selection, model: str = OPENROUTER_TEXT_MODEL) -> CoverCopy:
    """在生成图片前先生成封面标题。

    图片模型不擅长同时决定广告文案并准确渲染，因此此节点先明确文案，最终合规
    仍交由后续审核判断。
    """
    description = (video_description or "").strip()
    if not description:
        raise RuntimeError("缺少广告主题种子,无法生成封面标题")
    body = {
        "model": model,
        "temperature": 0.5,
        "max_tokens": AI_COVER_COPY_MAX_TOKENS,
        "response_format": {"type": "json_object"},
        "messages": _load_cover_copy_prompt_messages(description, selection),
    }
    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()
    parsed = _extract_json_object(_extract_chat_completion_text(resp.json()))
    candidates = _normalize_cover_copy_candidates(parsed.get("candidates"))
    best_candidate = _best_cover_copy_candidate(candidates)
    main_title = _clean_cover_copy_text(
        parsed.get("selected_title") or parsed.get("main_title"),
        max_len=64,
    )
    if len(main_title) > AI_COVER_COPY_TITLE_MAX_LEN:
        main_title = ""
    if not main_title and best_candidate:
        main_title = str(best_candidate.get("title") or "")
    if not main_title:
        raise RuntimeError(
            f"OpenRouter 标题节点未返回 {AI_COVER_COPY_TITLE_MAX_LEN} 字以内的 main_title"
        )
    highlight_terms = _normalize_highlight_terms(
        parsed.get("highlight_terms"),
        title=main_title,
    )
    return CoverCopy(
        main_title=main_title,
        hook_angle=str(parsed.get("hook_angle") or (best_candidate or {}).get("hook_angle") or "").strip()[:50],
        hook_point=str(parsed.get("hook_point") or "").strip()[:160],
        highlight_terms=highlight_terms,
        reason=str(parsed.get("reason") or "").strip()[:200],
        candidates=candidates,
    )


def _render_pattern_prompt(video_description: str, selection, cover_copy: CoverCopy | None = None) -> str:
    pattern = selection.pattern
    positive_examples = "、".join(pattern.positive_examples or []) or "无"
    negative_examples = "、".join(pattern.negative_examples or []) or "无"
    cover_copy_text = ""
    if cover_copy:
        cover_copy_text = f"""
【封面标题】
- 主标题:{cover_copy.main_title}
- 标题钩子:{cover_copy.hook_angle or "未标注"}
- 高亮词:{",".join(cover_copy.highlight_terms or []) or "无"}

图片中必须使用上面的主标题。主标题不能添加任何括号、引号、书名号或额外符号。只允许高亮【高亮词】中列出的词;如果高亮词为“无”,不要强行彩色高亮。不要生成副标题,也不要自行新增其他标题、角标或引导文字。
"""
    enriched_description = f"""【广告主题种子】
{video_description}
{cover_copy_text}

【选中的创意Pattern】
- pattern_key:{pattern.pattern_key}
- pattern_name:{pattern.pattern_name}
- hook_category:{pattern.hook_category}
- visual_template:{pattern.visual_template}
- title_hook_rule:{pattern.title_hook_rule}
- visual_direction:{pattern.visual_rule}
- relevance_rule:{pattern.relevance_rule}
- compliance_rule:{pattern.compliance_rule}
- positive_examples:{positive_examples}
- negative_examples:{negative_examples}

visual_direction 只是可选视觉方向参考,不是硬约束;如果它和画面主体多样性、非默认人物、非默认老人正脸冲突,以全局视觉策略为准。
请基于【广告主题种子】和【选中的创意Pattern】生成广告封面。视频内容只提供主题方向,画面可以做适合信息流点击的合理转译。"""
    return _render_prompt(enriched_description)


def build_generation_prompts(
    *,
    video_id: int,
    title: str,
    category: str,
    features: Iterable[VideoElementFeature],
    sanitized_description: str = "",
) -> list[GenerationPrompt]:
    """根据视频的 ODPS 主题构建一个生成提示词。"""
    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 build_pattern_generation_prompts(
    *,
    video_id: int,
    title: str,
    category: str,
    features: Iterable[VideoElementFeature],
    sanitized_description: str,
    crowd_package: str = "",
    placement: str = AI_IMAGE_PATTERN_PLACEMENT,
    top_k: int = AI_IMAGE_PATTERN_TOP_K,
    text_model: str = OPENROUTER_TEXT_MODEL,
) -> list[GenerationPrompt]:
    """使用模型选出的创意模式构建生成提示词。"""
    from tools.material_strategy_learning import select_creative_patterns

    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
    selections = select_creative_patterns(
        video_features=feature_list,
        crowd_package=crowd_package,
        placement=placement,
        include_draft=True,
        top_k=top_k,
        use_model=True,
        model=text_model,
    )
    if not selections:
        logger.info(
            "[ai_generated_material] video=%d no matched pattern, fallback to topic prompt",
            video_id,
        )
        return build_generation_prompts(
            video_id=video_id,
            title=title,
            category=category,
            features=feature_list,
            sanitized_description=video_description,
        )
    prompts: list[GenerationPrompt] = []
    for selection in selections[:max(1, int(top_k))]:
        cover_copy = None
        try:
            cover_copy = generate_cover_copy(video_description, selection, model=text_model)
        except Exception as e:
            message = (
                f"[ai_generated_material] video={video_id} pattern={selection.pattern.pattern_key} "
                f"cover copy generation failed:{e}"
            )
            if AI_COVER_COPY_REQUIRED:
                raise RuntimeError(message) from e
            logger.warning("%s, fallback to image prompt", message)
        topic_hit = dict(topic)
        topic_hit["original_standard_element"] = topic_text
        topic_hit["video_description"] = video_description
        topic_hit["pattern_selection"] = {
            "pattern_key": selection.pattern.pattern_key,
            "pattern_name": selection.pattern.pattern_name,
            "score": selection.score,
            "reasons": selection.reasons,
            "penalties": selection.penalties,
            "matched_features": selection.matched_features,
            "hook_category": selection.pattern.hook_category,
            "visual_template": selection.pattern.visual_template,
            "title_hook_rule": selection.pattern.title_hook_rule,
            "visual_rule": selection.pattern.visual_rule,
            "relevance_rule": selection.pattern.relevance_rule,
            "compliance_rule": selection.pattern.compliance_rule,
            "positive_examples": selection.pattern.positive_examples or [],
            "negative_examples": selection.pattern.negative_examples or [],
        }
        if cover_copy:
            topic_hit["cover_copy"] = cover_copy.to_dict()
        prompts.append(
            GenerationPrompt(
                prompt_type=selection.pattern.pattern_key,
                prompt_text=_render_pattern_prompt(video_description, selection, cover_copy),
                feature_hits=[topic_hit],
            )
        )
    return prompts


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)
            for column_name, alter_sql in AI_MATERIAL_REVIEW_COLUMNS_SQL.items():
                cur.execute("SHOW COLUMNS FROM ai_generated_material LIKE %s", (column_name,))
                if not cur.fetchone():
                    cur.execute(alter_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:
    """将模型输出标准化为 RGB JPEG，供腾讯素材上传。"""
    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,
    *,
    input_image_url: str = "",
) -> tuple[bytes, str, dict]:
    """通过 OpenRouter Images API 生成一张图片。

    Images API 支持直接传 aspect_ratio；但并非所有模型都支持 output_format，
    因此代码仍会把返回的位图标准化为 JPEG。
    """
    body = {
        "model": model,
        "prompt": prompt,
        "aspect_ratio": AI_IMAGE_ASPECT_RATIO,
        "resolution": AI_IMAGE_RESOLUTION,
        "n": 1,
    }
    if input_image_url:
        body["input_references"] = [
            {
                "type": "image_url",
                "image_url": {"url": input_image_url},
            }
        ]
    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 insert_and_review_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,
) -> tuple[GeneratedMaterialAsset, Any]:
    """持久化一个生成资产，并执行与生产一致的 AI 审核。"""
    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,
    )
    from tools.ai_material_review import (
        MaterialReviewResult,
        review_generated_material,
        update_material_review_result,
    )

    try:
        review = review_generated_material(
            image_url=oss_url,
            prompt_type=prompt.prompt_type,
            prompt_text=prompt.prompt_text,
            feature_hits=prompt.feature_hits,
        )
    except Exception as e:
        logger.exception(
            "[ai_generated_material] AI审核异常 account=%d adgroup=%d landing=%d asset=%d: %s",
            account_id, adgroup_id, landing.video_id, asset.id, e,
        )
        review = MaterialReviewResult(
            status="error",
            score=0,
            reason=f"AI审核异常:{e}",
            risk_tags=["review_error"],
            ocr_text="",
            raw={"error": str(e)},
        )
    update_material_review_result(asset.id, review)
    return asset, review


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 IN ('generated', 'prepared')
                  AND ai_review_status='pass'
                  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,
    use_pattern_selector: bool = AI_IMAGE_USE_PATTERN_SELECTOR,
    placement: str = AI_IMAGE_PATTERN_PLACEMENT,
    pattern_top_k: int = AI_IMAGE_PATTERN_TOP_K,
    text_model: str = OPENROUTER_TEXT_MODEL,
    skip_prompt_types: set[str] | None = None,
    max_new_assets: int | None = None,
) -> 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,
    )
    if use_pattern_selector:
        prompts = build_pattern_generation_prompts(
            video_id=landing.video_id,
            title=landing.title,
            category=landing.category,
            features=db_features,
            sanitized_description=sanitized_description,
            crowd_package=crowd_package,
            placement=placement,
            top_k=pattern_top_k,
            text_model=text_model,
        )
        logger.info(
            "[ai_generated_material] landing=%d pattern prompts=%s",
            landing.video_id, [p.prompt_type for p in prompts],
        )
    else:
        prompts = build_generation_prompts(
            video_id=landing.video_id,
            title=landing.title,
            category=landing.category,
            features=db_features,
            sanitized_description=sanitized_description,
        )
    if skip_prompt_types:
        prompts = [p for p in prompts if p.prompt_type not in skip_prompt_types]
    if max_new_assets is not None:
        prompts = prompts[:max(0, int(max_new_assets))]
    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, review = insert_and_review_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,
        )
        if not review.passed:
            logger.warning(
                "[ai_generated_material] AI审核未通过 account=%d adgroup=%d landing=%d asset=%d status=%s score=%d reason=%s",
                account_id, adgroup_id, landing.video_id, asset.id, review.status, review.score, review.reason,
            )
            continue
        assets.append(asset)
        logger.info(
            "[ai_generated_material] generated reviewed account=%d adgroup=%d landing=%d asset=%d type=%s score=%d url=%s",
            account_id, adgroup_id, landing.video_id, asset.id, prompt.prompt_type, review.score, oss_url,
        )
    return assets


def get_or_generate_assets_for_landing(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
    use_pattern_selector: bool = AI_IMAGE_USE_PATTERN_SELECTOR,
    placement: str = AI_IMAGE_PATTERN_PLACEMENT,
    pattern_top_k: int = AI_IMAGE_PATTERN_TOP_K,
    text_model: str = OPENROUTER_TEXT_MODEL,
) -> list[GeneratedMaterialAsset]:
    existing = load_available_generated_assets(account_id, adgroup_id, landing.video_id)
    target_count = max(1, int(pattern_top_k) if use_pattern_selector else 1)
    if len(existing) >= target_count:
        return existing[:target_count]
    skip_prompt_types = {asset.prompt_type for asset in existing}
    generated = generate_assets_for_landing(
        account_id=account_id,
        adgroup_id=adgroup_id,
        crowd_package=crowd_package,
        landing=landing,
        use_pattern_selector=use_pattern_selector,
        placement=placement,
        pattern_top_k=pattern_top_k,
        text_model=text_model,
        skip_prompt_types=skip_prompt_types,
        max_new_assets=target_count - len(existing),
    )
    return [*existing, *generated]


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()