|
"""
|
|
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
|
|
|
|
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
|
|
|
