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

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

import hashlib
from urllib.parse import parse_qs, urlparse
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]


def _dedupe_key(url: str) -> str:
    parsed = urlparse(url)
    return parsed._replace(query="", fragment="").geturl() or url


def _is_decorative_weixin_image(url: str) -> bool:
    parsed = urlparse(url)
    query = parse_qs(parsed.query)
    wx_fmt = (query.get("wx_fmt") or [""])[0].lower()
    path = parsed.path.lower()
    if wx_fmt == "gif" or path.endswith(".gif"):
        return True
    if "mmbiz_gif" in path:
        return True
    return False


def filter_weixin_article_images(image_urls: list[str]) -> list[str]:
    """Keep only article images that should become decode cards.

    Weixin detail returns all article image resources, including duplicated
    images and common account UI/animation GIFs such as "在看" prompts. Those
    should stay in the original article text, but they should not become
    knowledge trace cards.
    """

    kept: list[str] = []
    seen: set[str] = set()
    for url in image_urls:
        if not url:
            continue
        key = _dedupe_key(url)
        if key in seen:
            continue
        seen.add(key)
        if _is_decorative_weixin_image(url):
            continue
        kept.append(url)
    return kept


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=filter_weixin_article_images(images) or [candidate.cover_url],
            raw=candidate.raw,
        )