×

深度分析小红书城API接口,用Python脚本实现

万邦科技Lex 万邦科技Lex 发表于2025-08-19 16:00:30 浏览26 评论0

抢沙发发表评论

免费测试小红书API接口

小红书作为以 UGC 内容为核心的生活方式平台,其数据接口未正式对外开放。以下分析基于移动端 APP 的非官方 API(通过抓包解析),涵盖内容获取、搜索等核心功能,并提供 Python 实现方案。需注意:非官方接口存在稳定性风险,使用需遵守平台用户协议。

### 一、小红书接口核心特性分析

#### 1. 接口体系与功能域

通过抓包分析,小红书核心接口可分为以下几类:

-   **内容推荐**:首页推荐笔记(`/api/sns/v1/feed`)、关注页内容(`/api/sns/v1/follow/feed`);

-   **搜索功能**:笔记搜索(`/api/sns/v1/search/notes`)、用户搜索(`/api/sns/v1/search/users`);

-   **笔记详情**:单篇笔记内容、评论、点赞数据(`/api/sns/v1/note/{note_id}/detail`);

-   **用户信息**:用户主页、发布的笔记(`/api/sns/v1/user/{user_id}/profile`);

-   **互动操作**:点赞、收藏、评论(需登录态,如`/api/sns/v1/note/like`)。

#### 2. 认证与请求规范

-   **匿名接口**:部分推荐内容、公开笔记详情可匿名访问,但限制较多;

-   **登录态接口**:搜索、关注、互动等功能需携带登录凭证(关键参数为`x-s`和`x-t`,由 APP 生成的签名和时间戳);

-   **请求头核心字段**:

    -   `User-Agent`:模拟小红书 APP(如`Xiaomi Redmi K40/Android 12/XHS/7.63.0`);

    -   `x-s`:请求签名(动态生成,涉及时间戳、路径、参数加密);

    -   `x-t`:毫秒级时间戳(与签名对应);

    -   `Cookie`:包含登录态信息(如`web_session`)。

-   **参数加密**:部分接口参数(如搜索关键词)经过简单加密,签名`x-s`是反爬核心(算法未完全公开,需模拟或破解)。

#### 3. 典型接口示例(搜索笔记)

-   **请求 URL**:

-   **方法**:GET

-   **核心参数**:

    -   `keyword`:搜索关键词(URL 编码);

    -   `page`:页码;

    -   `size`:每页条数(默认 20);

    -   `sort`:排序方式(`general`默认,`hot`热门)。

-   **响应格式**:JSON,包含`items`(笔记列表)、`has_more`(是否有下一页)。

### 二、Python 脚本实现:小红书接口调用框架

以下实现基于公开笔记搜索和详情获取,处理签名生成(简化版)和登录态,适用于学习研究。

import requests

import json

import time

import random

import logging

import hashlib

from typing import Dict, List, Optional

from requests.exceptions import RequestException


# 配置日志

logging.basicConfig(

    level=logging.INFO,

    format="%(asctime)s - %(levelname)s - %(message)s"

)


