feat: 支持爬取小红书二级评论

base/base_crawler.py

@@ -55,6 +55,12 @@ class AbstractStore(ABC):
     # @abstractmethod
     async def store_creator(self, creator: Dict):
         pass
+    
+    # TODO support all platform
+    # only xhs is supported, so @abstractmethod is commented
+    # @abstractmethod
+    async def store_sub_comment(self, sub_comment_items: Dict):
+        pass
 
 class AbstractStoreImage(ABC):
     #TODO: support all platform

config/base_config.py

@@ -39,6 +39,9 @@ ENABLE_GET_IMAGES = False
 # 是否开启爬评论模式, 默认不开启爬评论
 ENABLE_GET_COMMENTS = False
 
+# 是否开启爬二级评论模式, 默认不开启爬二级评论, 目前仅支持 xhs
+ENABLE_GET_SUB_COMMENTS = False
+
 # 指定小红书需要爬虫的笔记ID列表
 XHS_SPECIFIED_ID_LIST = [
     "6422c2750000000027000d88",

media_platform/xhs/client.py

@@ -225,7 +225,7 @@ class XiaoHongShuClient(AbstactApiClient):
         }
         return await self.get(uri, params)
 
-    async def get_note_sub_comments(self, note_id: str, root_comment_id: str, num: int = 30, cursor: str = ""):
+    async def get_note_sub_comments(self, note_id: str, root_comment_id: str, num: int = 10, cursor: str = ""):
         """
         获取指定父评论下的子评论的API
         Args:
@@ -275,6 +275,39 @@ class XiaoHongShuClient(AbstactApiClient):
             await asyncio.sleep(crawl_interval)
             result.extend(comments)
         return result
+    
+    async def get_comment_all_sub_comments(self, note_id: str, root_comment_id: str, sub_comment_cursor: str, 
+                                    sub_comments: List[Dict], crawl_interval: float = 1.0,
+                                    callback: Optional[Callable] = None) -> List[Dict]:
+        """
+        获取指定笔记下指定一级评论下的所有二级评论, 该方法会一直查找一个一级评论下的所有二级评论信息
+        Args:
+            note_id: 笔记ID
+            root_comment_id: 一级评论ID
+            sub_comment_cursor: 二级评论的初始分页游标
+            sub_comments: 爬取一级评论默认携带的二级评论列表
+            crawl_interval: 爬取一次评论的延迟单位（秒）
+            callback: 一次评论爬取结束后
+
+        Returns:
+        
+        """
+        result = []
+        sub_comment_has_more = True
+        while sub_comment_has_more:
+            comments_res = await self.get_note_sub_comments(note_id, root_comment_id, 10, sub_comment_cursor)
+            sub_comment_has_more = comments_res.get("has_more", False)
+            sub_comment_cursor = comments_res.get("cursor", "")
+            if "comments" not in comments_res:
+                utils.logger.info(
+                    f"[XiaoHongShuClient.get_comment_all_sub_comments] No 'comments' key found in response: {comments_res}")
+                break
+            comments = comments_res["comments"]
+            if callback:
+                await callback(note_id, sub_comments)
+            await asyncio.sleep(crawl_interval)
+            result.extend(comments)
+        return result
 
     async def get_creator_info(self, user_id: str) -> Dict:
         """

media_platform/xhs/core.py

@@ -123,7 +123,9 @@ class XiaoHongShuCrawler(AbstractCrawler):
                         note_id_list.append(note_detail.get("note_id"))
                 page += 1
                 utils.logger.info(f"[XiaoHongShuCrawler.search] Note details: {note_details}")
-                await self.batch_get_note_comments(note_id_list)
+                note_comments = await self.batch_get_note_comments(note_id_list)
+                await self.batch_get_sub_comments(note_comments)
+                
 
     async def get_creators_and_notes(self) -> None:
         """Get creator's notes and retrieve their comment information."""
@@ -183,11 +185,11 @@ class XiaoHongShuCrawler(AbstractCrawler):
                     f"[XiaoHongShuCrawler.get_note_detail] have not fund note detail note_id:{note_id}, err: {ex}")
                 return None
 
-    async def batch_get_note_comments(self, note_list: List[str]):
+    async def batch_get_note_comments(self, note_list: List[str]) -> List[Dict]:
         """Batch get note comments"""
         if not config.ENABLE_GET_COMMENTS:
             utils.logger.info(f"[XiaoHongShuCrawler.batch_get_note_comments] Crawling comment mode is not enabled")
-            return
+            return None
 
         utils.logger.info(
             f"[XiaoHongShuCrawler.batch_get_note_comments] Begin batch get note comments, note list: {note_list}")
@@ -196,17 +198,59 @@ class XiaoHongShuCrawler(AbstractCrawler):
         for note_id in note_list:
             task = asyncio.create_task(self.get_comments(note_id, semaphore), name=note_id)
             task_list.append(task)
-        await asyncio.gather(*task_list)
+        note_comments = await asyncio.gather(*task_list)
+        # 这里是每个note_id的评论列表，可以将其合并成一个列表, 方便后续操作
+        note_comments = [comment for comments in note_comments for comment in comments]
+        return note_comments
 
