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

"""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"