The comment data for analysis includes user nickname, user ID, comment level, and comment content.

Data Extraction Process

The target content is retrieved via XHR requests, specifically through the comment endpoint. Pagination is managed using a cursor parameter. The logic for constructing the request URL is as follows:

def get_comment_url(note_id, cursor=None, page=1):
    base_url = 'https://edith.xiaohongshu.com/api/sns/web/v2/comment/page'
    query_params = {
        'note_id': note_id,
        'top_comment_id': '',
        'image_formats': 'jpg,webp,avif'
    }
    
    if page == 1:
        query_params['cursor'] = ''
    else:
        print("[INFO] Fetching next page")
        query_params['cursor'] = cursor
    
    # Construct the full URL with query parameters
    from urllib.parse import urlencode
    return f"{base_url}?{urlencode(query_params)}"

# Example usage for parsing the response and getting the next cursor
response_data = json.loads(json_text)
next_cursor_token = response_data.get('data', {}).get('cursor')
has_more_data = response_data.get('data', {}).get('has_more', False)

To determine if data extraction is complete, check the boolean value of has_more in the API response. A value of True indicates more comments are available.

Optimizing Data Collection

To improve efficiency, implement concurrent requests. Use a session management approach to handle multiple requests simultaneously.

import asyncio
import aiohttp

async def fetch_page(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def gather_comments(note_id, initial_cursor=''):
    async with aiohttp.ClientSession() as session:
        all_comments = []
        next_cursor = initial_cursor
        has_more = True
        
        while has_more:
            url = get_comment_url(note_id, cursor=next_cursor, page=2 if next_cursor else 1)
            page_data = await fetch_page(session, url)
            
            # Extract comments from the current page
            current_comments = page_data.get('data', {}).get('comments', [])
            all_comments.extend(process_comments(current_comments))
            
            # Update pagination state
            next_cursor = page_data.get('data', {}).get('cursor', '')
            has_more = page_data.get('data', {}).get('has_more', False)
        
        return all_comments

def process_comments(comment_list):
    processed = []
    for c in comment_list:
        processed.append({
            'nickname': c.get('user', {}).get('nickname'),
            'user_id': c.get('user', {}).get('user_id'),
            'level': c.get('level'),
            'content': c.get('content')
        })
    return processed