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

"""V4-M2: 快手 client search/detail/account_info normalization tests."""

from __future__ import annotations

import httpx
import pytest

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.kuaishou import (
    KUAISHOU_SEARCH_MAX_INTERVAL_SECONDS,
    KUAISHOU_SEARCH_MIN_INTERVAL_SECONDS,
    CrawapiKuaishouClient,
)


class FakeHttpClient:
    def __init__(self, responses):
        self.responses = list(responses)
        self.requests = []

    def post(self, url, json, headers, timeout):
        self.requests.append({"url": url, "json": json, "headers": headers, "timeout": timeout})
        response = self.responses.pop(0)
        if isinstance(response, Exception):
            raise response
        return response


def _response(status_code, data):
    return httpx.Response(
        status_code,
        json=data,
        request=httpx.Request("POST", "http://crawler.test/x"),
    )


def _query():
    return {
        "search_query_id": "q_001",
        "search_query": "早上好",
        "search_query_generation_method": "item_single",
        "discovery_start_source": "pattern_itemset",
    }


def _author_query():
    return {
        "search_query_id": "author_001",
        "platform_author_id": "3xfkwajatdh7p7i",
        "discovery_start_source": "pattern_itemset",
    }


def _client(responses):
    return CrawapiKuaishouClient(
        base_url="http://crawler.test",
        http_client=FakeHttpClient(responses),
    )


def _item(content_id="ks_001"):
    return {
        "channel_content_id": content_id,
        "content_link": f"https://www.kuaishou.com/short-video/{content_id}",
        "title": "早上好 #祝福",
        "body_text": "早安视频",
        "topic_list": ["早上好"],
        "content_type": "video",
        "video_url_list": [{"video_url": "https://v.kwaicdn.test/a.mp4"}],
        "channel_account_id": "3xfkwajatdh7p7i",
        "channel_account_name": "祝福账号",
        "view_count": 12345,
        "like_count": 234,
        "collect_count": 12,
        "comment_count": 34,
        "share_count": 56,
        "publish_timestamp": 1780904037000,
    }


def test_kuaishou_search_maps_canonical_fields():
    client = _client([
        _response(200, {"code": 0, "data": {"data": [_item()], "has_more": False}})
    ])

    result = client.search(_query())[0]

    assert client.http_client.requests[0]["url"].endswith("/crawler/kuai_shou/keyword_v2")
    assert client.http_client.requests[0]["json"] == {"keyword": "早上好"}
    assert result["platform"] == "kuaishou"
    assert result["platform_content_id"] == "ks_001"
    assert result["platform_content_url"].endswith("/ks_001")
    assert result["platform_author_id"] == "3xfkwajatdh7p7i"
    assert result["author_display_name"] == "祝福账号"
    assert result["play_url"] == "https://v.kwaicdn.test/a.mp4"
    assert result["video_url_candidates"][0]["url"] == "https://v.kwaicdn.test/a.mp4"
    assert result["video_url_candidates"][0]["source"] == "search"
    assert result["statistics"] == {
        "digg_count": 234,
        "comment_count": 34,
        "share_count": 56,
        "collect_count": 12,
        "play_count": 12345,
    }
    assert result["tags"] == ["#早上好"]
    assert result["create_time"] == 1780904037
    assert result["platform_raw_payload"]["channel_content_id"] == "ks_001"


def test_kuaishou_search_falls_back_to_detail_video_url():
    search_item = {**_item("ks_detail_fallback"), "video_url_list": []}
    detail_item = {**_item("ks_detail_fallback"), "video_url_list": [{"video_url": "https://v.kwaicdn.test/detail.mp4"}]}
    client = _client(
        [
            _response(200, {"code": 0, "data": {"data": [search_item], "has_more": False}}),
            _response(200, {"code": 0, "data": {"data": detail_item}}),
        ]
    )

    result = client.search(_query())[0]

    assert client.http_client.requests[1]["url"].endswith("/crawler/kuai_shou/detail")
    assert client.http_client.requests[1]["json"] == {"content_id": "ks_detail_fallback"}
    assert result["play_url"] == "https://v.kwaicdn.test/detail.mp4"
    assert result["platform_raw_payload"]["selected_video_url_path"] == "$.detail.video_url_list[0].video_url"
    assert result["platform_raw_payload"]["kuaishou_detail_fallback_status"] == "used"
    assert result["video_url_candidates"][0]["url"] == "https://v.kwaicdn.test/detail.mp4"
    assert "media_failure_reason" not in result


