Create-knowledge-find-decode/creation_knowledge/integrations/video_extract.py

"""原生整段视频提炼：把整段 mp4 经 OpenRouter base64 video_url 发给 Gemini，
按时间戳分段提炼 What/Why/How。取代抽帧作为视频内容提炼主路。

每个 segment → 一张段卡 Card(kind="segment", start, end)，写入 post.cards；
同时产出 ExtractedContent.cards=[{index, content}]，让下游 split 照常按【卡片N】溯源。

网络分裂：下载抖音视频需非新加坡出口，调 OpenRouter 需可达 Google。下载做成可注入
（downloader）或可传入本地文件（video_path），便于"下载节点 / 推理节点"分离。
"""
from __future__ import annotations

import base64
import logging
import os
import re
from typing import Any, Callable, Optional

import httpx

from creation_knowledge.config import Settings
from creation_knowledge.jsonio import extract_json_object
from creation_knowledge.models import Card, CardExtract, ExtractedContent, Post
from creation_knowledge.prompts import load_prompt

logger = logging.getLogger(__name__)

_REFERER = {
    "douyin": "https://www.douyin.com/",
    "kuaishou": "https://www.kuaishou.com/",
    "bilibili": "https://www.bilibili.com/",
    "shipinhao": "https://channels.weixin.qq.com/",
}
_IOS_UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15"


class VideoExtractError(RuntimeError):
    pass


def _mmss_to_sec(value: Any) -> Optional[float]:
    """'MM:SS' / 'HH:MM:SS' / 数字 → 秒。"""
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return float(value)
    parts = str(value).strip().split(":")
    try:
        nums = [float(p) for p in parts]
    except ValueError:
        return None
    sec = 0.0
    for n in nums:
        sec = sec * 60 + n
    return sec


def _default_download(url: str, platform: str, timeout: float = 180.0) -> bytes:
    headers = {"User-Agent": _IOS_UA, "Referer": _REFERER.get(platform, "")}
    with httpx.stream("GET", url, headers=headers, timeout=timeout,
                      follow_redirects=True) as r:
        r.raise_for_status()
        return b"".join(r.iter_bytes())


def _seg_content(seg: dict) -> str:
    parts = [seg.get("title") or ""]
    for label, key in (("What", "what"), ("Why", "why"), ("How", "how")):
        v = seg.get(key)
        if v and str(v).strip().lower() not in ("null", "none", ""):
            parts.append(f"{label}：{v}")
    return "。".join(p for p in parts if p)


def extract_video(
    post: Post,
    *,
    settings: Settings,
    http_post: Callable[..., Any] = httpx.post,
    video_path: Optional[str] = None,
    downloader: Optional[Callable[[str, str], bytes]] = None,
    timeout: float = 600.0,
) -> ExtractedContent:
    """对视频帖做原生整段提炼，就地写好 post.cards（段卡），返回 ExtractedContent。"""
    key = settings.openrouter_api_key
    if not key:
        raise VideoExtractError("missing OPENROUTER_API_KEY")

    # 1) 拿到 mp4 字节
    if video_path and os.path.exists(video_path):
        data = open(video_path, "rb").read()
        logger.info("video_extract 用本地文件 %s (%d bytes)", video_path, len(data))
    else:
        if not post.video_urls:
            raise VideoExtractError(f"post {post.id} 无 video_urls 且未提供 video_path")
        url = post.video_urls[0]
        if post.platform == "douyin" and "ratio=" in url:  # 偏好较小码率控成本
            url = re.sub(r"ratio=[^&]+", f"ratio={settings.douyin_ratio}", url)
        dl = downloader or _default_download
        try:
            data = dl(url, post.platform)
        except Exception as exc:
            raise VideoExtractError(f"视频下载失败: {exc}") from exc
    if not data:
        raise VideoExtractError("视频字节为空")

    # 2) base64 + 3) OpenRouter 原生视频
    data_url = "data:video/mp4;base64," + base64.b64encode(data).decode()
    prompt = load_prompt("extract_video").format()
    body = {"model": settings.video_model, "messages": [{"role": "user", "content": [
        {"type": "text", "text": prompt},
        {"type": "video_url", "video_url": {"url": data_url}},
    ]}]}
    try:
        resp = http_post(
            f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
            headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
            json=body, timeout=timeout)
        resp.raise_for_status()
        content = resp.json()["choices"][0]["message"]["content"]
    except httpx.HTTPError as exc:
        raise VideoExtractError(f"openrouter_http_error: {exc}") from exc
    except (KeyError, IndexError, TypeError, ValueError) as exc:
        raise VideoExtractError(f"openrouter_response_invalid: {exc}") from exc

    obj = extract_json_object(content)
    segments = obj.get("segments") or []

    # 4) 段卡 + 每段内容
    cards: list[Card] = []
    card_extracts: list[CardExtract] = []
    for i, seg in enumerate(segments, start=1):
        if not isinstance(seg, dict):
            continue
        cards.append(Card(index=i, kind="segment",
                          start=_mmss_to_sec(seg.get("start")),
                          end=_mmss_to_sec(seg.get("end"))))
        card_extracts.append(CardExtract(index=i, content=_seg_content(seg)))
    post.cards = cards  # 就地写入段卡，供 upsert 落库 + 前端展示
    return ExtractedContent(
        text=str(obj.get("overall") or obj.get("video_title") or ""),
        cards=card_extracts,
        is_empty=len(card_extracts) == 0,
    )