feat: 微博爬虫帖子搜索完成

media_platform/weibo/client.py

@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+# @Author  : relakkes@gmail.com
+# @Time    : 2023/12/23 15:40
+# @Desc    : 微博爬虫 API 请求 client
+
+import asyncio
+import json
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+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 SearchType
+
+
+class WeiboClient:
+    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://m.weibo.cn"
+        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("ok") != 1:
+            utils.logger.error(f"[WeiboClient.request] request {method}:{url} err, res:{data}")
+            raise DataFetchError(data.get("msg", "unkonw error"))
+        else:
+            return data.get("data", {})
+
+    async def get(self, uri: str, params=None) -> Dict:
+        final_uri = uri
+        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:
+        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("[WeiboClient.pong] Begin pong weibo...")
+        ping_flag = False
+        try:
+            pass
+        except Exception as e:
+            utils.logger.error(f"[BilibiliClient.pong] Pong weibo 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 get_note_by_keyword(
+            self,
+            keyword: str,
+            page: int = 1,
+            search_type: SearchType = SearchType.DEFAULT
+    ) -> Dict:
+        """
+        search note by keyword
+        :param keyword: 微博搜搜的关键词
+        :param page: 分页参数 -当前页码
+        :param search_type: 搜索的类型，见 weibo/filed.py 中的枚举SearchType
+        :return:
+        """
+        uri = "/api/container/getIndex"
+        containerid = f"100103type={search_type.value}&q={keyword}"
+        params = {
+            "containerid": containerid,
+            "page_type": "searchall",
+            "page": page,
+        }
+        return await self.get(uri, params)