""" bilibili_api.channel_series 用户合集与列表相关 """ import json from enum import Enum from typing import List, Union, Optional import httpx from .utils.utils import get_api, raise_for_statement from .utils.credential import Credential from .utils.network import Api, HEADERS API_USER = get_api("user") API = get_api("channel-series") channel_meta_cache = {} class ChannelOrder(Enum): """ 合集视频排序顺序。 + DEFAULT: 默认排序 + CHANGE : 升序排序 """ DEFAULT = "false" CHANGE = "true" class ChannelSeriesType(Enum): """ 合集与列表类型 + SERIES: 相同视频分类 + SEASON: 新概念多 P **SEASON 类合集与列表名字为`合集·XXX`,请注意区别** """ SERIES = 0 SEASON = 1 class ChannelSeries: """ 合集与列表类 Attributes: credential (Credential): 凭据类. Defaults to None. """ def __init__( self, uid: int = -1, type_: ChannelSeriesType = ChannelSeriesType.SERIES, id_: int = -1, credential: Union[Credential, None] = None, ): """ Args: uid(int) : 用户 uid. Defaults to -1. type_(ChannelSeriesType): 合集与列表类型. Defaults to ChannelSeriesType.SERIES. id_(int) : season_id 或 series_id. Defaults to -1. credential(Credential) : 凭证. Defaults to None. """ global channel_meta_cache raise_for_statement(id_ != -1) raise_for_statement(type_ != None) from .user import User self.__uid = uid self.is_new = type_.value self.id_ = id_ self.owner = User(self.__uid, credential=credential) self.credential = credential self.meta = None if not f"{type_.value}-{id_}" in channel_meta_cache.keys(): if self.is_new: api = API_USER["channel_series"]["season_info"] params = {"season_id": self.id_} else: api = API_USER["channel_series"]["info"] params = {"series_id": self.id_} resp = Api(**api).update_params(**params).result_sync if self.is_new: self.meta = resp["info"] self.meta["mid"] = resp["info"]["upper"]["mid"] self.__uid = self.meta["mid"] self.owner = User(self.__uid, credential=credential) else: self.meta = resp["meta"] self.__uid = self.meta["mid"] self.owner = User(self.__uid, credential=credential) else: self.meta = channel_meta_cache[f"{type_.value}-{id_}"] def get_meta(self) -> dict: """ 获取元数据 Returns: 调用 API 返回的结果 """ return self.meta # type: ignore async def get_videos( self, sort: ChannelOrder = ChannelOrder.DEFAULT, pn: int = 1, ps: int = 100 ) -> dict: """ 获取合集视频 Args: sort(ChannelOrder): 排序方式 pn(int) : 页数,默认为 1 ps(int) : 每一页显示的视频数量 Returns: 调用 API 返回的结果 """ if self.is_new: return await self.owner.get_channel_videos_season(self.id_, sort, pn, ps) else: return await self.owner.get_channel_videos_series(self.id_, sort, pn, ps) async def create_channel_series( name: str, aids: List[int] = [], keywords: List[str] = [], description: str = "", credential: Union[Credential, None] = None, ) -> dict: """ 新建一个视频列表 (旧版合集) Args: name (str): 列表名称。 aids (List[int]): 要加入列表的视频的 aid 列表。 keywords (List[str]): 列表的关键词。 description (str): 列表的描述。 credential (Credential | None): 凭据类。 Returns: dict: 调用 API 返回的结果 """ from .user import get_self_info credential = credential if credential else Credential() credential.raise_for_no_sessdata() credential.raise_for_no_bili_jct() api = API_USER["channel_series"]["create"] info = await get_self_info(credential) data = { "mid": info["mid"], "aids": ",".join(map(lambda x: str(x), aids)), "name": name, "keywords": ",".join(keywords), "description": description, } return await Api(**api, credential=credential).update_data(**data).result async def del_channel_series(series_id: int, credential: Credential) -> dict: """ 删除视频列表(旧版合集) Args: series_id (int) : 旧版合集 id。 credential (Credential): 凭据类。 Returns: dict: 调用 API 返回的结果 """ from .user import User, get_self_info credential.raise_for_no_sessdata() credential.raise_for_no_bili_jct() series_total = ChannelSeries( type_=ChannelSeriesType.SERIES, id_=series_id, credential=credential ).get_meta()["total"] self_uid = (await get_self_info(credential))["mid"] aids = [] pages = series_total // 20 + (1 if (series_total % 20 != 0) else 0) for page in range(1, pages + 1, 1): page_info = await User(self_uid, credential).get_channel_videos_series( series_id, pn=page, ps=20 ) for aid in page_info["aids"]: aids.append(aid) api = API_USER["channel_series"]["del_channel_series"] data = { "mid": self_uid, "series_id": series_id, "aids": ",".join(map(lambda x: str(x), aids)), } return await Api(**api, credential=credential).update_data(**data).result async def add_aids_to_series( series_id: int, aids: List[int], credential: Credential ) -> dict: """ 添加视频至视频列表(旧版合集) Args: series_id (int) : 旧版合集 id。 aids (List[int]) : 视频 aid 列表。 credential (Credential): 凭据类。 Returns: dict: 调用 API 返回的结果 """ from .user import get_self_info credential.raise_for_no_sessdata() credential.raise_for_no_bili_jct() self_info = await get_self_info(credential) api = API_USER["channel_series"]["add_channel_aids_series"] data = { "mid": self_info["mid"], "series_id": series_id, "aids": ",".join(map(lambda x: str(x), aids)), } return await Api(**api, credential=credential).update_data(**data).result async def del_aids_from_series( series_id: int, aids: List[int], credential: Credential ) -> dict: """ 从视频列表(旧版合集)删除视频 Args: series_id (int) : 旧版合集 id。 aids (List[int]) : 视频 aid 列表。 credential (Credential): 凭据类。 Returns: dict: 调用 API 返回的结果 """ from .user import get_self_info credential.raise_for_no_sessdata() credential.raise_for_no_bili_jct() self_info = await get_self_info(credential) api = API_USER["channel_series"]["del_channel_aids_series"] data = { "mid": self_info["mid"], "series_id": series_id, "aids": ",".join(map(lambda x: str(x), aids)), } return await Api(**api, credential=credential).update_data(**data).result async def set_follow_channel_season( season_id: int, status: bool = True, credential: Optional[Credential] = None ) -> dict: """ 设置是否订阅合集(新版) Args: season_id (int) : 合集 id status (bool): 是否订阅状态. Defaults to True. """ api = API["operate"]["fav"] if status else API["operate"]["unfav"] data = {"season_id": season_id} return await Api(**api, credential=credential).update_data(**data).result