Python 进阶爬虫：解析知识星球 API

一、知识星球 API 核心原理与接口分析

知识星球的前端页面采用动态加载技术（JavaScript 渲染），所有内容数据均通过后端 API 接口以 JSON 格式返回，前端再将数据渲染为可视化页面。因此，API 爬虫的核心逻辑是模拟前端请求，直接调用 API 接口获取原始 JSON 数据，而非解析 HTML 页面。

1.1 API 接口基础架构

知识星球的 API 接口遵循 RESTful 设计规范，核心请求域名为<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">https://api.zsxq.com</font>，所有接口均通过 HTTPS 协议传输，确保数据安全性。接口主要分为三大类：

• 认证类接口：用于获取登录凭证（Cookie/Token），是后续数据请求的基础；
• 内容类接口：获取星球列表、主题列表、主题详情、评论、点赞等核心内容数据；
• 用户类接口：获取用户信息、粉丝列表、关注列表等辅助数据。

所有接口的请求方式以<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">GET</font>和<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">POST</font>为主，请求参数包含公共参数（如时间戳、签名、设备信息）与业务参数（如星球 ID、主题 ID、分页参数），其中签名验证是知识星球反爬机制的核心，也是 API 爬虫的关键难点。

1.2 关键接口梳理（核心业务接口）

在实际爬虫开发中，我们重点关注以下高频核心接口，覆盖数据抓取的全流程：

1.3 反爬机制与签名验证解析

知识星球的 API 接口通过签名（signature） 机制防止非法请求，所有非公开接口的请求头或请求参数中必须包含合法的签名，否则会返回<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">401 Unauthorized</font>（未授权）或<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">403 Forbidden</font>（禁止访问）错误。签名的生成逻辑是知识星球 API 爬虫的核心，其生成规则如下：

示例：若请求接口为<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">/v1/groups</font>，参数为<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">count=20&end_time=1735872000000</font>，公共参数为<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">app_version=3.11.0&platform=ios&timestamp=1735872100000</font>，则待签名字符串为<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">/v1/groups&app_version=3.11.0&count=20&end_time=1735872000000&platform=ios&timestamp=1735872100000</font>，拼接密钥后 MD5 加密即为签名。

二、环境准备与依赖安装

在实现 API 爬虫前，需准备 Python 开发环境并安装必要的依赖库，核心依赖包括：

• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">requests</font>：用于发送 HTTP 请求，处理 API 接口调用；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">pycryptodome</font>：用于 MD5 签名生成（Python 内置 hashlib 也可实现，pycryptodome 兼容性更强）；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">json</font>：用于解析 API 返回的 JSON 数据（Python 内置，无需安装）；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">time</font>/<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">datetime</font>：用于时间戳生成与时间格式转换（Python 内置）。

2.1 环境要求

• Python 版本：3.8 及以上（建议 3.10+，兼容性更好）；
• 操作系统：Windows/MacOS/Linux 均可；
• 网络环境：可正常访问知识星球（需科学上网，若国内访问失败）。

三、Python 实现知识星球 API 爬虫

本节将分模块实现知识星球 API 爬虫，包括签名生成工具、登录凭证获取、核心接口请求、数据解析与存储，最终实现从星球列表到主题详情的全量数据抓取。

3.1 核心工具类：签名生成与请求封装

首先实现签名生成工具，封装请求头、参数处理与签名逻辑，确保所有 API 请求符合知识星球的验证规则。该工具类是整个爬虫的基础，需保证签名生成的准确性。

python

运行

import requests
import hashlib
import time
import json
from urllib.parse import urlencode
from datetime import datetime

class ZsxqApiSpider:
def __init__(self, cookie=None):
"""
初始化知识星球API爬虫
:param cookie: 登录后的Cookie（若未提供，需手动登录获取）
"""
# 基础配置
self.base_url = "https://api.zsxq.com"
self.app_version = "3.11.0" # 客户端版本，固定值
self.platform = "ios" # 平台类型，固定值
self.secret = "zsxqapi2020" # 知识星球内置密钥，核心参数
self.cookie = cookie # 登录凭证，必须提供

# 基础请求头（公共请求头，部分字段可固定）
self.headers = {
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Connection": "keep-alive",
"Cookie": self.cookie,
"Origin": "https://wx.zsxq.com",
"Referer": "https://wx.zsxq.com/"
}

def generate_signature(self, path, params=None):
"""
生成知识星球API签名（核心方法）
:param path: 接口路径（如/v1/groups）
:param params: 请求参数字典（GET参数）
:return: 签名字符串（32位小写MD5）
"""
# 1. 初始化公共参数
common_params = {
"app_version": self.app_version,
"platform": self.platform,
"timestamp": str(int(time.time() * 1000)) # 毫秒级时间戳
}

# 2. 合并公共参数与业务参数，并按键名升序排列
all_params = common_params.copy()
if params and isinstance(params, dict):
all_params.update(params)
# 按键名升序排序
sorted_params = sorted(all_params.items(), key=lambda x: x[0])
# 拼接为key=value格式
params_str = urlencode(sorted_params)

