Create-knowledge-find-decode/acquisition/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 random
import re
import time
from typing import Any, Callable, Optional
from urllib.parse import urljoin

import httpx

from core.config import Settings
from core.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。
    给 max_interval_seconds 时，每次间隔在 [min, max] 随机取（搜索接口加随机抖动躲限流）。"""

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

    def _interval(self) -> float:
        if self.max_interval_seconds is not None:
            return random.uniform(self.min_interval_seconds, self.max_interval_seconds)
        return self.min_interval_seconds

    def wait(self, bucket: str) -> None:
        last = self._last.get(bucket)
        if last is not None:
            remaining = self._interval() - (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:
        is_douyin = platform == "douyin"
        base_url = settings.piaoquantv_douyin_base_url if is_douyin else settings.aiddit_crawler_base_url
        body = {"content_id": content_id}
        if is_douyin:                                  # 抖音详情走 piaoquantv，需带 account_id + cookie_batch
            body["account_id"] = settings.piaoquantv_douyin_account_id
            body["cookie_batch"] = settings.piaoquantv_douyin_cookie_batch
        url = urljoin(base_url, cfg["path"])
        resp = client.post(
            url,
            json=body,
            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)