Agent/examples/auto_put_ad_mini/tools/external_recalled_material.py

"""模块 B 的外部召回素材处理。

仅对配置 ``material_source=external_recall`` 的账户启用此链路；历史召回与
既有文生图流程保持不变。
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Iterable

from tools.ai_generated_material import (
    OPENROUTER_IMAGE_MODEL,
    build_ai_image_object_key,
    generate_image_bytes,
    upload_image_to_oss,
)
from tools.ai_material_review import MaterialReviewResult, review_generated_material
from tools.material_recall import Material, recall_materials_for_video
from tools.video_recall import LandingVideo

logger = logging.getLogger(__name__)

EXTERNAL_RECALL_SOURCE_LABEL = os.getenv(
    "EXTERNAL_RECALL_SOURCE_LABEL", "外部合作"
).strip()
EXTERNAL_RECALL_CANDIDATE_LIMIT = int(
    os.getenv("EXTERNAL_RECALL_CANDIDATE_LIMIT", "300")
)
EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING = int(
    os.getenv("EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING", "3")
)
EXTERNAL_RECALL_UV_WINDOW_DAYS = int(
    os.getenv("EXTERNAL_RECALL_UV_WINDOW_DAYS", "90")
)
EXTERNAL_RECALL_MIN_UV = int(
    os.getenv("EXTERNAL_RECALL_MIN_UV", "500")
)
EXTERNAL_IMAGE_MODEL = os.getenv(
    "EXTERNAL_IMAGE_MODEL", OPENROUTER_IMAGE_MODEL
).strip()
EXTERNAL_CLEANUP_PROMPT_PATH = (
    Path(__file__).resolve().parents[1] / "prompts" / "external_material_cleanup.md"
)


CREATE_EXTERNAL_MATERIAL_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS external_recalled_material (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
    adgroup_id BIGINT NOT NULL COMMENT '广告ID',
    crowd_package VARCHAR(200) NOT NULL COMMENT '投放人群包名称',
    landing_video_id BIGINT NOT NULL COMMENT '承接视频ID',
    source_material_id VARCHAR(255) NOT NULL COMMENT '外部召回materialId/card_cover_id',
    source_image_url VARCHAR(1000) NOT NULL COMMENT '外部素材原图URL',
    source_title VARCHAR(1000) DEFAULT NULL COMMENT '外部素材标题',
    similarity_score DOUBLE NOT NULL DEFAULT 0 COMMENT '向量召回相似度',
    visit_uv_30d BIGINT NOT NULL DEFAULT 0 COMMENT '配置窗口内访问UV之和(兼容历史字段名)',
    uv_window_start VARCHAR(8) DEFAULT NULL COMMENT 'UV窗口开始分区',
    uv_window_end VARCHAR(8) DEFAULT NULL COMMENT 'UV窗口结束分区',
    recall_json MEDIUMTEXT DEFAULT NULL COMMENT '召回命中与原始素材JSON',
    edit_prompt MEDIUMTEXT NOT NULL COMMENT '去播放按钮图生图prompt',
    model VARCHAR(200) NOT NULL COMMENT '图生图模型',
    oss_object_key VARCHAR(500) DEFAULT NULL COMMENT '派生图OSS key',
    oss_url VARCHAR(1000) DEFAULT NULL COMMENT '派生图公网URL',
    status VARCHAR(50) NOT NULL DEFAULT 'generated' COMMENT 'generated/prepared/approved/rejected/hold/skip/posted_ok/post_failed/error',
    approval_status VARCHAR(50) DEFAULT NULL COMMENT '人工审批状态',
    tencent_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
    dynamic_creative_id BIGINT DEFAULT NULL COMMENT '腾讯动态创意ID',
    ai_review_status VARCHAR(50) DEFAULT NULL COMMENT 'AI审核状态',
    ai_review_score INT DEFAULT NULL COMMENT 'AI审核评分',
    ai_review_reason TEXT DEFAULT NULL COMMENT 'AI审核原因',
    ai_review_json MEDIUMTEXT DEFAULT NULL COMMENT 'AI审核JSON',
    error TEXT DEFAULT NULL COMMENT '处理错误',
    raw_response MEDIUMTEXT DEFAULT NULL COMMENT '图生图模型响应摘要',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    UNIQUE KEY uk_account_ad_landing_source
        (account_id, adgroup_id, landing_video_id, source_material_id),
    KEY idx_account_ad_status (account_id, adgroup_id, status),
    KEY idx_source_created (source_material_id, created_at),
    KEY idx_status_created (status, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='外部召回素材AI清理派生图'
"""


