YouTube 视频营销的工程化实践：Data API、数据看板与合规自动化

import os
import requests
from typing import Iterable

API_KEY = os.getenv("YOUTUBE_API_KEY")
BASE = "https://www.googleapis.com/youtube/v3"

def chunked(items: list, size: int = 50) -> Iterable[list]:
    for i in range(0, len(items), size):
        yield items[i:i + size]

def fetch_videos(video_ids: list[str]) -> list[dict]:
    results = []
    for batch in chunked(video_ids, 50):
        resp = requests.get(
            f"{BASE}/videos",
            params={
                "key": API_KEY,
                "id": ",".join(batch),
                "part": "snippet,statistics,contentDetails",
            },
            timeout=15,
        )
        resp.raise_for_status()
        results.extend(resp.json().get("items", []))
    return results

def parse_video(item: dict) -> dict:
    stats = item.get("statistics", {})
    snippet = item.get("snippet", {})
    return {
        "video_id": item["id"],
        "title": snippet.get("title", ""),
        "published_at": snippet.get("publishedAt", ""),
        "duration": item.get("contentDetails", {}).get("duration", ""),
        "views": int(stats.get("viewCount", 0)),
        "likes": int(stats.get("likeCount", 0)),
        "comments": int(stats.get("commentCount", 0)),
    }

from dataclasses import dataclass

@dataclass
class VideoMetrics:
    views: int
    likes: int
    comments: int
    duration_sec: int
    watch_time_sec: float = 0.0

    @property
    def engagement_rate(self) -> float:
        if self.views == 0:
            return 0.0
        return (self.likes + self.comments) / self.views

    @property
    def avg_view_duration(self) -> float:
        if self.views == 0:
            return 0.0
        return self.watch_time_sec / self.views

    @property
    def retention(self) -> float:
        if self.duration_sec == 0:
            return 0.0
        return self.avg_view_duration / self.duration_sec

def score(m: VideoMetrics, w=(0.5, 0.3, 0.2)) -> float:
    """加权评分：留存 > 互动 > 绝对播放量"""
    return (
        w[0] * m.retention
        + w[1] * min(m.engagement_rate * 20, 1.0)
        + w[2] * min(m.views / 10000, 1.0)
    )

import hashlib
import time
from dataclasses import dataclass, field

@dataclass
class Variant:
    name: str
    title: str
    thumbnail: str
    impressions: int = 0
    clicks: int = 0

    @property
    def ctr(self) -> float:
        return self.clicks / self.impressions if self.impressions else 0.0

@dataclass
class ABTest:
    video_id: str
    variants: list[Variant] = field(default_factory=list)
    started_at: float = field(default_factory=time.time)

    def snapshot(self) -> dict:
        return {
            "video_id": self.video_id,
            "variants": [
                {"name": v.name, "ctr": round(v.ctr, 4),
                 "impressions": v.impressions, "clicks": v.clicks}
                for v in self.variants
            ],
        }

    def winner(self, min_impressions: int = 5000) -> str | None:
        eligible = [v for v in self.variants
                    if v.impressions >= min_impressions]
        if len(eligible) < 2:
            return None
        return max(eligible, key=lambda v: v.ctr).name

def variant_key(video_id: str, name: str) -> str:
    raw = f"{video_id}:{name}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

import re

PII = re.compile(r"(\d{11}|\d{17}[\dXx]|[\w.+-]+@[\w-]+\.[\w.]+)")

BANNED_PHRASES = {
    "guaranteed income", "get rich quick", "100% safe",
    "cure", "miracle", "no risk",
}

def mask(text: str) -> str:
    return PII.sub("[REDACTED]", text)

def audit_metadata(title: str, description: str) -> list[str]:
    issues = []
    blob = f"{title} {description}".lower()
    for phrase in BANNED_PHRASES:
        if phrase in blob:
            issues.append(f"含承诺性或误导性用语：{phrase}")
    if "ai generated" not in blob and "ai-generated" not in blob:
        issues.append("未标注 AI 生成内容（如适用）")
    if PII.search(blob):
        issues.append("含个人信息，需脱敏")
    return issues

def audit_copyright(claims: list[dict]) -> list[str]:
    """claims 来自 Content ID 或人工登记"""
    return [
        f"视频 {c['video_id']} 存在版权主张：{c['asset']}"
        for c in claims if c.get("blocked")
    ]