"""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 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: """Rewrite ODPS topic into an ad-theme seed 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 清洗后描述为空") 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: """Generate the cover title before image generation. The image model is weak at deciding ad copy and then rendering it. This node makes the copy explicit while keeping final compliance to later review. """ 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]: """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 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]: """Build generation prompts using model-selected creative patterns.""" 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: """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 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]: """Persist one generated asset and run the same AI review used by production.""" 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()