Create-knowledge-find-decode/tests/test_platform_adapters.py

from __future__ import annotations

from types import SimpleNamespace

from acquisition.platforms import douyin as douyin_module
from acquisition.platforms import weixin as weixin_module
from acquisition.platforms import xiaohongshu as xhs_module
from acquisition.platforms.base import PlatformCandidate
from acquisition.platforms.douyin import DouyinAdapter
from acquisition.platforms.weixin import WeixinAdapter
from acquisition.platforms.xiaohongshu import XiaohongshuAdapter
from acquisition.search import SearchHit, SearchPage
from core.models import Post


def test_xiaohongshu_adapter_maps_search_and_detail(monkeypatch):
    settings = SimpleNamespace(search_content_type="图文")

    def fake_search(query, *, cursor, content_type, limit, settings, rate_limiter):
        assert query == "脚本 开头"
        assert cursor == ""
        assert content_type is None
        assert limit == 3
        return SearchPage(
            rows=[
                {
                    "id": "xhs-1",
                    "url": "https://xhs.test/1",
                    "title": "脚本开头",
                    "nick_name": "作者 A",
                    "cover_url": "https://img.test/cover.jpg",
                }
            ],
            raw_count=1,
            cursor=cursor,
        )

    def fake_fetch(source_id, *, settings, rate_limiter):
        assert source_id == "xhs-1"
        return Post(
            id="xhs_xhs-1",
            platform="xiaohongshu",
            url="https://xhs.test/detail/1",
            content_id="xhs-1",
            title="详情标题",
            content_type="图文",
            body_text="详情正文",
            image_urls=["https://img.test/1.jpg"],
            author_name="详情作者",
            raw={"source": "mock"},
        )

    monkeypatch.setattr(xhs_module, "search_xiaohongshu_page", fake_search)
    monkeypatch.setattr(xhs_module, "fetch_post_detail", fake_fetch)

    adapter = XiaohongshuAdapter()
    candidate = adapter.search("脚本 开头", settings=settings, limit=3, rate_limiter=None)[0]
    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)

    assert candidate.source_id == "xhs-1"
    assert candidate.author == "作者 A"
    assert candidate.raw["page_index"] == 1
    assert candidate.raw["page_rank"] == 1
    assert item.title == "详情标题"
    assert item.image_urls == ["https://img.test/1.jpg"]
    assert item.content_mode == "image_post"
    assert item.raw["source"] == "mock"


def test_xiaohongshu_adapter_search_pages_uses_next_cursor(monkeypatch):
    settings = SimpleNamespace(search_content_type="图文")
    cursors: list[str] = []

    def fake_search(query, *, cursor, content_type, limit, settings, rate_limiter):
        cursors.append(cursor)
        if cursor == "":
            return SearchPage(
                rows=[{"id": "xhs-1", "url": "https://xhs.test/1", "title": "第一页"}],
                raw_count=1,
                cursor=cursor,
                next_cursor="2",
                has_more=True,
            )
        return SearchPage(
            rows=[{"id": "xhs-2", "url": "https://xhs.test/2", "title": "第二页"}],
            raw_count=1,
            cursor=cursor,
            has_more=False,
        )

    monkeypatch.setattr(xhs_module, "search_xiaohongshu_page", fake_search)

    pages = list(
        XiaohongshuAdapter().search_pages(
            "脚本",
            settings=settings,
            limit=2,
            rate_limiter=None,
            max_pages=2,
        )
    )

    assert cursors == ["", "2"]
    assert [page.candidates[0].source_id for page in pages] == ["xhs-1", "xhs-2"]
    assert pages[1].candidates[0].raw["page_index"] == 2


def test_weixin_adapter_hashes_url_and_reads_detail(monkeypatch):
    settings = SimpleNamespace()

    def fake_search(query, *, cursor, limit, settings, rate_limiter):
        assert query == "公众号 选题"
        assert cursor == "0"
        assert limit == 2
        return SearchPage(
            rows=[
                {
                    "url": "https://mp.weixin.qq.com/s/abc",
                    "title": "公众号选题",
                    "nick_name": "公众号",
                    "cover_url": "https://img.test/wx.jpg",
                }
            ],
            raw_count=1,
            cursor=cursor,
        )

    def fake_detail(url, *, settings, rate_limiter):
        assert url == "https://mp.weixin.qq.com/s/abc"
        return "公众号正文", ["https://img.test/wx-1.jpg"]

    monkeypatch.setattr(weixin_module, "search_weixin_page", fake_search)
    monkeypatch.setattr(weixin_module, "fetch_weixin_detail", fake_detail)

    adapter = WeixinAdapter()
    candidate = adapter.search("公众号 选题", settings=settings, limit=2, rate_limiter=None)[0]
    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)

    assert candidate.platform == "weixin"
    assert len(candidate.source_id) == 16
    assert candidate.raw["page_index"] == 1
    assert item.body_text == "公众号正文"
    assert item.image_urls == ["https://img.test/wx-1.jpg"]
    assert item.content_mode == "article"