def test_kuaishou_metadata_search_does_not_probe_or_fetch_detail():
    probe_calls: list[str] = []
    search_item = {**_item("ks_metadata"), "video_url_list": []}
    client = CrawapiKuaishouClient(
        base_url="http://crawler.test",
        http_client=FakeHttpClient(
            [_response(200, {"code": 0, "data": {"data": [search_item], "has_more": False}})]
        ),
        video_url_probe_fn=lambda url, platform: probe_calls.append(url) or {},
    )

    [result] = client.search_full_page_metadata(_query())

    assert len(client.http_client.requests) == 1
    assert probe_calls == []
    assert result["platform_content_id"] == "ks_metadata"
    assert result["media_failure_reason"] == "no_valid_play_url"
    assert "_platform_search_item" in result


def test_kuaishou_hydrate_search_result_media_uses_detail_fallback():
    search_item = {**_item("ks_hydrate"), "video_url_list": []}
    detail_item = {
        **_item("ks_hydrate"),
        "video_url_list": [{"video_url": "https://v.kwaicdn.test/detail.mp4"}],
    }
    client = _client(
        [
            _response(200, {"code": 0, "data": {"data": [search_item], "has_more": False}}),
            _response(200, {"code": 0, "data": {"data": detail_item}}),
        ]
    )
    [metadata] = client.search_full_page_metadata(_query())

    hydrated = client.hydrate_search_result_media(metadata, _query())

    assert client.http_client.requests[1]["url"].endswith("/crawler/kuai_shou/detail")
    assert hydrated["play_url"] == "https://v.kwaicdn.test/detail.mp4"
    assert hydrated["platform_raw_payload"]["kuaishou_detail_fallback_status"] == "used"
    assert hydrated["video_url_candidates"][0]["url"] == "https://v.kwaicdn.test/detail.mp4"
    assert "_platform_search_item" not in hydrated


def test_kuaishou_search_keeps_no_valid_after_search_and_detail_fail():
    search_item = {**_item("ks_no_video"), "video_url_list": [], "image_url_list": []}
    detail_item = {**search_item, "content_link": "https://www.gifshow.com/fw/photo/ks_no_video"}
    client = _client(
        [
            _response(200, {"code": 0, "data": {"data": [search_item], "has_more": False}}),
            _response(200, {"code": 0, "data": {"data": detail_item}}),
        ]
    )

    result = client.search(_query())[0]

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


def test_kuaishou_search_limits_to_five_by_default():
    items = [_item(f"ks_{index}") for index in range(6)]
    client = _client([_response(200, {"code": 0, "data": {"data": items}})])

    results = client.search(_query())

    assert [result["platform_content_id"] for result in results] == [
        "ks_0",
        "ks_1",
        "ks_2",
        "ks_3",
        "ks_4",
    ]


def test_kuaishou_search_full_page_bypasses_local_limit():
    items = [_item(f"ks_{index}") for index in range(3)]
    client = _client([_response(200, {"code": 0, "data": {"data": items}})])
    client.max_results_per_query = 1

    results = client.search_full_page(_query())

    assert [result["platform_content_id"] for result in results] == ["ks_0", "ks_1", "ks_2"]


def test_kuaishou_from_env_search_limiter_is_ten_seconds(monkeypatch, tmp_path):
    monkeypatch.setenv("CONTENTFIND_API_CRAWAPI_BASE_URL", "http://crawler.test")

    client = CrawapiKuaishouClient.from_env(env_path=tmp_path / "missing.env")

    assert client.rate_limiter is not None
    assert client.rate_limiter.min_interval_seconds == KUAISHOU_SEARCH_MIN_INTERVAL_SECONDS
    assert client.rate_limiter.max_interval_seconds == KUAISHOU_SEARCH_MAX_INTERVAL_SECONDS


