Agent/examples/auto_put_ad_mini/debug_select_creative_patterns.py

"""Debug creative pattern selection for one video.

This script is read-only except the video feature cache may be populated when
ODPS lookup is enabled. It does not generate images, upload OSS files, or create
Tencent ads.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from dotenv import load_dotenv

_HERE = Path(__file__).parent
load_dotenv(_HERE / ".env")
sys.path.insert(0, str(_HERE))

from tools.material_strategy_learning import select_creative_patterns  # noqa: E402
from tools.video_feature_query import (  # noqa: E402
    fetch_video_element_features,
    read_cached_video_element_features,
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="调试单个视频的创意 pattern 选择")
    parser.add_argument("--video-id", type=int, required=True)
    parser.add_argument("--crowd-package", default="", help="保留展示字段,pattern选择不按人群包区分")
    parser.add_argument("--placement", default="")
    parser.add_argument("--top-k", type=int, default=3)
    parser.add_argument(
        "--include-draft",
        action="store_true",
        default=True,
        help="包含 DRAFT pattern,用于上线前调试评估",
    )
    parser.add_argument(
        "--approved-only",
        action="store_true",
        help="只看 APPROVED/enabled=1 的生产可用 pattern",
    )
    parser.add_argument(
        "--cached-only",
        action="store_true",
        help="只读本地视频特征缓存,不查 ODPS",
    )
    parser.add_argument(
        "--no-model",
        action="store_true",
        help="关闭模型选择,只看非语义稳定兜底排序",
    )
    parser.add_argument(
        "--model",
        default=None,
        help="覆盖 OPENROUTER_TEXT_MODEL,例如 google/gemini-2.5-flash",
    )
    return parser.parse_args()


def _feature_to_dict(feature) -> dict:
    return {
        "video_id": feature.video_id,
        "dt": feature.dt,
        "element_dimension": feature.element_dimension,
        "point_type": feature.point_type,
        "standard_element": feature.standard_element,
        "contribution_score": feature.contribution_score,
    }


def main() -> int:
    args = parse_args()
    if args.cached_only:
        feature_map = read_cached_video_element_features([args.video_id])
    else:
        feature_map = fetch_video_element_features([args.video_id])
    features = feature_map.get(args.video_id) or []
    selections = select_creative_patterns(
        video_features=features,
        crowd_package=args.crowd_package,
        placement=args.placement,
        include_draft=not args.approved_only and args.include_draft,
        top_k=args.top_k,
        use_model=not args.no_model,
        model=args.model,
    )
    payload = {
        "video_id": args.video_id,
        "crowd_package": args.crowd_package,
        "placement": args.placement,
        "selection_mode": "model_select" if not args.no_model else "stable_fallback",
        "feature_count": len(features),
        "features": [_feature_to_dict(feature) for feature in features],
        "selected_patterns": [
            {
                "pattern_version": item.pattern.pattern_version,
                "pattern_key": item.pattern.pattern_key,
                "pattern_name": item.pattern.pattern_name,
                "status": item.pattern.status,
                "enabled": item.pattern.enabled,
                "score": item.score,
                "reasons": item.reasons,
                "penalties": item.penalties,
                "matched_features": item.matched_features,
                "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,
                "positive_examples": item.pattern.positive_examples or [],
                "negative_examples": item.pattern.negative_examples or [],
            }
            for item in selections
        ],
    }
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())