# 3. 拼接待签名字符串：path + & + params_str + & + secret
sign_str = f"{path}&{params_str}&{self.secret}"

# 4. MD5加密生成签名
md5 = hashlib.md5()
md5.update(sign_str.encode("utf-8"))
signature = md5.hexdigest()

return signature, common_params["timestamp"]

def send_get_request(self, path, params=None):
"""
发送GET请求（封装签名与请求逻辑）
:param path: 接口路径
:param params: 请求参数
:return: 响应数据（字典格式）
"""
# 生成签名与时间戳
signature, timestamp = self.generate_signature(path, params)

# 更新请求头，添加签名与时间戳
self.headers["X-Signature"] = signature
self.headers["X-Timestamp"] = timestamp

# 拼接完整请求URL
url = f"{self.base_url}{path}"

try:
# 发送GET请求
response = requests.get(url, headers=self.headers, params=params, timeout=10)
response.raise_for_status() # 抛出HTTP错误（如404、500）
return response.json() # 返回JSON格式数据
except requests.exceptions.RequestException as e:
print(f"GET请求失败：{str(e)}")
return None

def send_post_request(self, path, data=None):
"""
发送POST请求（如点赞、评论）
:param path: 接口路径
:param data: POST请求体数据
:return: 响应数据
"""
signature, timestamp = self.generate_signature(path)
self.headers["X-Signature"] = signature
self.headers["X-Timestamp"] = timestamp
self.headers["Content-Type"] = "application/json;charset=UTF-8"

url = f"{self.base_url}{path}"
try:
response = requests.post(url, headers=self.headers, json=data, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"POST请求失败：{str(e)}")
return None

代码解析：

• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">__init__</font>方法：初始化基础配置，包括请求域名、固定参数、登录 Cookie 与请求头，Cookie 是登录凭证，必须从浏览器中手动获取；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">generate_signature</font>方法：严格按照知识星球的签名规则，实现参数排序、字符串拼接与 MD5 加密，返回合法签名与时间戳；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">send_get_request</font>/<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">send_post_request</font>方法：封装请求逻辑，自动添加签名与时间戳，处理请求异常，返回结构化 JSON 数据。

3.2 登录凭证（Cookie）获取

知识星球的 API 接口需要登录后才能访问，因此必须先获取登录后的 Cookie。获取步骤如下：

注意：Cookie 具有时效性，通常有效期为 1-3 个月，过期后需重新获取。

3.3 核心功能实现：星球与主题数据抓取

在工具类的基础上，实现具体的业务功能，包括获取星球列表、主题列表、主题详情，并将数据保存为 JSON 文件，方便后续分析。

python

运行

def get_my_groups(self, count=20):
"""
获取我的知识星球列表
:param count: 每页获取的星球数量（最大50）
:return: 星球列表数据
"""
path = "/v1/groups"
params = {"count": count}
response = self.send_get_request(path, params)
if response and response.get("succeeded"):
groups = response.get("resp_data", {}).get("groups", [])
print(f"成功获取{len(groups)}个星球")
return groups
else:
print("获取星球列表失败：", response.get("resp_err", "未知错误"))
return []

def get_group_topics(self, group_id, count=20, end_time=None):
"""
获取指定星球的主题列表
:param group_id: 星球ID
:param count: 每页主题数量
:param end_time: 分页时间戳（用于加载更多，首次为None）
:return: 主题列表数据
"""
path = f"/v1/groups/{group_id}/topics"
params = {"count": count}
if end_time:
params["end_time"] = end_time

response = self.send_get_request(path, params)
if response and response.get("succeeded"):
topics = response.get("resp_data", {}).get("topics", [])
# 提取下一页的end_time（用于分页加载）
next_end_time = response.get("resp_data", {}).get("end_time")
return topics, next_end_time
else:
print("获取主题列表失败：", response.get("resp_err", "未知错误"))
return [], None

def get_topic_detail(self, topic_id):
"""
获取主题详情（完整内容、图片、附件）
:param topic_id: 主题ID
:return: 主题详情数据
"""
path = f"/v1/topics/{topic_id}"
response = self.send_get_request(path)
if response and response.get("succeeded"):
topic_detail = response.get("resp_data", {}).get("topic", {})
return topic_detail
else:
print("获取主题详情失败：", response.get("resp_err", "未知错误"))
return {}

