bilibili_api/utils/network.py

"""

bilibili_api.utils.network



与网络请求相关的模块。能对会话进行管理（复用 TCP 连接）。

"""

import re
import json
import time
import atexit
import asyncio
import hashlib
import hmac
from functools import reduce
from urllib.parse import urlencode
from dataclasses import field, dataclass
from typing import Any, Dict, Union, Coroutine, Type
from inspect import iscoroutinefunction as isAsync
from urllib.parse import quote

import httpx
import aiohttp

from .sync import sync
from .. import settings
from .utils import get_api
from .credential import Credential
from ..exceptions import ApiException, ResponseCodeException, NetworkException

__httpx_session_pool: Dict[asyncio.AbstractEventLoop, httpx.AsyncClient] = {}
__aiohttp_session_pool: Dict[asyncio.AbstractEventLoop, aiohttp.ClientSession] = {}
__httpx_sync_session: httpx.Client = None
last_proxy = ""
wbi_mixin_key = ""
buvid3 = ""

# 获取密钥时的申必数组
OE = [
    46,
    47,
    18,
    2,
    53,
    8,
    23,
    32,
    15,
    50,
    10,
    31,
    58,
    3,
    45,
    35,
    27,
    43,
    5,
    49,
    33,
    9,
    42,
    19,
    29,
    28,
    14,
    39,
    12,
    38,
    41,
    13,
    37,
    48,
    7,
    16,
    24,
    55,
    40,
    61,
    26,
    17,
    0,
    1,
    60,
    51,
    30,
    4,
    22,
    25,
    54,
    21,
    56,
    59,
    6,
    63,
    57,
    62,
    11,
    36,
    20,
    34,
    44,
    52,
]
# 使用 Referer 和 UA 请求头以绕过反爬虫机制
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36 Edg/116.0.1938.54",
    "Referer": "https://www.bilibili.com",
}
API = get_api("credential")


def retry_sync(times: int = 3):
    """

    重试装饰器



    Args:

        times (int): 最大重试次数 默认 3 次 负数则一直重试直到成功



    Returns:

        Any: 原函数调用结果

    """

    def wrapper(func):
        def inner(*args, **kwargs):
            # 这里必须新建一个变量用来计数！！不能直接对 times 操作！！！
            nonlocal times
            loop = times
            while loop != 0:
                if loop != times and settings.request_log:
                    settings.logger.info("第 %d 次重试", times - loop)
                loop -= 1
                try:
                    return func(*args, **kwargs)
                except json.decoder.JSONDecodeError:
                    # json 解析错误 说明数据获取有误 再给次机会
                    continue
                except ResponseCodeException as e:
                    # -403 时尝试重新获取 wbi_mixin_key 可能过期了
                    if e.code == -403:
                        global wbi_mixin_key
                        wbi_mixin_key = ""
                        continue
                    # 不是 -403 错误直接报错
                    raise
            raise ApiException("重试达到最大次数")

        return inner

    return wrapper


def retry(times: int = 3):
    """

    重试装饰器



    Args:

        times (int): 最大重试次数 默认 3 次 负数则一直重试直到成功



    Returns:

        Any: 原函数调用结果

    """

    def wrapper(func: Coroutine):
        async def inner(*args, **kwargs):
            # 这里必须新建一个变量用来计数！！不能直接对 times 操作！！！
            nonlocal times
            loop = times
            while loop != 0:
                if loop != times and settings.request_log:
                    settings.logger.info("第 %d 次重试", times - loop)
                loop -= 1
                try:
                    return await func(*args, **kwargs)
                except json.decoder.JSONDecodeError:
                    # json 解析错误 说明数据获取有误 再给次机会
                    continue
                except ResponseCodeException as e:
                    # -403 时尝试重新获取 wbi_mixin_key 可能过期了
                    if e.code == -403:
                        global wbi_mixin_key
                        wbi_mixin_key = ""
                        continue
                    # 不是 -403 错误直接报错
                    raise
            raise ApiException("重试达到最大次数")

        return inner

    if isAsync(times):
        # 防呆不防傻 防止有人 @retry() 不打括号
        func = times
        times = 3
        return wrapper(func)

    return wrapper


