fix: 小红书创作者功能数据获取优化

media_platform/xhs/client.py

@@ -1,7 +1,7 @@
 import asyncio
 import json
 import re
-from typing import Callable, Dict, List, Optional
+from typing import Callable, Dict, List, Optional, Union, Any
 from urllib.parse import urlencode
 
 import httpx
@@ -28,6 +28,7 @@ class XHSClient:
         self.timeout = timeout
         self.headers = headers
         self._host = "https://edith.xiaohongshu.com"
+        self._domain = "https://www.xiaohongshu.com"
         self.IP_ERROR_STR = "网络连接异常，请检查网络设置或重启试试"
         self.IP_ERROR_CODE = 300012
         self.NOTE_ABNORMAL_STR = "笔记状态异常，请稍后查看"
@@ -63,7 +64,7 @@ class XHSClient:
         self.headers.update(headers)
         return self.headers
 
-    async def request(self, method, url, **kwargs) -> Dict:
+    async def request(self, method, url, **kwargs) -> Union[str, Any]:
         """
         封装httpx的公共请求方法，对请求响应做一些处理
         Args:
@@ -82,10 +83,10 @@ class XHSClient:
                 method, url, timeout=self.timeout,
                 **kwargs
             )
-        
+
         if return_response:
             return response.text
-        
+
         data: Dict = response.json()
         if data["success"]:
             return data.get("data", data.get("success", {}))
@@ -186,56 +187,6 @@ class XHSClient:
         }
         return await self.post(uri, data)
 
-    async def get_creator_info_and_notes(self, creator: str) -> Dict:
-        """
-        获取博主的信息和第一页的笔记
-        Args:
-            creator: 博主ID
-        Returns:
-            {"creator":{}, "notes":[]}
-        """
-        path = '/user/profile/'+creator
-        content = await self.request(method="GET", url=f"https://www.xiaohongshu.com{path}", return_response=True)
-        match = re.search(r'<script>window.__INITIAL_STATE__=(.+)<\/script>', content, re.M)
-
-        if match == None:
-            return {}
-        
-        info = json.loads(match.group(1).replace(':undefined', ':null'), strict=False)
-        if info == None:
-            return {}
-        
-        return {
-            'creator': info.get('user').get('userPageData'),
-            'notes': info.get('user').get('notes')[0],
-            'cursor': info.get('user').get('noteQueries')[0].get('cursor'),
-            'has_more_notes': info.get('user').get('noteQueries')[0].get('hasMore')
-        }
-
-    async def get_notes_by_creator(
-            self, creator: str,
-            cursor: str, 
-            page_size: int = 30
-    ) -> Dict:
-        """
-        获取博主的笔记
-        Args:
-            creator: 博主ID
-            cursor: 上一页最后一条笔记的ID
-            page_size: 分页数据长度
-
-        Returns:
-
-        """
-        uri = "/api/sns/web/v1/user_posted"
-        data = {
-            "user_id": creator,
-            "cursor": cursor,
-            "num": page_size,
-            "image_formats": "jpg,webp,avif"
-        }
-        return await self.get(uri, data)
-
     async def get_note_by_id(self, note_id: str) -> Dict:
         """
         获取笔记详情API
@@ -268,7 +219,7 @@ class XHSClient:
         params = {
             "note_id": note_id,
             "cursor": cursor,
-            "top_comment_id":"",
+            "top_comment_id": "",
             "image_formats": "jpg,webp,avif"
         }
         return await self.get(uri, params)
@@ -323,3 +274,76 @@ class XHSClient:
             await asyncio.sleep(crawl_interval)
             result.extend(comments)
         return result
+
+    async def get_creator_info(self, user_id: str) -> Dict:
+        """
+        通过解析网页版的用户主页HTML，获取用户个人简要信息
+        PC端用户主页的网页存在window.__INITIAL_STATE__这个变量上的，解析它即可
+        eg: https://www.xiaohongshu.com/user/profile/59d8cb33de5fb4696bf17217
+        """
+        uri = f"/user/profile/{user_id}"
+        html_content = await self.request("GET", self._domain + uri, return_response=True, headers=self.headers)
+        match = re.search(r'<script>window.__INITIAL_STATE__=(.+)<\/script>', html_content, re.M)
+
+        if match is None:
+            return {}
+
+        info = json.loads(match.group(1).replace(':undefined', ':null'), strict=False)
+        if info is None:
+            return {}
+        return info.get('user').get('userPageData')
+
+    async def get_notes_by_creator(
+            self, creator: str,
+            cursor: str,
+            page_size: int = 30
+    ) -> Dict:
+        """
+        获取博主的笔记
+        Args:
+            creator: 博主ID
+            cursor: 上一页最后一条笔记的ID
+            page_size: 分页数据长度
+
+        Returns:
+
+        """
+        uri = "/api/sns/web/v1/user_posted"
+        data = {
+            "user_id": creator,
+            "cursor": cursor,
+            "num": page_size,
+            "image_formats": "jpg,webp,avif"
+        }
+        return await self.get(uri, data)
+
+    async def get_all_notes_by_creator(self, user_id: str, crawl_interval: float = 1.0,
+                                       callback: Optional[Callable] = None) -> List[Dict]:
+        """
+        获取指定用户下的所有发过的帖子，该方法会一直查找一个用户下的所有帖子信息
+        Args:
+            user_id: 用户ID
+            crawl_interval: 爬取一次的延迟单位（秒）
+            callback: 一次分页爬取结束后的更新回调函数
+
+        Returns:
+
+        """
+        result = []
+        notes_has_more = True
+        notes_cursor = ""
+        while notes_has_more:
+            notes_res = await self.get_notes_by_creator(user_id, notes_cursor)
+            notes_has_more = notes_res.get("has_more", False)
+            notes_cursor = notes_res.get("cursor", "")
+            if "notes" not in notes_res:
+                utils.logger.info(f"[XHSClient.get_all_notes_by_creator] No 'notes' key found in response: {notes_res}")
+                break
+
+            notes = notes_res["notes"]
+            utils.logger.info(f"[XHSClient.get_all_notes_by_creator] got user_id:{user_id} notes len : {len(notes)}")
+            if callback:
+                await callback(notes)
+            await asyncio.sleep(crawl_interval)
+            result.extend(notes)
+        return result