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

import httpx
import pytest

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.douyin import (
    DOUYIN_SEARCH_MAX_INTERVAL_SECONDS,
    DOUYIN_SEARCH_MIN_INTERVAL_SECONDS,
    RAW_AUTHOR_ACCOUNT_KEY,
    RAW_AUTHOR_ID_KEY,
    RAW_CONTENT_ID_KEY,
    CrawapiDouyinClient,
    RateLimiter,
)


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://crawapi.test/endpoint"),
    )


def _client(responses, rate_limiter=None, fallback_base_url="", **overrides):
    options = {
        "base_url": "http://crawapi.test",
        "fallback_base_url": fallback_base_url,
        "keyword_path": "/crawler/dou_yin/keyword",
        "blogger_path": "/crawler/dou_yin/blogger",
        "default_crawapi_account_ref": "771431222",
        "http_client": FakeHttpClient(responses),
        "rate_limiter": rate_limiter,
    }
    options.update(overrides)
    return CrawapiDouyinClient(**options)


def _search_query(text="早上好祝福视频"):
    return {
        "search_query_id": "q_001",
        "search_query": text,
        "discovery_start_source": "pattern_itemset",
    }


def test_douyin_keyword_search_maps_content_fields():
    client = _client(
        [
            _response(
                200,
                {
                    "data": {
                        "data": [
                            {
                                RAW_CONTENT_ID_KEY: "7615247738577423622",
                                "desc": "早睡早起早上好",
                                "author": {
                                    "nickname": "让你心情变浅的兔子",
                                    RAW_AUTHOR_ID_KEY: "MS4wLjABAAAAdYc",
                                },
                                "statistics": {
                                    "digg_count": 1931,
                                    "comment_count": 45,
                                    "share_count": 1968,
                                    "collect_count": 462,
                                },
                                "text_extra": [{"hashtag_name": "早上好"}],
                            }
                        ],
                        "has_more": True,
                        "next_cursor": "10",
                    }
                },
            ),
        ]
    )

    results = client.search(_search_query())

    assert len(results) == 1
    result = results[0]
    assert result["content_discovery_id"] == "q_001_content_001"
    assert result["platform_content_id"] == "7615247738577423622"
    assert result["platform_author_id"] == "MS4wLjABAAAAdYc"
    assert result["tags"] == ["#早上好"]
    assert result["has_more"] is True
    assert result["next_cursor"] == "10"
    assert result["discovery_relation"] == "derived_from_pattern_demand"
    assert result["platform_auth_mode"] == "no_bearer"
    assert result["platform_raw_payload"][RAW_CONTENT_ID_KEY] == "7615247738577423622"
    assert client.http_client.requests[0]["json"][RAW_AUTHOR_ACCOUNT_KEY] == "771431222"
    # V3 清理: 画像调用链已砍,搜索一条内容只发 1 次 keyword 请求,不再追加画像请求。
    assert len(client.http_client.requests) == 1
    assert client.http_client.requests[0]["url"].endswith("/crawler/dou_yin/keyword")


def test_douyin_keyword_search_posts_piaoquantv_apifox_payload():
    client = _client([_response(200, {"data": {"data": [], "has_more": False}})])

    client.search(_search_query("宁艺卓"))

    assert client.http_client.requests[0]["json"] == {
        RAW_AUTHOR_ACCOUNT_KEY: "771431222",
        "keyword": "宁艺卓",
        "content_type": "视频",
        "sort_type": "综合排序",
        "publish_time": "不限",
        "duration": "不限",
        "cursor": "0",
        "cookie_batch": "default",
    }


def test_douyin_keyword_search_returns_empty_list():
    client = _client([_response(200, {"data": {"data": [], "has_more": False}})])

    results = client.search(_search_query("无结果"))

    assert results == []


