当前位置： 首页 > news >正文

新手避坑指南：手把手教你用Requests库爬取中国大学MOOC，从找API到存CSV

news 2026/6/3 4:42:39

从零开始实战：用Python Requests优雅爬取中国大学MOOC课程数据

1. 准备工作与环境搭建

在开始爬取中国大学MOOC数据之前，我们需要做好充分的准备工作。首先确保你的Python环境已经安装好Requests库，这是我们将要使用的主要工具。如果你还没有安装，可以通过以下命令快速安装：

pip install requests pandas

为什么选择Requests库？相比其他HTTP客户端库，Requests具有以下优势：

  • 简洁直观的API设计
  • 自动处理URL编码和会话管理
  • 内置JSON解码功能
  • 完善的超时和重试机制

必备工具清单：

  • Chrome/Firefox浏览器（用于分析网络请求）
  • Python 3.6+环境
  • 代码编辑器（VS Code/PyCharm等）
  • 网络抓包工具（浏览器开发者工具足够）

提示：建议在虚拟环境中操作，避免污染全局Python环境。可以使用python -m venv mooc_env创建虚拟环境。

2. 分析网站结构与API定位

2.1 使用开发者工具探查API

打开中国大学MOOC网站（https://www.icourse163.org/），按F12打开开发者工具，切换到"网络"(Network)选项卡。这里我们需要重点关注XHR/fetch请求，因为现代网站通常通过这类异步请求获取数据。

关键操作步骤：

  1. 勾选"保留日志"(Preserve log)
  2. 筛选XHR请求
  3. 浏览不同页面观察请求变化
  4. 查看请求头和响应内容

2.2 识别核心API接口

通过分析，我们发现几个关键API端点：

接口类型URL示例请求方式主要参数
课程分类.../listChannelCategoryDetail.rpcPOSTcsrfKey, includeALLChannels
课程列表.../searchCourseCardByChannelAndCategoryId.rpcPOSTcategoryId, pageIndex
课程评论.../getCourseEvaluatePaginationByCourseIdOrTermId.rpcPOSTcourseId, pageSize

常见陷阱：

  • 动态csrfKey处理（需要从首页或cookie获取）
  • 分页参数逻辑（有些接口pageIndex从0开始，有些从1开始）
  • 请求头验证（特别是User-Agent和Referer）

3. 构建健壮的爬虫代码

3.1 基础请求封装

我们先创建一个可复用的请求函数，处理通用逻辑：

import requests from urllib.parse import urljoin BASE_URL = "https://www.icourse163.org/" def make_request(url, data=None, headers=None): """封装基础请求函数""" try: full_url = urljoin(BASE_URL, url) default_headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Referer": BASE_URL } if headers: default_headers.update(headers) response = requests.post( full_url, data=data, headers=default_headers, timeout=10 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None

3.2 处理分页数据

分页是爬虫中常见的需求，这里展示一个通用的分页处理模式：

def fetch_paginated_data(api_url, params_template, start_page=1): """处理分页数据的通用函数""" page = start_page all_data = [] while True: # 动态替换页码参数 current_params = params_template.format(page=page) data = make_request(api_url, data=current_params) if not data or not data.get("result"): break page_data = data["result"].get("list", []) all_data.extend(page_data) # 判断是否还有下一页 total_pages = data["result"]["query"]["totlePageCount"] if page >= total_pages: break page += 1 # 避免请求过快 time.sleep(1) return all_data

3.3 数据存储方案

我们使用pandas来存储数据，它比标准csv模块更强大：

import pandas as pd def save_to_csv(data, filename, columns=None): """将数据保存为CSV文件""" df = pd.DataFrame(data) if columns: df = df[columns] # 只保留指定列 df.to_csv(filename, index=False, encoding="utf_8_sig")

4. 完整爬取流程实现

4.1 获取课程分类数据

def get_course_categories(): """获取所有课程分类""" api_url = "web/j/channelBean.listChannelCategoryDetail.rpc" params = "includeALLChannels=true&includeDefaultChannels=false" data = make_request(api_url, data=params) if data and data.get("result"): categories = [] for category in data["result"]["channelCategoryDetails"]: for channel in category["channels"]: categories.append({ "id": channel["id"], "name": channel["name"], "desc": channel.get("shortDesc", "") }) return categories return []

