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

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

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

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===== 所有数据抓取完成 =====")

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))