def test_douyin_keyword_search_uses_explicit_page_cursor():
    client = _client([_response(200, {"data": {"data": [], "has_more": False}})])

    client.search({**_search_query("下一页"), "page_cursor": "20"})

    assert client.http_client.requests[0]["json"]["cursor"] == "20"


def test_douyin_fetch_author_works_maps_fake_response():
    client = _client(
        [
            _response(
                200,
                {
                    "data": {
                        "data": [
                            {
                                RAW_CONTENT_ID_KEY: "7615247738577423001",
                                "desc": "作者作品",
                                "author": {
                                    "nickname": "作者",
                                    RAW_AUTHOR_ID_KEY: "MS4wLjABAAAA001",
                                },
                                "statistics": {"digg_count": 100},
                            }
                        ],
                        "has_more": False,
                    }
                },
            ),
        ]
    )

    results = client.fetch_author_works(
        {
            "search_query_id": "author_001",
            "search_query": "作者作品",
            "platform_author_id": "MS4wLjABAAAA001",
            "discovery_start_source": "pattern_itemset",
        }
    )

    # M5A 受控变化: 作者作品改打 blogger 接口,payload 用 account_id 三字段合同。
    assert results[0]["search_query_id"] == "author_001"
    assert results[0]["previous_discovery_step"] == "author_works"
    assert client.http_client.requests[0]["json"][RAW_AUTHOR_ACCOUNT_KEY] == "MS4wLjABAAAA001"


def test_douyin_non_keyword_requests_keep_aiddit_legacy_base_url():
    client = _client(
        [
            _response(
                200,
                {
                    "data": {
                        "data": [
                            {
                                RAW_CONTENT_ID_KEY: "7615247738577423001",
                                "desc": "作者作品",
                                "author": {RAW_AUTHOR_ID_KEY: "MS4wLjABAAAA001"},
                            }
                        ],
                        "has_more": False,
                    }
                },
            ),
        ],
        base_url="http://crawapi.piaoquantv.com",
        fallback_base_url="http://crawler.aiddit.com",
    )

    client.fetch_author_works(
        {
            "search_query_id": "author_001",
            "search_query": "作者作品",
            "platform_author_id": "MS4wLjABAAAA001",
            "discovery_start_source": "pattern_itemset",
        }
    )

    assert client.http_client.requests[0]["url"].startswith("http://crawler.aiddit.com/")
    assert client.http_client.requests[0]["url"].endswith("/crawler/dou_yin/blogger")
    assert len(client.http_client.requests) == 1


def test_douyin_keyword_search_http_error_is_sanitized():
    client = _client([_response(500, {"error": "server failed"})])

    with pytest.raises(RuntimeError, match="keyword_search failed: HTTP 500"):
        client.search(_search_query("接口失败"))


def test_douyin_keyword_search_business_error_is_failed():
    client = _client([_response(200, {"code": 22001, "msg": "强制登录", "data": None})])

    with pytest.raises(RuntimeError, match="keyword_search failed: business_error"):
        client.search(_search_query("账号态失败"))


def test_douyin_keyword_search_falls_back_to_candidate_on_unknown_business_error():
    client = _client(
        [
            _response(200, {"code": 10000, "msg": "未知错误", "data": None}),
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
        ],
        fallback_base_url="http://candidate.test",
    )

    assert client.search(_search_query("主站业务错误")) == []
    assert [request["url"] for request in client.http_client.requests] == [
        "http://crawapi.test/crawler/dou_yin/keyword",
        "http://candidate.test/crawler/dou_yin/keyword",
    ]


