File size: 6,416 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 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 |
"""
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"""
|