@dataclass(frozen=True)
class ExternalMaterialAsset:
    id: int
    source_material_id: str
    source_image_url: str
    source_title: str
    similarity_score: float
    visit_uv_30d: int
    uv_window_start: str
    uv_window_end: str
    oss_url: str
    recall_json: dict[str, Any]

    def to_material(self) -> Material:
        recall = self.recall_json
        return Material(
            material_id=f"external:{self.source_material_id}",
            score=self.similarity_score,
            title=self.source_title,
            cover=self.oss_url,
            cost=None,
            ctr=None,
            cvr=None,
            roi=None,
            impressions=None,
            quality_score=None,
            visit_uv_30d=self.visit_uv_30d,
            recall_strategy=str(recall.get("recall_strategy") or ""),
            recall_query_text=str(recall.get("recall_query_text") or ""),
            recall_config_code=str(recall.get("recall_config_code") or ""),
            recall_element_dimension=str(
                recall.get("recall_element_dimension") or ""
            ),
            recall_point_type=str(recall.get("recall_point_type") or ""),
            recall_standard_element=str(
                recall.get("recall_standard_element") or ""
            ),
            recall_hit_queries=list(recall.get("recall_hit_queries") or []),
            raw={
                "external_recalled_material_id": self.id,
                "source_material_id": self.source_material_id,
                "source_image_url": self.source_image_url,
                "visit_uv_30d": self.visit_uv_30d,
                "uv_window_start": self.uv_window_start,
                "uv_window_end": self.uv_window_end,
                "oss_url": self.oss_url,
            },
        )


def ensure_external_material_table() -> None:
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(CREATE_EXTERNAL_MATERIAL_TABLE_SQL)
        conn.commit()
    finally:
        conn.close()


def _source_image_url(material: Material) -> str:
    if material.cover:
        return material.cover
    images = (material.raw or {}).get("imageList") or []
    return str(images[0]) if images else ""


def _quote_sql(value: str) -> str:
    return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"


def _get_odps_client():
    from tools.odps_module import get_odps_client

    client = get_odps_client(project="loghubods")
    if client is None:
        raise RuntimeError("无法初始化ODPS客户端")
    return client


def _latest_partition_dt(client) -> str:
    table = client._odps.get_table("cooperate_top_cards_daily_v1")
    partition = table.get_max_partition()
    if partition is None:
        raise RuntimeError("cooperate_top_cards_daily_v1没有可用分区")
    name = getattr(partition, "name", "") or str(partition)
    for part in name.split(","):
        if part.strip().startswith("dt="):
            return part.split("=", 1)[1].strip(" '\"")
    raise RuntimeError(f"无法解析cooperate_top_cards_daily_v1最新分区:{name}")


def _query_visit_uv_30d(
    material_ids: Iterable[str],
) -> tuple[dict[str, int], str, str]:
    ids = sorted({str(value).strip() for value in material_ids if str(value).strip()})
    if not ids:
        return {}, "", ""
    client = _get_odps_client()
    end_dt = _latest_partition_dt(client)
    end_date = datetime.strptime(end_dt, "%Y%m%d").date()
    start_dt = (end_date - timedelta(days=EXTERNAL_RECALL_UV_WINDOW_DAYS - 1)).strftime(
        "%Y%m%d"
    )
    totals: dict[str, int] = {}
    for offset in range(0, len(ids), 500):
        batch = ids[offset:offset + 500]
        quoted = ",".join(_quote_sql(value) for value in batch)
        sql = f"""
SELECT card_cover_id, SUM(`访问uv`) AS visit_uv_30d
FROM loghubods.cooperate_top_cards_daily_v1
WHERE dt BETWEEN '{start_dt}' AND '{end_dt}'
  AND card_cover_id IN ({quoted})
GROUP BY card_cover_id
"""
        instance = client._odps.execute_sql(sql)
        instance.wait_for_success()
        with instance.open_reader(tunnel=False) as reader:
            for row in reader:
                totals[str(row[0])] = int(row[1] or 0)
    return totals, start_dt, end_dt


