rogerxavier's picture
Upload 258 files
0aee47a verified
"""
bilibili_api.comment
评论相关。
关于资源 ID(oid)的一些示例({}部分为应该传入的参数)。
+ 视频:AV 号:av{170001}。
+ 专栏:cv{9762979}。
+ 动态(画册类型):{116859542}。
+ 动态(纯文本):{497080393649439253}。
+ 课程:ep{5556}
+ 小黑屋: ban/{2600321}
"""
from enum import Enum
from typing import Union, Optional
from .utils.utils import get_api
from .utils.credential import Credential
from .utils.network import Api
from .exceptions.ArgsException import ArgsException
API = get_api("common")
class CommentResourceType(Enum):
"""
资源类型枚举。
+ VIDEO: 视频。
+ ARTICLE: 专栏。
+ DYNAMIC_DRAW: 画册。
+ DYNAMIC: 动态(画册也属于动态的一种,只不过画册还有一个专门的 ID)。
+ AUDIO:音频。
+ AUDIO_LIST:歌单。
+ CHEESE: 课程
+ BLACK_ROOM: 小黑屋
+ MANGA: 漫画
"""
VIDEO = 1
ARTICLE = 12
DYNAMIC_DRAW = 11
DYNAMIC = 17
AUDIO = 14
AUDIO_LIST = 19
CHEESE = 33
BLACK_ROOM = 6
MANGA = 22
class OrderType(Enum):
"""
评论排序方式枚举。
+ LIKE:按点赞数倒序。
+ TIME:按发布时间倒序。
"""
LIKE = 2
TIME = 0
class ReportReason(Enum):
"""
举报类型枚举
+ OTHER: 其他
+ SPAM_AD: 垃圾广告
+ PORNOGRAPHY: 色情
+ FLOOD: 刷屏
+ PROVOCATION: 引战
+ SPOILER: 剧透
+ POLITICS: 政治
+ PERSONAL_ATTACK: 人身攻击
+ IRRELEVANT_CONTENT: 内容不相关
+ ILLEGAL: 违法违规
+ VULGAR: 低俗
+ ILLEGAL_WEBSITE: 非法网站
+ GAMBLING_FRAUD: 赌博诈骗
+ SPREADING_FALSE_INFORMATION: 传播不实信息
+ INCITING_INFORMATION: 怂恿教唆信息
+ PRIVACY_VIOLATION: 侵犯隐私
+ FLOOR_TAKING: 抢楼
+ INAPPROPRIATE_CONTENT_FOR_MINORS: 青少年不良信息
"""
OTHER = 0
SPAM_AD = 1
PORNOGRAPHY = 2
FLOOD = 3
PROVOCATION = 4
SPOILER = 5
POLITICS = 6
PERSONAL_ATTACK = 7
IRRELEVANT_CONTENT = 8
ILLEGAL = 9
VULGAR = 10
ILLEGAL_WEBSITE = 11
GAMBLING_FRAUD = 12
SPREADING_FALSE_INFORMATION = 13
INCITING_INFORMATION = 14
PRIVACY_VIOLATION = 15
FLOOR_TAKING = 16
INAPPROPRIATE_CONTENT_FOR_MINORS = 17
class Comment:
"""
对单条评论的相关操作。
Attributes:
credential (Credential): 凭据类
"""
def __init__(
self,
oid: int,
type_: CommentResourceType,
rpid: int,
credential: Union[Credential, None] = None,
):
"""
Args:
oid (int) : 评论所在资源 ID。
type_ (ResourceType): 评论所在资源类型枚举。
rpid (int) : 评论 ID。
credential (Credential) : 凭据类. Defaults to None.
"""
self.__oid = oid
self.__rpid = rpid
self.__type = type_
self.credential = credential if credential else Credential()
def __get_data(self, status: bool) -> dict:
"""
获取通用请求载荷。
Args:
status (bool): 状态。
Returns:
dict: 请求载荷数据。
"""
return {
"oid": self.__oid,
"type": self.__type.value,
"rpid": self.__rpid,
"action": 1 if status else 0,
}
def get_rpid(self) -> int:
return self.__rpid
def get_type(self) -> CommentResourceType:
return self.__type
def get_oid(self) -> int:
return self.__oid
async def like(self, status: bool = True) -> dict:
"""
点赞评论。
Args:
status (bool, optional): 状态, Defaults to True.
Returns:
dict: 调用 API 返回的结果
"""
self.credential.raise_for_no_sessdata()
self.credential.raise_for_no_bili_jct()
api = API["comment"]["like"]
return (
await Api(**api, credential=self.credential)
.update_data(**self.__get_data(status))
.result
)
async def hate(self, status: bool = True) -> dict:
"""
点踩评论。
Args:
status (bool, optional): 状态, Defaults to True.
Returns:
dict: 调用 API 返回的结果
"""
self.credential.raise_for_no_sessdata()
self.credential.raise_for_no_bili_jct()
api = API["comment"]["hate"]
return (
await Api(**api, credential=self.credential)
.update_data(**self.__get_data(status))
.result
)
async def pin(self, status: bool = True) -> dict:
"""
置顶评论。
Args:
status (bool, optional): 状态, Defaults to True.
Returns:
dict: 调用 API 返回的结果
"""
self.credential.raise_for_no_sessdata()
self.credential.raise_for_no_bili_jct()
api = API["comment"]["pin"]
return (
await Api(**api, credential=self.credential)
.update_data(**self.__get_data(status))
.result
)
async def delete(self) -> dict:
"""
删除评论。
Returns:
dict: 调用 API 返回的结果
"""
self.credential.raise_for_no_sessdata()
self.credential.raise_for_no_bili_jct()
api = API["comment"]["del"]
data = self.__get_data(True)
del data["action"]
return await Api(**api, credential=self.credential).update_data(**data).result
async def get_sub_comments(self, page_index: int = 1) -> dict:
"""
获取子评论。即评论下的评论。
Args:
page_index (int, optional): 页码索引,从 1 开始。Defaults to 1.
Returns:
dict: 调用 API 返回的结果
"""
if page_index <= 0:
raise ArgsException("page_index 必须大于或等于 1")
api = API["comment"]["sub_reply"]
params = {
"pn": page_index,
"ps": 10,
"type": self.__type.value,
"oid": self.__oid,
"root": self.__rpid,
}
return (
await Api(**api, credential=self.credential).update_params(**params).result
)
async def report(
self, report_reason: ReportReason, content: Optional[str] = None
) -> dict:
"""
举报评论
Args:
report_reason (ReportReason): 举报类型枚举
content (str, optional): 其他举报备注内容仅 reason=ReportReason.OTHER 可用且不能为 None.
Error Code:
0: 成功
-101: 账号未登录
-102: 账号被封停
-111: csrf校验失败
-400: 请求错误
-403: 权限不足
-404: 无此项
-500: 服务器错误
-509: 请求过于频繁
12002: 评论区已关闭
12006: 没有该评论
12008: 已经举报过了
12009: 评论主体的type不合法
12019: 举报过于频繁
12077: 举报理由过长或过短
"""
self.credential.raise_for_no_sessdata()
api = API["comment"]["report"]
if content is not None and report_reason != ReportReason.OTHER:
raise ArgsException("content 只能在 report_reason=ReportReason.OTHER 时使用")
elif content is None and report_reason == ReportReason.OTHER:
raise ArgsException("report_reason=ReportReason.OTHER 时 content 不能为空")
data = {
"oid": self.__oid,
"type": self.__type.value,
"rpid": self.__rpid,
"reason": report_reason.value,
"content": content,
}
return await Api(**api, credential=self.credential).update_data(**data).result
async def send_comment(
text: str,
oid: int,
type_: CommentResourceType,
root: Union[int, None] = None,
parent: Union[int, None] = None,
credential: Union[None, Credential] = None,
) -> dict:
"""
通用发送评论 API。
说明 `root` 和 `parent`,假设评论的是视频,常见的评论有三种情况:
1. 只在视频下面发送评论:root=None, parent=None;
2. 回复视频下面的评论:root=评论 ID, parent=None;
3. 回复视频下面的评论中的评论:root=在哪条评论下评论的 ID, parent=回复哪条评论的 ID。
当 root 为空时,parent 必须为空。
Args:
text (str) : 评论内容。
oid (str) : 资源 ID。
type_ (CommentsResourceType) : 资源类型枚举。
root (int, optional): 根评论 ID, Defaults to None.
parent (int, optional): 父评论 ID, Defaults to None.
credential (Credential) : 凭据
Returns:
dict: 调用 API 返回的结果
"""
if credential is None:
credential = Credential()
credential.raise_for_no_sessdata()
credential.raise_for_no_bili_jct()
data = {
"oid": oid,
"type": type_.value,
"message": text,
"plat": 1,
}
if root is None and parent is None:
# 直接回复资源
pass
elif root is not None and parent is None:
# 回复资源下面的评论
data["root"] = root
data["parent"] = root
elif root is not None and parent is not None:
# 回复资源下面的评论的评论
data["root"] = root
data["parent"] = parent
else:
# root=None 时,parent 不得设置
raise ArgsException("root=None 时,parent 不得设置")
api = API["comment"]["send"]
return await Api(**api, credential=credential).update_data(**data).result
async def get_comments(
oid: int,
type_: CommentResourceType,
page_index: int = 1,
order: OrderType = OrderType.TIME,
credential: Union[Credential, None] = None,
) -> dict:
"""
获取资源评论列表。
第二页以及往后需要提供 `credential` 参数。
Args:
oid (int) : 资源 ID。
type_ (CommentsResourceType) : 资源类枚举。
page_index (int, optional) : 页码. Defaults to 1.
order (OrderType, optional) : 排序方式枚举. Defaults to OrderType.TIME.
credential (Credential, optional): 凭据。Defaults to None.
Returns:
dict: 调用 API 返回的结果
"""
if page_index <= 0:
raise ArgsException("page_index 必须大于或等于 1")
api = API["comment"]["get"]
params = {"pn": page_index, "type": type_.value, "oid": oid, "sort": order.value}
return await Api(**api, credential=credential).update_params(**params).result
async def get_comments_lazy(
oid: int,
type_: CommentResourceType,
# pagination_str: str = "",
pn: int = 1,
ps: int = 20,
order: OrderType = OrderType.TIME,
credential: Union[Credential, None] = None,
) -> dict:
"""
新版获取资源评论列表。
第二次以及往后需要提供 `credential` 参数。
Args:
oid (int) : 资源 ID。
type_ (CommentsResourceType) : 资源类枚举。
pagination_str (str, optional) : 分页依据 Defaults to `{"offset":""}`. 弃用 #658
pn (int, optional) : 页码. Defaults to 1.
ps (int, optional) : 每页数量. Defaults to 20.
order (OrderType, optional) : 排序方式枚举. Defaults to OrderType.TIME.
credential (Credential, optional): 凭据。Defaults to None.
Returns:
dict: 调用 API 返回的结果
"""
api = API["comment"]["reply_by_session_id"]
params = {
"oid": oid,
"type": type_.value,
"mode": order.value,
"next": pn - 1,
"ps": ps,
}
return await Api(**api, credential=credential).update_params(**params).result