Python API自动化实战：从浏览器抓包到课程数据采集脚本开发

1. 项目概述：从“看”到“用”的转变

最近在分析一个在线课程开放平台，手头有大量的课程目录、章节信息、视频播放地址需要整理。如果纯靠手动复制粘贴，那工作量简直不敢想，而且容易出错。相信很多做内容分析、竞品调研或者想批量下载学习资料的朋友都遇到过类似的场景。这时候，API（应用程序编程接口）就成了我们的“瑞士军刀”。简单来说，API就是平台对外提供的一个标准化的数据通道，允许我们通过编写程序（也就是脚本）来批量、自动地获取和处理数据，而不是在浏览器里点点点。

这个项目标题“在线课程开放平台API分析及脚本制作(一)”，核心就是完成从“观察者”到“使用者”的转变。我们不仅要看懂平台提供了哪些API，更要能写出稳定、高效的脚本来调用它们，把数据变成我们实际可用的资源。整个过程会涉及到网络请求分析、数据解析、错误处理、脚本工程化等多个环节，是提升自动化办公和数据处理能力的绝佳实战。

2. 核心思路与前期侦察

在动手写代码之前，充分的“侦察”工作至关重要。这决定了我们脚本的稳定性、效率以及是否会被平台的反爬机制拦截。盲目地开始写请求，很容易掉进坑里。

2.1 目标平台分析与接口定位

首先，我们需要明确目标。这个“在线课程开放平台”可能是一个慕课网站、一个企业内训系统，或者一个知识付费平台。不同的平台，其技术架构和开放程度天差地别。

第一步，判断API的开放性与类型：

1. 公开API：最理想的情况。平台官方提供了完整的API文档，说明了每个接口的地址、请求方法、参数和返回格式。这种情况下，我们只需要按照文档申请API Key（密钥）即可。但教育类平台出于内容版权和保护考虑，提供完整公开API的并不多。
2. 内部API（逆向工程）：更常见的情况。平台本身的前端网页或手机App，在运行时会通过Ajax或Fetch技术调用后端的API来获取数据、渲染页面。我们的目标就是找到这些被前端调用的“内部”接口。它们虽然没有官方文档，但功能是完整可用的。

对于第二种情况，我们的侦察工具就是浏览器的“开发者工具”（按F12打开）。

实操：使用浏览器开发者工具捕获API请求

1. 打开目标平台的课程列表页或某个具体的课程详情页。
2. 按下F12打开开发者工具，切换到Network（网络）标签页。
3. 刷新页面，或者进行翻页、点击章节等操作。
4. 在Network面板中，你会看到大量请求。我们需要重点关注XHR或Fetch类型的请求，这些通常是API接口。
5. 观察请求的URL、Method（GET/POST）、Headers（尤其是Authorization,Cookie,User-Agent等）以及Payload（POST请求的请求体）。
6. 点击某个请求，在Preview或Response标签页查看服务器返回的数据，通常是JSON格式，非常规整。

注意：有些平台会对API请求进行加密或签名，增加逆向难度。这时需要更深入的分析，可能涉及对前端JavaScript代码的调试。对于入门项目，我们优先选择那些请求参数清晰、返回数据明文的接口。

2.2 接口关键信息解析与记录

找到疑似获取课程数据的API后，我们需要像做实验一样记录下它的所有特征。我习惯用一个表格来整理：

关键点解析：

• 认证（Authorization）：绝大多数内部API都需要身份认证。最常见的是在Headers里携带Authorization: Bearer <你的token>或者直接使用Cookie。这个token或cookie通常在你登录网站后产生。脚本需要模拟登录过程来获取它，或者手动从浏览器中复制出来临时使用（不推荐长期）。
• 参数传递：GET请求的参数通常拼接在URL问号后面（如?page=1），称为Query参数。POST请求的参数通常放在请求体（Body）中，格式可能是JSON或Form Data。
• 返回格式：教育平台的API返回通常有固定的结构，比如{“code”: 0, “message”: “success”, “data”: {…}}。code为0表示成功，非0表示失败（如400，401，500等）。我们的脚本必须处理这些错误码。

2.3 工具选型与环境准备

工欲善其事，必先利其器。对于API分析和脚本编写，Python是目前最主流、生态最丰富的选择。

核心工具栈：

1. Python 3.8+：脚本语言主体。确保你的系统已安装。在命令行输入python --version或python3 --version检查。
2. Requests库：用于发送HTTP请求的黄金标准库。安装命令：pip install requests。
3. 浏览器开发者工具：如前所述，用于侦察。
4. JSON查看器：浏览器自带的Preview已足够，也可以使用jq命令行工具或在线格式化网站，方便阅读复杂的JSON数据。
5. 代码编辑器：VS Code, PyCharm, 甚至 Sublime Text 都可以。