-    async def get_comments(self, note_id: str, semaphore: asyncio.Semaphore):
+    async def get_comments(self, note_id: str, semaphore: asyncio.Semaphore) -> List[Dict]:
         """Get note comments with keyword filtering and quantity limitation"""
         async with semaphore:
             utils.logger.info(f"[XiaoHongShuCrawler.get_comments] Begin get note id comments {note_id}")
-            await self.xhs_client.get_note_all_comments(
+            return await self.xhs_client.get_note_all_comments(
                 note_id=note_id,
                 crawl_interval=random.random(),
                 callback=xhs_store.batch_update_xhs_note_comments
             )
+            
+    async def batch_get_sub_comments(self, note_comments: List[Dict]):
+        """Batch get note sub_comments"""
+        if not config.ENABLE_GET_SUB_COMMENTS:
+            utils.logger.info(f"[XiaoHongShuCrawler.batch_get_sub_comments] Crawling sub_comment mode is not enabled")
+            return
+
+        semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM)
+        task_list: List[Task] = []
+        for note_comment in note_comments:
+            note_id = note_comment.get("note_id")
+            sub_comments = note_comment.get("sub_comments")
+            if sub_comments:
+                await xhs_store.batch_update_xhs_note_sub_comments(note_id, sub_comments)
+                
+            sub_comment_has_more = note_comment.get("sub_comment_has_more")
+            if not sub_comment_has_more:
+                continue
+            
+            root_comment_id = note_comment.get("id")
+            sub_comment_cursor = note_comment.get("sub_comment_cursor") 
+            task = asyncio.create_task(self.get_sub_comments(note_id, root_comment_id,
+                                    sub_comment_cursor, sub_comments, semaphore), name=note_id)
+            task_list.append(task)
+        await asyncio.gather(*task_list)
+    
+    async def get_sub_comments(self, note_id: str, root_comment_id: str, sub_comment_cursor: str,
+                                    sub_comments: List[Dict], semaphore: asyncio.Semaphore) -> List[Dict]:
+        """Get note sub_comments with keyword filtering and quantity limitation"""
+        async with semaphore:
+            utils.logger.info(f"[XiaoHongShuCrawler.get_sub_comments] Begin get note id sub_comments {note_id}")
+            await self.xhs_client.get_comment_all_sub_comments(
+                note_id=note_id,
+                root_comment_id=root_comment_id,
+                sub_comment_cursor=sub_comment_cursor,
+                sub_comments=sub_comments,
+                crawl_interval=random.random(),
+                callback=xhs_store.batch_update_xhs_note_sub_comments
+            )
 
     @staticmethod
     def format_proxy_info(ip_proxy_info: IpInfoModel) -> Tuple[Optional[Dict], Optional[Dict]]:

schema/tables.sql