class XiaohongshuAPI:

    def __init__(self, cookie: str = ""):

        """

        初始化小红书API客户端

        :param cookie: 登录Cookie(从浏览器/APP抓包获取,含web_session等)

        """

        self.base_url = "https://edith.xiaohongshu.com"

        self.cookie = cookie

        # 设备信息(模拟安卓设备)

        self.device_id = f"android-{''.join(random.choices('0123456789abcdef', k=16))}"

        self.user_agents = [

            "com.xingin.xhs/7.63.0 (Linux; U; Android 12; zh-CN; Redmi K40; Build/SKQ1.211006.001; 64bit)",

            "com.xingin.xhs/7.62.0 (Linux; U; Android 13; zh-CN; MI 13; Build/TKQ1.221114.001; 64bit)"

        ]

        self.session = requests.Session()

        self._update_headers()


    def _update_headers(self):

        """更新请求头,包含动态参数"""

        timestamp = str(int(time.time() * 1000))

        headers = {

            "User-Agent": random.choice(self.user_agents),

            "Accept-Encoding": "gzip, deflate",

            "Host": "edith.xiaohongshu.com",

            "Connection": "keep-alive",

            "x-t": timestamp,

            "x-s": self._generate_sign(timestamp),  # 简化版签名(实际需破解官方算法)

            "x-device-id": self.device_id,

            "Cookie": self.cookie,

            "Content-Type": "application/json; charset=utf-8"

        }

        self.session.headers.update(headers)


    def _generate_sign(self, timestamp: str) -> str:

        """

        生成简化版x-s签名(仅用于示例,实际需根据官方算法逆向)

        官方签名涉及路径、参数、设备信息、密钥等,此处为模拟

        """

        sign_str = f"timestamp={timestamp}&device_id={self.device_id}&secret_key=xxx"  # 假设密钥

        return hashlib.md5(sign_str.encode()).hexdigest()


    def _random_sleep(self):

        """随机休眠,降低反爬风险"""

        time.sleep(random.uniform(2, 4))


    def search_notes(self, keyword: str, page: int = 1, size: int = 10) -> Optional[List[Dict]]:

        """

        搜索笔记

        :param keyword: 搜索关键词

        :param page: 页码

        :param size: 每页条数

        :return: 笔记列表(含标题、作者、点赞数等)

        """

        url = f"{self.base_url}/api/sns/v1/search/notes"

        params = {

            "keyword": keyword,

            "page": page,

            "size": size,

            "sort": "general",

            "note_type": 0

        }


        try:

            self._update_headers()

            response = self.session.get(url, params=params, timeout=10)

            response.raise_for_status()

            result = response.json()


            # 处理错误响应

            if result.get("success") is False:

                logging.error(f"搜索失败:{result.get('msg')},错误码:{result.get('code')}")

                return None


            # 解析笔记列表

            notes = result.get("data", {}).get("items", [])

            if not notes:

                logging.info("未找到相关笔记")

                return []


            parsed_notes = []

            for note in notes:

                note_info = note.get("note_card", {})

                parsed_notes.append({

                    "note_id": note_info.get("note_id"),

                    "title": note_info.get("title"),

                    "content": note_info.get("desc"),

                    "author": {

                        "user_id": note_info.get("user", {}).get("user_id"),

                        "name": note_info.get("user", {}).get("name")

                    },

                    "stats": {

                        "likes": note_info.get("like_count"),

                        "comments": note_info.get("comment_count"),

                        "collections": note_info.get("collect_count")

                    },

                    "image_urls": [img.get("url") for img in note_info.get("images", [])]

                })


            logging.info(f"搜索到{len(parsed_notes)}条笔记,关键词:{keyword},页码:{page}")

            self._random_sleep()

            return parsed_notes


        except RequestException as e:

            logging.error(f"搜索请求异常:{str(e)}")

            self._random_sleep()

            return None


    def get_note_detail(self, note_id: str) -> Optional[Dict]:

        """

        获取笔记详情

        :param note_id: 笔记ID(从搜索接口获取)

        :return: 笔记详情(含完整内容、评论等)

        """

        url = f"{self.base_url}/api/sns/v1/note/{note_id}/detail"


        try:

            self._update_headers()

            response = self.session.get(url, timeout=10)

            response.raise_for_status()

            result = response.json()


            if result.get("success") is False:

                logging.error(f"获取笔记详情失败:{result.get('msg')}")

                return None


            data = result.get("data", {})

            note_info = data.get("note", {})

            # 解析评论(前3条)

            comments = []

            for cmt in data.get("comments", [])[:3]:

                comments.append({

                    "user_name": cmt.get("user", {}).get("name"),

                    "content": cmt.get("content"),

                    "time": cmt.get("create_time")

                })


            return {

                "note_id": note_id,

                "title": note_info.get("title"),

                "content": note_info.get("desc"),

                "author": {

                    "user_id": note_info.get("user", {}).get("user_id"),

                    "name": note_info.get("user", {}).get("name"),

                    "avatar": note_info.get("user", {}).get("avatar_url")

                },

                "stats": {

                    "likes": note_info.get("like_count"),

                    "comments": note_info.get("comment_count"),

                    "collections": note_info.get("collect_count"),

                    "shares": note_info.get("share_count")

                },

                "image_urls": [img.get("url") for img in note_info.get("images", [])],

                "tags": [tag.get("name") for tag in note_info.get("tags", [])],

                "comments": comments

            }


        except RequestException as e:

            logging.error(f"笔记详情请求异常:{str(e)}")

            self._random_sleep()

            return None