环境验证：打开你的命令行（Windows的CMD/PowerShell，Mac/Linux的Terminal），依次执行以下命令，确保没有报错：

python --version pip show requests

如果遇到类似“python”不是内部或外部命令或“pip”不是内部或外部命令的错误，说明Python或pip没有正确安装或未添加到系统环境变量PATH中。这是新手最常见的坑，需要回头检查Python安装步骤，并勾选“Add Python to PATH”选项。

3. 脚本基础框架搭建与核心函数实现

侦察完毕，信息在手，现在开始搭建我们的脚本骨架。一个好的脚本应该是模块化、可配置、易维护的。

3.1 构建健壮的请求会话

直接使用requests.get()每次都是独立的请求，不利于管理Cookie和公共请求头。我们应该使用requests.Session()来创建一个会话对象。

import requests import json import time from typing import Optional, Dict, Any class CoursePlatformAPI: def __init__(self, base_url: str, auth_token: Optional[str] = None): """ 初始化API客户端 :param base_url: API的基础地址，如 https://api.example.com :param auth_token: 可选的认证token """ self.base_url = base_url.rstrip('/') # 移除末尾可能的斜杠 self.session = requests.Session() # 设置公共请求头，模拟浏览器行为 self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Content-Type': 'application/json; charset=utf-8', }) if auth_token: self.session.headers.update({'Authorization': f'Bearer {auth_token}'}) # 请求重试配置 self.max_retries = 3 self.retry_delay = 2 # 秒 def _make_request(self, method: str, endpoint: str, **kwargs) -> Optional[Dict[str, Any]]: """ 封装请求的核心方法，包含重试和错误处理逻辑 """ url = f"{self.base_url}/{endpoint.lstrip('/')}" for attempt in range(self.max_retries): try: response = self.session.request(method, url, **kwargs) response.raise_for_status() # 如果状态码不是200，会抛出HTTPError异常 # 尝试解析JSON响应 return response.json() except requests.exceptions.HTTPError as e: status_code = e.response.status_code print(f"HTTP错误! 状态码: {status_code}, URL: {url}") # 针对不同状态码处理 if status_code == 401: print("认证失败，请检查token是否有效或已过期。") return None elif status_code == 403: print("权限不足，访问被拒绝。") return None elif status_code == 404: print("请求的资源不存在。") return None elif status_code == 429: print("请求过于频繁，触发限流。等待后重试...") time.sleep(self.retry_delay * (attempt + 1)) continue elif status_code >= 500: print(f"服务器内部错误 ({status_code})，第{attempt+1}次重试...") time.sleep(self.retry_delay) continue else: print(f"未处理的HTTP错误: {e}") return None except requests.exceptions.ConnectionError: print(f"网络连接错误，第{attempt+1}次重试...") time.sleep(self.retry_delay) except requests.exceptions.Timeout: print(f"请求超时，第{attempt+1}次重试...") time.sleep(self.retry_delay) except json.JSONDecodeError: print(f"响应不是有效的JSON格式。原始文本: {response.text[:200]}...") return None except Exception as e: print(f"未知错误: {e}") return None print(f"请求失败，已达到最大重试次数 {self.max_retries}。") return None

代码解读与心得：

• 会话（Session）：使用Session()对象可以在多次请求间保持Cookie和连接，提升效率。
• 请求头（Headers）：设置User-Agent是基本操作，让请求看起来像来自浏览器。Accept和Content-Type告诉服务器我们想要和发送JSON数据。
• 错误处理：这是脚本稳定性的核心。response.raise_for_status()能自动检查HTTP状态码。我们对常见的错误码（401认证、403禁止、404未找到、429限流、5xx服务器错误）进行了分类处理。特别是429（Too Many Requests），对于爬虫类脚本非常常见，必须实现退避重试逻辑。
• 重试机制：网络不稳定或服务器临时故障是常态。简单的重试机制能大幅提升脚本的健壮性。这里用了指数退避的简化版（固定延迟）。
• JSON解析：使用response.json()解析，并用try-except包裹，防止服务器返回非JSON数据（如HTML错误页面）导致脚本崩溃。

3.2 实现核心业务接口函数

基于我们之前侦察记录的接口信息，现在可以实现具体的业务函数了。