4.2 获取指定分类下的课程

def get_courses_by_category(category_id): """根据分类ID获取课程列表""" api_url = "web/j/mocSearchBean.searchCourseCardByChannelAndCategoryId.rpc" params = { "mocCourseQueryVo": json.dumps({ "categoryId": -1, "categoryChannelId": category_id, "pageIndex": 1, "pageSize": 20 }) } return fetch_paginated_data(api_url, params)

4.3 获取课程详情与评论

def get_course_details(course_id): """获取课程详细信息""" api_url = f"web/j/courseBean.getMocCourseDetail.rpc?courseId={course_id}" return make_request(api_url) def get_course_comments(course_id): """获取课程评论""" api_url = "web/j/mocCourseV2RpcBean.getCourseEvaluatePaginationByCourseIdOrTermId.rpc" params = { "courseId": course_id, "pageIndex": 1, "pageSize": 20, "orderBy": 3 # 按时间排序 } return fetch_paginated_data(api_url, params)

5. 反爬策略应对方案

5.1 请求头优化

现代网站通常会检查以下请求头：

HEADERS = { "Accept": "application/json", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "zh-CN,zh;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "Origin": BASE_URL, "Pragma": "no-cache", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "X-Requested-With": "XMLHttpRequest" }

5.2 请求频率控制

实现智能延迟机制：

import random import time class RequestThrottler: def __init__(self, base_delay=1.0, random_factor=0.3): self.base_delay = base_delay self.random_factor = random_factor def wait(self): delay = self.base_delay * (1 + random.uniform(-self.random_factor, self.random_factor)) time.sleep(delay)

5.3 代理与重试机制

from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry def create_session(retries=3, backoff_factor=0.3): """创建带重试机制的会话""" session = requests.Session() retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=(500, 502, 504), ) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) return session

6. 数据清洗与存储优化

6.1 处理嵌套JSON结构

很多API返回的数据包含多层嵌套，我们需要展平：

def flatten_json(data, prefix=""): """展平嵌套的JSON结构""" flattened = {} for key, value in data.items(): if isinstance(value, dict): flattened.update(flatten_json(value, f"{prefix}{key}_")) else: flattened[f"{prefix}{key}"] = value return flattened

6.2 数据去重策略

使用内存高效的布隆过滤器：

from pybloom_live import ScalableBloomFilter class Deduplicator: def __init__(self): self.filter = ScalableBloomFilter(initial_capacity=1000) def is_duplicate(self, item_id): if item_id in self.filter: return True self.filter.add(item_id) return False

6.3 增量爬取实现

记录最后爬取位置：

import json import os class CrawlState: def __init__(self, state_file="crawl_state.json"): self.state_file = state_file self.state = self._load_state() def _load_state(self): if os.path.exists(self.state_file): with open(self.state_file) as f: return json.load(f) return {} def save_state(self): with open(self.state_file, "w") as f: json.dump(self.state, f) def get_last_crawl(self, key): return self.state.get(key, 0) def update_last_crawl(self, key, value): self.state[key] = value

7. 项目结构与代码组织

推荐的项目结构：

mooc_crawler/ ├── crawler/ # 核心爬虫代码 │ ├── __init__.py │ ├── api.py # API请求封装 │ ├── models.py # 数据模型 │ ├── storage.py # 存储处理 │ └── utils.py # 工具函数 ├── config.py # 配置文件 ├── main.py # 主程序入口 └── requirements.txt # 依赖列表

在main.py中实现主流程：

from crawler.api import MoocSpider from crawler.storage import save_data def main(): spider = MoocSpider() # 1. 获取所有分类 categories = spider.get_categories() save_data(categories, "categories.csv") # 2. 遍历分类获取课程 all_courses = [] for cat in categories: courses = spider.get_courses_by_category(cat["id"]) all_courses.extend(courses) save_data(all_courses, "courses.csv") # 3. 获取课程详情和评论 for course in all_courses[:10]: # 只爬取前10个作为示例 details = spider.get_course_details(course["id"]) comments = spider.get_course_comments(course["id"]) save_data([details], f"details_{course['id']}.csv") save_data(comments, f"comments_{course['id']}.csv") if __name__ == "__main__": main()

