bilibili_api/video_tag.py

"""
bilibili_api.video_tag

视频标签相关，部分的标签的 id 与同名的频道的 id 一模一样。
"""

from typing import Optional

import httpx

from .errors import *
from .utils.utils import get_api
from .utils.credential import Credential
from .utils.network import Api

API = get_api("video_tag")
API_video = get_api("video")


class Tag:
    """
    标签类
    """

    def __init__(
        self,
        tag_name: Optional[str] = None,
        tag_id: Optional[int] = None,
        credential: Optional[Credential] = None,
    ):
        """
        Args:
            tag_name   (str | None): 标签名. Defaults to None.

            tag_id     (int | None): 标签 id. Defaults to None.

            credential (Credential): 凭据类. Defaults to None.

        注意：tag_name 和 tag_id 任选一个传入即可。tag_id 优先使用。
        """
        if tag_id == None:
            if tag_name == None:
                raise ArgsException("tag_name 和 tag_id 需要提供一个。")
            self.__tag_id = self.__get_tag_info_sync(tag_name)["tag_id"]
        else:
            self.__tag_id = tag_id
        credential = credential if credential else Credential()
        self.credential = credential

    def get_tag_id(self) -> int:
        return self.__tag_id

    def __get_tag_info_sync(self, tag_name: str) -> dict:
        api = API["info"]["tag_info"]
        params = {"tag_name": tag_name}
        return Api(**api).update_params(**params).result_sync

    async def get_tag_info(self) -> dict:
        """
        获取标签信息。

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

    async def get_similar_tags(self) -> dict:
        """
        获取相关的标签

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

    # async def get_cards(self) -> dict:
    #     """
    #     获取标签下的视频/动态

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

    async def get_history_cards(self, offset_dynamic_id: int) -> dict:
        """
        获取标签下，指定dynamic_id的视频的后一个视频/动态作为起始的视频/动态

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

    async def subscribe_tag(self) -> dict:
        """
        关注标签。

        Returns:
            dict: 调用 API 返回的结果。
        """
        self.credential.raise_for_no_sessdata()
        self.credential.raise_for_no_bili_jct()

        api = API_video["operate"]["subscribe_tag"]

        data = {"tag_id": self.__tag_id}
        return await Api(**api, credential=self.credential).update_data(**data).result

    async def unsubscribe_tag(self) -> dict:
        """
        取关标签。

        Returns:
            dict: 调用 API 返回的结果。
        """
        self.credential.raise_for_no_sessdata()
        self.credential.raise_for_no_bili_jct()

        api = API_video["operate"]["unsubscribe_tag"]

        data = {"tag_id": self.__tag_id}
        return await Api(**api, credential=self.credential).update_data(**data).result