Agent/examples/auto_put_ad_mini/tools/material_strategy_learning.py

"""素材策略学习数据持久化。

本模块保存高消耗素材快照与视觉标注，并刻意与广告/创意创建链路解耦；导入
学习数据绝不能创建或修改腾讯广告。
"""

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