Spaces:
Sleeping
Sleeping
File size: 4,570 Bytes
0aee47a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
"""
bilibili_api.topic
话题相关
"""
from enum import Enum
from typing import Union, Optional
from . import dynamic
from .user import get_self_info
from .utils.utils import get_api
from .utils.credential import Credential
from .utils.network import Api
API = get_api("topic")
class TopicCardsSortBy(Enum):
"""
话题下内容排序方式
+ NEW: 最新
+ HOT: 最热
+ RECOMMEND: 推荐
"""
NEW = 3
HOT = 2
RECOMMEND = 1
async def get_hot_topics(numbers: int = 33) -> dict:
"""
获取动态页的火热话题
Args:
numbers (int): 话题数量. Defaults to 33.
Returns:
dict: 调用 API 返回的结果
"""
api = API["info"]["dynamic_page_topics"]
params = {"page_size": numbers}
return await Api(**api).update_params(**params).result
async def search_topic(keyword: str, ps: int = 20, pn: int = 1) -> dict:
"""
搜索话题
从动态页发布动态处的话题搜索框搜索话题
Args:
keyword (str): 搜索关键词
ps (int): 每页数量. Defaults to 20.
pn (int): 页数. Defaults to 1.
Returns:
dict: 调用 API 返回的结果
"""
api = API["info"]["search"]
params = {"keywords": keyword, "page_size": ps, "page_num": pn}
return await Api(**api).update_params(**params).result
class Topic:
"""
话题类
Attributes:
credential (Credential): 凭据类
"""
def __init__(self, topic_id: int, credential: Union[Credential, None] = None):
"""
Args:
topic_id (int) : 话题 id
credential (Credential): 凭据类
"""
self.__topic_id = topic_id
self.credential = credential if credential else Credential()
def get_topic_id(self) -> int:
"""
获取话题 id
Returns:
int: 话题 id
"""
return self.__topic_id
async def get_info(self) -> dict:
"""
获取话题简介
Returns:
dict: 调用 API 返回的结果
"""
api = API["info"]["info"]
params = {"topic_id": self.get_topic_id()}
return (
await Api(**api, credential=self.credential).update_params(**params).result
)
async def get_cards(
self,
ps: int = 100,
offset: Optional[str] = None,
sort_by: TopicCardsSortBy = TopicCardsSortBy.HOT,
) -> dict:
"""
获取话题下的内容
未登录无法使用热门排序字段即 TopicCardsSortBy.RECOMMEND
Args:
ps (int): 数据数量. Defaults to 100.
offset (Optional[str]): 偏移量. 生成格式为 f'{页码}_{页码*数据量]}' 如'2_40' Defaults to None.
sort_by (TopicCardsSortBy): 排序方式. Defaults to TopicCardsSortBy.HOT.
Returns:
dict: 调用 API 返回的结果
"""
api = API["info"]["cards"]
params = {
"topic_id": self.get_topic_id(),
"page_size": ps,
"sort_by": sort_by.value
}
if offset:
params.update({"offset": offset})
return (
await Api(**api, credential=self.credential).update_params(**params).result
)
async def like(self, status: bool = True) -> dict:
"""
设置点赞话题
Args:
status (bool): 是否设置点赞. Defaults to True.
Returns:
dict: 调用 API 返回的结果
"""
api = API["operate"]["like"]
data = {
"topic_id": self.get_topic_id(),
"action": "like" if status else "cancel_like",
"business": "topic",
"up_mid": (await get_self_info(self.credential))["mid"],
}
return await Api(**api, credential=self.credential).update_data(**data).result
async def set_favorite(self, status: bool = True) -> dict:
"""
设置收藏话题
Args:
status (bool): 是否设置收藏. Defaults to True.
Returns:
dict: 调用 API 返回的结果
"""
api = API["operate"]["add_favorite" if status else "cancel_favorite"]
data = {"topic_id": self.get_topic_id()}
return await Api(**api, credential=self.credential).update_data(**data).result
|