def rank_external_materials_by_visit_uv(
    materials: list[Material],
    min_uv: int | None = None,
) -> list[Material]:
    """补充配置窗口访问 UV，按 min_uv 过滤，再按 UV 和相似度排序。"""
    if not materials:
        return []
    totals, start_dt, end_dt = _query_visit_uv_30d(
        material.material_id for material in materials
    )
    for material in materials:
        material.visit_uv_30d = totals.get(material.material_id, 0)
        material.raw["visit_uv_30d"] = material.visit_uv_30d
        material.raw["uv_window_start"] = start_dt
        material.raw["uv_window_end"] = end_dt
    before = len(materials)
    if min_uv is not None:
        materials = [
            m for m in materials if (m.visit_uv_30d or 0) >= min_uv
        ]
    dropped_uv = before - len(materials)
    ranked = sorted(
        materials,
        key=lambda material: (material.visit_uv_30d or 0, material.score or 0),
        reverse=True,
    )
    logger.info(
        "[external_material] UV排序 candidates=%d dropped_uv=%d min_uv=%s days=%d window=%s-%s top=%s",
        before, dropped_uv, min_uv, EXTERNAL_RECALL_UV_WINDOW_DAYS, start_dt, end_dt,
        [
            (material.material_id, material.visit_uv_30d, round(material.score, 4))
            for material in ranked[:5]
        ],
    )
    return ranked


def recall_external_materials_for_video(
    landing: LandingVideo,
    *,
    element_features: Iterable,
) -> list[Material]:
    # 外部合作素材绕过历史成本门槛(apply_cost_filter=False):
    # 素材多为合作方提供、无本账户历史投放成本,以访问 UV 门槛把关。
    materials = recall_materials_for_video(
        landing,
        final_top_n=None,
        source_labels=[EXTERNAL_RECALL_SOURCE_LABEL],
        element_features=element_features,
        apply_cover_blacklist=False,
        apply_cost_filter=False,
    )
    ranked = rank_external_materials_by_visit_uv(
        materials,
        min_uv=EXTERNAL_RECALL_MIN_UV,
    )
    return ranked[:max(1, EXTERNAL_RECALL_CANDIDATE_LIMIT)]


def _load_cleanup_prompt() -> str:
    text = EXTERNAL_CLEANUP_PROMPT_PATH.read_text(encoding="utf-8").strip()
    if not text:
        raise RuntimeError(f"外部素材清理prompt为空:{EXTERNAL_CLEANUP_PROMPT_PATH}")
    return text


def _recall_json(material: Material) -> dict[str, Any]:
    return {
        "recall_strategy": material.recall_strategy,
        "recall_query_text": material.recall_query_text,
        "recall_config_code": material.recall_config_code,
        "recall_element_dimension": material.recall_element_dimension,
        "recall_point_type": material.recall_point_type,
        "recall_standard_element": material.recall_standard_element,
        "recall_hit_queries": material.recall_hit_queries,
        "source_raw": material.raw,
    }


def _asset_from_row(row: dict[str, Any]) -> ExternalMaterialAsset:
    try:
        recall = json.loads(row.get("recall_json") or "{}")
    except Exception:
        recall = {}
    return ExternalMaterialAsset(
        id=int(row["id"]),
        source_material_id=str(row["source_material_id"]),
        source_image_url=str(row.get("source_image_url") or ""),
        source_title=str(row.get("source_title") or ""),
        similarity_score=float(row.get("similarity_score") or 0),
        visit_uv_30d=int(row.get("visit_uv_30d") or 0),
        uv_window_start=str(row.get("uv_window_start") or ""),
        uv_window_end=str(row.get("uv_window_end") or ""),
        oss_url=str(row.get("oss_url") or ""),
        recall_json=recall,
    )


def _load_existing_row(
    account_id: int,
    adgroup_id: int,
    landing_video_id: int,
    source_material_id: str,
) -> dict[str, Any] | None:
    ensure_external_material_table()
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
SELECT *
FROM external_recalled_material
WHERE account_id=%s AND adgroup_id=%s AND landing_video_id=%s
  AND source_material_id=%s