8. 错误处理与日志记录

完善的错误处理系统：

import logging from functools import wraps def setup_logging(): """配置日志系统""" logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler("mooc_crawler.log"), logging.StreamHandler() ] ) def handle_errors(func): """错误处理装饰器""" @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: logging.error(f"Error in {func.__name__}: {str(e)}", exc_info=True) return None return wrapper

9. 性能优化技巧

9.1 异步请求实现

使用aiohttp实现异步请求：

import aiohttp import asyncio async def fetch_async(url, session, params=None): """异步请求函数""" try: async with session.post(url, data=params) as response: return await response.json() except Exception as e: print(f"Async request failed: {e}") return None async def crawl_multiple(urls): """并发爬取多个URL""" connector = aiohttp.TCPConnector(limit=10) # 控制并发量 timeout = aiohttp.ClientTimeout(total=30) async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: tasks = [fetch_async(url, session) for url in urls] return await asyncio.gather(*tasks)

9.2 内存优化策略

对于大数据量，使用生成器而非列表：

def paginated_generator(api_url, params_template): """生成器方式处理分页""" page = 1 while True: data = make_request(api_url, params_template.format(page=page)) if not data: break yield from data["result"]["list"] if page >= data["result"]["query"]["totlePageCount"]: break page += 1

9.3 数据分批存储

避免内存溢出：

def batch_save(data_generator, filename, batch_size=1000): """分批保存数据""" batch = [] for item in data_generator: batch.append(item) if len(batch) >= batch_size: save_to_csv(batch, filename, mode="a") batch = [] if batch: # 保存剩余数据 save_to_csv(batch, filename, mode="a")

10. 实际案例分析

10.1 处理动态csrfKey

观察发现csrfKey会变化，需要从首页获取：

def get_csrf_key(session): """从首页获取csrfKey""" response = session.get(BASE_URL) # 从cookie中获取 return session.cookies.get("NTESSTUDYSI")

10.2 处理特殊编码问题

部分接口返回的数据需要特殊解码：

def decode_response(text): """处理特殊编码的响应""" try: # 尝试直接解析JSON return json.loads(text) except json.JSONDecodeError: # 处理可能的unicode转义 decoded = text.encode("utf-8").decode("unicode-escape") return json.loads(decoded)

10.3 课程图片下载示例

def download_image(url, save_path): """下载课程封面图片""" try: response = requests.get(url, stream=True, timeout=10) if response.status_code == 200: with open(save_path, "wb") as f: for chunk in response.iter_content(1024): f.write(chunk) return True return False except Exception as e: print(f"下载图片失败: {e}") return False

11. 项目扩展思路

11.1 数据可视化分析

使用pandas和matplotlib进行简单分析：

import matplotlib.pyplot as plt def analyze_course_data(filename): """分析课程数据""" df = pd.read_csv(filename) # 按学校统计课程数量 school_counts = df["schoolName"].value_counts().head(10) plt.figure(figsize=(10, 6)) school_counts.plot(kind="barh") plt.title("Top 10 Universities by Course Count") plt.tight_layout() plt.savefig("course_distribution.png")

11.2 构建自动化爬虫系统

使用APScheduler实现定时任务：

from apscheduler.schedulers.blocking import BlockingScheduler def scheduled_crawl(): """定时爬取任务""" spider = MoocSpider() spider.run_full_crawl() if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(scheduled_crawl, "cron", hour=2) # 每天凌晨2点执行 scheduler.start()

11.3 数据API服务搭建

使用Flask提供数据接口：

from flask import Flask, jsonify app = Flask(__name__) @app.route("/api/courses") def get_courses(): data = pd.read_csv("courses.csv") return jsonify(data.to_dict(orient="records")) @app.route("/api/courses/<course_id>") def get_course(course_id): try: details = pd.read_csv(f"details_{course_id}.csv") comments = pd.read_csv(f"comments_{course_id}.csv") return jsonify({ "details": details.to_dict(orient="records"), "comments": comments.to_dict(orient="records") }) except FileNotFoundError: return jsonify({"error": "Course not found"}), 404

