Open Drelf2018 opened 3 months ago
z0z0r4
commented
3 months ago 基本上我是没空了，围观
直接从数据文件中读取
这个信息量有点大啊，接口定义有示例吗？
这个 class Api: 看的比较迷糊，自动导入 URL 和 method ？
最妙的是能识别到原来函数上标注的返回值类型
现在的函数不能拿到返回值类型吗...？没太看懂，还是说包装了装饰器之后
没看出啥意义，期待讲解，以 https://github.com/Nemo2011/bilibili-api/blob/48e6350ccd9596dadf65a6443927c9ccfcdfd6c6/bilibili_api/user.py#L529-L547 为例
顶多把三行的接口调用变成了一行（假如 class Api: 是通用的...
Drelf2018
commented
3 months ago import asyncio
from inspect import Parameter, signature
import httpx
from bilibili_api import HEADERS
class Api:
def __init__(self, method: str, url: str):
self.method = method
self.url = url
self.params = {}
self.data = {}
self.headers = HEADERS.copy()
def __call__(self, fn):
for key, param in signature(fn).parameters.items():
val = param.default
if val is Parameter.empty:
continue
if key == "data":
self.data = val
elif key == "headers":
self.headers.update(val)
else:
self.params[key] = val
return self.request
async def request(self, **kwargs):
data = self.data.copy()
if "data" in kwargs:
data.update(kwargs.pop("data"))
headers = self.headers.copy()
if "headers" in kwargs:
headers.update(kwargs.pop("headers"))
params = self.params.copy()
params.update(kwargs)
return httpx.request(self.method, self.url, params=params, data=data, headers=headers).json()["data"]
class Get(Api):
def __init__(self, url: str):
return super().__init__("Get", url)
@Api("GET", "https://api.bilibili.com/x/web-interface/view")
async def get_video_info(*, bvid: str) -> dict: ...
@Get("https://api.bilibili.com/x/v2/reply")
async def get_replies(*, oid: str, type: int = 11) -> dict: ...
async def main():
bvid = "BV1gz421m7mk"
info = await get_video_info(bvid=bvid)
print(f'{bvid} title: {info.get("title")}')
data = await get_replies(oid="315871473")
for reply in data["replies"]:
print(reply["member"]["uname"]+": "+reply["content"]["message"])
break
data = await get_replies(oid="909825870340816915", type=17)
for reply in data["replies"]:
print(reply["member"]["uname"]+": "+reply["content"]["message"])
break
if __name__ == "__main__":
asyncio.run(main())BV1gz421m7mk title: 小羊🐧振翅【Animelo Summer Live 2023 MyGO!!!!!部分】
-Dashstar-: ( ゜- ゜)つロ
死鱼眼八幡: 急急急急急急import asyncio
import json
import re
from enum import Enum, IntFlag
from functools import partial
from inspect import Parameter, signature
from bilibili_api import (HEADERS, Credential, ResponseCodeException, enc_wbi, get_mixin_key,
get_session)
from bilibili_api.comment import CommentResourceType, OrderType
from bilibili_api.user import AudioOrder, VideoOrder
class Flag(IntFlag):
WBI = 1 << 0
VERIFY = 1 << 1
NO_CSRF = 1 << 2
JSON_BODY = 1 << 3
IGNORE_CODE = 1 << 4
class Api:
def __init__(self, url: str, flag: Flag = 0):
self.method = self.__class__.__name__.upper()
self.url = url
self.flag= flag
self.params = {}
self.data = {}
self.headers = HEADERS.copy()
self.credential = Credential()
@property
def wbi(self):
return self.flag & Flag.WBI != 0
@property
def verify(self):
return self.flag & Flag.VERIFY != 0
@property
def no_csrf(self):
return self.flag & Flag.NO_CSRF != 0
@property
def json_body(self):
return self.flag & Flag.JSON_BODY != 0
@property
def ignore_code(self):
return self.flag & Flag.IGNORE_CODE != 0
def __call__(self, fn):
for key, param in signature(fn).parameters.items():
val = param.default
if val is Parameter.empty:
continue
if key == "data":
self.data = val
elif key == "headers":
self.headers.update(val)
elif key == "credential":
if val is not None:
self.credential = val
else:
if isinstance(val, Enum):
val = val.value
self.params[key] = val
return self.wrapper
async def wrapper(self, **kwargs):
data = self.data.copy()
if "data" in kwargs:
data.update(kwargs.pop("data"))
headers = self.headers.copy()
if "headers" in kwargs:
headers.update(kwargs.pop("headers"))
if "credential" in kwargs:
self.credential = kwargs.pop("credential")
params = self.params.copy()
for key, val in kwargs.items():
if isinstance(val, Enum):
val = val.value
params[key] = val
return await self.request(params=params, data=data, headers=headers)
async def request(self, params: dict, data: dict, headers: dict):
"""
向接口发送请求。
Returns:
接口未返回数据时，返回 None，否则返回该接口提供的 data 或 result 字段的数据。
"""
# 如果接口需要 Credential 且未传入则报错 (默认值为 Credential())
if self.verify:
self.credential.raise_for_no_sessdata()
# 请求为非 GET 且 no_csrf 不为 True 时要求 bili_jct
if self.method != "GET" and not self.no_csrf:
self.credential.raise_for_no_bili_jct()
# if settings.request_log:
# settings.logger.info(self)
# jsonp
if params.get("jsonp") == "jsonp":
params["callback"] = "callback"
if self.wbi:
# global wbi_mixin_key
# if wbi_mixin_key == "":
# wbi_mixin_key = await get_mixin_key()
# enc_wbi(params, wbi_mixin_key)
enc_wbi(params, await get_mixin_key())
# 自动添加 csrf
if not self.no_csrf and self.method in ["POST", "DELETE", "PATCH"]:
data["csrf"] = self.credential.bili_jct
data["csrf_token"] = self.credential.bili_jct
cookies = self.credential.get_cookies()
# if self.credential.buvid3 is None:
# global buvid3
# if buvid3 == "" and self.url != API["info"]["spi"]["url"]:
# buvid3 = (await get_spi_buvid())["b_3"]
# cookies["buvid3"] = buvid3
# else:
# cookies["buvid3"] = self.credential.buvid3
cookies["Domain"] = ".bilibili.com"
config = {
"url": self.url,
"method": self.method,
"data": data,
"params": params,
# "files": self.files,
"cookies": cookies,
"headers": headers,
}
# config.update(kwargs)
# print(config)
if self.json_body:
config["headers"]["Content-Type"] = "application/json"
config["data"] = json.dumps(config["data"])
session = get_session()
resp = await session.request(**config)
# 检查响应头 Content-Length
content_length = resp.headers.get("content-length")
if content_length and int(content_length) == 0:
return None
if "callback" in params:
# JSONP 请求
resp_data: dict = json.loads(re.match("^.*?({.*}).*$", resp.text, re.S).group(1))
else:
# JSON
resp_data: dict = json.loads(resp.text)
# 检查 code
if not self.ignore_code:
code = resp_data.get("code")
if code is None:
raise ResponseCodeException(-1, "API 返回数据未含 code 字段", resp_data)
if code != 0:
msg = resp_data.get("msg")
if msg is None:
msg = resp_data.get("message")
if msg is None:
msg = "接口未返回错误信息"
raise ResponseCodeException(code, msg, resp_data)
# elif settings.request_log:
# settings.logger.info(resp_data)
real_data = resp_data.get("data")
if real_data is None:
real_data = resp_data.get("result")
return real_data
class Get(Api):
"""
Get method api
"""
@Get("https://api.bilibili.com/x/web-interface/view")
async def get_info(*, bvid: str) -> dict:
"""
获取视频信息。
Returns:
dict: 调用 API 返回的结果。
"""
@Get("https://api.bilibili.com/x/v2/reply")
async def get_comments(
*,
oid: int,
type: CommentResourceType,
pn: int = 1,
order: OrderType = OrderType.TIME,
credential: Credential = None
) -> dict:
"""
获取资源评论列表。
第二页以及往后需要提供 `credential` 参数。
Args:
oid (int) : 资源 ID。
type (CommentsResourceType) : 资源类枚举。
pn (int, optional) : 页码. Defaults to 1.
order (OrderType, optional) : 排序方式枚举. Defaults to OrderType.TIME.
credential (Credential, optional) : 凭据。Defaults to None.
Returns:
dict: 调用 API 返回的结果
"""
class User:
def __init__(self, uid: int, credential: Credential = None):
self.get_audios = partial(self.get_audios, uid=uid, credential=credential)
self.get_videos = partial(self.get_videos, mid=uid, credential=credential)
@Get("https://api.bilibili.com/audio/music-service/web/song/upper")
async def get_audios(self, *, pn: int = 1, ps: int = 30, order: AudioOrder = AudioOrder.PUBDATE) -> dict:
"""
获取用户投稿音频。
Args:
order (AudioOrder, optional): 排序方式. Defaults to AudioOrder.PUBDATE.
pn (int, optional) : 页码数，从 1 开始。 Defaults to 1.
ps (int, optional) : 每一页的视频数. Defaults to 30.
Returns:
dict: 调用接口返回的内容。
"""
@Get("https://api.bilibili.com/x/space/wbi/arc/search", Flag.WBI)
async def get_videos(self, *,
tid: int = 0,
pn: int = 1,
ps: int = 30,
keyword: str = "",
order: VideoOrder = VideoOrder.PUBDATE,
web_location: int = 1550101,
dm_img_list: str = "[]",
dm_img_str: str = "V2ViR0wgMS4wIChPcGVuR0wgRVMgMi4wIENocm9taXVtKQ",
dm_cover_img_str: str = "QU5HTEUgKEludGVsLCBJbnRlbChSKSBVSEQgR3JhcGhpY3MgNjMwICgweDAwMDAzRTlCKSBEaXJlY3QzRDExIHZzXzVfMCBwc181XzAsIEQzRDExKUdvb2dsZSBJbmMuIChJbnRlbC",
) -> dict:
"""
获取用户投稿视频信息。
Args:
tid (int, optional) : 分区 ID. Defaults to 0（全部）.
pn (int, optional) : 页码，从 1 开始. Defaults to 1.
ps (int, optional) : 每一页的视频数. Defaults to 30.
keyword (str, optional) : 搜索关键词. Defaults to "".
order (VideoOrder, optional): 排序方式. Defaults to VideoOrder.PUBDATE
Returns:
dict: 调用接口返回的内容。
"""
async def main():
credential = Credential(...)
user = User(12344667, credential=credential)
data = await user.get_videos()
for v in data["list"]["vlist"]:
print("【"+v["title"]+"】封面:", v["pic"])
break
data = await get_comments(oid=909825870340816915, type=CommentResourceType.DYNAMIC)
for reply in data["replies"]:
print(reply["member"]["uname"]+":", reply["content"]["message"])
break
if __name__ == "__main__":
asyncio.run(main())【手术前一天的神楽mea】封面: http://i1.hdslb.com/bfs/archive/f302d499a075067eb5a90ff7ddd498348dbac278.jpg
死鱼眼八幡: 急急急急急急
v0.1.0
这是一段我想实现的代码，是能获取到结果的，最妙的是能识别到原来函数上标注的返回值类型。
函数体内部不需要写任何代码，只要在参数那里标注好类型，剩下的就会直接从数据文件中读取。
调用时必须要通过指名的方式，这样才能把值放在正确的
key里面。