LIMIT 1
""",
                (account_id, adgroup_id, landing_video_id, source_material_id),
            )
            return cur.fetchone()
    finally:
        conn.close()


def _insert_generated_asset(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
    material: Material,
    source_image_url: str,
    prompt: str,
    object_key: str,
    oss_url: str,
    raw_response: dict[str, Any],
) -> int:
    ensure_external_material_table()
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
INSERT INTO external_recalled_material
    (account_id, adgroup_id, crowd_package, landing_video_id,
     source_material_id, source_image_url, source_title, similarity_score,
     visit_uv_30d, uv_window_start, uv_window_end, recall_json,
     edit_prompt, model, oss_object_key, oss_url, raw_response)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
                (
                    account_id,
                    adgroup_id,
                    crowd_package,
                    landing.video_id,
                    material.material_id,
                    source_image_url,
                    material.title,
                    material.score,
                    int(material.visit_uv_30d or 0),
                    str(material.raw.get("uv_window_start") or ""),
                    str(material.raw.get("uv_window_end") or ""),
                    json.dumps(_recall_json(material), ensure_ascii=False, default=str),
                    prompt,
                    EXTERNAL_IMAGE_MODEL,
                    object_key,
                    oss_url,
                    json.dumps(raw_response, ensure_ascii=False, default=str)[:16000000],
                ),
            )
            asset_id = int(cur.lastrowid)
        conn.commit()
        return asset_id
    finally:
        conn.close()


def _update_review(asset_id: int, review) -> None:
    from db.connection import get_connection

    status = "generated" if review.status == "pass" else review.status
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
UPDATE external_recalled_material
SET status=%s, ai_review_status=%s, ai_review_score=%s,
    ai_review_reason=%s, ai_review_json=%s, error=%s,
    updated_at=CURRENT_TIMESTAMP
WHERE id=%s
""",
                (
                    status,
                    review.status,
                    review.score,
                    review.reason,
                    json.dumps(review.raw, ensure_ascii=False, default=str),
                    review.reason if review.status == "error" else None,
                    asset_id,
                ),
            )
        conn.commit()
    finally:
        conn.close()


def _review_asset(
    *,
    asset_id: int,
    oss_url: str,
    prompt: str,
    landing: LandingVideo,
    material: Material,
):
    try:
        review = review_generated_material(
            image_url=oss_url,
            prompt_type="external_cleanup",
            prompt_text=prompt,
            feature_hits=[
                {
                    "landing_video_id": landing.video_id,
                    "landing_title": landing.title,
                    "source_material_id": material.material_id,
                    "source_title": material.title,
                    "visit_uv_30d": int(material.visit_uv_30d or 0),
                    "recall_hit_queries": material.recall_hit_queries,
                }
            ],
        )
    except Exception as exc:
        logger.exception(
            "[external_material] AI审核异常 asset=%d source=%s:%s",
            asset_id, material.material_id, exc,
        )
        review = MaterialReviewResult(
            status="error",
            score=0,
            reason=f"AI审核异常:{exc}",
            risk_tags=["review_error"],
            ocr_text="",
            raw={"error": str(exc)},
        )
    _update_review(asset_id, review)
    return review


