Spaces:
Sleeping
Sleeping
""" | |
bilibili_api.creative_center | |
创作中心相关。 | |
务必携带 Credential 信息,否则无法获取到数据。 | |
""" | |
from enum import Enum | |
from typing import List, Union, Optional | |
from datetime import datetime | |
from .video_zone import VideoZoneTypes | |
from .utils.utils import get_api | |
from .utils.credential import Credential | |
from .utils.network import Api | |
API = get_api("creative_center") | |
class GraphPeriod(Enum): | |
""" | |
统计图表的时间段。 | |
+ YESTERDAY: 昨天 | |
+ WEEK: 近一周 | |
+ MONTH: 近一月 | |
+ THREE_MONTH: 近三月 | |
+ TOTAL: 历史累计 | |
""" | |
YESTERDAY = -1 | |
WEEK = 0 | |
MONTH = 1 | |
THREE_MONTH = 2 | |
TOTAL = 3 | |
class FanGraphPeriod(Enum): | |
""" | |
粉丝统计图表的时间段。 | |
+ YESTERDAY: 昨天 | |
+ WEEK: 近一周 | |
+ MONTH: 近一月 | |
+ THREE_MONTH: 近三月 | |
""" | |
YESTERDAY = -1 | |
WEEK = 0 | |
MONTH = 1 | |
THREE_MONTH = 2 | |
class GraphType(Enum): | |
""" | |
统计图表的类型。 | |
+ PLAY: 播放量 | |
+ VISITOR: 访问量 | |
+ FAN: 粉丝数 | |
+ LIKE: 点赞数 | |
+ FAV: 收藏数 | |
+ SHARE: 分享数 | |
+ COMMENT: 评论数 | |
+ DAMKU: 弹幕数 | |
+ COIN: 投币数 | |
+ ELEC: 充电数 | |
""" | |
PLAY = "play" | |
VISITOR = "visitor" | |
FAN = "fan" | |
LIKE = "like" | |
FAV = "fav" | |
SHARE = "share" | |
COMMENT = "comment" | |
DAMKU = "dm" | |
COIN = "coin" | |
ELEC = "elec" | |
class FanGraphType(Enum): | |
""" | |
粉丝统计图表的类型。 | |
+ ALL_FANS: 粉丝总量 | |
+ FAN: 新增粉丝 | |
+ FOLLOW: 新增关注 | |
+ UNFOLLOW: 取消关注 | |
""" | |
ALL_FANS = "all_fans" | |
NEW_FANS = "fan" | |
FOLLOW = "follow" | |
UNFOLLOW = "unfollow" | |
class ArticleInfoType(Enum): | |
""" | |
文章统计信息的类型。 | |
+ READ: 阅读 | |
+ COMMENT: 评论 | |
+ SHARE: 分享 | |
+ COIN: 投币 | |
+ FAV: 收藏 | |
+ LIKE: 点赞 | |
""" | |
READ = 1 | |
COMMENT = 2 | |
SHARE = 3 | |
COIN = 4 | |
FAV = 5 | |
LIKE = 6 | |
class Copyright(Enum): | |
""" | |
稿件播放完成率对比的版权类型。 | |
+ ALL: 全部 | |
+ ORIGINAL: 原创 | |
+ REPRINT: 转载 | |
""" | |
ALL = 0 | |
ORIGINAL = 1 | |
REPRINT = 2 | |
class UploadManagerOrder(Enum): | |
""" | |
内容管理排序字段。 | |
+ CLICK: 点击 | |
+ STOW: 收藏 | |
+ SENDDATE: 上传日期 | |
+ DM_COUNT: 弹幕数量 | |
+ COMMENT_COUNT: 评论数量 | |
""" | |
CLICK = "click" | |
STOW = "stow" | |
SENDDATE = "senddate" | |
DM_COUNT = "dm_count" | |
COMMENT_COUNT = "scores" # wtf the scores means | |
class UploadManagerStatus(Enum): | |
""" | |
内容管理稿件状态字段。 | |
+ ALL: 全部稿件 | |
+ PUBED: 已通过 | |
+ IS_PUBING: 进行中 | |
+ NOT_PUBED: 未通过 | |
""" | |
ALL = "is_pubing,pubed,not_pubed" | |
PUBED = "pubed" | |
IS_PUBING = "is_pubing" | |
NOT_PUBED = "not_pubed" | |
class UploadManagerSort(Enum): | |
""" | |
内容管理文章排序字段。 | |
+ CREATED_TIME: 创建日期 | |
+ LIKE: 点赞 | |
+ COMMENT: 评论 | |
+ FAV: 收藏 | |
+ COIN: 投币 | |
""" | |
CREATED_TIME = 1 | |
LIKE = 2 | |
COMMENT = 3 | |
FAV = 5 | |
COIN = 6 | |
class UploadManagerArticleStatus(Enum): | |
""" | |
内容管理文章状态字段。 | |
+ ALL: 全部稿件 | |
+ PUBED: 已通过 | |
+ IS_PUBING: 进行中 | |
+ NOT_PUBED: 未通过 | |
""" | |
ALL = 0 | |
PUBED = 2 | |
IS_PUBING = 1 | |
NOT_PUBED = 3 | |
class ArchiveType(Enum): | |
""" | |
评论管理中的稿件类型。 | |
+ VIDEO: 视频 | |
+ ARTICLE: 文章 | |
+ AUDIO: 音频 | |
""" | |
VIDEO = 1 # 视频 | |
ARTICLE = 12 # 文章 | |
AUDIO = 14 # 音频 | |
class CommentManagerOrder(Enum): | |
""" | |
评论管理中的排序字段。 | |
+ RECENTLY: 最近 | |
+ LIKE: 点赞 | |
+ REPLY: 回复 | |
""" | |
RECENTLY = 1 # 最近 | |
LIKE = 2 # 点赞 | |
REPLY = 3 # 回复 | |
class DanmakuMode(Enum): | |
""" | |
弹幕模式。 | |
+ ROLL: 滚动 | |
+ BOTTOM: 底端 | |
+ TOP: 顶端 | |
+ REVERSE: 逆向 | |
+ ADVANCED: 高级 | |
+ CODE: 代码 | |
+ BAS: BAS 补充注释 | |
""" | |
ROLL = 1 # 滚动 | |
BOTTOM = 4 # 底端 | |
TOP = 5 # 顶端 | |
REVERSE = 6 # 逆向 | |
ADVANCED = 7 # 高级 | |
CODE = 8 # 代码 | |
BAS = 9 # BAS | |
class DanmakuPool(Enum): | |
""" | |
子弹幕池类型。 | |
+ NORMAL: 普通 | |
+ SUBTITLE: 字幕 | |
+ SPECIAL: 特殊 | |
""" | |
NORMAL = 0 # 普通 | |
SUBTITLE = 1 # 字幕 | |
SPECIAL = 2 # 特殊 | |
class DanmakuOrder(Enum): | |
""" | |
弹幕排序依据 | |
+ CTIME: 发送时间 | |
+ LIKE_COUNT: 点赞数 | |
""" | |
CTIME = "ctime" | |
LIKE_COUNT = "like_count" | |
class DanmakuSort(Enum): | |
""" | |
弹幕排序顺序 | |
+ DESC: 降序 | |
+ ASC: 升序 | |
""" | |
DESC = "desc" | |
ASC = "asc" | |
class DanmakuType(Enum): | |
""" | |
弹幕筛选类型 | |
+ ALL: 全部 | |
+ PROTECT: 保护弹幕 | |
""" | |
ALL = 0 | |
PROTECT = 2 | |
async def get_compare(credential: Credential) -> dict: | |
""" | |
获取对比数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 视频对比数据。 | |
""" | |
api = API["overview"]["compare"] | |
return await Api(**api, credential=credential).result | |
async def get_graph( | |
credential: Credential, | |
period: GraphPeriod = GraphPeriod.WEEK, | |
graph_type: GraphType = GraphType.PLAY, | |
) -> dict: | |
""" | |
获取统计图表数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
period (GraphPeriod): 时间段。 | |
graph_type (GraphType): 图表类型。 | |
Returns: | |
dict: 视频统计图表数据。 | |
""" | |
api = API["overview"]["graph"] | |
params = { | |
"period": period.value, | |
"s_locale": "zh_CN", | |
"type": graph_type.value, | |
} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_overview( | |
credential: Credential, period: GraphPeriod = GraphPeriod.WEEK | |
) -> dict: | |
""" | |
获取概览数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
period (GraphPeriod): 时间段。 | |
Returns: | |
dict: 视频概览数据。 | |
""" | |
api = API["overview"]["num"] | |
# 不知道 tab 的作用是什么,但是不传会报错 | |
params = {"period": period.value, "s_locale": "zh_CN", "tab": 0} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_video_survey(credential: Credential) -> dict: | |
""" | |
获取视频各分区中占比排行。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 视频分区排行数据。 | |
""" | |
api = API["data-up"]["video"]["survey"] | |
params = {"type": 1} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_video_playanalysis( | |
credential: Credential, copyright: Copyright = Copyright.ALL | |
) -> dict: | |
""" | |
获取稿件播放完成率对比。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
copyright (Copyright): 版权类型。 | |
Returns: | |
dict: 稿件播放完成率对比数据。 | |
""" | |
api = API["data-up"]["video"]["playanalysis"] | |
params = {"copyright": copyright.value} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_video_source(credential: Credential) -> dict: | |
""" | |
获取稿件播放来源分布。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 视频来源分布数据。 | |
""" | |
api = API["data-up"]["video"]["source"] | |
params = {"s_locale": "zh_CN"} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_fan_overview( | |
credential: Credential, period: FanGraphPeriod = FanGraphPeriod.WEEK | |
) -> dict: | |
""" | |
获取粉丝概览数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
period (FanGraphPeriod): 时间段。 | |
Returns: | |
dict: 粉丝概览数据。 | |
""" | |
api = API["data-up"]["fan"]["overview"] | |
params = {"period": period.value} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_fan_graph( | |
credential: Credential, | |
period: FanGraphPeriod = FanGraphPeriod.WEEK, | |
graph_type: FanGraphType = FanGraphType.ALL_FANS, | |
) -> dict: | |
""" | |
获取粉丝图表数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
period (FanGraphPeriod): 时间段。 | |
graph_type (FanGraphType): 图表类型。 | |
Returns: | |
dict: 粉丝图表数据。 | |
""" | |
api = API["data-up"]["fan"]["graph"] | |
params = {"period": period.value, "type": graph_type.value} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_article_overview(credential: Credential) -> dict: | |
""" | |
获取文章概览数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 文章概览数据。 | |
""" | |
api = API["data-up"]["article"]["overview"] | |
return await Api(**api, credential=credential).result | |
async def get_article_graph( | |
credential: Credential, graph_type: ArticleInfoType = ArticleInfoType.READ | |
) -> dict: | |
""" | |
获取文章图表数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
graph_type (ArticleInfoType): 图表类型。 | |
Returns: | |
dict: 文章图表数据。 | |
""" | |
api = API["data-up"]["article"]["graph"] | |
params = {"type": graph_type.value} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_article_rank( | |
credential: Credential, rank_type: ArticleInfoType = ArticleInfoType.READ | |
) -> dict: | |
""" | |
获取文章排行数据。 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
rank_type (ArticleInfoType): 排行依据。 | |
Returns: | |
dict: 文章排行数据。 | |
""" | |
api = API["data-up"]["article"]["rank"] | |
params = {"type": rank_type.value} | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_article_source(credential: Credential) -> dict: | |
""" | |
获取文章阅读终端数据 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 文章阅读终端数据。 | |
""" | |
api = API["data-up"]["article"]["source"] | |
return await Api(**api, credential=credential).result | |
""" | |
upload manager | |
https://member.bilibili.com/platform/upload-manager | |
""" | |
async def get_video_draft_upload_manager_info(credential: Credential) -> dict: | |
""" | |
获取内容管理视频草稿信息 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 内容管理视频草稿信息。 | |
""" | |
api = API["upload-manager"]["video_draft"] | |
return await Api(**api, credential=credential).result | |
async def get_video_upload_manager_info( | |
credential: Credential, | |
is_interative: bool = False, | |
pn: int = 1, | |
ps: int = 10, | |
order: UploadManagerOrder = UploadManagerOrder.CLICK, | |
tid: Union[VideoZoneTypes, None, int] = None, | |
status: UploadManagerStatus = UploadManagerStatus.ALL, | |
) -> dict: | |
""" | |
获取内容管理视频信息 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
is_interative (bool): 是否为互动视频 | |
pn (int): 页码 | |
ps (int): 每页项数 | |
tid: (VideoZoneTypes, None, int): 分区 | |
status (UploadManagerStatus): 稿件状态 | |
order (UploadManagerOrder): 稿件排序 | |
Returns: | |
dict: 内容管理视频信息。 | |
""" | |
params = { | |
"pn": pn, | |
"ps": ps, | |
"coop": 1, | |
"status": status.value, | |
"order": order.value, | |
"interactive": 2 if is_interative else 1, | |
"tid": tid.value if isinstance(tid, Enum) else tid, | |
} | |
api = API["upload-manager"]["video"] | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_article_upload_manager_info( | |
credential: Credential, | |
status: UploadManagerArticleStatus = UploadManagerArticleStatus.ALL, | |
sort: UploadManagerSort = UploadManagerSort.CREATED_TIME, | |
pn: int = 1, | |
) -> dict: | |
""" | |
获取内容管理文章信息 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
pn (int): 页码 | |
status (UploadManagerArticleStatus): 稿件状态 | |
sort (UploadManagerSort): 稿件排序 | |
Returns: | |
dict: 内容管理文章信息。 | |
""" | |
params = {"pn": pn, "group": status.value, "sort": sort.value, "mobi_app": "pc"} | |
api = API["upload-manager"]["article"] | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_article_list_upload_manager_info(credential: Credential) -> dict: | |
""" | |
获取内容管理文章信息 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
Returns: | |
dict: 内容管理文集信息。 | |
""" | |
api = API["upload-manager"]["article_list"] | |
return await Api(**api, credential=credential).result | |
""" | |
comment manager | |
https://member.bilibili.com/platform/comment | |
""" | |
async def get_comments( | |
credential: Credential, | |
oid: Optional[int] = None, | |
keyword: Optional[str] = None, | |
archive_type: ArchiveType = ArchiveType.VIDEO, | |
order: CommentManagerOrder = CommentManagerOrder.RECENTLY, | |
filter: int = -1, | |
pn: int = 1, | |
ps: int = 10, | |
charge_plus_filter: bool = False, | |
) -> dict: | |
""" | |
获取评论 | |
Args: | |
credentials (Credential): Credential 凭据。 | |
oid (Optional, int): 指定稿件 | |
keyword (Optional, str): 关键词 | |
archive_type (ArchiveType): 稿件类型 | |
order (CommentManagerOrder): 排序字段 | |
filter (int): 筛选器,作用未知 | |
pn (int): 页码 | |
ps (int): 每页项数 | |
charge_plus_filter (bool): charge_plus_filter | |
Returns: | |
dict: 评论管理评论信息。 | |
""" | |
params = { | |
"order": order.value, | |
"filter": filter, | |
"type": archive_type.value, | |
"pn": pn, | |
"ps": ps, | |
"charge_plus_filter": charge_plus_filter, | |
} | |
if keyword is not None: | |
params["keyword"] = keyword | |
if oid is not None: | |
params["oid"] = oid | |
api = API["comment-manager"]["fulllist"] | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def del_comments( | |
credential: Credential, | |
oid: Union[int, List[int]], | |
rpid: Union[int, List[int]], | |
archive_type: ArchiveType = ArchiveType.VIDEO, | |
): | |
""" | |
删除评论 | |
每个评论对应一个 oid | |
Args: | |
credentials (Credential): Credential 凭据。 | |
oid (int, lsit): 指定稿件 | |
rpid (int, lsit): 指定评论 | |
archive_type (ArchiveType): 稿件类型 | |
""" | |
data = { | |
"oid": ",".join(oid) if isinstance(oid, list) else oid, | |
"type": archive_type.value, | |
"rpid": ",".join(rpid) if isinstance(rpid, list) else rpid, | |
"jsonp": "jsonp", | |
"csrf": credential.bili_jct, | |
} | |
api = API["comment-manager"]["del"] | |
return await Api(**api, credential=credential).update_data(**data).result | |
""" | |
danmaku manager | |
https://member.bilibili.com/platform/inter-active/danmu | |
""" | |
async def get_recently_danmakus( | |
credential: Credential, pn: int = 1, ps: int = 50 | |
) -> dict: | |
""" | |
最近弹幕 | |
Args: | |
credential (Credential): Credential 凭据。 | |
pn (int): 页码。 | |
ps (int): 每页项数。 | |
Returns: | |
dict: 弹幕管理最近弹幕信息。 | |
""" | |
params = {"pn": pn, "ps": ps} | |
api = API["danmaku-manager"]["recent"] | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def get_danmakus( | |
credential: Credential, | |
oid: int, | |
select_type: DanmakuType = DanmakuType.ALL, | |
archive_type: ArchiveType = ArchiveType.VIDEO, | |
mids: Optional[Union[int, List[int]]] = None, | |
keyword: Optional[str] = None, | |
progress_from: Optional[int] = None, | |
progress_to: Optional[int] = None, | |
ctime_from: Optional[datetime] = None, | |
ctime_to: Optional[datetime] = None, | |
modes: Optional[Union[DanmakuMode, List[DanmakuMode]]] = None, | |
pools: Optional[Union[DanmakuPool, List[DanmakuPool]]] = None, | |
attrs=None, # 未知参数,我在高级筛选里面找不到 | |
order: DanmakuOrder = DanmakuOrder.CTIME, | |
sort: DanmakuSort = DanmakuSort.DESC, | |
pn: int = 1, | |
ps: int = 50, | |
cp_filter: bool = False, | |
) -> dict: | |
""" | |
弹幕搜索 | |
Args: | |
credential (Credential): Credential 凭据 | |
oid (int): 稿件oid,用逗号分隔 | |
select_type (DanmakuType): 弹幕类型 | |
archive_type (ArchiveType): 稿件类型 | |
mids (list[int], int): 用户mids,用逗号分隔或者直接 int | |
keyword (str): 关键词 | |
progress_from (int): 进度开始 | |
progress_to (int): 进度结束 | |
ctime_from (datetime.datetime): 创建时间起始 | |
ctime_to (datetime.datetime): 创建时间结束 | |
modes (DanmakuMode): 弹幕模式。 | |
pool (DanmakuPool): 弹幕池 | |
attrs (Unknown): 弹幕属性,未知参数 | |
order (DanmakuOrder): 排序字段 | |
sort (DanmakuSort): 排序方式 | |
pn (int): 页码。 | |
ps (int): 每页项数。 | |
cp_filter (bool): 是否过滤CP弹幕。未知参数,默认为 False | |
Returns: | |
dict: 弹幕搜索结果 | |
""" | |
params = { | |
"oid": oid, | |
"type": archive_type.value, | |
"mids": ",".join(mids) if isinstance(mids, list) else mids, | |
"select_type": select_type.value, | |
"keyword": keyword, | |
"progress_from": progress_from, | |
"progress_to": progress_to, | |
"ctime_from": ctime_from.strftime("%d-%m-%Y %H:%M:%S") | |
if ctime_from is not None | |
else None, | |
"ctime_to": ctime_to.strftime("%d-%m-%Y %H:%M:%S") | |
if ctime_to is not None | |
else None, | |
"modes": ( | |
",".join([mode.value for mode in modes]) | |
if isinstance(modes, list) | |
else modes.value | |
) | |
if modes is not None | |
else None, | |
"pool": ( | |
",".join([pool.value for pool in pools]) | |
if isinstance(pools, list) | |
else pools.value | |
) | |
if pools is not None | |
else None, | |
"attrs": attrs, | |
"order": order.value, | |
"sort": sort.value, | |
"pn": pn, | |
"ps": ps, | |
"cp_filter": cp_filter, | |
} | |
api = API["danmaku-manager"]["search"] | |
return await Api(**api, credential=credential).update_params(**params).result | |
async def del_danmaku( | |
credential: Credential, oid: int, dmids: Union[int, List[int]] | |
) -> dict: | |
""" | |
删除弹幕 | |
Args: | |
oid (int): 稿件 oid | |
dmids (list[int], int): 弹幕 id,可以传入列表和 int | |
""" | |
return await edit_danmaku_state(credential=credential, oid=oid, dmids=dmids, state=1) | |
async def edit_danmaku_state( | |
credential: Credential, | |
oid: int, | |
dmids: Union[int, List[int]], | |
state: Optional[int] = None, | |
) -> dict: | |
""" | |
操作弹幕状态 | |
Args: | |
oid (int): 稿件 oid | |
dmids (list[int], int): 弹幕 id,可以传入列表和 int | |
state (int, Optional): 弹幕状态 1 删除 2 保护 3 取消保护 | |
Returns: | |
dict: API 返回信息 | |
""" | |
data = { | |
"type": 1, | |
"oid": oid, | |
"dmids": ",".join(dmids) if isinstance(dmids, list) else dmids, | |
"state": state, | |
} | |
api = API["danmaku-manager"]["state"] | |
return await Api(**api, credential=credential).update_data(**data).result | |
async def edit_danmaku_pool( | |
credential: Credential, | |
oid: int, | |
dmids: Union[int, List[int]], | |
is_subtitle: bool = True, | |
) -> dict: | |
""" | |
操作弹幕池 | |
Args: | |
oid (int): 稿件 oid | |
dmids (list[int], int): 弹幕 id,可以传入列表和 int | |
is_subtitle (bool): 是否为字幕 | |
Returns: | |
dict: API 返回信息 | |
""" | |
data = { | |
"type": 1, | |
"oid": oid, | |
"dmids": ",".join(dmids) if isinstance(dmids, list) else dmids, | |
"pool": 1 if is_subtitle else 0, | |
} | |
api = API["danmaku-manager"]["pool"] | |
return await Api(**api, credential=credential).update_data(**data).result | |