def test_douyin_keyword_search_falls_back_to_crawler_payload_on_force_login():
    client = _client(
        [
            _response(200, {"code": 22001, "msg": "抖音搜索异常: 强制登录", "data": None}),
            _response(
                200,
                {
                    "code": 0,
                    "data": {
                        "data": [
                            {
                                RAW_CONTENT_ID_KEY: "7615247738577423622",
                                "desc": "国风传统",
                                "author": {
                                    "nickname": "作者",
                                    RAW_AUTHOR_ID_KEY: "MS4wLjABAAAAdYc",
                                },
                                "statistics": {
                                    "digg_count": 10,
                                    "comment_count": 2,
                                    "share_count": 3,
                                    "collect_count": 4,
                                },
                            }
                        ],
                        "has_more": True,
                        "next_cursor": "12",
                    },
                },
            ),
        ],
        fallback_base_url="http://crawler.aiddit.com",
    )

    results = client.search(_search_query("国风传统"))

    assert client.http_client.requests[0]["json"] == {
        RAW_AUTHOR_ACCOUNT_KEY: "771431222",
        "keyword": "国风传统",
        "content_type": "视频",
        "sort_type": "综合排序",
        "publish_time": "不限",
        "duration": "不限",
        "cursor": "0",
        "cookie_batch": "default",
    }
    assert client.http_client.requests[1]["json"] == {
        "content_type": "视频",
        "keyword": "国风传统",
        "cursor": "",
        "sort_type": "最多点赞",
    }
    assert results[0]["platform_content_id"] == "7615247738577423622"
    assert results[0]["platform_author_id"] == "MS4wLjABAAAAdYc"
    assert results[0]["statistics"]["digg_count"] == 10
    assert results[0]["has_more"] is True
    assert results[0]["next_cursor"] == "12"


def test_douyin_keyword_search_does_not_fallback_on_parameter_error():
    client = _client(
        [
            _response(200, {"code": 10001, "msg": "参数异常: 参数校验不通过", "data": None}),
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
        ],
        fallback_base_url="http://candidate.test",
    )

    with pytest.raises(RuntimeError, match="business_error code=10001"):
        client.search(_search_query("参数错误"))
    assert len(client.http_client.requests) == 1


def test_douyin_keyword_search_falls_back_to_candidate_on_network_error():
    client = _client(
        [
            httpx.ConnectError("network failed"),
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
        ],
        fallback_base_url="http://candidate.test",
    )

    assert client.search(_search_query("网络失败候选")) == []
    assert [request["url"] for request in client.http_client.requests] == [
        "http://crawapi.test/crawler/dou_yin/keyword",
        "http://candidate.test/crawler/dou_yin/keyword",
    ]


def test_douyin_keyword_search_candidate_uses_same_rate_limit_bucket():
    limiter = FakeRateLimiter()
    client = _client(
        [
            _response(200, {"code": 10000, "msg": "未知错误", "data": None}),
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
        ],
        rate_limiter=limiter,
        fallback_base_url="http://candidate.test",
    )

    client.search(_search_query("候选限流"))

    assert limiter.buckets == ["douyin_search", "douyin_search"]


def test_douyin_keyword_search_keeps_host_attempts_when_candidate_fails():
    client = _client(
        [
            _response(200, {"code": 10000, "msg": "未知错误", "data": None}),
            _response(500, {"error": "server failed"}),
        ],
        fallback_base_url="http://candidate.test",
    )

    with pytest.raises(RuntimeError, match="HTTP 500") as exc_info:
        client.search(_search_query("主备都失败"))
    attempts = exc_info.value.detail["crawapi_host_attempts"]
    assert [attempt["role"] for attempt in attempts] == ["primary", "candidate"]
    assert [attempt["provider"] for attempt in attempts] == ["piaoquantv", "crawler"]
    assert attempts[0]["business_code"] == "10000"
    assert attempts[0]["request_payload_summary"]["duration"] == "不限"
    assert attempts[0]["request_payload_summary"]["cookie_batch"] == "default"
    assert attempts[1]["status_code"] == 500


def test_douyin_keyword_search_network_error_is_sanitized():
    client = _client([httpx.ConnectError("network failed")])

    with pytest.raises(RuntimeError, match="keyword_search failed: network_error"):
        client.search(_search_query("网络失败"))