def get_or_create_external_asset(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
    material: Material,
) -> Material | None:
    existing = _load_existing_row(
        account_id,
        adgroup_id,
        landing.video_id,
        material.material_id,
    )
    if existing:
        if (
            existing.get("oss_url")
            and existing.get("status") in {"generated", "error"}
            and existing.get("ai_review_status") in {None, "", "error"}
        ):
            review = _review_asset(
                asset_id=int(existing["id"]),
                oss_url=str(existing["oss_url"]),
                prompt=str(existing.get("edit_prompt") or _load_cleanup_prompt()),
                landing=landing,
                material=material,
            )
            existing = _load_existing_row(
                account_id,
                adgroup_id,
                landing.video_id,
                material.material_id,
            )
            if review.passed and existing:
                return _asset_from_row(existing).to_material()
        if (
            existing.get("ai_review_status") == "pass"
            and existing.get("status") in {"generated", "prepared"}
            and existing.get("oss_url")
        ):
            logger.info(
                "[external_material] 复用派生图 account=%d adgroup=%d landing=%d source=%s id=%s",
                account_id, adgroup_id, landing.video_id, material.material_id,
                existing["id"],
            )
            return _asset_from_row(existing).to_material()
        logger.info(
            "[external_material] 已有不可用派生记录 source=%s status=%s review=%s",
            material.material_id, existing.get("status"),
            existing.get("ai_review_status"),
        )
        return None

    source_image_url = _source_image_url(material)
    if not source_image_url:
        logger.warning(
            "[external_material] source=%s 没有可编辑图片URL", material.material_id
        )
        return None

    prompt = _load_cleanup_prompt()
    image_bytes, content_type, raw_response = generate_image_bytes(
        prompt,
        model=EXTERNAL_IMAGE_MODEL,
        input_image_url=source_image_url,
    )
    source_hash = hashlib.sha256(material.material_id.encode("utf-8")).hexdigest()[:12]
    object_key = build_ai_image_object_key(
        account_id=account_id,
        landing_video_id=landing.video_id,
        prompt_type=f"external_cleanup_{source_hash}",
        extension="jpg",
    )
    oss_url = upload_image_to_oss(image_bytes, content_type, object_key)
    asset_id = _insert_generated_asset(
        account_id=account_id,
        adgroup_id=adgroup_id,
        crowd_package=crowd_package,
        landing=landing,
        material=material,
        source_image_url=source_image_url,
        prompt=prompt,
        object_key=object_key,
        oss_url=oss_url,
        raw_response=raw_response,
    )
    review = _review_asset(
        asset_id=asset_id,
        oss_url=oss_url,
        prompt=prompt,
        landing=landing,
        material=material,
    )
    logger.info(
        "[external_material] 派生图完成 account=%d adgroup=%d landing=%d source=%s "
        "asset=%d uv_window=%d review=%s score=%d url=%s",
        account_id, adgroup_id, landing.video_id, material.material_id,
        asset_id, int(material.visit_uv_30d or 0), review.status, review.score, oss_url,
    )
    if not review.passed:
        return None
    row = _load_existing_row(
        account_id,
        adgroup_id,
        landing.video_id,
        material.material_id,
    )
    return _asset_from_row(row).to_material() if row else None


def prepare_external_material_for_landing(
    *,
    account_id: int,
    adgroup_id: int,
    crowd_package: str,
    landing: LandingVideo,
    element_features: Iterable,
    excluded_material_ids: set[str],
) -> Material | None:
    ranked = recall_external_materials_for_video(
        landing,
        element_features=element_features,
    )
    attempted = 0
    for material in ranked:
        external_id = f"external:{material.material_id}"
        if external_id in excluded_material_ids:
            continue
        attempted += 1
        try:
            processed = get_or_create_external_asset(
                account_id=account_id,
                adgroup_id=adgroup_id,
                crowd_package=crowd_package,
                landing=landing,
                material=material,
            )
        except Exception as exc:
            logger.exception(
                "[external_material] 图生图失败 landing=%d source=%s:%s",
                landing.video_id, material.material_id, exc,
            )
            processed = None
        if processed is not None:
            return processed
        if attempted >= max(1, EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING):
            break
    logger.info(
        "[external_material] landing=%d 无可用派生素材 recalled=%d attempted=%d",
        landing.video_id, len(ranked), attempted,
    )
    return None


def update_external_material_status(
    record: dict,
    status: str,
    *,
    dynamic_creative_id: int | str | None = None,
    tencent_image_id: str = "",
    error: str = "",
) -> None:
    asset_id = record.get("_external_recalled_material_id")
    if not asset_id:
        return
    ensure_external_material_table()
    from db.connection import get_connection

    normalized_status = {
        "approve": "approved",
        "reject": "rejected",
    }.get(status, status)
    approval_status = {
        "approve": "approved",
        "reject": "rejected",
        "hold": "hold",
        "skip": "skip",
    }.get(status, normalized_status)
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
UPDATE external_recalled_material
SET status=%s, approval_status=%s,
    dynamic_creative_id=COALESCE(%s, dynamic_creative_id),
    tencent_image_id=COALESCE(NULLIF(%s, ''), tencent_image_id),
    error=COALESCE(NULLIF(%s, ''), error), updated_at=CURRENT_TIMESTAMP
WHERE id=%s
""",
                (
                    normalized_status,
                    approval_status,
                    int(dynamic_creative_id) if dynamic_creative_id else None,
                    tencent_image_id,
                    error[:2000] if error else "",
                    int(asset_id),
                ),
            )
        conn.commit()
    finally:
        conn.close()