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

"""V3-M1C: 视频号 client search/normalization/retry/blocked tests."""

from __future__ import annotations

import httpx
import pytest

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.shipinhao import (
    SHIPINHAO_SEARCH_MAX_INTERVAL_SECONDS,
    SHIPINHAO_SEARCH_MIN_INTERVAL_SECONDS,
    CrawapiShipinhaoClient,
)


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})
        return self.responses.pop(0)


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": "彩虹",
        "discovery_start_source": "pattern_itemset",
    }


def _client(responses):
    sleeps: list[float] = []
    client = CrawapiShipinhaoClient(
        base_url="http://crawler.test",
        http_client=FakeHttpClient(responses),
        sleep_fn=sleeps.append,
    )
    return client, sleeps


_SUCCESS = {
    "code": 0,
    "data": {
        "has_more": True,
        "next_cursor": 12,
        "data": [
            {
                "channel_content_id": "finderobj_abc",
                "title": "圆形彩虹 #彩虹 #见者好运",
                "content_type": "video",
                "video_url_list": [{"video_url": "https://findermp.video.qq.com/x"}],
                "channel_account_id": "acc_123",
                "channel_account_name": "掌上巴彦淖尔",
                "like_count": 92,
                "publish_timestamp": 1780904037000,
            }
        ],
    },
}
_FAIL_25011 = {"code": 25011, "msg": "视频号接口异常: 获取搜索结果失败", "data": None}


def test_shipinhao_search_maps_canonical_fields():
    client, _ = _client([_response(200, _SUCCESS)])
    result = client.search(_query())[0]
    assert result["platform"] == "shipinhao"
    assert result["platform_content_id"] == "finderobj_abc"
    assert result["platform_author_id"] == "acc_123"
    assert result["author_display_name"] == "掌上巴彦淖尔"
    assert result["tags"] == ["#彩虹", "#见者好运"]
    assert result["play_url"] == "https://findermp.video.qq.com/x"
    assert result["statistics"]["digg_count"] == 92
    assert result["create_time"] == 1780904037  # ms -> s
    assert result["has_more"] is True
    assert result["next_cursor"] == "12"


def test_shipinhao_search_falls_back_to_title_research_for_same_content_id():
    first = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_need_retry",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
                    "channel_account_id": "acc_123",
                    "channel_account_name": "掌上巴彦淖尔",
                }
            ]
        },
    }
    second = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_need_retry",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "video_url_list": [{"video_url": "https://findermp.video.qq.com/video.mp4"}],
                    "channel_account_id": "acc_123",
                    "channel_account_name": "掌上巴彦淖尔",
                }
            ]
        },
    }
    client, _ = _client([_response(200, first), _response(200, second)])

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

    assert client.http_client.requests[1]["json"] == {"keyword": "当狗唱dj #心墙", "cursor": ""}
    assert result["play_url"] == "https://findermp.video.qq.com/video.mp4"
    assert result["platform_raw_payload"]["selected_video_url_path"] == "$.research.video_url_list[0].video_url"
    assert result["platform_raw_payload"]["shipinhao_research_status"] == "used"
    assert "media_failure_reason" not in result


def test_shipinhao_metadata_search_does_not_probe_or_title_research():
    probe_calls: list[str] = []
    first = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_metadata",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
                }
            ]
        },
    }
    client = CrawapiShipinhaoClient(
        base_url="http://crawler.test",
        http_client=FakeHttpClient([_response(200, first)]),
        sleep_fn=lambda seconds: None,
        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"] == "finderobj_metadata"
    assert result["media_failure_reason"] == "no_valid_play_url"
    assert "_platform_search_item" in result


def test_shipinhao_hydrate_search_result_media_uses_title_research():
    first = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_hydrate",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
                }
            ]
        },
    }
    second = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_hydrate",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "video_url_list": [{"video_url": "https://findermp.video.qq.com/video.mp4"}],
                }
            ]
        },
    }
    client, _ = _client([_response(200, first), _response(200, second)])
    [metadata] = client.search_full_page_metadata(_query())

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

    assert client.http_client.requests[1]["json"] == {"keyword": "当狗唱dj #心墙", "cursor": ""}
    assert hydrated["play_url"] == "https://findermp.video.qq.com/video.mp4"
    assert hydrated["platform_raw_payload"]["shipinhao_research_status"] == "used"
    assert "_platform_search_item" not in hydrated


