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_filters_duplicate_and_decorative_gif_images(monkeypatch): settings = SimpleNamespace() def fake_detail(url, *, settings, rate_limiter): return "公众号正文", [ "https://mmbiz.qpic.cn/mmbiz_png/a/640?wx_fmt=png", "https://mmbiz.qpic.cn/mmbiz_png/a/640?wx_fmt=png", "https://mmbiz.qpic.cn/mmbiz_png/b/640?wx_fmt=png", "https://mmbiz.qpic.cn/mmbiz_gif/c/640?wx_fmt=gif", "https://mmbiz.qpic.cn/mmbiz_gif/d/640", ] monkeypatch.setattr(weixin_module, "fetch_weixin_detail", fake_detail) candidate = PlatformCandidate( rank=1, platform="weixin", source_id="wx1", url="https://mp.weixin.qq.com/s/abc", title="公众号选题", cover_url="https://img.test/cover.jpg", ) item = WeixinAdapter().fetch_detail(candidate, settings=settings, rate_limiter=None) assert item.image_urls == [ "https://mmbiz.qpic.cn/mmbiz_png/a/640?wx_fmt=png", "https://mmbiz.qpic.cn/mmbiz_png/b/640?wx_fmt=png", ] 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")