def test_douyin_keyword_search_bad_json_is_sanitized():
    client = CrawapiDouyinClient(
        base_url="http://crawapi.test",
        keyword_path="/crawler/dou_yin/keyword",
        http_client=FakeHttpClient(
            [
                httpx.Response(
                    200,
                    content=b"not json",
                    request=httpx.Request("POST", "http://crawapi.test/endpoint"),
                )
            ]
        ),
    )

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


def test_douyin_keyword_search_can_limit_results_per_query():
    client = CrawapiDouyinClient(
        base_url="http://crawapi.test",
        keyword_path="/crawler/dou_yin/keyword",
        max_results_per_query=1,
        http_client=FakeHttpClient(
            [
                _response(
                    200,
                    {
                        "data": {
                            "data": [
                                {RAW_CONTENT_ID_KEY: "1", "author": {}, "statistics": {}},
                                {RAW_CONTENT_ID_KEY: "2", "author": {}, "statistics": {}},
                            ]
                        }
                    },
                ),
            ]
        ),
    )

    results = client.search(_search_query("限量"))

    assert [result["platform_content_id"] for result in results] == ["1"]
    assert len(client.http_client.requests) == 1


def test_douyin_keyword_search_default_limit_is_five():
    client = CrawapiDouyinClient(
        base_url="http://crawapi.test",
        keyword_path="/crawler/dou_yin/keyword",
        http_client=FakeHttpClient(
            [
                _response(
                    200,
                    {
                        "data": {
                            "data": [
                                {RAW_CONTENT_ID_KEY: str(index), "author": {}, "statistics": {}}
                                for index in range(6)
                            ]
                        }
                    },
                ),
            ]
        ),
    )

    results = client.search(_search_query("默认限量"))

    assert [result["platform_content_id"] for result in results] == ["0", "1", "2", "3", "4"]


def test_douyin_keyword_search_full_page_bypasses_local_limit():
    client = CrawapiDouyinClient(
        base_url="http://crawapi.test",
        keyword_path="/crawler/dou_yin/keyword",
        max_results_per_query=1,
        http_client=FakeHttpClient(
            [
                _response(
                    200,
                    {
                        "data": {
                            "data": [
                                {RAW_CONTENT_ID_KEY: str(index), "author": {}, "statistics": {}}
                                for index in range(3)
                            ]
                        }
                    },
                ),
            ]
        ),
    )

    results = client.search_full_page(_search_query("整页"))

    assert [result["platform_content_id"] for result in results] == ["0", "1", "2"]


def _author_query(author_id="MS4wLjABAAAA001", **extra):
    return {
        "search_query_id": "author_001",
        "search_query": "作者作品",
        "platform_author_id": author_id,
        "discovery_start_source": "pattern_itemset",
        **extra,
    }


def _blogger_response(items=None, has_more=True, next_cursor="20"):
    return _response(
        200,
        {
            "code": 0,
            "data": {"data": items or [], "has_more": has_more, "next_cursor": next_cursor},
        },
    )


class FakeRateLimiter:
    def __init__(self):
        self.buckets = []

    def wait(self, bucket):
        self.buckets.append(bucket)


def test_fetch_author_works_posts_to_blogger_path():
    client = _client([_blogger_response()])

    client.fetch_author_works(_author_query())

    assert client.http_client.requests[0]["url"].endswith("/crawler/dou_yin/blogger")


def test_fetch_author_works_payload_uses_account_id_from_platform_author_id():
    client = _client([_blogger_response()])

    client.fetch_author_works(_author_query("MS4wLjABAAAA999"))

    payload = client.http_client.requests[0]["json"]
    assert payload == {
        RAW_AUTHOR_ACCOUNT_KEY: "MS4wLjABAAAA999",
        "sort_type": "最新",
        "cursor": "",
    }


