Create-knowledge-find-decode/decode_content/readers/video.py

"""Native whole-video reader for creation knowledge decode."""
from __future__ import annotations

import base64
import logging
import os
import re
import time
from pathlib import Path
from typing import Any, Callable, Optional

import httpx

from core.config import Settings
from core.jsonio import extract_json_object
from core.media_download import download_media_bytes
from core.models import Card, CardExtract, ExtractedContent, Post
from core.prompts import load_prompt
from pipeline.tracing import TraceContext, TraceWriter, hash_prompt, redact_headers, timed_ms

logger = logging.getLogger(__name__)

class VideoExtractError(RuntimeError):
    pass


def _mmss_to_sec(value: Any) -> Optional[float]:
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return float(value)
    parts = str(value).strip().split(":")
    try:
        nums = [float(part) for part in parts]
    except ValueError:
        return None
    sec = 0.0
    for number in nums:
        sec = sec * 60 + number
    return sec


def _default_download(url: str, platform: str, timeout: float = 180.0) -> bytes:
    return download_media_bytes(url, platform, timeout=timeout)


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


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,
    trace_writer: TraceWriter | None = None,
    trace_context: TraceContext | None = None,
) -> ExtractedContent:
    key = settings.openrouter_api_key
    if not key:
        raise VideoExtractError("missing OPENROUTER_API_KEY")

    if oss_video_url:
        media_url = oss_video_url
        public_url = public_url or oss_video_url
    else:
        if post.platform == "bilibili" and not (video_path and os.path.exists(video_path)):
            raise VideoExtractError("B站视频暂未接入：需取 voice_data 音轨 + ffmpeg 合并")

        if video_path and os.path.exists(video_path):
            data = open(video_path, "rb").read()
            logger.info("video_extract using local file %s (%d bytes)", video_path, len(data))
        else:
            if not post.video_urls:
                raise VideoExtractError(f"post {post.id} has no video_urls and no 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("视频字节为空")

        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 saved whole video %s (%d bytes)", save_path, len(data))

        media_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": media_url}},
                ],
            }
        ],
    }
    started = time.perf_counter()
    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
    try:
        resp = http_post(
            f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
            headers=headers,
            json=body,
            timeout=timeout,
        )
        resp.raise_for_status()
        response_json = resp.json()
        content = response_json["choices"][0]["message"]["content"]
    except httpx.HTTPError as exc:
        if trace_writer is not None:
            trace_writer.llm_call(
                context=trace_context or TraceContext(stage="decode", substage="read_video"),
                stage="decode",
                substage="read_video",
                provider="openrouter",
                model_name=settings.video_model,
                endpoint=f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
                prompt_name="extract_video",
                prompt_hash=hash_prompt(prompt),
                request_payload={"headers": redact_headers(headers), **body},
                status="failed",
                error_message=str(exc),
                latency_ms=timed_ms(started),
            )
        raise VideoExtractError(f"openrouter_http_error: {exc}") from exc
    except (KeyError, IndexError, TypeError, ValueError) as exc:
        if trace_writer is not None:
            trace_writer.llm_call(
                context=trace_context or TraceContext(stage="decode", substage="read_video"),
                stage="decode",
                substage="read_video",
                provider="openrouter",
                model_name=settings.video_model,
                endpoint=f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
                prompt_name="extract_video",
                prompt_hash=hash_prompt(prompt),
                request_payload={"headers": redact_headers(headers), **body},
                status="failed",
                error_message=str(exc),
                latency_ms=timed_ms(started),
            )
        raise VideoExtractError(f"openrouter_response_invalid: {exc}") from exc

    obj = extract_json_object(content)
    segments = obj.get("segments") or []
    cards: list[Card] = []
    card_extracts: list[CardExtract] = []
    for idx, segment in enumerate(segments, start=1):
        if not isinstance(segment, dict):
            continue
        cards.append(
            Card(
                index=idx,
                kind="segment",
                url=public_url,
                start=_mmss_to_sec(segment.get("start")),
                end=_mmss_to_sec(segment.get("end")),
            )
        )
        card_extracts.append(CardExtract(index=idx, content=_seg_content(segment)))
    post.cards = cards
    if trace_writer is not None:
        trace_writer.llm_call(
            context=trace_context or TraceContext(stage="decode", substage="read_video"),
            stage="decode",
            substage="read_video",
            provider="openrouter",
            model_name=settings.video_model,
            endpoint=f"{settings.openrouter_base_url.rstrip('/')}/chat/completions",
            prompt_name="extract_video",
            prompt_hash=hash_prompt(prompt),
            request_payload={"headers": redact_headers(headers), **body},
            response_payload=response_json,
            parsed_payload={
                "segment_count": len(card_extracts),
                "overall": obj.get("overall"),
                "video_title": obj.get("video_title"),
            },
            status="done",
            latency_ms=timed_ms(started),
        )
    return ExtractedContent(
        text=str(obj.get("overall") or obj.get("video_title") or ""),
        cards=card_extracts,
        is_empty=len(card_extracts) == 0,
    )


def read_video(
    post: Post,
    *,
    settings: Settings,
    http_post: Callable[..., Any] | None = None,
    video_path: str | None = None,
    downloader: Callable[[str, str], bytes] | None = None,
    save_path: Path | None = None,
    public_url: str | None = None,
    oss_video_url: str | None = None,
    trace_writer: TraceWriter | None = None,
    trace_context: TraceContext | None = None,
) -> ExtractedContent:
    kwargs: dict[str, Any] = {
        "settings": settings,
        "video_path": video_path,
        "downloader": downloader,
        "save_path": save_path,
        "public_url": public_url or oss_video_url,
        "oss_video_url": oss_video_url,
        "trace_writer": trace_writer,
        "trace_context": trace_context,
    }
    if http_post is not None:
        kwargs["http_post"] = http_post
    return extract_video(post, **kwargs)


__all__ = ["VideoExtractError", "extract_video", "read_video"]