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