# 示例调用

if __name__ == "__main__":

    # 1. 从抓包获取Cookie(需登录后获取,含web_session等字段)

    COOKIE = "web_session=xxx; xhsTrackerId=xxx; ..."  # 替换为实际Cookie


    # 2. 初始化客户端

    xhs = XiaohongshuAPI(cookie=COOKIE)


    # 3. 搜索笔记(示例:关键词"旅行攻略")

    notes = xhs.search_notes(keyword="旅行攻略", page=1, size=5)

    if notes:

        print("搜索到的笔记(前2条):")

        for note in notes[:2]:

            print(f"\n笔记ID:{note['note_id']}")

            print(f"标题:{note['title']}")

            print(f"内容:{note['content'][:50]}...")

            print(f"点赞:{note['stats']['likes']} | 评论:{note['stats']['comments']}")


    # 4. 获取第一条笔记的详情

    if notes and len(notes) > 0:

        first_note_id = notes[0]["note_id"]

        detail = xhs.get_note_detail(first_note_id)

        if detail:

            print(f"\n\n笔记详情({detail['title']}):")

            print(f"完整内容:{detail['content']}")

            print(f"标签:{','.join(detail['tags'])}")

            print("热门评论:")

            for cmt in detail["comments"]:

                print(f"- {cmt['user_name']}:{cmt['content']}")

### 三、关键技术点解析

#### 1. 签名机制与反爬应对

-   **`x-s`签名**:小红书核心反爬措施,生成逻辑涉及时间戳(`x-t`)、设备 ID、请求路径、参数及私有密钥,需通过逆向 APP 获取算法(示例中为简化版)。实际使用可参考开源项目(如`xhs-signature`)的逆向实现。

-   **动态请求头**:每次请求更新`User-Agent`、`x-t`(时间戳)、`x-s`(签名),模拟真实 APP 行为。

-   **Cookie 管理**:登录态 Cookie(如`web_session`)是访问大部分接口的前提,需从抓包工具(如 Charles、Fiddler)中获取。

#### 2. 接口数据解析

-   **搜索接口**:响应中`data.items`包含笔记卡片信息,提取`note_id`(用于详情查询)、标题、内容摘要、互动数据等。

-   **详情接口**:返回完整笔记内容、作者信息、标签、评论等,可进一步解析图片 URL(需处理防盗链)。

#### 3. 扩展与限制

-   **分页获取**:通过`page`参数循环调用搜索接口,直至`has_more`为`false`。

-   **内容限制**:未登录状态下可获取的内容有限,部分笔记需关注作者才能查看。

-   **频率控制**:建议单 IP 请求间隔≥2 秒,批量获取需使用代理池避免封禁。

### 四、风险与注意事项

1.  **合规性**:非官方 API 调用可能违反小红书用户协议,商业用途需联系平台授权;

1.  **接口稳定性**:`x-s`签名算法可能频繁更新,导致脚本失效,需定期维护;

1.  **法律风险**:大规模爬取内容可能涉及侵犯知识产权或个人信息,需遵守《网络安全法》;

1.  **反爬升级**:频繁请求可能导致账号封禁或 IP 拉黑,建议仅用于个人学习。

该实现可用于学习 API 逆向、反爬机制分析等场景,通过完善签名算法和代理池可提升稳定性。实际应用中需优先考虑平台官方合作渠道(如有)。


群贤毕至

访客