"""M2 离线测试:用 5 个真实 capture 验证 parse_detail_response 字段映射。""" from __future__ import annotations import json from pathlib import Path import pytest from acquisition.crawler import ( PLATFORMS, detect_platform_and_id, parse_content_id, parse_detail_response, ) 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")) @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"