@dataclass
class Api:
    """

    用于请求的 Api 类



    Args:

        url (str): 请求地址



        method (str): 请求方法



        comment (str, optional): 注释. Defaults to "".



        wbi (bool, optional): 是否使用 wbi 鉴权. Defaults to False.



        verify (bool, optional): 是否验证凭据. Defaults to False.



        no_csrf (bool, optional): 是否不使用 csrf. Defaults to False.



        json_body (bool, optional): 是否使用 json 作为载荷. Defaults to False.



        ignore_code (bool, optional): 是否忽略返回值 code 的检验. Defaults to False.



        data (dict, optional): 请求载荷. Defaults to {}.



        params (dict, optional): 请求参数. Defaults to {}.



        credential (Credential, optional): 凭据. Defaults to Credential().

    """

    url: str
    method: str
    comment: str = ""
    wbi: bool = False
    verify: bool = False
    no_csrf: bool = False
    json_body: bool = False
    ignore_code: bool = False
    data: dict = field(default_factory=dict)
    params: dict = field(default_factory=dict)
    files: dict = field(default_factory=dict)
    headers: dict = field(default_factory=dict)
    credential: Credential = field(default_factory=Credential)

    def __post_init__(self) -> None:
        self.method = self.method.upper()
        self.original_data = self.data.copy()
        self.original_params = self.params.copy()
        self.data = {k: "" for k in self.data.keys()}
        self.params = {k: "" for k in self.params.keys()}
        self.files = {k: "" for k in self.files.keys()}
        self.headers = {k: "" for k in self.headers.keys()}
        if self.credential is None:
            self.credential = Credential()
        self.__result = None

    def __setattr__(self, __name: str, __value: Any) -> None:
        """

        每次更新参数都要把 __result 清除

        """
        if self.initialized and __name != "_Api__result":
            self.__result = None
        return super().__setattr__(__name, __value)

    @property
    def initialized(self):
        return "_Api__result" in self.__dict__

    @property
    async def result(self) -> Union[int, str, dict]:
        """

        异步获取请求结果



        `self.__result` 用来暂存数据 参数不变时获取结果不变

        """
        if self.__result is None:
            self.__result = await self.request()
        return self.__result

    @property
    def result_sync(self) -> Union[int, str, dict]:
        """

        通过 `sync` 同步获取请求结果



        一般用于非协程内同步获取数据



        `self.__result` 用来暂存数据 参数不变时获取结果不变

        """
        if self.__result is None:
            self.__result = self.request_sync()
        return self.__result

    def update_data(self, **kwargs) -> "Api":
        """

        毫无亮点的更新 data

        """
        self.data = kwargs
        self.__result = None
        return self

    def update_params(self, **kwargs) -> "Api":
        """

        毫无亮点的更新 params

        """
        self.params = kwargs
        self.__result = None
        return self

    def update_files(self, **kwargs) -> "Api":
        """

        毫无亮点的更新 files

        """
        self.files = kwargs
        self.__result = None
        return self

    def update_headers(self, **kwargs) -> "Api":
        """

        毫无亮点的更新 headers

        """
        self.headers = kwargs
        self.__result = None
        return self

    def update(self, **kwargs) -> "Api":
        """

        毫无亮点的自动选择更新，不包括 files, headers

        """
        if self.method == "GET":
            return self.update_params(**kwargs)
        else:
            return self.update_data(**kwargs)

    def _prepare_params_data(self) -> None:
        new_params, new_data = {}, {}
        for key, value in self.params.items():
            if isinstance(value, bool):
                new_params[key] = int(value)
            elif value != None:
                new_params[key] = value
        for key, value in self.data.items():
            if isinstance(value, bool):
                new_params[key] = int(value)
            elif value != None:
                new_data[key] = value
        self.params, self.data = new_params, new_data

    def _prepare_request_sync(self, **kwargs) -> dict:
        """

        准备请求的配置参数



        Args:

            **kwargs: 其他额外的请求配置参数



        Returns:

            dict: 包含请求的配置参数

        """
        # 如果接口需要 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 self.params.get("jsonp") == "jsonp":
            self.params["callback"] = "callback"

        if self.wbi:
            global wbi_mixin_key
            if wbi_mixin_key == "":
                wbi_mixin_key = get_mixin_key_sync()
            enc_wbi(self.params, wbi_mixin_key)

        # 自动添加 csrf
        if (
            not self.no_csrf
            and self.verify
            and self.method in ["POST", "DELETE", "PATCH"]
        ):
            self.data["csrf"] = self.credential.bili_jct
            self.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 = get_spi_buvid_sync()["b_3"]
            cookies["buvid3"] = buvid3
        else:
            cookies["buvid3"] = self.credential.buvid3
        # cookies["Domain"] = ".bilibili.com"

        config = {
            "url": self.url,
            "method": self.method,
            "data": self.data,
            "params": self.params,
            "files": self.files,
            "cookies": cookies,
            "headers": HEADERS.copy() if len(self.headers) == 0 else self.headers,
            "timeout": settings.timeout,
        }
        config.update(kwargs)

        if self.json_body:
            config["headers"]["Content-Type"] = "application/json"
            config["data"] = json.dumps(config["data"])

        return config

    async def _prepare_request(self, **kwargs) -> dict:
        """

        准备请求的配置参数



        Args:

            **kwargs: 其他额外的请求配置参数



        Returns:

            dict: 包含请求的配置参数

        """
        # 如果接口需要 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 self.params.get("jsonp") == "jsonp":
            self.params["callback"] = "callback"

        if self.wbi:
            global wbi_mixin_key
            if wbi_mixin_key == "":
                wbi_mixin_key = await get_mixin_key()
            enc_wbi(self.params, wbi_mixin_key)

        # 自动添加 csrf
        if (
            not self.no_csrf
            and self.verify
            and self.method in ["POST", "DELETE", "PATCH"]
        ):
            self.data["csrf"] = self.credential.bili_jct
            self.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": self.data,
            "params": self.params,
            "files": self.files,
            "cookies": cookies,
            "headers": HEADERS.copy() if len(self.headers) == 0 else self.headers,
            "timeout": settings.timeout
            if settings.http_client == settings.HTTPClient.HTTPX
            else aiohttp.ClientTimeout(total=settings.timeout),
        }
        config.update(kwargs)

        if self.json_body:
            config["headers"]["Content-Type"] = "application/json"
            config["data"] = json.dumps(config["data"])

        if settings.http_client == settings.HTTPClient.AIOHTTP and not self.json_body:
            config["data"].update(config["files"])
            if config["data"] != {}:
                data = aiohttp.FormData()
                for key, val in config["data"].items():
                    data.add_field(key, val)
                config["data"] = data
            config.pop("files")
            if settings.proxy != "":
                config["proxy"] = settings.proxy
        elif settings.http_client == settings.HTTPClient.AIOHTTP:
            # 舍去 files
            config.pop("files")

        return config

    def _get_resp_text_sync(self, resp: httpx.Response):
        return resp.text

    async def _get_resp_text(self, resp: Union[httpx.Response, aiohttp.ClientResponse]):
        if isinstance(resp, httpx.Response):
            return resp.text
        else:
            return await resp.text()

    @retry_sync(times=settings.wbi_retry_times)
    def request_sync(self, raw: bool = False, **kwargs) -> Union[int, str, dict]:
        """

        向接口发送请求。



        Returns:

            接口未返回数据时，返回 None，否则返回该接口提供的 data 或 result 字段的数据。

        """
        self._prepare_params_data()
        config = self._prepare_request_sync(**kwargs)
        session = get_httpx_sync_session()
        resp = session.request(**config)
        try:
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise NetworkException(resp.status_code, str(resp.status_code))
        real_data = self._process_response(
            resp, self._get_resp_text_sync(resp), raw=raw
        )
        return real_data

    @retry(times=settings.wbi_retry_times)
    async def request(self, raw: bool = False, **kwargs) -> Union[int, str, dict]:
        """

        向接口发送请求。



        Returns:

            接口未返回数据时，返回 None，否则返回该接口提供的 data 或 result 字段的数据。

        """
        self._prepare_params_data()
        config = await self._prepare_request(**kwargs)
        session: Union[httpx.AsyncClient, aiohttp.ClientSession]
        # 判断http_client的类型
        if settings.http_client == settings.HTTPClient.HTTPX:
            session = get_session()
            resp = await session.request(**config)
            try:
                resp.raise_for_status()
            except httpx.HTTPStatusError as e:
                raise NetworkException(resp.status_code, str(resp.status_code))
            real_data = self._process_response(
                resp, await self._get_resp_text(resp), raw=raw
            )
            return real_data
        elif settings.http_client == settings.HTTPClient.AIOHTTP:
            session = get_aiohttp_session()
            async with session.request(**config) as resp:
                try:
                    resp.raise_for_status()
                except aiohttp.ClientResponseError as e:
                    raise NetworkException(e.status, e.message)
                real_data = self._process_response(
                    resp, await self._get_resp_text(resp), raw=raw
                )
                return real_data

    def _process_response(

        self,

        resp: Union[httpx.Response, aiohttp.ClientResponse],

        resp_text: str,

        raw: bool = False,

    ) -> Union[int, str, dict]:
        """

        处理接口的响应数据

        """
        # 检查响应头 Content-Length
        content_length = resp.headers.get("content-length")
        if content_length and int(content_length) == 0:
            return None

        if "callback" in self.params:
            # JSONP 请求
            resp_data: dict = json.loads(
                re.match("^.*?({.*}).*$", resp_text, re.S).group(1)
            )
        else:
            # JSON
            resp_data: dict = json.loads(resp_text)

        if raw:
            return resp_data

        OK = resp_data.get("OK")

        # 检查 code
        if not self.ignore_code:
            if OK is None:
                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 OK != 1:
                raise ResponseCodeException(-1, "API 返回数据 OK 不为 1", resp_data)
        elif settings.request_log:
            settings.logger.info(resp_data)

        real_data = resp_data.get("data") if OK is None else resp_data
        if real_data is None:
            real_data = resp_data.get("result")
        return real_data

    @classmethod
    def from_file(cls, path: str, credential: Union[Credential, None] = None):
        """

        以 json 文件生成对象



        Args:

            path (str): 例如 user.info.info



            credential (Credential, Optional): 凭据类. Defaults to None.



        Returns:

            api (Api): 从文件中读取的 api 信息

        """
        path_list = path.split(".")
        api = get_api(path_list.pop(0))
        for key in path_list:
            api = api.get(key)
        return cls(credential=credential, **api)


