Create-knowledge-find-decode/acquisition/platforms/xiaohongshu.py

"""Xiaohongshu formal acquisition adapter."""
from __future__ import annotations

from typing import Any, Iterable

from acquisition.content_mode import infer_content_mode
from acquisition.crawler import fetch_post_detail
from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
from acquisition.search import search_xiaohongshu_page
from core.config import Settings


class XiaohongshuAdapter:
    platform = "xiaohongshu"

    def search(
        self,
        query: str,
        *,
        settings: Settings,
        limit: int,
        rate_limiter: Any,
    ) -> list[PlatformCandidate]:
        pages = self.search_pages(
            query,
            settings=settings,
            limit=limit,
            rate_limiter=rate_limiter,
            max_pages=1,
        )
        return [candidate for page in pages for candidate in page.candidates]

    def search_pages(
        self,
        query: str,
        *,
        settings: Settings,
        limit: int,
        rate_limiter: Any,
        max_pages: int = 2,
    ) -> Iterable[PlatformSearchPage]:
        cursor = ""
        rank = 1
        seen: set[str] = set()
        for page_index in range(1, max_pages + 1):
            page = search_xiaohongshu_page(
                query,
                cursor=cursor,
                content_type=None,
                limit=limit,
                settings=settings,
                rate_limiter=rate_limiter,
            )
            candidates: list[PlatformCandidate] = []
            for page_rank, row in enumerate(page.rows, start=1):
                source_id = row.get("id") or ""
                if not source_id or source_id in seen:
                    continue
                seen.add(source_id)
                raw = {
                    **row,
                    "page_index": page_index,
                    "page_rank": page_rank,
                    "source_cursor": page.cursor,
                    "request_payload": page.request_payload,
                }
                candidates.append(
                    PlatformCandidate(
                        rank=rank,
                        platform=self.platform,
                        source_id=source_id,
                        url=row.get("url") or "",
                        title=row.get("title") or "",
                        author=row.get("nick_name") or "",
                        cover_url=row.get("cover_url") or "",
                        raw=raw,
                    )
                )
                rank += 1
            yield PlatformSearchPage(
                page_index=page_index,
                candidates=candidates,
                raw_count=page.raw_count,
                cursor=page.cursor,
                next_cursor=page.next_cursor,
                has_more=page.has_more,
                provider=page.provider,
                request_payload=page.request_payload,
            )
            if not page.has_more or not page.next_cursor or page.next_cursor == cursor:
                break
            cursor = page.next_cursor

    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,
        )
        image_urls = post.image_urls or ([candidate.cover_url] if candidate.cover_url else [])
        content_mode = infer_content_mode(
            platform=self.platform,
            content_type=post.content_type,
            body_text=post.body_text,
            image_urls=image_urls,
            video_urls=post.video_urls,
            raw={**candidate.raw, "detail_content_type": post.content_type},
        )
        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 "",
            content_mode=content_mode,
            title=post.title or candidate.title,
            author=post.author_name or candidate.author,
            body_text=post.body_text or "",
            image_urls=image_urls,
            video_urls=post.video_urls,
            raw=post.raw if isinstance(post.raw, dict) else {},
        )