Agent/examples/auto_put_ad_mini/tools/material_strategy_learning.py

"""Material strategy learning persistence.

This module stores high-consumption material snapshots and visual annotations.
It is intentionally separate from the ad/creative creation pipeline: importing
learning data must not create or modify Tencent ads.
"""

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:
    """Upsert initial DRAFT creative patterns learned from Top100 analysis."""
    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]:
    """Load selectable patterns. Production should keep 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:
    """Stable fallback score when the model selector is unavailable.

    This intentionally does not use keyword relevance matching. Pattern
    relevance is the model node's responsibility. Fallback only keeps the system
    available and auditable.
    """
    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])

    # Weak prior from historical material volume, not semantic relevance.
    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:
    # Keep the same precedence as 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]:
    """Use an LLM node to choose from pre-existing candidate patterns.

    The model can only rerank/select candidates produced from DB. It cannot
    invent pattern keys or bypass compliance/risk context.
    """
    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]:
    """Select creative patterns for one video.

    Production shape:
    1. Load patterns from DB.
    2. Send all available candidates to the model for semantic selection.
    3. If the model fails, use a non-semantic stable fallback.

    crowd_package is intentionally ignored: material style selection is shared
    across audience packages.
    """
    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:
    """Import the existing local high-consumption material analysis into DB."""
    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,
    )