async def check_valid(credential: Credential) -> bool:
    """

    检查 cookies 是否有效



    Args:

        credential (Credential): 凭据类



    Returns:

        bool: cookies 是否有效

    """
    data = await get_nav(credential)
    return data["isLogin"]


async def get_spi_buvid() -> dict:
    """

    获取 buvid3 / buvid4



    Returns:

        dict: 账号相关信息

    """
    return await Api(**API["info"]["spi"]).result


def get_spi_buvid_sync() -> dict:
    """

    同步获取 buvid3 / buvid4



    Returns:

        dict: 账号相关信息

    """
    return Api(**API["info"]["spi"]).result_sync


def get_nav_sync(credential: Union[Credential, None] = None):
    """

    获取导航



    Args:

        credential (Credential, Optional): 凭据类. Defaults to None



    Returns:

        dict: 账号相关信息

    """
    return Api(credential=credential, **API["info"]["valid"]).result_sync


async def get_nav(credential: Union[Credential, None] = None):
    """

    获取导航



    Args:

        credential (Credential, Optional): 凭据类. Defaults to None



    Returns:

        dict: 账号相关信息

    """
    return await Api(credential=credential, **API["info"]["valid"]).result


def get_mixin_key_sync() -> str:
    """

    获取混合密钥



    Returns:

        str: 新获取的密钥

    """
    data = get_nav_sync()
    wbi_img: Dict[str, str] = data["wbi_img"]

    # 为什么要把里的 lambda 表达式换成函数 这不是一样的吗
    # split = lambda key: wbi_img.get(key).split("/")[-1].split(".")[0]
    def split(key):
        return wbi_img.get(key).split("/")[-1].split(".")[0]

    ae = split("img_url") + split("sub_url")
    le = reduce(lambda s, i: s + (ae[i] if i < len(ae) else ""), OE, "")
    return le[:32]


