rogerxavier's picture
Upload 258 files
0aee47a verified
raw
history blame
6.42 kB
"""
bilibili_api.login_func
登录功能相关
"""
import enum
import threading
from typing import Tuple, Union
from . import login
from .utils.utils import get_api
from .exceptions import LoginError
import urllib.parse
from .utils.picture import Picture
from .utils.credential import Credential
API = get_api("login")
class QrCodeLoginEvents(enum.Enum):
"""
二维码登录状态枚举
+ SCAN: 未扫描二维码
+ CONF: 未确认登录
+ TIMEOUT: 二维码过期
+ DONE: 成功
"""
SCAN = "scan"
CONF = "confirm"
TIMEOUT = "timeout"
DONE = "done"
def get_qrcode() -> Tuple[Picture, str]:
"""
获取二维码及登录密钥(后面有用)
Returns:
Tuple[Picture, str]: 第一项是二维码图片地址(本地缓存)和登录密钥。登录密钥需要保存。
"""
login_data = login.update_qrcode_data()
login_key = login_data["qrcode_key"]
img = login.make_qrcode(login_data["url"])
return (Picture.from_file(img), login_key)
def check_qrcode_events(login_key: str) -> Tuple[QrCodeLoginEvents, Union[str, Credential]]:
"""
检查登录状态。(建议频率 1s,这个 API 也有风控!)
Args:
login_key (str): 登录密钥(get_qrcode 的返回值第二项)
Returns:
Tuple[QrCodeLoginEvents, str|Credential]: 状态(第一项)和信息(第二项)(如果成功登录信息为凭据类)
"""
events = login.login_with_key(login_key)
if events["code"] == 86101:
return QrCodeLoginEvents.SCAN, events["message"]
elif events["code"] == 86090:
return QrCodeLoginEvents.CONF, events["message"]
elif events["code"] == 86038:
return QrCodeLoginEvents.TIMEOUT, events["message"]
elif events["code"] == 0:
url: str = events["data"]["url"]
cookies_list = url.split("?")[1].split("&")
sessdata = ""
bili_jct = ""
dede = ""
for cookie in cookies_list:
if cookie[:8] == "SESSDATA":
sessdata = cookie[9:]
if cookie[:8] == "bili_jct":
bili_jct = cookie[9:]
if cookie[:11].upper() == "DEDEUSERID=":
dede = cookie[11:]
c = Credential(sessdata, bili_jct, dedeuserid=dede)
return QrCodeLoginEvents.DONE, c
else:
raise LoginError(events["message"])
def get_tv_qrcode() -> Tuple[Picture, str]:
"""
获取 TV 端登录二维码及登录密钥(后面有用)
Returns:
Tuple[Picture, str]: 第一项是二维码图片地址(本地缓存)和登录密钥。登录密钥需要保存。
"""
qrcode_data = login.update_tv_qrcode_data()
qrcode_url = qrcode_data["url"]
auth_code = qrcode_data["auth_code"]
img = login.make_qrcode(qrcode_url)
return (Picture.from_file(img), auth_code)
def check_tv_qrcode_events(auth_code: str) -> Tuple[QrCodeLoginEvents, Union[str, Credential]]:
"""
检查登录状态。
Args:
auth_code (str): 登录密钥
Returns:
Tuple[QrCodeLoginEvents, str|Credential]: 状态(第一项)和信息(第二项)(如果成功登录信息为凭据类)
"""
events = login.verify_tv_login_status(auth_code=auth_code)
if events["code"] == 86039:
return QrCodeLoginEvents.SCAN, events["message"]
elif events["code"] == 86038:
return QrCodeLoginEvents.TIMEOUT, events["message"]
elif events["code"] == 0:
c = login.parse_tv_resp(events["data"])
return QrCodeLoginEvents.DONE, c
else:
raise LoginError(events["message"])
def start_geetest_server() -> "ServerThreadModel":
"""
验证码服务打开服务器
Returns:
ServerThread: 服务进程,将自动开启
返回值内函数及属性:
(继承:threading.Thread)
- url (str) : 验证码服务地址
- start (Callable): 开启进程
- stop (Callable): 结束进程
``` python
print(start_geetest_server().url)
```
"""
return login.start_server() # type: ignore
def close_geetest_server() -> None:
"""
关闭极验验证服务(打开极验验证服务后务必关闭掉它,否则会卡住)
"""
return login.close_server()
def done_geetest() -> bool:
"""
检查是否完成了极验验证。
如果没有完成极验验证码就开始短信登录发送短信,那么可能会让你的项目卡住。
Returns:
bool: 是否完成极验验证
"""
result = login.get_result()
if result != -1:
return True
else:
return False
def safecenter_start_geetest_server() -> "ServerThreadModel":
"""
登录验证专用函数:验证码服务打开服务器
Returns:
ServerThread: 服务进程,将自动开启
返回值内函数及属性:
(继承:threading.Thread)
- url (str) : 验证码服务地址
- start (Callable): 开启进程
- stop (Callable): 结束进程
``` python
print(start_geetest_server().url)
```
"""
return login.safecenter_start_server() # type: ignore
def safecenter_close_geetest_server() -> None:
"""
登录验证专用函数:关闭极验验证服务(打开极验验证服务后务必关闭掉它,否则会卡住)
"""
return login.safecenter_close_server()
def safecenter_done_geetest() -> bool:
"""
登录验证专用函数:检查是否完成了极验验证。
如果没有完成极验验证码就开始短信登录发送短信,那么可能会让你的项目卡住。
Returns:
bool: 是否完成极验验证
"""
result = login.safecenter_get_result()
if result != -1:
return True
else:
return False
COUNTRIES_LIST = login.get_countries_list()
countries_list = COUNTRIES_LIST
class ServerThreadModel(threading.Thread):
"""
A simple model for bilibili_api.utils.captcha._start_server.ServerThread.
"""
url: str
def __init__(self, *args, **kwargs):
...
def stop(self):
"""Stop the server and this thread nicely"""