Agent/examples/auto_put_ad_mini/tools/video_risk.py

"""承接视频风险标签查询与审核。

在模块 B 创建 landing plan(xcx/save)之前调用:
  video_id → piaoquantv getVideoTagIds → 风险等级 → pass/block

接口返回的 data 是 tag id 列表。只有 config.VIDEO_RISK_TAG_LEVELS 中的 tag
参与风险等级计算,其他 tag 保留到 unknown_tag_ids 便于排查,但不默认阻断。
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, asdict
from typing import Any

import httpx

from config import (
    VIDEO_RISK_API_TIMEOUT_SECONDS,
    VIDEO_RISK_API_URL,
    VIDEO_RISK_CHECK_ENABLED,
    VIDEO_RISK_MAX_ALLOWED_LEVEL,
    VIDEO_RISK_TAG_LEVELS,
)

logger = logging.getLogger(__name__)


@dataclass
class VideoRiskResult:
    video_id: int
    passed: bool
    max_level: int
    risk_tag_ids: list[str]
    unknown_tag_ids: list[str]
    reason: str
    raw_tag_ids: list[str]

    def to_record_fields(self) -> dict[str, Any]:
        """转成可合并进候选记录的落表字段 dict。"""
        return {
            "landing_risk_passed": self.passed,
            "landing_risk_level": self.max_level,
            "landing_risk_tag_ids": ",".join(self.risk_tag_ids),
            "landing_unknown_tag_ids": ",".join(self.unknown_tag_ids),
            "landing_risk_reason": self.reason,
            "_landing_risk": asdict(self),
        }


def _normalize_tag_ids(data: Any) -> list[str]:
    """把接口返回的 data 规范化成字符串 tag id 列表。"""
    if data is None:
        return []
    if isinstance(data, list):
        return [str(x) for x in data if x is not None]
    return [str(data)]


def query_video_risk_tags(video_id: int) -> list[str]:
    """查询视频风险 tag ids。

    Raises:
        RuntimeError: HTTP / 接口 code 非 0 / data 非预期时抛出。
    """
    resp = httpx.post(
        VIDEO_RISK_API_URL,
        json={"videoId": int(video_id)},
        headers={"Content-Type": "application/json"},
        timeout=VIDEO_RISK_API_TIMEOUT_SECONDS,
    )
    resp.raise_for_status()
    payload = resp.json()
    if payload.get("code") != 0:
        raise RuntimeError(
            f"video risk api failed video_id={video_id} "
            f"code={payload.get('code')} msg={payload.get('msg')}"
        )
    return _normalize_tag_ids(payload.get("data"))


def evaluate_video_risk(video_id: int, tag_ids: list[str]) -> VideoRiskResult:
    """把 tag ids 映射成最高风险等级并判断是否可用。"""
    risk_tag_ids: list[str] = []
    unknown_tag_ids: list[str] = []
    levels: list[int] = []

    for tag_id in tag_ids:
        level = VIDEO_RISK_TAG_LEVELS.get(str(tag_id))
        if level is None:
            unknown_tag_ids.append(str(tag_id))
            continue
        risk_tag_ids.append(str(tag_id))
        levels.append(int(level))

    max_level = max(levels) if levels else 0
    passed = max_level <= VIDEO_RISK_MAX_ALLOWED_LEVEL
    if not VIDEO_RISK_CHECK_ENABLED:
        passed = True
        reason = "risk check disabled"
    elif max_level == 0:
        reason = "no mapped risk tags"
    elif passed:
        reason = f"risk level {max_level} <= allowed {VIDEO_RISK_MAX_ALLOWED_LEVEL}"
    else:
        reason = f"risk level {max_level} > allowed {VIDEO_RISK_MAX_ALLOWED_LEVEL}"

    return VideoRiskResult(
        video_id=int(video_id),
        passed=passed,
        max_level=max_level,
        risk_tag_ids=risk_tag_ids,
        unknown_tag_ids=unknown_tag_ids,
        reason=reason,
        raw_tag_ids=[str(x) for x in tag_ids],
    )


def check_video_risk(video_id: int) -> VideoRiskResult:
    """查询并评估视频风险。

    接口异常时返回 passed=False,让调用方跳过该 landing,避免在风险未知时创建 plan。
    """
    if not VIDEO_RISK_CHECK_ENABLED:
        return evaluate_video_risk(int(video_id), [])

    try:
        tag_ids = query_video_risk_tags(int(video_id))
        result = evaluate_video_risk(int(video_id), tag_ids)
        logger.info(
            "[video_risk] video_id=%s passed=%s level=%s risk_tags=%s unknown_tags=%s reason=%s",
            video_id,
            result.passed,
            result.max_level,
            result.risk_tag_ids,
            result.unknown_tag_ids,
            result.reason,
        )
        return result
    except Exception as e:
        logger.warning("[video_risk] video_id=%s 查询失败,跳过该视频:%s", video_id, e)
        return VideoRiskResult(
            video_id=int(video_id),
            passed=False,
            max_level=999,
            risk_tag_ids=[],
            unknown_tag_ids=[],
            reason=f"risk api error: {e}",
            raw_tag_ids=[],
        )