"""帖子详情拉取：多平台 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/?... 链接。""" 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)