"""素材策略学习数据持久化。 本模块保存高消耗素材快照与视觉标注，并刻意与广告/创意创建链路解耦；导入 学习数据绝不能创建或修改腾讯广告。 """ from __future__ import annotations import csv import hashlib import json import logging import os import re from dataclasses import dataclass from datetime import date from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pathlib import Path from typing import Any, Iterable, Sequence import httpx logger = logging.getLogger(__name__) CREATE_STRATEGY_LEARNING_TABLES_SQL = [ """ CREATE TABLE IF NOT EXISTS material_performance_snapshot_run ( id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键', run_id VARCHAR(64) NOT NULL COMMENT '快照任务ID', window_start DATE NOT NULL COMMENT '统计窗口开始日期', window_end DATE NOT NULL COMMENT '统计窗口结束日期', top_n INT NOT NULL DEFAULT 5000 COMMENT '拉取TopN', source VARCHAR(64) NOT NULL DEFAULT 'odps' COMMENT '数据来源', sql_file VARCHAR(255) DEFAULT NULL COMMENT 'SQL文件路径', row_count INT NOT NULL DEFAULT 0 COMMENT '明细行数', total_cost_fen BIGINT NOT NULL DEFAULT 0 COMMENT '窗口内总消耗(分)', status VARCHAR(32) NOT NULL DEFAULT 'SUCCESS' COMMENT '任务状态', error_message MEDIUMTEXT DEFAULT NULL COMMENT '错误信息', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_run_id (run_id), KEY idx_window (window_start, window_end) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='高消耗素材表现快照任务' """, """ CREATE TABLE IF NOT EXISTS material_performance_snapshot_item ( id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键', run_id VARCHAR(64) NOT NULL COMMENT '快照任务ID', rank_no INT NOT NULL COMMENT '消耗排名', account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID', ad_id BIGINT NOT NULL COMMENT '广告ID', creative_id BIGINT NOT NULL COMMENT '创意ID', creative_name VARCHAR(255) DEFAULT NULL COMMENT '创意名称', ad_name VARCHAR(255) DEFAULT NULL COMMENT '广告名称', video_id BIGINT DEFAULT NULL COMMENT '承接视频ID', title VARCHAR(512) DEFAULT NULL COMMENT '素材标题', image_url VARCHAR(1024) DEFAULT NULL COMMENT '素材图片URL', image_hash VARCHAR(64) DEFAULT NULL COMMENT '图片内容hash', crowd_package VARCHAR(255) DEFAULT NULL COMMENT '人群包名称', optimization_goal VARCHAR(128) DEFAULT NULL COMMENT '优化目标', bid_amount_fen BIGINT DEFAULT NULL COMMENT '出价(分)', day_amount_fen BIGINT DEFAULT NULL COMMENT '日预算(分)', cost_fen BIGINT NOT NULL DEFAULT 0 COMMENT '消耗(分)', impressions BIGINT NOT NULL DEFAULT 0 COMMENT '曝光', clicks BIGINT NOT NULL DEFAULT 0 COMMENT '点击', ctr DECIMAL(10, 6) DEFAULT NULL COMMENT '点击率', key_page_view_count BIGINT NOT NULL DEFAULT 0 COMMENT '关键页访问次数', key_page_rate DECIMAL(10, 6) DEFAULT NULL COMMENT '关键页访问/点击', conversions_count BIGINT NOT NULL DEFAULT 0 COMMENT '转化数', conversion_rate DECIMAL(10, 6) DEFAULT NULL COMMENT '转化率', active_days INT NOT NULL DEFAULT 0 COMMENT '有消耗天数', first_dt VARCHAR(16) DEFAULT NULL COMMENT '首次投放日期', last_dt VARCHAR(16) DEFAULT NULL COMMENT '最后投放日期', raw_json MEDIUMTEXT DEFAULT NULL COMMENT '原始行JSON', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_run_creative (run_id, creative_id), KEY idx_creative (creative_id), KEY idx_image_hash (image_hash), KEY idx_video (video_id), KEY idx_package_cost (crowd_package, cost_fen) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='高消耗素材表现快照明细' """, """ CREATE TABLE IF NOT EXISTS material_visual_annotation ( id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键', image_hash VARCHAR(64) NOT NULL COMMENT '图片内容hash', image_url VARCHAR(1024) NOT NULL COMMENT '素材图片URL', annotation_version VARCHAR(64) NOT NULL COMMENT '标注版本', annotator VARCHAR(64) NOT NULL COMMENT '标注来源', creative_id BIGINT DEFAULT NULL COMMENT '样本创意ID', visual_template VARCHAR(128) DEFAULT NULL COMMENT '视觉模板', hook_category VARCHAR(255) DEFAULT NULL COMMENT '标题钩子分类', title_text VARCHAR(512) DEFAULT NULL COMMENT '图片/素材标题', title_length INT DEFAULT NULL COMMENT '标题长度', scene_type VARCHAR(128) DEFAULT NULL COMMENT '场景类型', person_type VARCHAR(128) DEFAULT NULL COMMENT '人物/主体估计', has_human TINYINT DEFAULT NULL COMMENT '是否有人物', text_area_level VARCHAR(32) DEFAULT NULL COMMENT '文字面积估计等级', color_style VARCHAR(128) DEFAULT NULL COMMENT '颜色/调性', button_like_element TINYINT NOT NULL DEFAULT 0 COMMENT '是否疑似按钮诱导', fake_ui_risk TINYINT NOT NULL DEFAULT 0 COMMENT '假界面风险', official_policy_risk TINYINT NOT NULL DEFAULT 0 COMMENT '官方/政策承诺风险', medical_health_risk TINYINT NOT NULL DEFAULT 0 COMMENT '医疗健康风险', politics_sensitive_risk TINYINT NOT NULL DEFAULT 0 COMMENT '涉政/国家情绪风险', celebrity_or_history_risk TINYINT NOT NULL DEFAULT 0 COMMENT '名人/历史人物风险', strong_inducement_risk TINYINT NOT NULL DEFAULT 0 COMMENT '强诱导风险', greeting_blessing_risk TINYINT NOT NULL DEFAULT 0 COMMENT '早晚安/祝福风险', compliance_level VARCHAR(32) NOT NULL DEFAULT 'caution' COMMENT '生成可学习等级', learnable_points MEDIUMTEXT DEFAULT NULL COMMENT '可学习点', avoid_points MEDIUMTEXT DEFAULT NULL COMMENT '避让点', raw_annotation MEDIUMTEXT DEFAULT NULL COMMENT '原始标注JSON', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_image_version (image_hash, annotation_version), KEY idx_creative (creative_id), KEY idx_template (visual_template), KEY idx_compliance (compliance_level) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='素材图片结构化标注' """, """ CREATE TABLE IF NOT EXISTS material_creative_pattern ( id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键', pattern_version VARCHAR(64) NOT NULL COMMENT '策略版本', pattern_key VARCHAR(128) NOT NULL COMMENT '策略唯一键', pattern_name VARCHAR(128) NOT NULL COMMENT '策略名称', hook_category VARCHAR(128) NOT NULL COMMENT '标题钩子类型', visual_template VARCHAR(128) NOT NULL COMMENT '视觉模板', applicable_crowd_packages VARCHAR(1024) DEFAULT NULL COMMENT '适用人群包', applicable_placements VARCHAR(1024) DEFAULT NULL COMMENT '适用版位', target_age_min INT DEFAULT NULL COMMENT '适用年龄下限', target_age_max INT DEFAULT NULL COMMENT '适用年龄上限', title_hook_rule MEDIUMTEXT NOT NULL COMMENT '标题钩子规则', visual_rule MEDIUMTEXT NOT NULL COMMENT '视觉规则', relevance_rule MEDIUMTEXT NOT NULL COMMENT '视频相关性规则', compliance_rule MEDIUMTEXT NOT NULL COMMENT '合规规则', selector_config MEDIUMTEXT DEFAULT NULL COMMENT '选择器配置JSON:match/exclude/score', positive_examples MEDIUMTEXT DEFAULT NULL COMMENT '正例JSON', negative_examples MEDIUMTEXT DEFAULT NULL COMMENT '反例JSON', source_run_id VARCHAR(64) DEFAULT NULL COMMENT '来源快照ID', source_material_count INT NOT NULL DEFAULT 0 COMMENT '来源素材数', source_total_cost_fen BIGINT NOT NULL DEFAULT 0 COMMENT '来源素材消耗(分)', status VARCHAR(32) NOT NULL DEFAULT 'DRAFT' COMMENT 'DRAFT/APPROVED/REJECTED', reviewed_by VARCHAR(64) DEFAULT NULL COMMENT '审核人', reviewed_at TIMESTAMP NULL DEFAULT NULL COMMENT '审核时间', enabled TINYINT NOT NULL DEFAULT 0 COMMENT '是否启用', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_version_key (pattern_version, pattern_key), KEY idx_status_enabled (status, enabled), KEY idx_hook_template (hook_category, visual_template) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='AI生成素材可学习创意模式' """, """ CREATE TABLE IF NOT EXISTS material_strategy_learning_report ( id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键', report_id VARCHAR(64) NOT NULL COMMENT '报告ID', run_id VARCHAR(64) NOT NULL COMMENT '快照任务ID', report_version VARCHAR(64) NOT NULL COMMENT '报告版本', summary MEDIUMTEXT NOT NULL COMMENT '报告摘要', top_patterns MEDIUMTEXT DEFAULT NULL COMMENT '核心模式JSON', risk_summary MEDIUMTEXT DEFAULT NULL COMMENT '风险摘要JSON', recommended_actions MEDIUMTEXT DEFAULT NULL COMMENT '建议动作JSON', report_path VARCHAR(512) DEFAULT NULL COMMENT '本地报告路径', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_report_id (report_id), KEY idx_run_id (run_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='素材策略学习周报' """, ] STRATEGY_LEARNING_MIGRATIONS_SQL = [ """ ALTER TABLE material_performance_snapshot_item MODIFY COLUMN bid_amount_fen BIGINT DEFAULT NULL COMMENT '出价(分)' """, """ ALTER TABLE material_performance_snapshot_item MODIFY COLUMN day_amount_fen BIGINT DEFAULT NULL COMMENT '日预算(分)' """, ] @dataclass(frozen=True) class ImportResult: run_id: str snapshot_rows: int annotation_rows: int report_rows: int @dataclass(frozen=True) class CreativePattern: pattern_version: str pattern_key: str pattern_name: str hook_category: str visual_template: str title_hook_rule: str visual_rule: str relevance_rule: str compliance_rule: str selector_config: dict[str, Any] | None = None applicable_crowd_packages: str = "" applicable_placements: str = "" target_age_min: int | None = 60 target_age_max: int | None = 75 positive_examples: list[str] | None = None negative_examples: list[str] | None = None source_run_id: str | None = None source_material_count: int = 0 source_total_cost_fen: int = 0 status: str = "DRAFT" enabled: int = 0 @dataclass(frozen=True) class PatternSelection: pattern: CreativePattern score: float reasons: list[str] penalties: list[str] matched_features: list[str] DEFAULT_PATTERN_VERSION = "seed_20260708_v1" DEFAULT_SOURCE_RUN_ID = "material_30d_20260607_20260706_top5000" DEFAULT_PATTERN_SEED_PATH = ( Path(__file__).resolve().parents[1] / "configs" / "material_creative_patterns_seed.json" ) DEFAULT_PATTERN_SELECTOR_PROMPT_PATH = ( Path(__file__).resolve().parents[1] / "prompts" / "ai_pattern_selector.md" ) OPENROUTER_CHAT_COMPLETIONS_URL = os.getenv( "OPENROUTER_CHAT_COMPLETIONS_URL", "https://openrouter.ai/api/v1/chat/completions", ) OPENROUTER_TEXT_MODEL = os.getenv("OPENROUTER_TEXT_MODEL", "google/gemini-3-flash-preview") PATTERN_SELECTOR_MIN_MODEL_SCORE = float(os.getenv("PATTERN_SELECTOR_MIN_MODEL_SCORE", "75")) def ensure_strategy_learning_tables() -> None: from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: for sql in CREATE_STRATEGY_LEARNING_TABLES_SQL: cur.execute(sql) for sql in STRATEGY_LEARNING_MIGRATIONS_SQL: cur.execute(sql) cur.execute("SHOW COLUMNS FROM material_creative_pattern LIKE 'selector_config'") if not cur.fetchone(): cur.execute( """ ALTER TABLE material_creative_pattern ADD COLUMN selector_config MEDIUMTEXT DEFAULT NULL COMMENT '选择器配置JSON:match/exclude/score' AFTER compliance_rule """ ) conn.commit() finally: conn.close() def _pattern_from_row(row: dict[str, Any]) -> CreativePattern: def _loads_list(value: Any) -> list[str]: if not value: return [] try: parsed = json.loads(str(value)) except json.JSONDecodeError: return [] if isinstance(parsed, list): return [str(item) for item in parsed] return [] def _loads_dict(value: Any) -> dict[str, Any]: if not value: return {} if isinstance(value, dict): return value try: parsed = json.loads(str(value)) except json.JSONDecodeError: return {} return parsed if isinstance(parsed, dict) else {} return CreativePattern( pattern_version=str(row.get("pattern_version") or ""), pattern_key=str(row.get("pattern_key") or ""), pattern_name=str(row.get("pattern_name") or ""), hook_category=str(row.get("hook_category") or ""), visual_template=str(row.get("visual_template") or ""), applicable_crowd_packages=str(row.get("applicable_crowd_packages") or ""), applicable_placements=str(row.get("applicable_placements") or ""), target_age_min=_as_int(row.get("target_age_min")), target_age_max=_as_int(row.get("target_age_max")), title_hook_rule=str(row.get("title_hook_rule") or ""), visual_rule=str(row.get("visual_rule") or ""), relevance_rule=str(row.get("relevance_rule") or ""), compliance_rule=str(row.get("compliance_rule") or ""), selector_config=_loads_dict(row.get("selector_config")), positive_examples=_loads_list(row.get("positive_examples")), negative_examples=_loads_list(row.get("negative_examples")), source_run_id=str(row.get("source_run_id") or "") or None, source_material_count=_as_required_int(row.get("source_material_count")), source_total_cost_fen=_as_required_int(row.get("source_total_cost_fen")), status=str(row.get("status") or "DRAFT"), enabled=_as_required_int(row.get("enabled")), ) def _load_seed_patterns(path: Path = DEFAULT_PATTERN_SEED_PATH) -> list[CreativePattern]: raw_patterns = json.loads(path.read_text(encoding="utf-8")) patterns: list[CreativePattern] = [] for raw in raw_patterns: patterns.append(CreativePattern( pattern_version=str(raw.get("pattern_version") or DEFAULT_PATTERN_VERSION), pattern_key=str(raw["pattern_key"]), pattern_name=str(raw["pattern_name"]), hook_category=str(raw.get("hook_category") or ""), visual_template=str(raw.get("visual_template") or ""), applicable_crowd_packages=str(raw.get("applicable_crowd_packages") or ""), applicable_placements=str(raw.get("applicable_placements") or ""), target_age_min=_as_int(raw.get("target_age_min")) or 60, target_age_max=_as_int(raw.get("target_age_max")) or 75, title_hook_rule=str(raw.get("title_hook_rule") or ""), visual_rule=str(raw.get("visual_rule") or ""), relevance_rule=str(raw.get("relevance_rule") or ""), compliance_rule=str(raw.get("compliance_rule") or ""), selector_config=raw.get("selector_config") or {}, positive_examples=[str(v) for v in raw.get("positive_examples") or []], negative_examples=[str(v) for v in raw.get("negative_examples") or []], source_run_id=str(raw.get("source_run_id") or "") or None, source_material_count=_as_required_int(raw.get("source_material_count")), source_total_cost_fen=_as_required_int(raw.get("source_total_cost_fen")), status=str(raw.get("status") or "DRAFT"), enabled=_as_required_int(raw.get("enabled")), )) return patterns def seed_default_draft_patterns(seed_path: Path = DEFAULT_PATTERN_SEED_PATH) -> int: """更新或插入从 Top100 分析得到的初始 DRAFT 创意模式。""" ensure_strategy_learning_tables() from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.executemany( """ INSERT INTO material_creative_pattern ( pattern_version, pattern_key, pattern_name, hook_category, visual_template, applicable_crowd_packages, applicable_placements, target_age_min, target_age_max, title_hook_rule, visual_rule, relevance_rule, compliance_rule, selector_config, positive_examples, negative_examples, source_run_id, source_material_count, source_total_cost_fen, status, enabled ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s ) ON DUPLICATE KEY UPDATE pattern_name=VALUES(pattern_name), hook_category=VALUES(hook_category), visual_template=VALUES(visual_template), applicable_crowd_packages=VALUES(applicable_crowd_packages), applicable_placements=VALUES(applicable_placements), target_age_min=VALUES(target_age_min), target_age_max=VALUES(target_age_max), title_hook_rule=VALUES(title_hook_rule), visual_rule=VALUES(visual_rule), relevance_rule=VALUES(relevance_rule), compliance_rule=VALUES(compliance_rule), selector_config=VALUES(selector_config), positive_examples=VALUES(positive_examples), negative_examples=VALUES(negative_examples), source_run_id=VALUES(source_run_id), source_material_count=VALUES(source_material_count), source_total_cost_fen=VALUES(source_total_cost_fen), status=VALUES(status), enabled=VALUES(enabled), updated_at=CURRENT_TIMESTAMP """, [ ( p.pattern_version, p.pattern_key, p.pattern_name, p.hook_category, p.visual_template, p.applicable_crowd_packages or None, p.applicable_placements or None, p.target_age_min, p.target_age_max, p.title_hook_rule, p.visual_rule, p.relevance_rule, p.compliance_rule, _json_dumps(p.selector_config or {}), _json_dumps(p.positive_examples or []), _json_dumps(p.negative_examples or []), p.source_run_id, p.source_material_count, p.source_total_cost_fen, p.status, p.enabled, ) for p in _load_seed_patterns(seed_path) ], ) conn.commit() finally: conn.close() return len(_load_seed_patterns(seed_path)) def load_creative_patterns(include_draft: bool = False) -> list[CreativePattern]: """读取可选创意模式；生产环境应保持 include_draft=False。""" ensure_strategy_learning_tables() from db.connection import get_connection if include_draft: where_sql = "status IN ('DRAFT', 'APPROVED')" else: where_sql = "status = 'APPROVED' AND enabled = 1" conn = get_connection() try: with conn.cursor() as cur: cur.execute( f""" SELECT * FROM material_creative_pattern WHERE {where_sql} ORDER BY enabled DESC, source_total_cost_fen DESC, id ASC """ ) rows = cur.fetchall() or [] finally: conn.close() return [_pattern_from_row(row) for row in rows] def _feature_texts(video_features: Sequence[Any]) -> list[str]: texts: list[str] = [] for feature in video_features: element_dimension = str(getattr(feature, "element_dimension", "") or "") point_type = str(getattr(feature, "point_type", "") or "") standard_element = str(getattr(feature, "standard_element", "") or "") text = " ".join(part for part in [element_dimension, point_type, standard_element] if part) if text: texts.append(text) return texts def _contains_any(text: str, keywords: Sequence[str]) -> bool: return any(keyword and keyword in text for keyword in keywords) def _selection_context_bonus( pattern: CreativePattern, *, placement: str, ) -> tuple[float, list[str], list[str]]: score = 0.0 reasons: list[str] = [] penalties: list[str] = [] if pattern.status == "APPROVED" and pattern.enabled: score += 20 reasons.append("approved_enabled") elif pattern.status == "DRAFT": score += 5 reasons.append("draft_for_review") if pattern.applicable_placements: placements = [p.strip() for p in pattern.applicable_placements.split(",") if p.strip()] if placement and placement in placements: score += 6 reasons.append("placement_matched") else: score -= 6 penalties.append("placement_not_matched") return score, reasons, penalties def _fallback_score_pattern( pattern: CreativePattern, *, feature_texts: Sequence[str], placement: str, ) -> PatternSelection: """模型选择器不可用时采用稳定的兜底评分。 这里刻意不使用关键词相关性匹配；模式相关性由模型节点负责，兜底逻辑只保证 系统可用且可审计。 """ joined_features = " ".join(feature_texts) score = 10.0 context_score, reasons, penalties = _selection_context_bonus( pattern, placement=placement, ) score += context_score reasons = ["fallback_candidate", *reasons] matched_features: list[str] = list(feature_texts[:3]) # 仅使用历史素材量作为弱先验，不代表语义相关性。 if pattern.source_total_cost_fen: score += min(15.0, pattern.source_total_cost_fen / 10_000_000) reasons.append("historical_cost_prior") if pattern.source_material_count: score += min(5.0, pattern.source_material_count / 2) reasons.append("historical_sample_count_prior") if not feature_texts: score -= 15 penalties.append("no_video_features") deduped_matches = [] seen_matches: set[str] = set() for text in matched_features: if text not in seen_matches: seen_matches.add(text) deduped_matches.append(text) return PatternSelection( pattern=pattern, score=round(score, 2), reasons=reasons, penalties=penalties, matched_features=deduped_matches[:5], ) def _openrouter_api_key() -> str: # 保持与 tools/ai_generated_material.py 相同的密钥读取优先级。 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,无法用模型选择pattern") return key 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: match = re.search(r"\{.*\}", raw, flags=re.S) if not match: raise parsed = json.loads(match.group(0)) if not isinstance(parsed, dict): raise ValueError("模型返回不是JSON object") return parsed def _load_pattern_selector_policy(path: Path = DEFAULT_PATTERN_SELECTOR_PROMPT_PATH) -> str: return path.read_text(encoding="utf-8").strip() def _model_rerank_pattern_selections( *, feature_texts: Sequence[str], placement: str, candidates: Sequence[PatternSelection], top_k: int, model: str | None = None, ) -> list[PatternSelection]: """使用 LLM 节点从既有候选模式中选择。 模型只能重排或选择数据库产生的候选，不能编造模式键，也不能绕过合规和风险上下文。 """ if not candidates: return [] model_name = model or OPENROUTER_TEXT_MODEL candidate_payload = [ { "pattern_key": item.pattern.pattern_key, "pattern_name": item.pattern.pattern_name, "hook_category": item.pattern.hook_category, "visual_template": item.pattern.visual_template, "title_hook_rule": item.pattern.title_hook_rule, "visual_rule": item.pattern.visual_rule, "relevance_rule": item.pattern.relevance_rule, "compliance_rule": item.pattern.compliance_rule, "fallback_score": item.score, "fallback_reasons": item.reasons, "fallback_penalties": item.penalties, "positive_examples": item.pattern.positive_examples or [], "negative_examples": item.pattern.negative_examples or [], } for item in candidates ] prompt_payload = { "selection_policy": _load_pattern_selector_policy(), "video_features": list(feature_texts), "placement": placement, "top_k": top_k, "candidate_patterns": candidate_payload, "output_schema": { "selected": [ { "pattern_key": "候选pattern_key", "score": "0-100整数,表示模型选择置信度", "reason": "为什么这个pattern最匹配视频", "matched_features": ["命中的视频特征文本"], "risk_notes": ["需要注意的合规风险,没有则空数组"], } ], "no_match_example": {"selected": []}, }, } resp = httpx.post( OPENROUTER_CHAT_COMPLETIONS_URL, headers={ "Authorization": f"Bearer {_openrouter_api_key()}", "Content-Type": "application/json", }, json={ "model": model_name, "messages": [ { "role": "system", "content": "你是广告创意策略选择器,只做结构化JSON输出。", }, { "role": "user", "content": json.dumps(prompt_payload, ensure_ascii=False), }, ], "temperature": 0.2, }, timeout=45, ) resp.raise_for_status() data = resp.json() content = (((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "") parsed = _extract_json_object(content) selected = parsed.get("selected") or [] by_key = {item.pattern.pattern_key: item for item in candidates} out: list[PatternSelection] = [] seen: set[str] = set() for raw in selected: if not isinstance(raw, dict): continue key = str(raw.get("pattern_key") or "") if not key or key in seen or key not in by_key: continue seen.add(key) base = by_key[key] try: model_score = float(raw.get("score")) except (TypeError, ValueError): model_score = base.score if model_score < PATTERN_SELECTOR_MIN_MODEL_SCORE: continue reason = str(raw.get("reason") or "").strip() risk_notes = [str(v) for v in raw.get("risk_notes") or [] if str(v)] matched = [str(v) for v in raw.get("matched_features") or [] if str(v)] out.append(PatternSelection( pattern=base.pattern, score=round(model_score, 2), reasons=[*base.reasons, "model_selected", *(["model_reason:" + reason] if reason else [])], penalties=[*base.penalties, *["model_risk:" + note for note in risk_notes]], matched_features=matched or base.matched_features, )) if len(out) >= top_k: break return out def select_creative_patterns( *, video_features: Sequence[Any], crowd_package: str = "", placement: str = "", include_draft: bool = False, top_k: int = 3, use_model: bool = True, model: str | None = None, ) -> list[PatternSelection]: """为一个视频选择创意模式。 生产流程：先从数据库读取模式，再把全部可用候选交给模型做语义选择；模型 失败时使用不依赖语义的稳定兜底。此处刻意忽略 crowd_package，因为素材风格 选择在不同人群包之间共享。 """ patterns = load_creative_patterns(include_draft=include_draft) feature_texts = _feature_texts(video_features) placement = str(placement or "") selections = [ _fallback_score_pattern( pattern, feature_texts=feature_texts, placement=placement, ) for pattern in patterns ] selections.sort(key=lambda item: item.score, reverse=True) if use_model: try: return _model_rerank_pattern_selections( feature_texts=feature_texts, placement=placement, candidates=selections, top_k=max(1, int(top_k)), model=model, ) except Exception as e: logger.warning("[pattern_selector] model selection failed, fallback to stable score: %s", e) return selections[:max(1, int(top_k))] def _read_csv(path: Path) -> list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as f: return list(csv.DictReader(f)) def _read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def _json_dumps(data: Any) -> str: return json.dumps(data, ensure_ascii=False, separators=(",", ":")) def _as_int(value: Any) -> int | None: text = str(value or "").strip() if not text: return None try: return int(Decimal(text)) except (InvalidOperation, ValueError): return None def _as_required_int(value: Any, default: int = 0) -> int: parsed = _as_int(value) return default if parsed is None else parsed def _as_decimal(value: Any, places: str = "0.000001") -> Decimal | None: text = str(value or "").strip() if not text: return None try: return Decimal(text).quantize(Decimal(places), rounding=ROUND_HALF_UP) except (InvalidOperation, ValueError): return None def _yuan_to_fen(value: Any) -> int: text = str(value or "").strip() if not text: return 0 try: yuan = Decimal(text) except (InvalidOperation, ValueError): return 0 return int((yuan * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP)) def _bool_int(value: Any) -> int: return 1 if str(value or "").strip().lower() in {"1", "true", "yes", "y", "是"} else 0 def _risk_flags(row: dict[str, str]) -> set[str]: raw = row.get("risk_flags") or "" return {part.strip() for part in raw.split(",") if part.strip()} def _file_or_url_hash(image_path: str, image_url: str, base_dir: Path) -> str: path = Path(image_path) if image_path else Path() if image_path and not path.is_absolute(): path = base_dir / path if image_path and path.exists(): return hashlib.sha256(path.read_bytes()).hexdigest() return hashlib.sha256(str(image_url or "").encode("utf-8")).hexdigest() def _annotation_hash_map(annotation_rows: Iterable[dict[str, str]], base_dir: Path) -> dict[int, str]: out: dict[int, str] = {} for row in annotation_rows: creative_id = _as_int(row.get("creative_id")) if creative_id is None: continue out[creative_id] = _file_or_url_hash( row.get("image_path") or "", row.get("image_url") or "", base_dir, ) return out def import_current_material_analysis( *, run_id: str, window_start: date, window_end: date, performance_csv: Path, performance_summary_json: Path, visual_annotations_csv: Path, visual_summary_json: Path, report_path: Path, sql_file: str, top_n: int = 5000, annotation_version: str = "top100_rule_v1", annotator: str = "rule_contact_sheet_review", report_version: str = "top100_visual_v1", ) -> ImportResult: """将已有的本地高消耗素材分析导入数据库。""" ensure_strategy_learning_tables() performance_rows = _read_csv(performance_csv) performance_summary = _read_json(performance_summary_json) annotation_rows = _read_csv(visual_annotations_csv) visual_summary = _read_json(visual_summary_json) image_hash_by_creative = _annotation_hash_map(annotation_rows, Path.cwd()) from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ INSERT INTO material_performance_snapshot_run ( run_id, window_start, window_end, top_n, source, sql_file, row_count, total_cost_fen, status, error_message ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE window_start=VALUES(window_start), window_end=VALUES(window_end), top_n=VALUES(top_n), source=VALUES(source), sql_file=VALUES(sql_file), row_count=VALUES(row_count), total_cost_fen=VALUES(total_cost_fen), status=VALUES(status), error_message=VALUES(error_message), updated_at=CURRENT_TIMESTAMP """, ( run_id, window_start, window_end, top_n, "odps", sql_file, int(performance_summary.get("rows") or len(performance_rows)), _yuan_to_fen(performance_summary.get("total_cost_yuan")), "SUCCESS", None, ), ) snapshot_values = [] for idx, row in enumerate(performance_rows, start=1): creative_id = _as_required_int(row.get("creative_id")) rank_no = _as_int(row.get("rank")) or idx snapshot_values.append( ( run_id, rank_no, _as_required_int(row.get("account_id")), _as_required_int(row.get("ad_id")), creative_id, row.get("creative_name") or None, row.get("ad_name") or None, _as_int(row.get("video_id")), row.get("title") or None, row.get("image_url") or None, image_hash_by_creative.get(creative_id), row.get("package_name") or None, row.get("optimization_goal") or None, _as_int(row.get("bid_amount")), _as_int(row.get("day_amount")), _yuan_to_fen(row.get("cost_yuan")), _as_required_int(row.get("view_count")), _as_required_int(row.get("valid_click_count")), _as_decimal(row.get("ctr")), _as_required_int(row.get("key_page_view_count")), _as_decimal(row.get("key_page_rate")), _as_required_int(row.get("conversions_count")), _as_decimal(row.get("conversion_rate")), _as_required_int(row.get("active_days")), row.get("first_dt") or None, row.get("last_dt") or None, _json_dumps(row), ) ) cur.executemany( """ INSERT INTO material_performance_snapshot_item ( run_id, rank_no, account_id, ad_id, creative_id, creative_name, ad_name, video_id, title, image_url, image_hash, crowd_package, optimization_goal, bid_amount_fen, day_amount_fen, cost_fen, impressions, clicks, ctr, key_page_view_count, key_page_rate, conversions_count, conversion_rate, active_days, first_dt, last_dt, raw_json ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s ) ON DUPLICATE KEY UPDATE rank_no=VALUES(rank_no), account_id=VALUES(account_id), ad_id=VALUES(ad_id), creative_name=VALUES(creative_name), ad_name=VALUES(ad_name), video_id=VALUES(video_id), title=VALUES(title), image_url=VALUES(image_url), image_hash=VALUES(image_hash), crowd_package=VALUES(crowd_package), optimization_goal=VALUES(optimization_goal), bid_amount_fen=VALUES(bid_amount_fen), day_amount_fen=VALUES(day_amount_fen), cost_fen=VALUES(cost_fen), impressions=VALUES(impressions), clicks=VALUES(clicks), ctr=VALUES(ctr), key_page_view_count=VALUES(key_page_view_count), key_page_rate=VALUES(key_page_rate), conversions_count=VALUES(conversions_count), conversion_rate=VALUES(conversion_rate), active_days=VALUES(active_days), first_dt=VALUES(first_dt), last_dt=VALUES(last_dt), raw_json=VALUES(raw_json), updated_at=CURRENT_TIMESTAMP """, snapshot_values, ) annotation_values = [] for row in annotation_rows: flags = _risk_flags(row) image_hash = _file_or_url_hash( row.get("image_path") or "", row.get("image_url") or "", Path.cwd(), ) annotation_values.append( ( image_hash, row.get("image_url") or "", annotation_version, annotator, _as_int(row.get("creative_id")), row.get("visual_template") or None, row.get("hook_categories") or None, row.get("title") or None, len(row.get("title") or ""), row.get("scene_type") or None, row.get("main_subject_est") or None, _bool_int(row.get("has_human_est")), row.get("text_area_ratio_est") or None, row.get("dominant_tone") or None, _bool_int(row.get("button_like_element_est")), 1 if "fake_ui" in flags or "fake_button" in flags else 0, 1 if "policy_money_claim_risk" in flags else 0, 1 if "medical_health_risk" in flags else 0, 1 if "politics_country_sensitive" in flags else 0, 1 if "celebrity_or_history_person" in flags else 0, 1 if "strong_inducement" in flags or "button_like_inducement" in flags else 0, 1 if "greeting_blessing_filtered" in flags else 0, row.get("risk_level_for_generation") or "caution", row.get("learnable_points") or None, row.get("avoid_points") or None, _json_dumps(row), ) ) cur.executemany( """ INSERT INTO material_visual_annotation ( image_hash, image_url, annotation_version, annotator, creative_id, visual_template, hook_category, title_text, title_length, scene_type, person_type, has_human, text_area_level, color_style, button_like_element, fake_ui_risk, official_policy_risk, medical_health_risk, politics_sensitive_risk, celebrity_or_history_risk, strong_inducement_risk, greeting_blessing_risk, compliance_level, learnable_points, avoid_points, raw_annotation ) VALUES ( %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s ) ON DUPLICATE KEY UPDATE image_url=VALUES(image_url), annotator=VALUES(annotator), creative_id=VALUES(creative_id), visual_template=VALUES(visual_template), hook_category=VALUES(hook_category), title_text=VALUES(title_text), title_length=VALUES(title_length), scene_type=VALUES(scene_type), person_type=VALUES(person_type), has_human=VALUES(has_human), text_area_level=VALUES(text_area_level), color_style=VALUES(color_style), button_like_element=VALUES(button_like_element), fake_ui_risk=VALUES(fake_ui_risk), official_policy_risk=VALUES(official_policy_risk), medical_health_risk=VALUES(medical_health_risk), politics_sensitive_risk=VALUES(politics_sensitive_risk), celebrity_or_history_risk=VALUES(celebrity_or_history_risk), strong_inducement_risk=VALUES(strong_inducement_risk), greeting_blessing_risk=VALUES(greeting_blessing_risk), compliance_level=VALUES(compliance_level), learnable_points=VALUES(learnable_points), avoid_points=VALUES(avoid_points), raw_annotation=VALUES(raw_annotation), updated_at=CURRENT_TIMESTAMP """, annotation_values, ) report_id = f"{run_id}_{report_version}" summary_text = report_path.read_text(encoding="utf-8") if report_path.exists() else "" cur.execute( """ INSERT INTO material_strategy_learning_report ( report_id, run_id, report_version, summary, top_patterns, risk_summary, recommended_actions, report_path ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE run_id=VALUES(run_id), report_version=VALUES(report_version), summary=VALUES(summary), top_patterns=VALUES(top_patterns), risk_summary=VALUES(risk_summary), recommended_actions=VALUES(recommended_actions), report_path=VALUES(report_path), updated_at=CURRENT_TIMESTAMP """, ( report_id, run_id, report_version, summary_text, _json_dumps(visual_summary.get("visual_template_counts") or {}), _json_dumps( { "risk_level_counts": visual_summary.get("risk_level_counts") or {}, "risk_flag_counts": visual_summary.get("risk_flag_counts") or [], } ), _json_dumps( [ "学习大字信息差结构,不要复制历史标题", "避免伪按钮、假界面、强诱导、涉政和医疗恐吓", "生成素材仍以承接视频ODPS内容特征为主", ] ), str(report_path), ), ) conn.commit() finally: conn.close() return ImportResult( run_id=run_id, snapshot_rows=len(performance_rows), annotation_rows=len(annotation_rows), report_rows=1, )