def test_fetch_author_works_uses_page_cursor():
    client = _client([_blogger_response()])

    client.fetch_author_works(_author_query(page_cursor="20"))

    assert client.http_client.requests[0]["json"]["cursor"] == "20"


def test_fetch_author_works_normalizes_author_work_fields():
    client = _client(
        [
            _blogger_response(
                items=[
                    {
                        RAW_CONTENT_ID_KEY: "7615247738577423001",
                        "desc": "作者作品",
                        "author": {"nickname": "作者", RAW_AUTHOR_ID_KEY: "MS4wLjABAAAA001"},
                        "statistics": {"digg_count": 100},
                        "create_time": 1733000000,
                    }
                ]
            ),
        ]
    )

    results = client.fetch_author_works(_author_query())

    assert results[0]["platform_content_id"] == "7615247738577423001"
    assert results[0]["platform_author_id"] == "MS4wLjABAAAA001"
    assert results[0]["statistics"]["digg_count"] == 100
    assert results[0]["create_time"] == 1733000000
    assert results[0]["previous_discovery_step"] == "author_works"
    assert results[0]["content_metadata_source"] == "douyin_blogger"
    assert len(client.http_client.requests) == 1


def test_from_env_reads_blogger_path_and_sort_type(monkeypatch, tmp_path):
    monkeypatch.setenv("CONTENTFIND_API_CRAWAPI_BASE_URL", "http://crawapi.test")
    monkeypatch.setenv("CONTENTFIND_API_CRAWAPI_FALLBACK_BASE_URL", "http://candidate.test")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_KEYWORD_PATH", "/crawler/dou_yin/keyword")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_BLOGGER_PATH", "/crawler/dou_yin/blogger")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_ACCOUNT_WORKS_DEFAULT_SORT_TYPE", "最热")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_PIAOQUANTV_ACCOUNT_ID", "7450041106378522636")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_PIAOQUANTV_DURATION", "不限")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_PIAOQUANTV_COOKIE_BATCH", "default")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_CRAWLER_SORT_TYPE", "最多点赞")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_CRAWLER_CURSOR_DEFAULT", "")

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

    assert client.base_url == "http://crawapi.test/"
    assert client.fallback_base_url == "http://candidate.test/"
    assert client.blogger_path == "crawler/dou_yin/blogger"
    assert client.piaoquantv_account_id == "7450041106378522636"
    assert client.piaoquantv_duration == "不限"
    assert client.piaoquantv_cookie_batch == "default"
    assert client.crawler_sort_type == "最多点赞"
    assert client.crawler_cursor_default == ""
    assert client.default_account_works_sort_type == "最热"
    assert client.max_results_per_query == 5
    assert isinstance(client.rate_limiter, RateLimiter)
    assert client.rate_limiter.min_interval_seconds == DOUYIN_SEARCH_MIN_INTERVAL_SECONDS
    assert client.rate_limiter.max_interval_seconds == DOUYIN_SEARCH_MAX_INTERVAL_SECONDS


def test_from_env_reads_max_results_override(monkeypatch, tmp_path):
    monkeypatch.setenv("CONTENTFIND_API_CRAWAPI_BASE_URL", "http://crawapi.test")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_KEYWORD_PATH", "/crawler/dou_yin/keyword")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_BLOGGER_PATH", "/crawler/dou_yin/blogger")
    monkeypatch.setenv("CONTENTFIND_DOUYIN_MAX_RESULTS_PER_QUERY", "2")

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

    assert client.max_results_per_query == 2


def test_rate_limiter_waits_between_keyword_calls():
    clock = {"now": 0.0}
    sleeps = []

    def fake_sleep(seconds):
        sleeps.append(seconds)
        clock["now"] += seconds

    limiter = RateLimiter(min_interval_seconds=30.0, now_fn=lambda: clock["now"], sleep_fn=fake_sleep)

    limiter.wait("douyin_search")
    limiter.wait("douyin_search")

    assert sleeps == [30.0]


