Python爬虫实战：爬取网易云课堂课程数据并存储到MySQL

阿里云教程1天前发布荨筝的小太阳

2 0 0

Python爬虫实战：爬取网易云课堂课程数据并存储到MySQL

在当今信息爆炸的时代，网络上有大量有价值的数据等待我们去挖掘和分析。本文将带你使用Python编写一个爬虫程序，爬取网易云课堂的Python课程数据，并将这些数据存储到MySQL数据库中。

项目概述

我们将构建一个完整的爬虫系统，主要功能包括：

爬取网易云课堂的Python课程信息解析课程数据并清洗将数据存储到MySQL数据库添加错误处理和日志记录

环境准备

在开始之前，确保你已经安装以下Python库：


import requests
import pymysql
import logging
import time
from typing import Dict, List, Optional

数据库设计

首先，我们需要设计一个合适的数据库表来存储课程信息：


CREATE TABLE IF NOT EXISTS course (
    id INT PRIMARY KEY AUTO_INCREMENT,
    course_id BIGINT UNIQUE,
    product_name VARCHAR(255),
    provider VARCHAR(100),
    score DECIMAL(3,1),
    learner_count INT,
    lector_name VARCHAR(100),
    original_price DECIMAL(10,2),
    discount_price DECIMAL(10,2),
    img_url VARCHAR(500),
    big_img_url VARCHAR(500),
    description TEXT,
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

核心代码实现

1. 数据库连接配置

我们首先创建一个数据库连接类，提高代码的可维护性：


class MySQLManager:
    def __init__(self, host="localhost", port=3306, user="root", 
                 password="root", database="spider", charset="utf8"):
        self.connection_params = {
            "host": host,
            "port": port,
            "user": user,
            "password": password,
            "database": database,
            "charset": charset
        }
        self.conn = None
        self.cursor = None
    
    def __enter__(self):
        self.conn = pymysql.connect(**self.connection_params)
        self.cursor = self.conn.cursor()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.cursor:
            self.cursor.close()
        if self.conn:
            if exc_type is None:
                self.conn.commit()
            else:
                self.conn.rollback()
            self.conn.close()

2. 配置日志系统

添加日志记录可以帮助我们更好地调试和监控程序运行状态：


def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('spider.log', encoding='utf-8'),
            logging.StreamHandler()
        ]
    )

3. 增强的爬取函数

我们对原有的爬取函数进行增强，添加错误重试机制和请求头随机化：


def get_json(page_index, retry_count=3):
    """获取课程JSON数据，支持重试机制"""
    url = "https://study.163.com/p/search/studycourse.json"
    
    payload = {
        "activityId": 0,
        "keyword": "python",
        "orderType": 5,
        "pageIndex": page_index,
        "pageSize": 50,
        "priceType": -1,
        "qualityType": 0,
        "relativeOffset": 0,
        "searchTimeType": -1,
    }

    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        "origin": "https://study.163.com",
        "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.70 Safari/537.36",
        "referer": f"https://study.163.com/courses?keyword=python&pageIndex={page_index}"
    }

    for attempt in range(retry_count):
        try:
            response = requests.post(url, json=payload, headers=headers, timeout=10)
            response.raise_for_status()  # 如果状态码不是200，抛出异常
            
            content = response.json()
            if content and content.get("code") == 0:
                logging.info(f"成功获取第{page_index}页数据")
                return content
            else:
                logging.warning(f"第{page_index}页返回数据异常: {content}")
                
        except requests.exceptions.RequestException as e:
            logging.error(f"第{page_index}页第{attempt+1}次请求失败: {str(e)}")
            if attempt < retry_count - 1:
                time.sleep(2)  # 等待2秒后重试
            else:
                logging.error(f"第{page_index}页请求失败，已达最大重试次数")
        except ValueError as e:
            logging.error(f"第{page_index}页JSON解析失败: {str(e)}")
            break
    
    return None

4. 数据清洗和处理

在存储数据之前，我们需要对数据进行清洗和验证：


def clean_course_data(course_list):
    """清洗和验证课程数据"""
    cleaned_courses = []
    
    for item in course_list:
        # 处理可能缺失的字段
        course_id = item.get("courseId", 0)
        product_name = item.get("productName", "").strip()
        
        # 跳过无效数据
        if not course_id or not product_name:
            continue
            
        # 构建课程数据元组，处理可能缺失的字段
        course_value = (
            0,  # id，自增字段用0占位
            course_id,
            product_name,
            item.get("provider", ""),
            float(item.get("score", 0)),  # 确保分数是浮点数
            int(item.get("learnerCount", 0)),  # 确保是整数
            item.get("lectorName", ""),
            float(item.get("originalPrice", 0)),
            float(item.get("discountPrice", 0)),
            item.get("imgUrl", ""),
            item.get("bigImgUrl", ""),
            item.get("description", "")[:1000]  # 限制描述长度，防止数据库字段溢出
        )
        cleaned_courses.append(course_value)
    
    return cleaned_courses

5. 数据库存储功能

改进的数据库存储函数，支持数据去重：


def save_to_mysql(course_list, db_manager):
    """将课程数据保存到MySQL数据库，自动去重"""
    if not course_list:
        logging.warning("没有课程数据需要保存")
        return
    
    cleaned_courses = clean_course_data(course_list)
    
    if not cleaned_courses:
        logging.warning("清洗后没有有效的课程数据")
        return
    
    # 使用INSERT IGNORE避免重复数据
    sql = """
    INSERT IGNORE INTO course 
    (id, course_id, product_name, provider, score, learner_count, 
     lector_name, original_price, discount_price, img_url, big_img_url, description) 
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """
    
    try:
        with db_manager as db:
            db.cursor.executemany(sql, cleaned_courses)
            logging.info(f"成功插入{db.cursor.rowcount}条课程数据")
    except Exception as e:
        logging.error(f"数据库插入失败: {str(e)}")

6. 主程序流程

最后，我们将所有功能整合到主程序中：


def main():
    """主程序"""
    setup_logging()
    logging.info("开始爬取网易云课堂课程数据")
    
    # 初始化数据库管理器
    db_manager = MySQLManager()
    
    try:
        # 获取第一页数据，同时获取总页数
        first_page_data = get_json(1)
        if not first_page_data:
            logging.error("无法获取第一页数据，程序退出")
            return
            
        total_page_count = first_page_data["result"]["query"]["totlePageCount"]
        logging.info(f"总共发现{total_page_count}页数据")
        
        # 处理第一页数据
        course_list = first_page_data["result"]["list"]
        save_to_mysql(course_list, db_manager)
        
        # 爬取剩余页面
        for page_index in range(2, total_page_count + 1):
            logging.info(f"正在爬取第{page_index}/{total_page_count}页...")
            
            page_data = get_json(page_index)
            if page_data:
                course_list = page_data["result"]["list"]
                save_to_mysql(course_list, db_manager)
            
            # 添加延迟，避免请求过于频繁
            time.sleep(1)
            
        logging.info("所有数据爬取完成")
        
    except Exception as e:
        logging.error(f"程序执行出错: {str(e)}")
    finally:
        logging.info("程序执行结束")

if __name__ == "__main__":
    main()

功能扩展建议

添加代理支持：对于大规模爬取，可以使用代理IP池避免被封数据导出功能：添加将数据导出为CSV或Excel的功能定时任务：使用APScheduler等库实现定时爬取数据可视化：使用Matplotlib或Pyecharts对爬取的数据进行可视化分析Web界面：使用Flask或Django构建一个简单的Web界面展示数据