# 接上面的 CoursePlatformAPI 类 def get_course_list(self, page: int = 1, size: int = 20, category: Optional[str] = None) -> Optional[Dict]: """ 获取课程列表 """ params = {'page': page, 'size': size} if category: params['category'] = category return self._make_request('GET', 'v1/courses', params=params) def get_course_detail(self, course_id: str) -> Optional[Dict]: """ 获取课程详细信息，包括章节列表 """ return self._make_request('GET', f'v1/course/{course_id}') def get_video_play_info(self, video_id: str, client_type: str = 'web') -> Optional[Dict]: """ 获取视频播放信息（如m3u8地址、清晰度列表等） 注意：这通常是POST请求，且参数在Body中 """ payload = { 'videoId': video_id, 'clientType': client_type # 可能还需要其他参数，如timestamp, sign等，根据实际接口调整 } return self._make_request('POST', 'v1/video/play', json=payload) def search_courses(self, keyword: str, page: int = 1) -> Optional[Dict]: """ 搜索课程 """ # 假设搜索接口参数不同 return self._make_request('GET', 'v1/search/courses', params={'q': keyword, 'page': page})

参数化设计心得：将接口参数设计为函数参数，而不是硬编码在URL里，使得脚本非常灵活。例如，get_course_list可以轻松地遍历所有分页：for page in range(1, total_pages+1): api.get_course_list(page=page)。

3.3 数据解析与持久化存储

拿到数据（JSON）后，我们需要从中提取有用的信息，并保存下来。通常我们会保存为结构化的文件，如CSV或JSON Lines。

import csv import os class DataProcessor: @staticmethod def parse_course_list(response_data: Dict) -> List[Dict]: """ 从课程列表API响应中解析出课程基本信息列表 """ courses = [] # 实际路径需要根据API返回的真实JSON结构调整 # 例如：response_data['data']['list'] course_items = response_data.get('data', {}).get('list', []) for item in course_items: course = { 'id': item.get('id'), 'title': item.get('title', '').strip(), 'instructor': item.get('teacherName') or item.get('instructor', ''), 'price': item.get('price', 0), 'student_count': item.get('studyCount') or item.get('studentCount', 0), 'category': item.get('categoryName', ''), 'cover_url': item.get('coverUrl', ''), 'update_time': item.get('updateTime', '') } courses.append(course) return courses @staticmethod def save_to_csv(data_list: List[Dict], filename: str): """ 将字典列表保存为CSV文件 """ if not data_list: print("数据列表为空，不保存文件。") return # 从第一条数据获取所有字段作为表头 fieldnames = data_list[0].keys() # 确保输出目录存在 os.makedirs('output', exist_ok=True) filepath = os.path.join('output', filename) with open(filepath, 'w', newline='', encoding='utf-8-sig') as csvfile: # utf-8-sig支持Excel中文 writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() writer.writerows(data_list) print(f"数据已保存至: {filepath}") @staticmethod def save_to_json(data, filename: str): """ 将数据保存为JSON文件 """ os.makedirs('output', exist_ok=True) filepath = os.path.join('output', filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"JSON数据已保存至: {filepath}")

存储格式选择建议：

• CSV：适合存储规整的表格数据，如课程列表。优点是可以用Excel直接打开，轻量。缺点是不适合存储嵌套结构（如一个课程下的章节列表）。
• JSON/JSON Lines：适合存储复杂的、嵌套的原始数据或处理后的对象。json.dump保存的格式可读性好。JSON Lines（每行一个JSON对象）则适合流式处理大数据。
• 数据库（SQLite/MySQL）：如果数据量很大，或需要复杂查询，上数据库是更好的选择。Python的sqlite3库是内置的，无需安装额外服务。

4. 实战：编写一个完整的课程信息抓取脚本

现在，我们把所有模块组合起来，写一个可以运行的完整脚本。这个脚本的目标是：抓取某个分类下的所有课程基本信息，并保存下来。

# main.py import time from course_api import CoursePlatformAPI, DataProcessor # 假设我们把上面的类放在course_api.py def main(): # 1. 配置（这些信息需要你从浏览器侦察中获得） BASE_URL = "https://api.example.com" # 替换为实际API基础地址 # 如何获取TOKEN？通常需要模拟登录。这里我们先手动从浏览器复制一个临时token。 # 警告：此token会过期。生产脚本需要实现完整的登录流程。 AUTH_TOKEN = "your_long_jwt_token_here" TARGET_CATEGORY = "programming" # 2. 初始化API客户端 print("初始化API客户端...") api = CoursePlatformAPI(base_url=BASE_URL, auth_token=AUTH_TOKEN) # 3. 获取第一页，试探总页数 print("获取第一页课程列表...") first_page_data = api.get_course_list(page=1, size=50, category=TARGET_CATEGORY) if not first_page_data or first_page_data.get('code') != 0: print("获取课程列表失败，请检查网络和认证信息。") return total_courses = first_page_data.get('data', {}).get('total', 0) page_size = 50 total_pages = (total_courses + page_size - 1) // page_size # 向上取整计算总页数 print(f"发现总课程数: {total_courses}, 总页数: {total_pages}") all_courses = [] # 4. 循环抓取所有页 for page in range(1, total_pages + 1): print(f"正在抓取第 {page}/{total_pages} 页...") if page > 1: # 第一页已经抓过了 page_data = api.get_course_list(page=page, size=page_size, category=TARGET_CATEGORY) if not page_data or page_data.get('code') != 0: print(f"第 {page} 页抓取失败，跳过。") continue else: page_data = first_page_data courses = DataProcessor.parse_course_list(page_data) all_courses.extend(courses) # 礼貌性延迟，避免请求过快触发反爬 time.sleep(1) # 5. 保存结果 print(f"共抓取到 {len(all_courses)} 门课程信息。") if all_courses: DataProcessor.save_to_csv(all_courses, f'courses_{TARGET_CATEGORY}.csv') # 也可以保存原始JSON数据供后续深度分析 # DataProcessor.save_to_json(all_courses, f'courses_{TARGET_CATEGORY}_raw.json') print("任务完成！") if __name__ == "__main__": main()

