Create-knowledge-find-decode/tests/test_video_extract.py

"""原生整段视频提取离线测试：注入假 http_post 返回 segments，断言段卡 + 内容。"""
from __future__ import annotations

import json

from core.config import PgConfig, Settings
from creation_knowledge.integrations.video_extract import extract_video
from core.models import Post

SEGMENTS = json.dumps({
    "video_title": "AI视频教程", "overall": "教 AI 出片",
    "segments": [
        {"start": "00:00", "end": "00:18", "title": "问题与简介",
         "knowledge_types": ["what"], "what": "常见翻车", "why": None, "how": None},
        {"start": "00:19", "end": "00:48", "title": "去噪原理",
         "knowledge_types": ["why"], "what": None, "why": "噪声去噪类似雕刻", "how": None},
    ]}, ensure_ascii=False)


class _Resp:
    def __init__(self, content): self._c = content
    def raise_for_status(self): return None
    def json(self): return {"choices": [{"message": {"content": self._c}}]}


def _settings() -> Settings:
    return Settings(
        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
        aiddit_crawler_base_url="x", crawler_timeout=30,
        openrouter_timeout_seconds=90,
        openrouter_model="google/gemini-3-flash-preview",
        openrouter_base_url="https://openrouter.ai/api/v1", openrouter_api_key="k",
        llm_model="m", max_cards=12,
        frames_dir="runtime/frames", douyin_ratio="540p", data_dir="")


def _fake_post():
    captured = {}

    def post(url, headers=None, json=None, timeout=None):
        captured["url"] = url
        captured["json"] = json
        return _Resp(SEGMENTS)

    return post, captured


def test_video_extract_makes_segment_cards(tmp_path):
    vf = tmp_path / "v.mp4"
    vf.write_bytes(b"\x00\x00\x00\x18ftypmp42fakevideobytes")
    fake, captured = _fake_post()
    post = Post(id="dy_1", platform="douyin", url="u", content_id="1",
                content_type="video", video_urls=["http://x/play?ratio=default"])

    out = extract_video(post, settings=_settings(), http_post=fake, video_path=str(vf))

    # ExtractedContent.cards = 每段一条内容
    assert len(out.cards) == 2
    assert out.cards[0].index == 1 and "问题" in out.cards[0].content
    assert out.is_empty is False and out.text  # overall 进 text

    # post.cards = 段卡，带 start/end（秒）
    assert len(post.cards) == 2
    assert post.cards[0].kind == "segment"
    assert post.cards[0].start == 0.0 and post.cards[1].start == 19.0
    assert post.cards[1].end == 48.0
    assert post.cards[0].url is None  # 段卡默认无图

    # 请求体含 base64 video_url
    parts = captured["json"]["messages"][0]["content"]
    assert any(p["type"] == "video_url"
               and p["video_url"]["url"].startswith("data:video/mp4;base64,")
               for p in parts)


def test_video_extract_saves_and_sets_url(tmp_path):
    """传 save_path/public_url：整段视频落盘 + 段卡 url 指向整段。"""
    vf = tmp_path / "src.mp4"
    vf.write_bytes(b"\x00\x00\x00\x18ftypmp42fakevideobytes")
    save_path = tmp_path / "out" / "video.mp4"
    fake, _ = _fake_post()
    post = Post(id="dy_1", platform="douyin", url="u", content_id="1",
                content_type="video", video_urls=["http://x/play"])

    extract_video(post, settings=_settings(), http_post=fake, video_path=str(vf),
                  save_path=save_path, public_url="/data/B/douyin/dy_1/video.mp4")

    assert save_path.exists() and save_path.read_bytes() == vf.read_bytes()
    assert all(c.url == "/data/B/douyin/dy_1/video.mp4" for c in post.cards)


def test_extract_video_oss_direct_feed():
    """传 oss_video_url：不下载、不 base64，请求体直接用该 http 直链；段卡 url 指向它。"""
    fake, captured = _fake_post()
    downloads = []

    def spy_downloader(url, platform):
        downloads.append(url)
        return b"should-not-be-called"

    post = Post(id="dy_2", platform="douyin", url="u", content_id="2",
                content_type="video", video_urls=["http://cdn/raw.mp4?sig=tmp"])

    out = extract_video(post, settings=_settings(), http_post=fake,
                        downloader=spy_downloader,
                        oss_video_url="https://res.cybertogether.net/crawler/video/x.mp4")

    assert downloads == []  # 未触发任何下载
    parts = captured["json"]["messages"][0]["content"]
    vparts = [p for p in parts if p["type"] == "video_url"]
    assert vparts and vparts[0]["video_url"]["url"] == "https://res.cybertogether.net/crawler/video/x.mp4"
    assert not vparts[0]["video_url"]["url"].startswith("data:")  # 不是 base64
    assert out.cards and len(post.cards) == 2
    assert all(c.url == "https://res.cybertogether.net/crawler/video/x.mp4" for c in post.cards)