async def get_mixin_key() -> str:
    """

    获取混合密钥



    Returns:

        str: 新获取的密钥

    """
    data = await get_nav()
    wbi_img: Dict[str, str] = data["wbi_img"]

    # 为什么要把里的 lambda 表达式换成函数 这不是一样的吗
    # split = lambda key: wbi_img.get(key).split("/")[-1].split(".")[0]
    def split(key):
        return wbi_img.get(key).split("/")[-1].split(".")[0]

    ae = split("img_url") + split("sub_url")
    le = reduce(lambda s, i: s + (ae[i] if i < len(ae) else ""), OE, "")
    return le[:32]


def enc_wbi(params: dict, mixin_key: str):
    """

    更新请求参数



    Args:

        params (dict): 原请求参数



        mixin_key (str): 混合密钥

    """
    params.pop("w_rid", None)  # 重试时先把原有 w_rid 去除
    params["wts"] = int(time.time())
    # web_location 因为没被列入参数可能炸一些接口 比如 video.get_ai_conclusion
    params["web_location"] = 1550101
    Ae = urlencode(sorted(params.items()))
    params["w_rid"] = hashlib.md5((Ae + mixin_key).encode(encoding="utf-8")).hexdigest()


def hmac_sha256(key: str, message: str) -> str:
    """

    使用HMAC-SHA256算法对给定的消息进行加密

    :param key: 密钥

    :param message: 要加密的消息

    :return: 加密后的哈希值

    """
    # 将密钥和消息转换为字节串
    key = key.encode("utf-8")
    message = message.encode("utf-8")

    # 创建HMAC对象，使用SHA256哈希算法
    hmac_obj = hmac.new(key, message, hashlib.sha256)

    # 计算哈希值
    hash_value = hmac_obj.digest()

    # 将哈希值转换为十六进制字符串
    hash_hex = hash_value.hex()

    return hash_hex