脚本运行与调试：

1. 将上面的CoursePlatformAPI、DataProcessor类代码保存为course_api.py。
2. 将main()函数代码保存为main.py，放在同一目录。
3. 修改BASE_URL、AUTH_TOKEN等配置为你侦察到的真实值。
4. 在命令行中运行：python main.py。

重要提示：关于认证Token的获取：上面的脚本假设你已经有了一个有效的Token。在实际中，获取Token通常需要模拟登录。这涉及到分析平台的登录接口（通常是POST一个包含用户名、密码的请求），处理可能存在的验证码、加密参数等。这是一个更高级的话题，但核心步骤依然是：用开发者工具抓取登录请求 -> 用Python的Requests库模拟这个请求 -> 从响应中提取Token（通常在返回的JSON里或Set-Cookie头中）。切记，任何自动化操作都必须遵守目标平台的robots.txt协议和服务条款，尊重版权，仅将数据用于个人学习或合规的分析目的。

5. 常见问题排查与脚本优化技巧

在实际操作中，你几乎一定会遇到各种问题。下面是我踩过坑后总结的一些排查思路和优化技巧。

5.1 请求失败问题排查清单

5.2 提升脚本稳定性与效率的进阶技巧

1. 使用配置文件：将BASE_URL、AUTH_TOKEN、请求间隔等配置项写入一个单独的config.yaml或config.ini文件，方便管理和修改，避免硬编码。
2. 实现Token自动刷新：写一个login()方法，当检测到401错误时，自动调用登录接口获取新Token，并更新Session的请求头。
3. 添加日志系统：使用Python内置的logging模块替代print。可以设置不同级别（DEBUG, INFO, WARNING, ERROR），将日志输出到文件，方便后期排查问题。

   import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', filename='course_spider.log') logger = logging.getLogger(__name__) # 使用时：logger.info(f“正在抓取第{page}页”)

4. 处理速率限制与礼貌爬取：除了固定延迟，可以加入随机延迟，使请求模式更接近人类。对于大规模抓取，务必遵守robots.txt，并在非高峰时段进行。

   import random time.sleep(random.uniform(1, 3)) # 在1到3秒间随机休眠

5. 断点续传：如果抓取大量数据中途中断，会很麻烦。可以在脚本中记录已成功抓取的页码或课程ID到文件，重启时读取这个文件，跳过已抓取的部分。
6. 异步请求提升速度：对于大量独立的API调用（如获取几千门课程的详情），使用同步请求 (requests) 会非常慢。可以考虑使用aiohttp库进行异步并发请求，能极大提升效率。但异步编程复杂度更高，且需注意目标服务器的并发承受能力。

5.3 安全与合规性再强调

最后，也是最重要的一点，我们必须时刻牢记边界。API脚本是一把双刃剑。

• 合规使用：仅用于学习、测试或个人数据分析。绝对不要用于恶意爬取、侵犯版权、攻击服务器或进行商业数据盗用。
• 尊重版权：课程视频、讲义等内容通常受版权保护。抓取播放地址用于个人离线学习可能处于灰色地带，批量下载或传播则可能违法。
• 关注robots.txt：访问https://目标网站/robots.txt，查看网站是否禁止爬虫访问某些路径。
• 控制影响：将请求频率控制在极低水平，避免对目标平台的正常服务造成任何影响。

写API脚本的过程，是一个极佳的学习路径：从网络协议（HTTP）到数据交换格式（JSON），从编程语言（Python）到工程实践（错误处理、日志、配置管理）。当你成功运行起第一个脚本，将杂乱的数据变成整洁的表格时，那种成就感就是驱动我们不断探索的动力。在下一部分，我们可以探讨更深入的话题，比如如何处理需要加密签名的复杂API，如何模拟登录获取持久会话，以及如何将抓取的数据进行更深入的分析和可视化。