| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667 |
- """Xiaohongshu formal acquisition adapter."""
- from __future__ import annotations
- from typing import Any
- from acquisition.crawler import fetch_post_detail
- from acquisition.platforms.base import PlatformCandidate, PlatformItem
- from acquisition.search import search_xiaohongshu
- from core.config import Settings
- class XiaohongshuAdapter:
- platform = "xiaohongshu"
- def search(
- self,
- query: str,
- *,
- settings: Settings,
- limit: int,
- rate_limiter: Any,
- ) -> list[PlatformCandidate]:
- rows = search_xiaohongshu(
- query,
- content_type=settings.search_content_type,
- limit=limit,
- settings=settings,
- rate_limiter=rate_limiter,
- )
- return [
- PlatformCandidate(
- rank=i,
- platform=self.platform,
- source_id=row.get("id") or "",
- url=row.get("url") or "",
- title=row.get("title") or "",
- author=row.get("nick_name") or "",
- cover_url=row.get("cover_url") or "",
- raw=row,
- )
- for i, row in enumerate(rows, start=1)
- ]
- def fetch_detail(
- self,
- candidate: PlatformCandidate,
- *,
- settings: Settings,
- rate_limiter: Any,
- ) -> PlatformItem:
- post = fetch_post_detail(
- candidate.source_id or candidate.url,
- settings=settings,
- rate_limiter=rate_limiter,
- )
- return PlatformItem(
- platform=self.platform,
- source_id=post.content_id or candidate.source_id,
- url=post.url or candidate.url,
- content_type=post.content_type or "图文",
- title=post.title or candidate.title,
- author=post.author_name or candidate.author,
- body_text=post.body_text or "",
- image_urls=post.image_urls or [candidate.cover_url],
- video_urls=post.video_urls,
- raw=post.raw if isinstance(post.raw, dict) else {},
- )
|
