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,
    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),
        )