"""M2 离线测试：用 5 个真实 capture 验证 parse_detail_response 字段映射。""" from __future__ import annotations import json from pathlib import Path import pytest from acquisition.crawler import ( PLATFORMS, RateLimiter, detect_platform_and_id, fetch_post_detail, parse_content_id, parse_detail_response, ) from core.config import PgConfig, Settings FIXTURES = Path(__file__).parent / "fixtures" # content_id -> (期望作者, 标题里应出现的子串) EXPECTED = { "67e4bdf50000000006028a59": ("海狸教自媒体运营", "短视频脚本"), "698481e1000000000a02a7c1": ("Irvin是个编剧（接稿版）", "叙事结构"), "67e2e39b0000000003028ff0": ("拾意", "剧本创作"), "699308fa0000000016009697": ("方圆的增长飞轮", "选题"), "680659e8000000001a007a11": ("故事设计原理拆解学习"[:2], "故事设计"), } def _load(content_id: str) -> dict: return json.loads((FIXTURES / f"xhs_case_{content_id}.json").read_text("utf-8")) def _settings() -> Settings: return Settings( pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"), aiddit_crawler_base_url="http://crawler.test", crawler_timeout=30, openrouter_timeout_seconds=90, openrouter_model="m", openrouter_base_url="b", openrouter_api_key="k", llm_model="m", max_cards=12, frames_dir="f", douyin_ratio="540p", data_dir="", ) class _Resp: def __init__(self, payload): self._p = payload def raise_for_status(self): return None def json(self): return self._p class _FakeClient: def __init__(self, payload): self.payload = payload self.calls = [] self.closed = False def post(self, url, json=None, headers=None, timeout=None): self.calls.append({"url": url, "body": json}) return _Resp(self.payload) def close(self): self.closed = True class _RecordingLimiter: def __init__(self): self.buckets = [] def wait(self, bucket: str): self.buckets.append(bucket) @pytest.mark.parametrize("content_id", list(EXPECTED)) def test_parse_detail_fields(content_id: str): # 重抓的 xhs_case_*.json 即原始响应 {code,msg,data}，无 response 包裹 response = _load(content_id) post = parse_detail_response(response, fallback_content_id=content_id) assert post.id == f"xhs_{content_id}" assert post.platform == "xiaohongshu" assert post.content_id == content_id assert content_id in post.url assert post.title, "title 不应为空" _, title_sub = EXPECTED[content_id] assert title_sub in post.title assert post.raw.get("code") == 0 def test_authors(): for cid, (author, _) in EXPECTED.items(): post = parse_detail_response(_load(cid), fallback_content_id=cid) if cid in ("67e4bdf50000000006028a59", "698481e1000000000a02a7c1", "67e2e39b0000000003028ff0", "699308fa0000000016009697"): assert post.author_name == author def test_shiyi_posts_text_sparse(): """拾意两条正文几乎只有话题串——M3 多模态提取存在的理由。""" for cid in ("67e2e39b0000000003028ff0", "680659e8000000001a007a11"): post = parse_detail_response(_load(cid), fallback_content_id=cid) stripped = post.body_text.replace("#", "").replace("话题", "").strip() assert len(stripped) < 40, f"{cid} body_text 应很稀疏: {post.body_text!r}" def test_parse_content_id_from_url(): url = ("https://www.xiaohongshu.com/explore/67e4bdf50000000006028a59" "?xsec_token=ABC=&xsec_source=pc_like") assert parse_content_id(url) == "67e4bdf50000000006028a59" assert parse_content_id("67e4bdf50000000006028a59") == "67e4bdf50000000006028a59" @pytest.mark.parametrize("url,platform,cid", [ # 长链 ("https://www.xiaohongshu.com/explore/67e4bdf50000000006028a59?xsec_token=A", "xiaohongshu", "67e4bdf50000000006028a59"), ("https://www.xiaohongshu.com/discovery/item/680659e8000000001a007a11", "xiaohongshu", "680659e8000000001a007a11"), ("https://www.douyin.com/video/7612631899479648866", "douyin", "7612631899479648866"), ("https://www.douyin.com/search/x?modal_id=7612631899479648866&type=general", "douyin", "7612631899479648866"), ("https://www.gifshow.com/fw/photo/3xepqgwddgbickc", "kuaishou", "3xepqgwddgbickc"), ("https://www.kuaishou.com/short-video/3xepqgwddgbickc", "kuaishou", "3xepqgwddgbickc"), ("https://www.bilibili.com/video/BV1qd4y1x76w/?spm_id_from=333", "bilibili", "BV1qd4y1x76w"), # 裸 id 兜底 ("67e4bdf50000000006028a59", "xiaohongshu", "67e4bdf50000000006028a59"), ("7612631899479648866", "douyin", "7612631899479648866"), ("BV1qd4y1x76w", "bilibili", "BV1qd4y1x76w"), ]) def test_detect_platform_and_id(url, platform, cid): assert detect_platform_and_id(url) == (platform, cid) def test_parse_detail_platform_prefix(): """同一 parse 函数 + platform 参数 → 平台名与 id 前缀正确。""" response = _load("67e4bdf50000000006028a59") for platform, expected_prefix in [(p, c["prefix"]) for p, c in PLATFORMS.items()]: post = parse_detail_response(response, platform=platform, fallback_content_id="67e4bdf50000000006028a59") assert post.platform == platform assert post.id == f"{expected_prefix}_67e4bdf50000000006028a59" def test_fetch_douyin_detail_piaoquantv_body_and_provider(): payload = {"code": 0, "data": {"data": { "channel_content_id": "761", "content_link": "https://douyin.test/video/761", "content_type": "video", "title": "视频", "video_url_list": [{"video_url": "https://video.test/761.mp4"}], }}} client = _FakeClient(payload) limiter = _RecordingLimiter() post = fetch_post_detail( "761", platform="douyin", provider="piaoquantv", settings=_settings(), http_client=client, rate_limiter=limiter, ) assert post.provider == "piaoquantv" assert post.raw["detail_provider"] == "piaoquantv" assert client.calls[0]["url"].startswith("http://crawapi.piaoquantv.com/") assert client.calls[0]["body"]["content_id"] == "761" assert client.calls[0]["body"]["account_id"] == "7450041106378522636" assert client.calls[0]["body"]["cookie_batch"] == "default" assert limiter.buckets == ["douyin"] def test_fetch_douyin_detail_aiddit_body_and_provider(): payload = {"code": 0, "data": {"data": { "channel_content_id": "762", "content_link": "https://douyin.test/video/762", "content_type": "video", "title": "视频", "video_url_list": [{"video_url": "https://video.test/762.mp4"}], }}} client = _FakeClient(payload) post = fetch_post_detail( "762", platform="douyin", provider="aiddit", settings=_settings(), http_client=client, rate_limiter=_RecordingLimiter(), ) assert post.provider == "aiddit" assert client.calls[0]["url"].startswith("http://crawler.test/") assert client.calls[0]["body"] == {"content_id": "762"} def test_fetch_douyin_detail_rejects_unknown_provider(): with pytest.raises(RuntimeError): fetch_post_detail( "762", platform="douyin", provider="other", settings=_settings(), http_client=_FakeClient({}), rate_limiter=RateLimiter(min_interval_seconds=0.0), )