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

"""帖子详情拉取：多平台 detail 接口（小红书/抖音/快手/B站）。

crawler.aiddit.com 各平台 detail 返回字段已归一化（channel_content_id/title/content_type/
body_text/image_url_list/video_url_list/channel_account_*），所以 parse 一套复用；差异只在
detail path、URL→content_id 解析、平台名/id 前缀。详见 数据接口与来源/视频音频取数实测.md。

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

import re
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

RATE_LIMIT_SECONDS = 15.0

# 平台路由：detail path + Post.id 前缀
PLATFORMS = {
    "xiaohongshu": {"path": "/crawler/xiao_hong_shu/detail", "prefix": "xhs"},
    "douyin": {"path": "/crawler/dou_yin/detail", "prefix": "dy"},
    "kuaishou": {"path": "/crawler/kuai_shou/detail", "prefix": "ks"},
    "bilibili": {"path": "/crawler/bilibili/detail", "prefix": "bili"},
}


def _last_seg(s: str) -> str:
    return s.split("?", 1)[0].rstrip("/").split("/")[-1]


def detect_platform_and_id(content_id_or_url: str) -> tuple[str, str]:
    """从 URL（或裸 id）识别平台 + content_id。不支持短链（v.douyin.com 等需先跟随重定向）。"""
    s = content_id_or_url.strip()
    low = s.lower()
    if "xiaohongshu.com" in low or "/explore/" in low or "/discovery/item/" in low:
        return "xiaohongshu", parse_content_id(s)
    if "douyin.com" in low:
        m = (re.search(r"modal_id=(\d+)", s) or re.search(r"/video/(\d+)", s)
             or re.search(r"/note/(\d+)", s) or re.search(r"(\d{15,})", s))
        return "douyin", m.group(1) if m else _last_seg(s)
    if "kuaishou.com" in low or "gifshow.com" in low:
        m = re.search(r"/(?:fw/photo|short-video|photo|f)/([A-Za-z0-9_-]+)", s)
        return "kuaishou", m.group(1) if m else _last_seg(s)
    if "bilibili.com" in low or "b23.tv" in low:
        m = re.search(r"(BV[0-9A-Za-z]+)", s)
        return "bilibili", m.group(1) if m else _last_seg(s)
    # 裸 id 兜底：BV→B站；纯长数字→抖音；其余→小红书（向后兼容）
    if s.startswith("BV"):
        return "bilibili", s
    if re.fullmatch(r"\d{15,}", s):
        return "douyin", s
    return "xiaohongshu", parse_content_id(s)


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, *, platform: str = "xiaohongshu",
                          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")

    prefix = PLATFORMS.get(platform, PLATFORMS["xiaohongshu"])["prefix"]
    content_id = inner.get("channel_content_id") or fallback_content_id
    link = inner.get("content_link") or ""
    images = _image_urls(inner)
    # 图文帖：每张图一张卡片（1-based）；视频帖的段卡由 extract_video 在提取时写入，覆盖封面卡
    cards = [Card(index=i, kind="image", url=u) for i, u in enumerate(images, start=1)]
    return Post(
        id=f"{prefix}_{content_id}",
        platform=platform,
        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:
    """真实拉取一条帖子详情，自动识别平台（小红书/抖音/快手/B站），返回 Post。"""
    settings = settings or Settings.from_env(env_file)
    platform, content_id = detect_platform_and_id(content_id_or_url)
    if not content_id:
        raise CrawlerError(f"cannot parse content_id from: {content_id_or_url}")
    cfg = PLATFORMS[platform]

    rate_limiter = rate_limiter or RateLimiter()
    rate_limiter.wait(f"{platform}_detail")

    owns_client = http_client is None
    client = http_client or httpx.Client()
    try:
        url = urljoin(settings.crawler_base_url, cfg["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, platform=platform, fallback_content_id=content_id)