def save_data_to_json(self, data, filename):
"""
将数据保存为JSON文件
:param data: 要保存的数据（字典/列表）
:param filename: 文件名（如groups.json）
"""
try:
with open(filename, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
print(f"数据已保存至{filename}")
except Exception as e:
print(f"保存数据失败：{str(e)}")

# 将方法绑定到类中
ZsxqApiSpider.get_my_groups = get_my_groups
ZsxqApiSpider.get_group_topics = get_group_topics
ZsxqApiSpider.get_topic_detail = get_topic_detail
ZsxqApiSpider.save_data_to_json = save_data_to_json

代码解析：

• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">get_my_groups</font>：调用星球列表接口，返回当前账号加入的所有星球，包含星球 ID、名称等核心信息；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">get_group_topics</font>：根据星球 ID 获取主题列表，支持分页加载（通过<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">end_time</font>参数实现），返回主题 ID、标题、摘要等；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">get_topic_detail</font>：根据主题 ID 获取完整详情，包括富文本内容、图片直链、附件下载地址；
• <font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">save_data_to_json</font>：将结构化数据保存为 JSON 文件，保留原始数据结构，方便后续处理。

3.4 主程序：全流程数据抓取

编写主程序，实现从「获取星球列表→遍历星球→获取主题列表→获取主题详情→保存数据」的全流程，同时添加分页逻辑，确保抓取所有数据。

python

运行

if __name__ == "__main__":
# 1. 配置登录Cookie（替换为你自己的Cookie）
ZSXQ_COOKIE = "zsxq_access_token=XXX; zsxqsessionid=XXX; …" # 替换为实际Cookie

# 2. 初始化爬虫
spider = ZsxqApiSpider(cookie=ZSXQ_COOKIE)

# 3. 获取我的星球列表并保存
print("===== 开始获取星球列表 =====")
groups = spider.get_my_groups(count=50)
if groups:
spider.save_data_to_json(groups, "zsxq_groups.json")

# 4. 遍历每个星球，抓取主题列表与详情
for group in groups:
group_id = group.get("group_id")
group_name = group.get("name", "未知星球")
print(f"\\n===== 开始抓取星球：{group_name}（ID：{group_id}）=====")

all_topics = []
end_time = None
page = 1

# 分页抓取主题列表（直到无更多数据）
while True:
print(f"正在抓取第{page}页主题…")
topics, next_end_time = spider.get_group_topics(group_id, count=50, end_time=end_time)
if not topics:
break

# 遍历每个主题，获取详情
for topic in topics:
topic_id = topic.get("topic_id")
print(f"正在获取主题详情：{topic_id}")
topic_detail = spider.get_topic_detail(topic_id)
if topic_detail:
all_topics.append(topic_detail)

# 更新分页参数
end_time = next_end_time
page += 1
time.sleep(1) # 延时1秒，避免请求过快触发反爬

# 保存当前星球的所有主题数据
if all_topics:
filename = f"zsxq_topics_{group_id}.json"
spider.save_data_to_json(all_topics, filename)

print("\\n===== 所有数据抓取完成 =====")

代码解析：

• 主程序首先配置 Cookie，初始化爬虫；
• 先获取星球列表并保存，再遍历每个星球，通过分页循环抓取所有主题；
• 对每个主题调用详情接口，获取完整数据后统一保存，添加<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">time.sleep(1)</font>延时，降低请求频率，避免触发反爬；
• 最终每个星球的主题数据单独保存为 JSON 文件，便于管理。

四、反爬规避与爬虫优化

知识星球的反爬机制除了签名验证，还包括请求频率限制、IP 封禁、Cookie 过期检测，为保证爬虫的稳定性，需进行以下优化：

4.1 请求频率控制

• 避免短时间内大量请求，在接口调用之间添加延时（<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">time.sleep(1-3)</font>）；
• 控制单次请求的<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">count</font>参数（建议 20-50），避免单次请求数据量过大触发限流。

4.2 IP 代理使用

若频繁请求导致 IP 被封禁，可使用代理 IP 池，在<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">send_get_request</font>中添加代理配置：

python

运行

import requests

# 代理配置信息
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

# 构建代理字典（包含认证信息）
# 格式：http://用户名:密码@代理主机:端口
proxies = {
"http": f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}",
"https": f"https://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}"
}

# 请求头（根据实际需求补充）
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}

# 示例：发送带代理的请求
url = "https://www.example.com" # 替换为目标URL
params = {} # 替换为实际请求参数

try:
# 发送请求时添加proxies参数
response = requests.get(url, headers=headers, params=params, proxies=proxies, timeout=10)
response.raise_for_status() # 抛出HTTP错误状态码异常
print("请求成功！状态码：", response.status_code)
print("响应内容：", response.text[:500]) # 打印前500字符
except requests.exceptions.RequestException as e:
print("请求失败：", str(e))

4.3 Cookie 自动刷新

当 Cookie 过期时，爬虫会返回<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">401</font>错误，可通过定时任务（如<font style="color:rgb(0, 0, 0);background-color:rgba(0, 0, 0, 0);">APScheduler</font>）定期重新获取 Cookie，或实现自动登录逻辑（需处理验证码，复杂度较高）。

五、总结与合规提醒

本文通过解析知识星球 API 的核心原理，实现了从签名生成、接口请求到数据存储的全流程 API 爬虫，相比传统网页爬虫，API 爬虫具有数据提取精准、效率高、稳定性强的优势，是进阶爬虫的核心技能。通过本文的代码，你可以快速实现知识星球数据的批量抓取，为内容分析、数据挖掘提供基础支撑。

核心知识点回顾