rogerxavier's picture
Upload 258 files
0aee47a verified
raw
history blame
10.3 kB
"""
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