12. 法律与道德考量

在开发和使用网络爬虫时，必须注意：

  1. 遵守robots.txt：检查目标网站的爬虫协议
  2. 控制请求频率：避免对服务器造成过大压力
  3. 尊重数据版权：明确数据使用范围和目的
  4. 用户隐私保护：不收集、存储或传播敏感个人信息
  5. 商业使用限制：未经许可不得将数据用于商业用途

重要提示：本教程仅用于教育目的，实际爬取前请确保遵守中国大学MOOC的使用条款和相关法律法规。建议控制爬取频率，避免对网站正常运营造成影响。

http://www.zskr.cn/news/1451491.html

相关文章：

  • OpenCore Legacy Patcher：让老Mac焕发新生的开源神器
  • 2026年口碑好的浙江模内喷漆注塑/IMC注塑/PUR注塑/汽车外饰件注塑优质厂家推荐榜 - 品牌宣传支持者
  • 【信息科学与工程学】【数据科学】数据科学领域-第三篇 数学基础07 群论02
  • 蓝速科技智能会议预约屏：打通钉钉飞书，终结会议室“撞车”难题
  • 告别重启！SpringBoot + Protobuf动态解析实战：在线更新.proto文件并实时解析MQTT数据
  • Sora 2非遗训练数据集构建指南：含2176小时田野影像、89种方言语音标注及文化语义对齐标准（附工信部备案编号）
  • Windows窗口置顶神器：3步解决多窗口遮挡问题
  • 2026年比较好的板式换热器清洗机/换热器高压清洗机/双面全自动换热片清洗机/换热片自动清洗机长期合作厂家推荐 - 行业平台推荐
  • 【VSCode】使用指南（自用）
  • 为什么你的Claude总在关键节点“随机跳转”？——决策树分支坍缩现象的3种检测工具与2小时修复流程
  • GD32F330时钟树实战工程：含多源切换、PLL配置与外设时钟分配
  • Persimmon-8B-Chat vs 其他开源模型：在昇腾平台上的对比评测
  • 高数函数定义域避坑指南：从‘狗不能为零’到‘整体思想’，手把手教你识别并解决3大易错题型
  • 保姆级教程：在银河麒麟V10 SP3 ARM64服务器上，用yum downloadonly搞定Docker 26.1离线安装包
  • 建筑平台JS逆向
  • STM32F407调试神器：用CubeMX+Keil5快速搞定串口printf打印（避坑指南）
  • 数据科学实战：从问题定义到成果展示的完整项目流程解析
  • Matlab一键运行的PSO优化BP神经网络回归预测工具包（含示例数据与全流程可视化）
  • 保姆级教程：用UE5材质系统手搓一个下雨天水坑的真实涟漪（附完整节点图）
  • 抖音直播数据抓取神器：5分钟快速上手实时弹幕监控工具
  • FastJson2.0.49 + Spring 6整合指南：手把手配置HttpMessageConverter（附常见错误排查）
  • 如何用Pulover‘s Macro Creator实现Windows自动化：完全指南
  • Elsevier Tracker：科研投稿状态追踪的实用指南
  • 为什么说Qwen-Image-Edit-Rapid-AIO是AI图像编辑的革命性突破？3步解锁专业级创作
  • AI Agent 面试题 907：如何设计Agent在特定行业的安全审计机制？
  • Windows/Mac上Anaconda Navigator启动失败的保姆级修复指南（2024最新）
  • Unity性能优化：别再滥用material了！sharedMaterial和material的内存陷阱与实战避坑
  • 2026年比较好的塑料模具/六角模具/护坡模具用户口碑推荐厂家 - 品牌宣传支持者
  • YOLOv5项目实战：让检测框‘说中文’——从数据标注到模型部署的全流程详解
  • 告别重复代码！用Vue3+TS给Uniapp项目封装一个像axios一样好用的uni.request