"""承接视频风险标签查询与审核。 在模块 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=[], )