def test_kuaishou_from_env_supports_blogger_path(monkeypatch, tmp_path):
    monkeypatch.setenv("CONTENTFIND_API_CRAWAPI_BASE_URL", "http://crawler.test")
    monkeypatch.setenv("CONTENTFIND_KUAISHOU_BLOGGER_PATH", "/custom/kuaishou/blogger")

    client = CrawapiKuaishouClient.from_env(env_path=tmp_path / "missing.env")

    assert client.blogger_path == "custom/kuaishou/blogger"


def test_kuaishou_detail_maps_canonical_fields():
    detail = _item("ks_detail")
    detail["share_count"] = 78
    client = _client([_response(200, {"code": 0, "data": {"data": detail}})])

    result = client.fetch_detail("ks_detail")

    assert client.http_client.requests[0]["url"].endswith("/crawler/kuai_shou/detail")
    assert client.http_client.requests[0]["json"] == {"content_id": "ks_detail"}
    assert result["platform"] == "kuaishou"
    assert result["platform_content_id"] == "ks_detail"
    assert result["platform_content_url"].endswith("/ks_detail")
    assert result["statistics"]["play_count"] == 12345
    assert result["statistics"]["share_count"] == 78
    assert result["play_url"] == "https://v.kwaicdn.test/a.mp4"
    assert result["create_time"] == 1780904037


def test_kuaishou_account_info_maps_profile_snapshot():
    account = {
        "channel_account_id": "3xfkwajatdh7p7i",
        "ks_id": "ksid_001",
        "digit_id": "123456",
        "account_link": "https://www.kuaishou.com/profile/3xfkwajatdh7p7i",
        "account_name": "祝福账号",
        "avatar_url": "https://avatar.test/a.jpg",
        "gender": "unknown",
        "description": "每天祝福",
        "tags": ["北京", "白羊座"],
        "follower_count": 1000,
        "publish_count": 88,
        "like_count": 9000,
        "update_timestamp": 1780904037000,
    }
    client = _client([_response(200, {"code": 0, "data": {"data": account}})])

    result = client.fetch_account_info("3xfkwajatdh7p7i", is_cache=False)

    assert client.http_client.requests[0]["url"].endswith("/crawler/kuai_shou/account_info")
    assert client.http_client.requests[0]["json"] == {
        "account_id": "3xfkwajatdh7p7i",
        "is_cache": False,
    }
    assert result["platform_author_id"] == "3xfkwajatdh7p7i"
    assert result["author_display_name"] == "祝福账号"
    assert result["profile_snapshot"]["tags"] == ["北京", "白羊座"]
    assert "tags" not in {key for key in result if key != "profile_snapshot"}


def test_kuaishou_fetch_author_works_maps_canonical_video_items():
    client = _client([
        _response(200, {"code": 0, "data": {"data": [_item("ks_author_001")], "has_more": True, "next_cursor": "20"}})
    ])

    result = client.fetch_author_works(_author_query())[0]

    assert client.http_client.requests[0]["url"].endswith("/crawler/kuai_shou/blogger")
    assert client.http_client.requests[0]["json"] == {"account_id": "3xfkwajatdh7p7i"}
    assert result["platform"] == "kuaishou"
    assert result["platform_content_id"] == "ks_author_001"
    assert result["platform_author_id"] == "3xfkwajatdh7p7i"
    assert result["platform_content_format"] == "video"
    assert result["play_url"] == "https://v.kwaicdn.test/a.mp4"
    assert result["statistics"]["play_count"] == 12345
    assert result["has_more"] is True
    assert result["next_cursor"] == "20"
    assert result["previous_discovery_step"] == "author_works"
    assert result["content_metadata_source"] == "kuaishou_blogger"


def test_kuaishou_http_429_maps_to_platform_rate_limited():
    client = _client([_response(429, {"error": "too many"})])

    with pytest.raises(ContentAgentError) as exc_info:
        client.search(_query())

    assert exc_info.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED
    assert exc_info.value.detail["status_code"] == 429


def test_kuaishou_bad_json_is_sanitized():
    client = CrawapiKuaishouClient(
        base_url="http://crawler.test",
        http_client=FakeHttpClient(
            [
                httpx.Response(
                    200,
                    content=b"not json",
                    request=httpx.Request("POST", "http://crawler.test/x"),
                )
            ]
        ),
    )

    with pytest.raises(RuntimeError, match="keyword_search failed: bad_json"):
        client.search(_query())