bilibili_api/topic.py

"""
bilibili_api.topic

话题相关
"""

from enum import Enum
from typing import Union, Optional

from . import dynamic
from .user import get_self_info
from .utils.utils import get_api
from .utils.credential import Credential
from .utils.network import Api

API = get_api("topic")


class TopicCardsSortBy(Enum):
    """
    话题下内容排序方式

    + NEW: 最新
    + HOT: 最热
    + RECOMMEND: 推荐
    """

    NEW = 3
    HOT = 2
    RECOMMEND = 1


async def get_hot_topics(numbers: int = 33) -> dict:
    """
    获取动态页的火热话题

    Args:
        numbers (int): 话题数量. Defaults to 33.

    Returns:
        dict: 调用 API 返回的结果
    """
    api = API["info"]["dynamic_page_topics"]
    params = {"page_size": numbers}
    return await Api(**api).update_params(**params).result


async def search_topic(keyword: str, ps: int = 20, pn: int = 1) -> dict:
    """
    搜索话题

    从动态页发布动态处的话题搜索框搜索话题

    Args:
        keyword (str): 搜索关键词

        ps      (int): 每页数量. Defaults to 20.

        pn      (int): 页数. Defaults to 1.

    Returns:
        dict: 调用 API 返回的结果
    """
    api = API["info"]["search"]
    params = {"keywords": keyword, "page_size": ps, "page_num": pn}
    return await Api(**api).update_params(**params).result


class Topic:
    """
    话题类

    Attributes:
        credential (Credential): 凭据类
    """

    def __init__(self, topic_id: int, credential: Union[Credential, None] = None):
        """
        Args:
            topic_id   (int)       : 话题 id

            credential (Credential): 凭据类
        """
        self.__topic_id = topic_id
        self.credential = credential if credential else Credential()

    def get_topic_id(self) -> int:
        """
        获取话题 id

        Returns:
            int: 话题 id
        """
        return self.__topic_id

    async def get_info(self) -> dict:
        """
        获取话题简介

        Returns:
            dict: 调用 API 返回的结果
        """
        api = API["info"]["info"]
        params = {"topic_id": self.get_topic_id()}
        return (
            await Api(**api, credential=self.credential).update_params(**params).result
        )

    async def get_cards(
        self,
        ps: int = 100,
        offset: Optional[str] = None,
        sort_by: TopicCardsSortBy = TopicCardsSortBy.HOT,
    ) -> dict:
        """
        获取话题下的内容

        未登录无法使用热门排序字段即 TopicCardsSortBy.RECOMMEND

        Args:
            ps (int): 数据数量. Defaults to 100.

            offset (Optional[str]): 偏移量. 生成格式为 f'{页码}_{页码*数据量]}' 如'2_40' Defaults to None.

            sort_by (TopicCardsSortBy): 排序方式. Defaults to TopicCardsSortBy.HOT.

        Returns:
            dict: 调用 API 返回的结果
        """
        api = API["info"]["cards"]
        params = {
            "topic_id": self.get_topic_id(),
            "page_size": ps,
            "sort_by": sort_by.value
        }
        if offset:
            params.update({"offset": offset})
        return (
            await Api(**api, credential=self.credential).update_params(**params).result
        )


    async def like(self, status: bool = True) -> dict:
        """
        设置点赞话题

        Args:
            status (bool): 是否设置点赞. Defaults to True.

        Returns:
            dict: 调用 API 返回的结果
        """
        api = API["operate"]["like"]
        data = {
            "topic_id": self.get_topic_id(),
            "action": "like" if status else "cancel_like",
            "business": "topic",
            "up_mid": (await get_self_info(self.credential))["mid"],
        }
        return await Api(**api, credential=self.credential).update_data(**data).result

    async def set_favorite(self, status: bool = True) -> dict:
        """
        设置收藏话题

        Args:
            status (bool): 是否设置收藏. Defaults to True.

        Returns:
            dict: 调用 API 返回的结果
        """
        api = API["operate"]["add_favorite" if status else "cancel_favorite"]
        data = {"topic_id": self.get_topic_id()}
        return await Api(**api, credential=self.credential).update_data(**data).result