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

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

import hashlib
from typing import Any

from acquisition.platforms.base import PlatformCandidate, PlatformItem
from acquisition.search import fetch_weixin_detail, search_weixin
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]:
        rows = search_weixin(
            query,
            limit=limit,
            settings=settings,
            rate_limiter=rate_limiter,
        )
        return [
            PlatformCandidate(
                rank=i,
                platform=self.platform,
                source_id=_hash(row.get("url") 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:
        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="图文",
            title=candidate.title,
            author=candidate.author,
            body_text=body_text,
            image_urls=images or [candidate.cover_url],
            raw=candidate.raw,
        )