content-find-agent-new/tests/test_platform_video_url.py

from __future__ import annotations

from content_agent.integrations.platform_video_url import (
    classify_url_candidate,
    find_url_candidates,
    select_video_url,
)


def _probe(status: str = "verified"):
    def probe(url: str, platform: str):
        return {
            "http_status": 206 if status == "verified" else None,
            "content_type": "video/mp4" if status == "verified" else "",
            "content_range": "bytes 0-99/100" if status == "verified" else "",
            "range_bytes": 100 if status == "verified" else 0,
            "looks_like_video": status == "verified",
            "probe_status": status,
        }

    return probe


def test_classifies_urls_without_ad_substring_false_positive():
    payload = {
        "content_link": "https://www.example.com/item",
        "image_url_list": [{"image_url": "https://img.example/a.jpg"}],
        "avatar": "https://wx.qlogo.cn/avatar",
        "bgm_data": {"play_url": "https://audio.example/a.m4a"},
        "ads": {"video_url": "https://ad.example/ad.mp4"},
        "video_url_list": [
            {
                "video_url": (
                    "https://findermp.video.qq.com/251/20302/stodownload?"
                    "head=1&token=abc"
                )
            }
        ],
    }

    by_path = {item.path: item for item in find_url_candidates(payload)}

    assert by_path["$.content_link"].kind == "page"
    assert by_path["$.image_url_list[0].image_url"].kind == "image"
    assert by_path["$.avatar"].kind == "avatar"
    assert by_path["$.bgm_data.play_url"].kind == "audio"
    assert by_path["$.ads.video_url"].kind == "ad_or_material"
    assert by_path["$.video_url_list[0].video_url"].kind == "video"
    assert classify_url_candidate("$.video_url", "https://x/stodownload?head=1")[0] == "video"


def test_kuaishou_prefers_verified_detail_video_over_search():
    result = select_video_url(
        "kuaishou",
        [
            ("search", {"video_url_list": [{"video_url": "https://v23-3.kwaicdn.com/search.mp4"}]}),
            ("detail", {"video_url_list": [{"video_url": "https://tymov2.a.kwimgs.com/detail.mp4"}]}),
        ],
        probe_fn=_probe(),
    )

    assert result["play_url"] == "https://tymov2.a.kwimgs.com/detail.mp4"
    assert result["platform_raw_payload"]["selected_video_url_path"] == "$.detail.video_url_list[0].video_url"
    assert result["platform_raw_payload"]["selected_video_url_probe_status"] == "verified"
    assert [candidate["url"] for candidate in result["video_url_candidates"]] == [
        "https://tymov2.a.kwimgs.com/detail.mp4",
        "https://v23-3.kwaicdn.com/search.mp4",
    ]
    assert "video_url_candidates" not in result["platform_raw_payload"]


def test_shipinhao_prefers_findermp_video_and_filters_non_video():
    result = select_video_url(
        "shipinhao",
        [
            (
                "search",
                {
                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
                    "video_url_list": [{"video_url": "https://findermp.video.qq.com/video.mp4"}],
                },
            )
        ],
        probe_fn=_probe(),
    )

    assert result["play_url"] == "https://findermp.video.qq.com/video.mp4"
    assert result["platform_raw_payload"]["video_url_candidate_counts"] == {
        "image": 1,
        "video": 1,
    }


def test_probe_failure_falls_back_to_strong_video_candidate():
    result = select_video_url(
        "kuaishou",
        [("search", {"video_url_list": [{"video_url": "https://v23-3.kwaicdn.com/search.mp4"}]})],
        probe_fn=_probe("failed"),
    )

    assert result["play_url"] == "https://v23-3.kwaicdn.com/search.mp4"
    assert result["platform_raw_payload"]["selected_video_url_probe_status"] == "failed_fallback"
    assert result["media_failure_reason"] is None


def test_douyin_demotes_v11_weba_when_better_candidate_exists():
    result = select_video_url(
        "douyin",
        [
            (
                "search",
                {
                    "video": {
                        "play_addr": {
                            "url_list": [
                                "https://v11-weba.douyinvod.com/video.mp4",
                                "https://v26-web.douyinvod.com/video.mp4",
                            ]
                        }
                    }
                },
            )
        ],
        probe_fn=_probe(),
    )

    assert result["play_url"] == "https://v26-web.douyinvod.com/video.mp4"
    assert result["platform_raw_payload"]["selected_video_url_host"] == "v26-web.douyinvod.com"
    assert [candidate["host"] for candidate in result["video_url_candidates"]] == [
        "v26-web.douyinvod.com",
        "v11-weba.douyinvod.com",
    ]


def test_video_url_candidates_are_capped_to_selected_plus_two():
    result = select_video_url(
        "kuaishou",
        [
            (
                "search",
                {
                    "video_url_list": [
                        {"video_url": f"https://v{index}.kwaicdn.com/{index}.mp4"}
                        for index in range(5)
                    ]
                },
            )
        ],
        probe_fn=_probe(),
    )

    assert len(result["video_url_candidates"]) == 3
    assert result["video_url_candidates"][0]["url"] == result["play_url"]


def test_no_video_candidate_returns_no_valid_play_url():
    result = select_video_url(
        "kuaishou",
        [("search", {"image_url_list": [{"image_url": "https://img.example/a.jpg"}]})],
        probe_fn=_probe(),
    )

    assert result["play_url"] is None
    assert result["media_failure_reason"] == "no_valid_play_url"
    assert result["platform_raw_payload"]["no_valid_play_url"] is True
    assert result["video_url_candidates"] == []