# -*- coding: utf-8 -*- # @Author : relakkes@gmail.com # @Time : 2023/12/2 18:44 # @Desc : bilibili 请求客户端 import asyncio import json from typing import Any, Callable, Dict, List, Optional, Tuple from urllib.parse import urlencode import httpx from playwright.async_api import BrowserContext, Page from tools import utils from .exception import DataFetchError from .field import CommentOrderType, SearchOrderType from .help import BilibiliSign class BilibiliClient: def __init__( self, timeout=10, proxies=None, *, headers: Dict[str, str], playwright_page: Page, cookie_dict: Dict[str, str], ): self.proxies = proxies self.timeout = timeout self.headers = headers self._host = "https://api.bilibili.com" self.playwright_page = playwright_page self.cookie_dict = cookie_dict async def request(self, method, url, **kwargs) -> Any: async with httpx.AsyncClient(proxies=self.proxies) as client: response = await client.request( method, url, timeout=self.timeout, **kwargs ) data: Dict = response.json() if data.get("code") != 0: raise DataFetchError(data.get("message", "unkonw error")) else: return data.get("data", {}) async def pre_request_data(self, req_data: Dict) -> Dict: """ 发送请求进行请求参数签名 需要从 localStorage 拿 wbi_img_urls 这参数，值如下： https://i0.hdslb.com/bfs/wbi/7cd084941338484aae1ad9425b84077c.png-https://i0.hdslb.com/bfs/wbi/4932caff0ff746eab6f01bf08b70ac45.png :param req_data: :return: """ if not req_data: return {} img_key, sub_key = await self.get_wbi_keys() return BilibiliSign(img_key, sub_key).sign(req_data) async def get_wbi_keys(self) -> Tuple[str, str]: """ 获取最新的 img_key 和 sub_key :return: """ local_storage = await self.playwright_page.evaluate("() => window.localStorage") wbi_img_urls = local_storage.get("wbi_img_urls", "") or local_storage.get( "wbi_img_url") + "-" + local_storage.get("wbi_sub_url") if wbi_img_urls and "-" in wbi_img_urls: img_url, sub_url = wbi_img_urls.split("-") else: resp = await self.request(method="GET", url=self._host + "/x/web-interface/nav") img_url: str = resp['wbi_img']['img_url'] sub_url: str = resp['wbi_img']['sub_url'] img_key = img_url.rsplit('/', 1)[1].split('.')[0] sub_key = sub_url.rsplit('/', 1)[1].split('.')[0] return img_key, sub_key async def get(self, uri: str, params=None, enable_params_sign: bool = True) -> Dict: final_uri = uri if enable_params_sign: params = await self.pre_request_data(params) if isinstance(params, dict): final_uri = (f"{uri}?" f"{urlencode(params)}") return await self.request(method="GET", url=f"{self._host}{final_uri}", headers=self.headers) async def post(self, uri: str, data: dict) -> Dict: data = await self.pre_request_data(data) json_str = json.dumps(data, separators=(',', ':'), ensure_ascii=False) return await self.request(method="POST", url=f"{self._host}{uri}", data=json_str, headers=self.headers) async def pong(self) -> bool: """get a note to check if login state is ok""" utils.logger.info("Begin pong bilibili...") ping_flag = False try: check_login_uri = "/x/web-interface/nav" response = await self.get(check_login_uri) if response.get("isLogin"): utils.logger.info("use cache login state get web interface successfull!") ping_flag = True except Exception as e: utils.logger.error(f"Pong bilibili failed: {e}, and try to login again...") ping_flag = False return ping_flag async def update_cookies(self, browser_context: BrowserContext): cookie_str, cookie_dict = utils.convert_cookies(await browser_context.cookies()) self.headers["Cookie"] = cookie_str self.cookie_dict = cookie_dict async def search_video_by_keyword(self, keyword: str, page: int = 1, page_size: int = 20, order: SearchOrderType = SearchOrderType.DEFAULT): """ KuaiShou web search api :param keyword: 搜索关键词 :param page: 分页参数具体第几页 :param page_size: 每一页参数的数量 :param order: 搜索结果排序，默认位综合排序 :return: """ uri = "/x/web-interface/wbi/search/type" post_data = { "search_type": "video", "keyword": keyword, "page": page, "page_size": page_size, "order": order.value } return await self.get(uri, post_data) async def get_video_info(self, video_id: str) -> Dict: """ Bilibli web video detail api :param video_id: :return: """ uri = "/x/web-interface/view/detail" params = { "aid": video_id } return await self.get(uri, params, enable_params_sign=False) async def get_video_comments(self, video_id: str, order_mode: CommentOrderType = CommentOrderType.DEFAULT, next: int = 0 ) -> Dict: """get video comments :param video_id: 视频 ID :param order_mode: 排序方式 :param next: 评论页选择 :return: """ uri = "/x/v2/reply/wbi/main" post_data = { "oid": video_id, "mode": order_mode.value, "type": 1, "ps": 20, "next": next } return await self.get(uri, post_data) async def get_video_all_comments(self, video_id: str, crawl_interval: float = 1.0, is_fetch_sub_comments=False, callback: Optional[Callable] = None, ): """ get video all comments include sub comments :param video_id: :param crawl_interval: :param is_fetch_sub_comments: :param callback: :return: """ result = [] is_end = False next_page =0 while not is_end: comments_res = await self.get_video_comments(video_id, CommentOrderType.DEFAULT, next_page) curson_info: Dict = comments_res.get("cursor") comment_list: List[Dict] = comments_res.get("replies", []) is_end = curson_info.get("is_end") next_page = curson_info.get("next") if callback: # 如果有回调函数，就执行回调函数 await callback(video_id, comment_list) await asyncio.sleep(crawl_interval) if not is_fetch_sub_comments: result.extend(comment_list) continue # todo handle get sub comments return result