def test_shipinhao_search_keeps_no_valid_when_title_research_does_not_match():
    first = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_need_retry",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "image_url_list": [{"image_url": "https://findermp.video.qq.com/cover.jpg"}],
                }
            ]
        },
    }
    second = {
        "code": 0,
        "data": {
            "data": [
                {
                    "channel_content_id": "finderobj_other",
                    "title": "当狗唱dj #心墙",
                    "content_type": "video",
                    "video_url_list": [{"video_url": "https://findermp.video.qq.com/video.mp4"}],
                }
            ]
        },
    }
    client, _ = _client([_response(200, first), _response(200, second)])

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


def test_shipinhao_search_default_limit_is_five():
    items = [
        {
            "channel_content_id": f"finderobj_{index}",
            "title": "圆形彩虹",
            "content_type": "video",
        }
        for index in range(6)
    ]
    success = {"code": 0, "data": {"data": items}}
    client, _ = _client([_response(200, success)])

    results = client.search(_query())

    assert [result["platform_content_id"] for result in results] == [
        "finderobj_0",
        "finderobj_1",
        "finderobj_2",
        "finderobj_3",
        "finderobj_4",
    ]


def test_shipinhao_search_full_page_bypasses_local_limit():
    items = [
        {
            "channel_content_id": f"finderobj_{index}",
            "title": "圆形彩虹",
            "content_type": "video",
        }
        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] == [
        "finderobj_0",
        "finderobj_1",
        "finderobj_2",
    ]


def test_shipinhao_search_retries_on_25011_then_succeeds():
    client, sleeps = _client([_response(200, _FAIL_25011), _response(200, _SUCCESS)])
    result = client.search(_query())
    assert len(result) == 1
    assert sleeps == [1]  # one backoff before the successful retry


def test_shipinhao_search_does_not_retry_empty_result():
    empty = {"code": 0, "data": {"has_more": False, "next_cursor": "", "data": []}}
    client, sleeps = _client([_response(200, empty)])
    assert client.search(_query()) == []
    assert sleeps == []


def test_shipinhao_search_raises_after_exhausted():
    client, sleeps = _client([_response(200, _FAIL_25011) for _ in range(3)])
    with pytest.raises(ContentAgentError) as exc:
        client.search(_query())
    assert exc.value.error_code == ErrorCode.PLATFORM_REQUEST_FAILED
    assert sleeps == [1, 2]  # backoff before attempts 2 and 3


def test_shipinhao_fetch_author_works_blocked_returns_empty():
    client, _ = _client([])
    assert client.fetch_author_works({"platform_author_id": "acc_123"}) == []
    assert client.http_client.requests == []


def test_shipinhao_from_env_reads_limit_override(monkeypatch, tmp_path):
    monkeypatch.setenv("CONTENTFIND_API_CRAWAPI_BASE_URL", "http://crawler.test")
    monkeypatch.setenv("CONTENTFIND_SHIPINHAO_MAX_RESULTS_PER_QUERY", "2")

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

    assert client.max_results_per_query == 2


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

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

    assert client.max_results_per_query == 5


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

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

    assert client.rate_limiter is not None
    assert client.rate_limiter.min_interval_seconds == SHIPINHAO_SEARCH_MIN_INTERVAL_SECONDS
    assert client.rate_limiter.max_interval_seconds == SHIPINHAO_SEARCH_MAX_INTERVAL_SECONDS


def test_shipinhao_account_info_is_not_reliable_capability():
    client, _ = _client([])

    assert not hasattr(client, "fetch_account_info")