async def get_bili_ticket() -> str:
    """

    获取 bili_ticket，但目前没用到，暂时不启用



    https://github.com/SocialSisterYi/bilibili-API-collect/issues/903



    Returns:

        str: bili_ticket

    """
    o = hmac_sha256("XgwSnGZ1p", f"ts{int(time.time())}")
    url = "https://api.bilibili.com/bapis/bilibili.api.ticket.v1.Ticket/GenWebTicket"
    params = {
        "key_id": "ec02",
        "hexsign": o,
        "context[ts]": f"{int(time.time())}",
        "csrf": "",
    }
    return (
        await Api(method="POST", url=url, no_csrf=True).update_params(**params).result
    )["ticket"]


def get_httpx_sync_session() -> httpx.Client:
    """

    获取当前模块的 httpx.Client 对象，用于自定义请求



    Returns:

        httpx.Client

    """
    global __httpx_sync_session
    global last_proxy

    if __httpx_sync_session is None or last_proxy != settings.proxy:
        if settings.proxy != "":
            last_proxy = settings.proxy
            proxies = {"all://": settings.proxy}
            session = httpx.Client(proxies=proxies)  # type: ignore
        else:
            last_proxy = ""
            session = httpx.Client()
        __httpx_sync_session = session

    return __httpx_sync_session


def set_httpx_sync_session(session: httpx.Client) -> None:
    """

    用户手动设置 Session



    Args:

        session (httpx.Client):  httpx.Client 实例。

    """
    global __httpx_sync_session
    __httpx_sync_session = session


def get_session() -> httpx.AsyncClient:
    """

    获取当前模块的 httpx.AsyncClient 对象，用于自定义请求



    Returns:

        httpx.AsyncClient

    """
    global __httpx_session_pool, last_proxy
    loop = asyncio.get_event_loop()
    session = __httpx_session_pool.get(loop, None)
    if session is None or last_proxy != settings.proxy:
        if settings.proxy != "":
            last_proxy = settings.proxy
            proxies = {"all://": settings.proxy}
            session = httpx.AsyncClient(proxies=proxies)  # type: ignore
        else:
            last_proxy = ""
            session = httpx.AsyncClient()
        __httpx_session_pool[loop] = session

    return session


def set_session(session: httpx.AsyncClient) -> None:
    """

    用户手动设置 Session



    Args:

        session (httpx.AsyncClient):  httpx.AsyncClient 实例。

    """
    loop = asyncio.get_event_loop()
    __httpx_session_pool[loop] = session


def get_aiohttp_session() -> aiohttp.ClientSession:
    """

    获取当前模块的 aiohttp.ClientSession 对象，用于自定义请求



    Returns:

        aiohttp.ClientSession

    """
    loop = asyncio.get_event_loop()
    session = __aiohttp_session_pool.get(loop, None)
    if session is None:
        session = aiohttp.ClientSession(
            loop=loop, connector=aiohttp.TCPConnector(), trust_env=True
        )
        __aiohttp_session_pool[loop] = session

    return session


def set_aiohttp_session(session: aiohttp.ClientSession) -> None:
    """

    用户手动设置 Session



    Args:

        session (aiohttp.ClientSession):  aiohttp.ClientSession 实例。

    """
    loop = asyncio.get_event_loop()
    __aiohttp_session_pool[loop] = session


def to_form_urlencoded(data: dict) -> str:
    temp = []
    for [k, v] in data.items():
        temp.append(f'{k}={quote(str(v)).replace("/", "%2F")}')

    return "&".join(temp)


@atexit.register
def __clean() -> None:
    """

    程序退出清理操作。

    """
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        return

    async def __clean_task():
        s0 = __aiohttp_session_pool.get(loop, None)
        if s0 is not None:
            await s0.close()
        s1 = __httpx_session_pool.get(loop, None)
        if s1 is not None:
            await s1.aclose()

    if loop.is_closed():
        loop.run_until_complete(__clean_task())
    else:
        loop.create_task(__clean_task())