Create-knowledge-find-decode/creation_knowledge/integrations/crawler.py

"""帖子详情拉取：小红书 detail 接口。

POST {base}/crawler/xiao_hong_shu/detail  body={"content_id": ...}  无 token，限流 ≥15s。
逻辑（限流 + code==0 信封校验）对齐 ContentFindAgentNew 的 crawapi_http.post_crawapi_json，
但不依赖其错误体系，保持本项目自包含。

拆成两层：parse_detail_response 纯函数（可离线用 fixture 测）+ fetch_post_detail 负责 HTTP。
"""
from __future__ import annotations

import time
from typing import Any, Callable, Optional
from urllib.parse import urljoin

import httpx

from creation_knowledge.config import Settings
from creation_knowledge.models import Card, Post

DETAIL_PATH = "/crawler/xiao_hong_shu/detail"
RATE_LIMIT_SECONDS = 15.0


class CrawlerError(RuntimeError):
    pass


class RateLimiter:
    """同一 bucket 两次调用间隔 ≥ min_interval。对齐 CFA RateLimiter。"""

    def __init__(
        self,
        min_interval_seconds: float = RATE_LIMIT_SECONDS,
        now_fn: Callable[[], float] = time.monotonic,
        sleep_fn: Callable[[float], None] = time.sleep,
    ) -> None:
        self.min_interval_seconds = min_interval_seconds
        self.now_fn = now_fn
        self.sleep_fn = sleep_fn
        self._last: dict[str, float] = {}

    def wait(self, bucket: str) -> None:
        last = self._last.get(bucket)
        if last is not None:
            remaining = self.min_interval_seconds - (self.now_fn() - last)
            if remaining > 0:
                self.sleep_fn(remaining)
        self._last[bucket] = self.now_fn()


def parse_content_id(content_id_or_url: str) -> str:
    """接受裸 content_id 或 https://www.xiaohongshu.com/explore/<id>?... 链接。"""
    s = content_id_or_url.strip()
    if "/explore/" in s:
        s = s.split("/explore/", 1)[1]
    if "/discovery/item/" in content_id_or_url:
        s = content_id_or_url.split("/discovery/item/", 1)[1]
    return s.split("?", 1)[0].strip("/").strip()


def _image_urls(inner: dict) -> list[str]:
    out: list[str] = []
    seen: set[str] = set()
    for x in inner.get("image_url_list") or []:
        u = x.get("image_url") if isinstance(x, dict) else x
        if u and u not in seen:
            seen.add(u)
            out.append(u)
    return out


def _video_urls(inner: dict) -> list[str]:
    out: list[str] = []
    for x in inner.get("video_url_list") or []:
        u = x.get("video_url") if isinstance(x, dict) else x
        if u:
            out.append(u)
    return out


def parse_detail_response(response: dict, fallback_content_id: str = "") -> Post:
    """把 detail 接口返回（信封 {code,msg,data:{data:{...}}}）解析成 Post。"""
    if not isinstance(response, dict):
        raise CrawlerError("bad_response: not a dict")
    code = response.get("code")
    if code not in (0, "0"):
        raise CrawlerError(f"business_error: code={code} msg={response.get('msg')}")
    inner = ((response.get("data") or {}).get("data")) or {}
    if not inner:
        raise CrawlerError("empty_detail: data.data is empty")

    content_id = inner.get("channel_content_id") or fallback_content_id
    link = inner.get("content_link") or (
        f"https://www.xiaohongshu.com/explore/{content_id}" if content_id else ""
    )
    images = _image_urls(inner)
    # 图文帖：每张图就是一张卡片（1-based；视频帧卡片在 pipeline 抽帧后补入）
    cards = [Card(index=i, kind="image", url=u) for i, u in enumerate(images, start=1)]
    return Post(
        id=f"xhs_{content_id}",
        platform="xiaohongshu",
        url=link,
        content_id=content_id,
        title=inner.get("title") or "",
        content_type=inner.get("content_type") or "",
        body_text=inner.get("body_text") or "",
        topic_list=list(inner.get("topic_list") or []),
        image_urls=images,
        video_urls=_video_urls(inner),
        cards=cards,
        author_id=inner.get("channel_account_id"),
        author_name=inner.get("channel_account_name"),
        raw=response,
    )


def fetch_post_detail(
    content_id_or_url: str,
    *,
    settings: Optional[Settings] = None,
    http_client: Any = None,
    rate_limiter: Optional[RateLimiter] = None,
    env_file: str = ".env",
) -> Post:
    """真实拉取一条小红书帖子详情，返回 Post。"""
    settings = settings or Settings.from_env(env_file)
    content_id = parse_content_id(content_id_or_url)
    if not content_id:
        raise CrawlerError(f"cannot parse content_id from: {content_id_or_url}")

    rate_limiter = rate_limiter or RateLimiter()
    rate_limiter.wait("xhs_detail")

    owns_client = http_client is None
    client = http_client or httpx.Client()
    try:
        url = urljoin(settings.crawler_base_url, DETAIL_PATH)
        resp = client.post(
            url,
            json={"content_id": content_id},
            headers={"Content-Type": "application/json"},
            timeout=settings.crawler_timeout,
        )
        resp.raise_for_status()
        data = resp.json()
    except httpx.HTTPError as exc:
        raise CrawlerError(f"http_error: {exc}") from exc
    except ValueError as exc:
        raise CrawlerError("bad_json") from exc
    finally:
        if owns_client:
            client.close()

    return parse_detail_response(data, fallback_content_id=content_id)