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 pathlib import Path
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/",
    "xiaohongshu": "https://www.xiaohongshu.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,
    save_path: Optional[Path] = None,
    public_url: Optional[str] = None,
    oss_video_url: Optional[str] = None,
) -> ExtractedContent:
    """对视频帖做原生整段提炼，就地写好 post.cards（段卡），返回 ExtractedContent。

    save_path/public_url 给定时：把下载到的整段 mp4 写到 save_path（单次下载，复用
    内存字节），并把每张段卡 url 设为 public_url（整段共用，前端按 start/end 加 #t 播片段）。
    两者默认 None＝不落盘、行为不变。

    oss_video_url 给定时：直接把该公网直链喂模型（已验证 Gemini 可直读 OSS http 视频链），
    跳过下载/落盘/base64；段卡 url 默认用它。默认 None＝走原 下载+base64 路径（fixture/兜底）。
    """
    key = settings.openrouter_api_key
    if not key:
        raise VideoExtractError("missing OPENROUTER_API_KEY")

    if oss_video_url:
        # OSS 直喂：公网直链直接进 video_url，跳过下载+落盘+base64
        media_url = oss_video_url
        public_url = public_url or oss_video_url   # 段卡指向 OSS 直链
    else:
        # B站留接口位：DASH 分轨（video m4s + voice_data 音轨 m4s）需 curl 不跟随重定向下载
        # 两轨 + ffmpeg 合并，单 URL 下载器拿不到音频。待接入；可先传入预合并的本地 video_path。
        if post.platform == "bilibili" and not (video_path and os.path.exists(video_path)):
            raise VideoExtractError(
                "B站视频暂未接入：需取 voice_data 音轨 + ffmpeg 合并（见 数据接口与来源/接口台账/B站.md）")

        # 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("视频字节为空")

        # 1.5) 可选落盘：复用内存字节，单次下载即保存（避免二次下载/抖音链接过期）
        if save_path is not None:
            save_path.parent.mkdir(parents=True, exist_ok=True)
            Path(save_path).write_bytes(data)
            logger.info("video_extract 保存整段视频 %s (%d bytes)", save_path, len(data))

        # 2) base64
        media_url = "data:video/mp4;base64," + base64.b64encode(data).decode()

    # 3) OpenRouter 原生视频
    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": media_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",
                          url=public_url,  # 整段视频公网路径；前端按 start/end 加 #t 播片段
                          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,
    )