"""原生整段视频提炼：把整段 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, )