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

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

import hashlib
from typing import Any, Iterable

from acquisition.content_mode import ARTICLE
from acquisition.platforms.base import PlatformCandidate, PlatformItem, PlatformSearchPage
from acquisition.search import fetch_weixin_detail, search_weixin_page
from core.config import Settings


def _hash(value: str, n: int = 16) -> str:
    return hashlib.md5(value.encode("utf-8")).hexdigest()[:n]


class WeixinAdapter:
    platform = "weixin"

    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 = "0"
        rank = 1
        seen: set[str] = set()
        for page_index in range(1, max_pages + 1):
            page = search_weixin_page(
                query,
                cursor=cursor,
                limit=limit,
                settings=settings,
                rate_limiter=rate_limiter,
            )
            candidates: list[PlatformCandidate] = []
            for page_rank, row in enumerate(page.rows, start=1):
                url = row.get("url") or ""
                source_id = _hash(url)
                if not url 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=url,
                        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:
        body_text, images = fetch_weixin_detail(
            candidate.url,
            settings=settings,
            rate_limiter=rate_limiter,
        )
        return PlatformItem(
            platform=self.platform,
            source_id=candidate.source_id or _hash(candidate.url),
            url=candidate.url,
            content_type="图文",
            content_mode=ARTICLE,
            title=candidate.title,
            author=candidate.author,
            body_text=body_text,
            image_urls=images or [candidate.cover_url],
            raw=candidate.raw,
        )