feat: B站爬虫签名实现

media_platform/bilibili/client.py

@@ -0,0 +1,163 @@
+# -*- 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, Optional
+from urllib.parse import urlencode
+
+import httpx
+from playwright.async_api import BrowserContext, Page
+
+from tools import utils
+
+from .help import BilibiliSign
+from .exception import DataFetchError
+
+
+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:
+        """
+        img_key, sub_key = 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", "")
+        img_url, sub_url = wbi_img_urls.split("-")
+        if not img_url or not sub_url:
+            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) -> Dict:
+        final_uri = uri
+        params = 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 = 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 kuaishou...")
+        ping_flag = False
+        try:
+            pass
+        except Exception as e:
+            utils.logger.error(f"Pong kuaishou 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_info_by_keyword(self, keyword: str, pcursor: str):
+        """
+        KuaiShou web search api
+        :param keyword: search keyword
+        :param pcursor: limite page curson
+        :return:
+        """
+        post_data = {
+        }
+        return await self.post("", post_data)
+
+    async def get_video_info(self, photo_id: str) -> Dict:
+        """
+        Kuaishou web video detail api
+        :param photo_id:
+        :return:
+        """
+        post_data = {
+        }
+        return await self.post("", post_data)
+
+    async def get_video_comments(self, photo_id: str, pcursor: str = "") -> Dict:
+        """get video comments
+        :param photo_id: photo id you want to fetch
+        :param pcursor: last you get pcursor, defaults to ""
+        :return:
+        """
+        post_data = {
+        }
+        return await self.post("", post_data)
+
+    async def get_video_all_comments(self, photo_id: str, crawl_interval: float = 1.0, is_fetch_sub_comments=False,
+                                     callback: Optional[Callable] = None, ):
+        """
+        get video all comments include sub comments
+        :param photo_id:
+        :param crawl_interval:
+        :param is_fetch_sub_comments:
+        :param callback:
+        :return:
+        """
+
+        result = []
+        pcursor = ""
+        while pcursor != "no_more":
+            comments_res = await self.get_video_comments(photo_id, pcursor)
+            vision_commen_list = comments_res.get("visionCommentList", {})
+            pcursor = vision_commen_list.get("pcursor", "")
+            comments = vision_commen_list.get("rootComments", [])
+
+            if callback:  # 如果有回调函数，就执行回调函数
+                await callback(photo_id, comments)
+
+            await asyncio.sleep(crawl_interval)
+            if not is_fetch_sub_comments:
+                result.extend(comments)
+                continue
+            # todo handle get sub comments
+        return result