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

"""Douyin 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_keyword
from core.config import Settings


class DouyinAdapter:
    platform = "douyin"

    def search(
        self,
        query: str,
        *,
        settings: Settings,
        limit: int,
        rate_limiter: Any,
    ) -> list[PlatformCandidate]:
        ids = search_keyword(
            query,
            platform=self.platform,
            content_type="视频",
            limit=limit,
            settings=settings,
            rate_limiter=rate_limiter,
        )
        return [
            PlatformCandidate(
                rank=i,
                platform=self.platform,
                source_id=content_id,
                raw={"id": content_id},
            )
            for i, content_id in enumerate(ids, start=1)
        ]

    def fetch_detail(
        self,
        candidate: PlatformCandidate,
        *,
        settings: Settings,
        rate_limiter: Any,
    ) -> PlatformItem:
        post = fetch_post_detail(
            candidate.source_id,
            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 "video",
            title=post.title or candidate.title,
            author=post.author_name or candidate.author,
            body_text=post.body_text or "",
            image_urls=post.image_urls,
            video_urls=post.video_urls,
            raw=post.raw if isinstance(post.raw, dict) else {},
        )