@@ -281,4 +281,27 @@ CREATE TABLE `xhs_note_comment` (
   KEY `idx_xhs_note_co_create__204f8d` (`create_time`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='小红书笔记评论';
 
+-- ----------------------------
+-- Table structure for xhs_note_sub_comment
+-- ----------------------------
+DROP TABLE IF EXISTS `xhs_note_sub_comment`;
+CREATE TABLE `xhs_note_sub_comment` (
+  `id` int NOT NULL AUTO_INCREMENT COMMENT '自增ID',
+  `user_id` varchar(64) NOT NULL COMMENT '用户ID',
+  `nickname` varchar(64) DEFAULT NULL COMMENT '用户昵称',
+  `avatar` varchar(255) DEFAULT NULL COMMENT '用户头像地址',
+  `ip_location` varchar(255) DEFAULT NULL COMMENT '评论时的IP地址',
+  `add_ts` bigint NOT NULL COMMENT '记录添加时间戳',
+  `last_modify_ts` bigint NOT NULL COMMENT '记录最后修改时间戳',
+  `comment_id` varchar(64) NOT NULL COMMENT '评论ID',
+  `target_comment_id` varchar(64) NOT NULL COMMENT '被评论的评论ID',
+  `create_time` bigint NOT NULL COMMENT '评论时间戳',
+  `note_id` varchar(64) NOT NULL COMMENT '笔记ID',
+  `content` longtext NOT NULL COMMENT '评论内容',
+  `pictures` varchar(512) DEFAULT NULL,
+  PRIMARY KEY (`id`),
+  KEY `idx_xhs_note_co_comment_8e8349` (`comment_id`),
+  KEY `idx_xhs_note_co_create__204f8d` (`create_time`)
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='小红书笔记二级评论';
+
 SET FOREIGN_KEY_CHECKS = 1;

store/xhs/__init__.py

@@ -89,6 +89,34 @@ async def update_xhs_note_comment(note_id: str, comment_item: Dict):
     }
     utils.logger.info(f"[store.xhs.update_xhs_note_comment] xhs note comment:{local_db_item}")
     await XhsStoreFactory.create_store().store_comment(local_db_item)
+    
+async def batch_update_xhs_note_sub_comments(note_id: str, comments: List[Dict]):
+    if not comments:
+        return
+    for comment_item in comments:
+        await update_xhs_note_sub_comment(note_id, comment_item)
+
+
+async def update_xhs_note_sub_comment(note_id: str, comment_item: Dict):
+    user_info = comment_item.get("user_info", {})
+    target_comment = comment_item.get("target_comment")
+    comment_id = comment_item.get("id")
+    comment_pictures = [item.get("url_default", "") for item in comment_item.get("pictures", [])]
+    local_db_item = {
+        "comment_id": comment_id,
+        "target_comment_id": target_comment.get('id'),
+        "create_time": comment_item.get("create_time"),
+        "ip_location": comment_item.get("ip_location"),
+        "note_id": note_id,
+        "content": comment_item.get("content"),
+        "user_id": user_info.get("user_id"),
+        "nickname": user_info.get("nickname"),
+        "avatar": user_info.get("image"),
+        "pictures": ",".join(comment_pictures),
+        "last_modify_ts": utils.get_current_timestamp(),
+    }
+    utils.logger.info(f"[store.xhs.update_xhs_note_sub_comment] xhs note comment:{local_db_item}")
+    await XhsStoreFactory.create_store().store_sub_comment(local_db_item)
 
 
 async def save_creator(user_id: str, creator: Dict):

store/xhs/xhs_store_impl.py

@@ -81,6 +81,17 @@ class XhsCsvStoreImplement(AbstractStore):
 
         """
         await self.save_data_to_csv(save_item=creator, store_type="creator")
+    
+    async def store_sub_comment(self, sub_comment_items: Dict):
+        """
+        Xiaohongshu sub_comments CSV storage implementation
+        Args:
+            sub_comment_items: sub_comments item dict
+
+        Returns:
+
+        """
+        await self.save_data_to_csv(save_item=sub_comment_items, store_type="sub_comments")
 
 
 class XhsDbStoreImplement(AbstractStore):
@@ -142,6 +153,26 @@ class XhsDbStoreImplement(AbstractStore):
             await add_new_creator(creator)
         else:
             await update_creator_by_user_id(user_id, creator)
+            
+    async def store_sub_comment(self, sub_comment_items: Dict):
+        """
+        Xiaohongshu content DB storage implementation
+        Args:
+            sub_comment_items: comment item dict
+
+        Returns:
+
+        """
+        from .xhs_store_sql import (add_new_sub_comment,
+                                    query_sub_comment_by_comment_id,
+                                    update_sub_comment_by_comment_id)
+        comment_id = sub_comment_items.get("comment_id")
+        comment_detail: Dict = await query_sub_comment_by_comment_id(comment_id=comment_id)
+        if not comment_detail:
+            sub_comment_items["add_ts"] = utils.get_current_timestamp()
+            await add_new_sub_comment(sub_comment_items)
+        else:
+            await update_sub_comment_by_comment_id(comment_id, comment_item=sub_comment_items)
 
 
 class XhsJsonStoreImplement(AbstractStore):
@@ -214,3 +245,14 @@ class XhsJsonStoreImplement(AbstractStore):
 
         """
         await self.save_data_to_json(creator, "creator")
+        
+    async def store_sub_comment(self, sub_comment_items: Dict):
+        """
+        sub_comment JSON storage implementatio
+        Args:
+            sub_comment_items:
+
+        Returns:
+
+        """
+        await self.save_data_to_json(sub_comment_items, "sub_comments")

store/xhs/xhs_store_sql.py

@@ -145,4 +145,50 @@ async def update_creator_by_user_id(user_id: str, creator_item: Dict) -> int:
     """
     async_db_conn: AsyncMysqlDB = media_crawler_db_var.get()
     effect_row: int = await async_db_conn.update_table("xhs_creator", creator_item, "user_id", user_id)
+    return effect_row
+
+
+async def query_sub_comment_by_comment_id(comment_id: str) -> Dict:
+    """
+    查询一条二级评论内容
+    Args:
+        comment_id:
+
+    Returns:
+
+    """
+    async_db_conn: AsyncMysqlDB = media_crawler_db_var.get()
+    sql: str = f"select * from xhs_note_sub_comment where comment_id = '{comment_id}'"
+    rows: List[Dict] = await async_db_conn.query(sql)
+    if len(rows) > 0:
+        return rows[0]
+    return dict()
+
+
+async def add_new_sub_comment(comment_item: Dict) -> int:
+    """
+    新增一条二级评论记录
+    Args:
+        comment_item:
+
+    Returns:
+
+    """
+    async_db_conn: AsyncMysqlDB = media_crawler_db_var.get()
+    last_row_id: int = await async_db_conn.item_to_table("xhs_note_sub_comment", comment_item)
+    return last_row_id
+
+
+async def update_sub_comment_by_comment_id(comment_id: str, comment_item: Dict) -> int:
+    """
+    更新增一条二级评论记录
+    Args:
+        comment_id:
+        comment_item:
+
+    Returns:
+
+    """
+    async_db_conn: AsyncMysqlDB = media_crawler_db_var.get()
+    effect_row: int = await async_db_conn.update_table("xhs_note_sub_comment", comment_item, "comment_id", comment_id)
     return effect_row