Spaces:
Sleeping
Sleeping
""" | |
bilibili_api.search | |
搜索 | |
""" | |
import json | |
from enum import Enum | |
from typing import List, Union, Callable | |
from .utils.utils import get_api | |
from .video_zone import VideoZoneTypes | |
from .utils.network import Api, get_session | |
from .utils.credential import Credential | |
API = get_api("search") | |
class SearchObjectType(Enum): | |
""" | |
搜索对象。 | |
+ VIDEO : 视频 | |
+ BANGUMI : 番剧 | |
+ FT : 影视 | |
+ LIVE : 直播 | |
+ ARTICLE : 专栏 | |
+ TOPIC : 话题 | |
+ USER : 用户 | |
+ LIVEUSER : 直播间用户 | |
""" | |
VIDEO = "video" | |
BANGUMI = "media_bangumi" | |
FT = "media_ft" | |
LIVE = "live" | |
ARTICLE = "article" | |
TOPIC = "topic" | |
USER = "bili_user" | |
LIVEUSER = "live_user" | |
PHOTO = "photo" | |
class OrderVideo(Enum): | |
""" | |
视频搜索类型 | |
+ TOTALRANK : 综合排序 | |
+ CLICK : 最多点击 | |
+ PUBDATE : 最新发布 | |
+ DM : 最多弹幕 | |
+ STOW : 最多收藏 | |
+ SCORES : 最多评论 | |
Ps: Api 中 的 order_sort 字段决定顺序还是倒序 | |
""" | |
TOTALRANK = "totalrank" | |
CLICK = "click" | |
PUBDATE = "pubdate" | |
DM = "dm" | |
STOW = "stow" | |
SCORES = "scores" | |
class OrderLiveRoom(Enum): | |
""" | |
直播间搜索类型 | |
+ NEWLIVE 最新开播 | |
+ ONLINE 综合排序 | |
""" | |
NEWLIVE = "live_time" | |
ONLINE = "online" | |
class OrderArticle(Enum): | |
""" | |
文章的排序类型 | |
+ TOTALRANK : 综合排序 | |
+ CLICK : 最多点击 | |
+ PUBDATE : 最新发布 | |
+ ATTENTION : 最多喜欢 | |
+ SCORES : 最多评论 | |
""" | |
TOTALRANK = "totalrank" | |
PUBDATE = "pubdate" | |
CLICK = "click" | |
ATTENTION = "attention" | |
SCORES = "scores" | |
class OrderUser(Enum): | |
""" | |
搜索用户的排序类型 | |
+ FANS : 按照粉丝数量排序 | |
+ LEVEL : 按照等级排序 | |
""" | |
FANS = "fans" | |
LEVEL = "level" | |
class OrderCheese(Enum): | |
""" | |
课程搜索排序类型 | |
+ RECOMMEND: 综合 | |
+ SELL : 销量最高 | |
+ NEW : 最新上架 | |
+ CHEEP : 售价最低 | |
""" | |
RECOMMEND = -1 | |
SELL = 1 | |
NEW = 2 | |
CHEEP = 3 | |
class CategoryTypePhoto(Enum): | |
""" | |
相册分类 | |
+ All 全部 | |
+ DrawFriend 画友 | |
+ PhotoFriend 摄影 | |
""" | |
All = 0 | |
DrawFriend = 2 | |
PhotoFriend = 1 | |
class CategoryTypeArticle(Enum): | |
""" | |
文章分类 | |
+ All 全部 | |
+ Anime 动画 | |
+ Game 游戏 | |
+ TV 电视 | |
+ Life 生活 | |
+ Hobby 兴趣 | |
+ LightNovel 轻小说 | |
+ Technology 科技 | |
""" | |
All = 0 | |
Anime = 2 | |
Game = 1 | |
TV = 28 | |
Life = 3 | |
Hobby = 29 | |
LightNovel = 16 | |
Technology = 17 | |
async def search(keyword: str, page: int = 1) -> dict: | |
""" | |
只指定关键字在 web 进行搜索,返回未经处理的字典 | |
Args: | |
keyword (str): 搜索关键词 | |
page (int): 页码. Defaults to 1. | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
api = API["search"]["web_search"] | |
params = {"keyword": keyword, "page": page} | |
return await Api(**api).update_params(**params).result | |
async def search_by_type( | |
keyword: str, | |
search_type: Union[SearchObjectType, None] = None, | |
order_type: Union[OrderUser, OrderLiveRoom, OrderArticle, OrderVideo, None] = None, | |
time_range: int = -1, | |
video_zone_type: Union[int, VideoZoneTypes, None] = None, | |
order_sort: Union[int, None] = None, | |
category_id: Union[CategoryTypeArticle, CategoryTypePhoto, int, None] = None, | |
page: int = 1, | |
page_size: int = 42, | |
debug_param_func: Union[Callable, None] = None, | |
) -> dict: | |
""" | |
指定分区,类型,视频长度等参数进行搜索,返回未经处理的字典 | |
类型:视频(video)、番剧(media_bangumi)、影视(media_ft)、直播(live)、直播用户(liveuser)、专栏(article)、话题(topic)、用户(bili_user) | |
Args: | |
debug_param_func (Callable | None, optional) : 参数回调器,用来存储或者什么的 | |
order_sort (int | None, optional) : 用户粉丝数及等级排序顺序 默认为0 由高到低:0 由低到高:1 | |
category_id (CategoryTypeArticle | CategoryTypePhoto | int | None, optional) : 专栏/相簿分区筛选,指定分类,只在相册和专栏类型下生效 | |
time_range (int, optional) : 指定时间,自动转换到指定区间,只在视频类型下生效 有四种:10分钟以下,10-30分钟,30-60分钟,60分钟以上 | |
video_zone_type (int | ZoneTypes | None, optional) : 话题类型,指定 tid (可使用 channel 模块查询) | |
order_type (OrderUser | OrderLiveRoom | OrderArticle | OrderVideo | None, optional): 排序分类类型 | |
keyword (str) : 搜索关键词 | |
search_type (SearchObjectType | None, optional) : 搜索类型 | |
page (int, optional) : 页码 | |
page_size (int, optional) : 每一页的数据大小 | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
params = {"keyword": keyword, "page": page, "page_size": page_size} | |
if search_type: | |
params["search_type"] = search_type.value | |
else: | |
raise ValueError("Missing arg:search_type") | |
# params["search_type"] = SearchObjectType.VIDEO.value | |
# category_id | |
if ( | |
search_type.value == SearchObjectType.ARTICLE.value | |
or search_type.value == SearchObjectType.PHOTO.value | |
): | |
if category_id: | |
if isinstance(category_id, int): | |
params["category_id"] = category_id | |
else: | |
params["category_id"] = category_id.value | |
# time_code | |
if search_type.value == SearchObjectType.VIDEO.value: | |
if time_range > 60: | |
time_code = 4 | |
elif 30 < time_range <= 60: | |
time_code = 3 | |
elif 10 < time_range <= 30: | |
time_code = 2 | |
elif 0 < time_range <= 10: | |
time_code = 1 | |
else: | |
time_code = 0 | |
params["duration"] = time_code | |
# zone_type | |
if video_zone_type: | |
if isinstance(video_zone_type, int): | |
params["tids"] = video_zone_type | |
elif isinstance(video_zone_type, VideoZoneTypes): | |
params["tids"] = video_zone_type.value | |
else: | |
params["tids"] = video_zone_type | |
# order_type | |
if order_type: | |
params["order"] = order_type.value | |
# order_sort | |
if search_type.value == SearchObjectType.USER.value: | |
params["order_sort"] = order_sort | |
if debug_param_func: | |
debug_param_func(params) | |
api = API["search"]["web_search_by_type"] | |
return await Api(**api).update_params(**params).result | |
async def get_default_search_keyword() -> dict: | |
""" | |
获取默认的搜索内容 | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
api = API["search"]["default_search_keyword"] | |
return await Api(**api).result | |
async def get_hot_search_keywords() -> dict: | |
""" | |
获取热搜 | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
api = API["search"]["hot_search_keywords"] | |
sess = get_session() | |
return json.loads((await sess.request("GET", api["url"])).text) | |
async def get_suggest_keywords(keyword: str) -> List[str]: | |
""" | |
通过一些文字输入获取搜索建议。类似搜索词的联想。 | |
Args: | |
keyword(str): 搜索关键词 | |
Returns: | |
List[str]: 关键词列表 | |
""" | |
keywords = [] | |
sess = get_session() | |
api = API["search"]["suggest"] | |
params = {"term": keyword} | |
res = await Api(**api).update_params(**params).result | |
for key in res["tag"]: | |
keywords.append(key["value"]) | |
return keywords | |
async def search_games(keyword: str) -> dict: | |
""" | |
搜索游戏特用函数 | |
Args: | |
keyword (str): 搜索关键词 | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
api = API["search"]["game"] | |
params = {"keyword": keyword} | |
return await Api(**api).update_params(**params).result | |
async def search_manga( | |
keyword: str, page_num: int = 1, page_size: int = 9, credential: Credential = None | |
): | |
""" | |
搜索漫画特用函数 | |
Args: | |
keyword (str): 搜索关键词 | |
page_num (int): 页码. Defaults to 1. | |
page_size (int): 每一页的数据大小. Defaults to 9. | |
credential (Credential): 凭据类. Defaults to None. | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
credential = credential if credential else Credential() | |
api = API["search"]["manga"] | |
data = {"key_word": keyword, "page_num": page_num, "page_size": page_size} | |
return ( | |
await Api(**api, credential=credential, no_csrf=True).update_data(**data).result | |
) | |
async def search_cheese( | |
keyword: str, | |
page_num: int = 1, | |
page_size: int = 30, | |
order: OrderCheese = OrderCheese.RECOMMEND, | |
): | |
""" | |
搜索课程特用函数 | |
Args: | |
keyword (str) : 搜索关键词 | |
page_num (int) : 页码. Defaults to 1. | |
page_size (int) : 每一页的数据大小. Defaults to 30. | |
order (OrderCheese): 排序方式. Defaults to OrderCheese.RECOMMEND | |
Returns: | |
dict: 调用 API 返回的结果 | |
""" | |
api = API["search"]["cheese"] | |
params = { | |
"word": keyword, | |
"page": page_num, | |
"page_size": page_size, | |
"sort_type": order.value, | |
} | |
return await Api(**api).update_params(**params).result | |