def test_douyin_rate_limiter_supports_ten_to_twelve_second_jitter():
    clock = {"now": 0.0}
    sleeps = []

    def fake_sleep(seconds):
        sleeps.append(seconds)
        clock["now"] += seconds

    limiter = RateLimiter(
        min_interval_seconds=DOUYIN_SEARCH_MIN_INTERVAL_SECONDS,
        max_interval_seconds=DOUYIN_SEARCH_MAX_INTERVAL_SECONDS,
        now_fn=lambda: clock["now"],
        sleep_fn=fake_sleep,
        random_fn=lambda lo, hi: lo,
    )

    limiter.wait("douyin_search")
    limiter.wait("douyin_search")

    assert sleeps == [10.0]


def test_search_chain_uses_shared_search_bucket():
    limiter = FakeRateLimiter()
    client = _client(
        [
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
        ],
        rate_limiter=limiter,
    )

    client.search(_search_query("关键词"))
    client.search({**_search_query("关键词"), "page_cursor": "10"})

    assert limiter.buckets == ["douyin_search", "douyin_search"]


def test_blogger_uses_separate_bucket_from_search_chain():
    limiter = FakeRateLimiter()
    client = _client(
        [
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
            _blogger_response(),
        ],
        rate_limiter=limiter,
    )

    client.search(_search_query("关键词"))
    client.fetch_author_works(_author_query())

    assert limiter.buckets == ["douyin_search", "douyin_blogger"]


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

    with pytest.raises(ContentAgentError) as exc_info:
        client.search(_search_query("被限流"))
    assert exc_info.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED
    assert exc_info.value.detail["status_code"] == 429


def test_http_429_does_not_fallback_to_candidate():
    client = _client(
        [
            _response(429, {"error": "too many"}),
            _response(200, {"code": 0, "data": {"data": [], "has_more": False}}),
        ],
        fallback_base_url="http://candidate.test",
    )

    with pytest.raises(ContentAgentError) as exc_info:
        client.search(_search_query("候选也不能绕限流"))
    assert exc_info.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED
    assert len(client.http_client.requests) == 1
    assert exc_info.value.detail["crawapi_host_attempts"][0]["role"] == "primary"


def test_business_rate_limit_code_maps_to_platform_rate_limited(monkeypatch):
    from content_agent.integrations import douyin

    monkeypatch.setattr(douyin, "RATE_LIMIT_BUSINESS_CODES", {"30005"})
    client = _client([_response(200, {"code": 30005, "msg": "ok", "data": None})])

    with pytest.raises(ContentAgentError) as exc_info:
        client.search(_search_query("业务限流"))
    assert exc_info.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED
    assert exc_info.value.detail["business_code"] == "30005"


def test_rate_limit_message_token_maps_to_platform_rate_limited():
    client = _client([_response(200, {"code": 1, "msg": "请求频繁,请稍后再试", "data": None})])

    with pytest.raises(ContentAgentError) as exc_info:
        client.search(_search_query("消息限流"))
    assert exc_info.value.error_code == ErrorCode.PLATFORM_RATE_LIMITED


def test_force_login_without_rate_limit_code_is_not_rate_limited():
    client = _client([_response(200, {"code": 22001, "msg": "强制登录", "data": None})])

    with pytest.raises(RuntimeError, match="business_error"):
        client.search(_search_query("强制登录"))


def test_bad_json_is_not_rate_limited():
    client = _client(
        [
            httpx.Response(
                200, content=b"not json",
                request=httpx.Request("POST", "http://crawapi.test/endpoint"),
            )
        ]
    )

    with pytest.raises(RuntimeError, match="bad_json"):
        client.search(_search_query("坏响应"))


def test_plain_500_is_not_rate_limited():
    client = _client([_response(500, {"error": "server failed"})])

    with pytest.raises(RuntimeError, match="HTTP 500"):
        client.search(_search_query("普通失败"))