bilibili_api/login_func.py

"""
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"""