def test_weixin_adapter_search_pages_uses_next_cursor(monkeypatch):
    settings = SimpleNamespace()
    cursors: list[str] = []

    def fake_search(query, *, cursor, limit, settings, rate_limiter):
        cursors.append(cursor)
        if cursor == "0":
            return SearchPage(
                rows=[{"url": "https://mp.weixin.qq.com/s/a", "title": "第一页"}],
                raw_count=1,
                cursor=cursor,
                next_cursor="1",
                has_more=True,
            )
        return SearchPage(
            rows=[{"url": "https://mp.weixin.qq.com/s/b", "title": "第二页"}],
            raw_count=1,
            cursor=cursor,
            has_more=False,
        )

    monkeypatch.setattr(weixin_module, "search_weixin_page", fake_search)

    pages = list(
        WeixinAdapter().search_pages(
            "公众号",
            settings=settings,
            limit=2,
            rate_limiter=None,
            max_pages=2,
        )
    )

    assert cursors == ["0", "1"]
    assert len(pages) == 2
    assert pages[1].candidates[0].raw["page_index"] == 2


def test_douyin_adapter_maps_video_search_and_detail(monkeypatch):
    settings = SimpleNamespace()

    def fake_search(query, *, content_type, limit, settings, rate_limiter):
        assert query == "短视频 讲法"
        assert content_type == "视频"
        assert limit == 1
        return [
            SearchHit(
                content_id="dy-1",
                provider="piaoquantv",
                raw={"aweme_id": "dy-1"},
                request_payload={"keyword": query},
            )
        ]

    def fake_fetch(source_id, *, platform, provider, settings, rate_limiter):
        assert source_id == "dy-1"
        assert platform == "douyin"
        assert provider == "piaoquantv"
        return Post(
            id="dy_dy-1",
            platform="douyin",
            provider=provider,
            url="https://douyin.test/1",
            content_id="dy-1",
            title="短视频讲法",
            content_type="video",
            body_text="视频文案",
            video_urls=["https://video.test/1.mp4"],
            raw={"source": "mock"},
        )

    monkeypatch.setattr(douyin_module, "search_douyin_hits", fake_search)
    monkeypatch.setattr(douyin_module, "fetch_post_detail", fake_fetch)

    adapter = DouyinAdapter()
    candidate = adapter.search("短视频 讲法", settings=settings, limit=1, rate_limiter=None)[0]
    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)

    assert candidate.source_id == "dy-1"
    assert candidate.provider == "piaoquantv"
    assert candidate.raw["search_provider"] == "piaoquantv"
    assert candidate.raw["original"] == {"aweme_id": "dy-1"}
    assert item.content_type == "video"
    assert item.provider == "piaoquantv"
    assert item.content_mode == "video_post"
    assert item.video_urls == ["https://video.test/1.mp4"]


def test_douyin_adapter_search_pages_preserves_provider_and_page_metadata(monkeypatch):
    settings = SimpleNamespace()
    cursors_seen: list[dict[str, str]] = []

    def fake_search_page(query, *, provider, cursors, content_type, limit, settings, rate_limiter):
        cursors_seen.append(dict(cursors))
        return SearchPage(
            rows=[
                SearchHit(
                    content_id="dy-1",
                    provider="aiddit",
                    raw={"aweme_id": "dy-1"},
                    request_payload={"keyword": query},
                )
            ],
            raw_count=1,
            cursor="",
            provider="aiddit",
            request_payload={"keyword": query},
            raw_metadata={"cursors": {"aiddit": "20"}},
        )

    monkeypatch.setattr(douyin_module, "search_douyin_page", fake_search_page)

    page = next(
        iter(
            DouyinAdapter().search_pages(
                "短视频",
                settings=settings,
                limit=1,
                rate_limiter=None,
                max_pages=1,
            )
        )
    )

    assert cursors_seen == [{}]
    assert page.provider == "aiddit"
    assert page.candidates[0].provider == "aiddit"
    assert page.candidates[0].raw["page_index"] == 1
    assert page.candidates[0].raw["search_provider"] == "aiddit"


def test_douyin_adapter_falls_back_to_image_post(monkeypatch):
    settings = SimpleNamespace()

    def fake_search(query, *, content_type, limit, settings, rate_limiter):
        assert content_type == "视频"
        return [SearchHit(content_id="dy-img-1", provider="aiddit")]

    def fake_fetch(source_id, *, platform, provider, settings, rate_limiter):
        assert provider == "aiddit"
        return Post(
            id="dy_dy-img-1",
            platform="douyin",
            provider=provider,
            url="https://douyin.test/img/1",
            content_id="dy-img-1",
            title="图片内容",
            content_type="image",
            body_text="图片正文",
            image_urls=["https://img.test/dy.jpg"],
            raw={"source": "mock"},
        )

    monkeypatch.setattr(douyin_module, "search_douyin_hits", fake_search)
    monkeypatch.setattr(douyin_module, "fetch_post_detail", fake_fetch)

    item = DouyinAdapter().fetch_detail(
        DouyinAdapter().search("图片", settings=settings, limit=1, rate_limiter=None)[0],
        settings=settings,
        rate_limiter=None,
    )

    assert item.content_mode == "image_post"
    assert item.provider == "aiddit"
    assert item.image_urls == ["https://img.test/dy.jpg"]


def test_douyin_adapter_requires_provider_for_detail():
    try:
        DouyinAdapter().fetch_detail(
            PlatformCandidate(rank=1, platform="douyin", source_id="dy-1"),
            settings=SimpleNamespace(),
            rate_limiter=None,
        )
    except RuntimeError as exc:
        assert "missing search provider" in str(exc)
    else:
        raise AssertionError("missing provider should fail")