"""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"]