rogerxavier's picture
Upload 258 files
0aee47a verified
raw
history blame
8.28 kB
"""
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