Python爬虫 - 豆瓣电影排行榜数据爬取、处理与存储

前言

在当今数据驱动的时代，信息的获取与分析变得尤为重要。电影作为一种广受欢迎的文化产品，其相关数据的挖掘与分析不仅能帮助观众更好地选择影片，还能为电影行业提供有价值的市场洞察。本文将详细介绍如何利用 Python 爬虫技术从豆瓣电影网站获取电影排行榜数据，并将其存储到 MySQL 数据库中，以便后续进行统计分析和探索性数据分析（EDA）。

我们将首先分析爬取的需求，明确要获取的电影分类及其详细信息。接着，使用 Python 的 requests 和 BeautifulSoup 库编写爬虫代码，抓取所需数据，并将其保存为 JSON 格式。随后，我们将设计相应的 MySQL 数据表结构，以便将爬取到的原始数据存储到数据库中。最后，借助 pandas 库，我们将处理和清洗数据，确保其格式符合数据库要求，并将最终数据导入 MySQL。

一、使用版本

二、需求分析

进行分析的时候，首先在如下图网络处分析，查找是否有获取数据的API接口；如果没有，再从页面元素标签上去分析。

1. 分析要爬取的内容

1.1 分析要爬取的分类

如下图所示，此处要爬取的是豆瓣电影排行榜的分类排行榜的每个分类下的所有数据。

右击剧情，点击检查。

可以看到他的链接地址标签<a href="/typerank?type_name=剧情&type=11&interval_id=100:90&action=">剧情</a>，此地址中包含type_name、type、interval_id、action这几个参数。

1.2 分析要爬取的单个电影的数据

如下图所示，点击后进入单个电影内容页面。

如下图所示，可以看到单个电影的详细信息，我们需要获取这些信息。

1.3 分析如何获取单个电影数据

1.3.1 预览数据

如下面两幅图所示，在这个接口的预览中可以获取到单个电影的详细数据。

1.3.2 查看请求网址、方法及请求头信息

在标头部分查看请求网址、方法，以及下面的请求头信息。

如上图所示，请求网址和请求头为：

top_list_base_url = 'https://movie.douban.com/j/chart/top_list'

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36'
}

1.3.3 查看请求参数

在载荷部分查看请求参数。

• interval_id和type：是分类中的数据，需要从分析要爬取的分类获取。
• start=0：指定从哪个位置开始获取电影列表。0意味着从第一条记录开始。如果你想要获取第二页的数据，你可以将此值设置为20（假设每页显示20条记录），以此类推。
• limit=20：定义了一次请求返回的最大电影数量。这里是20，所以每次请求最多会返回20部电影的信息。
  

如上图所示，请求参数为：

payload = {
    'type': '11',
    'interval_id': '100:90',
    'action': '',
    'start': '0',
    'limit': '20'
}

2. 数据用途

2.1 统计分析

1. 评分分布：

   • 分析电影评分的分布情况，比如平均分、最高分、最低分等。
   • 可以使用直方图或箱线图来可视化评分分布。

2. 观看人数：

   • 研究投票数（vote_count），了解该电影的受欢迎程度。
   • 对比不同电影之间的投票数差异，找出最受欢迎的作品。

3. 类型偏好：

   • 分析不同类型（types）的电影数量及其平均评分，判断观众对哪种类型的偏好。

4. 地区影响：

   • 探讨来自不同制作地区（regions）的电影在评分上的差异，以及各地区的贡献度。

2.2 探索性数据分析 (EDA)

1. 时间趋势：

   • 根据发布日期（release_date）分析电影评分随时间的变化趋势，观察是否存在特定年代的高分电影聚集现象。

2. 演员影响力：

   • 分析主演（actors）参与的电影数量及平均评分，评估明星效应。
   • 利用网络图展示演员间的合作关系，识别哪些演员经常合作。

3. 导演/编剧风格：

   • 如果有导演或编剧信息，可以进一步分析他们的作品特点，如评分、类型选择等。

2.3 高级分析

1. 预测模型：

   • 构建机器学习模型来预测电影评分，考虑因素包括演员阵容、类型、地区、上映日期等。
   • 使用回归分析预测票房收入（如果有相关数据）。

2. 文本挖掘与情感分析：

   • 如果有评论数据，可以执行文本挖掘，提取热门话题、关键词等。
   • 实施情感分析，评估观众对电影的整体态度是正面还是负面。

3. 关联规则学习：

   • 发现不同类型、演员、地区之间的关联模式，例如某些类型的电影是否更倾向于出现在特定地区。

4. 聚类分析：

   • 将电影根据其特征（如评分、类型、地区等）分为不同的群组，帮助理解市场细分。

5. 社交网络分析：

   • 建立演员、导演等创作人员之间的社交网络，分析合作频率和紧密度。

对于给定的数据片段，你还可以做如下具体操作：

• 计算评分的众数：从rating字段中解析出用户给出的具体分数，并计算最常见的评分。
• 检查播放状态的影响：对比is_playable为True和False的电影评分是否有显著区别。
• 查看封面图片的流行度：虽然这需要额外的数据源支持，但可以假设封面设计也会影响电影的关注度。

三、编写代码

1. 获取分类链接中的参数

爬取每个分类链接地址中的type_name、type、interval_id。

from urllib.parse import urlparse, parse_qs

import requests
from bs4 import BeautifulSoup

# 设置请求头，模拟浏览器访问，避免被服务器识别为爬虫
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36'
}


# 获取网页文本内容，获取失败返回None
def fetch_page_content(url, headers):
    try:
        response = requests.get(url=url, headers=headers)
        if response.status_code == 200:
            return response.text
        else:
            print(f"检索页面失败，状态码：{response.status_code}")
            return None
    except Exception as e:
        print(f"发生错误：{e}")
        return None


# 使用BeautifulSoup解析HTML文档
def parse_html(content):
    return BeautifulSoup(content, features='lxml')


# 提取电影类型链接，包含<a>标签的列表
def extract_movie_type_links(soup):
    return soup.select('.types > span > a')


# 从给定的URL中提取查询参数，并返回一个字典。
def get_url_parameters(url):
    parsed_url = urlparse(url)
    query_params = parse_qs(parsed_url.query)
    return query_params


# 处理电影类型链接，构建完整的URL并提取查询参数，返回每个电影类型的查询参数字典列表。
def process_movie_types(base_url, a_list):
    type_dicts = []
    for a in a_list:
        href = a.attrs.get('href')
        full_url = base_url + href if href.startswith('/') else href  # 确保链接是完整的URL
        type_dict = get_url_parameters(full_url)
        type_dicts.append(type_dict)
    return type_dicts


def get_type_params_list():
    chart_base_url = 'https://movie.douban.com/chart'
    # 获取页面内容
    content = fetch_page_content(chart_base_url, headers)
    if not content:
        return
    # 解析HTML文档
    soup = parse_html(content)
    # 提取电影类型链接
    a_list = extract_movie_type_links(soup)
    # 处理电影类型链接并提取查询参数
    return process_movie_types(chart_base_url, a_list)


if __name__ == '__main__':
    type_params_list = get_type_params_list()
    print(type_params_list)

打印结果为：

[{'type_name': ['剧情'], 'type': ['11'], 'interval_id': ['100:90']}, {'type_name': ['喜剧'], 'type': ['24'], 'interval_id': ['100:90']}, {'type_name': ['动作'], 'type': ['5'], 'interval_id': ['100:90']}, {'type_name': ['爱情'], 'type': ['13'], 'interval_id': ['100:90']}, {'type_name': ['科幻'], 'type': ['17'], 'interval_id': ['100:90']}, {'type_name': ['动画'], 'type': ['25'], 'interval_id': ['100:90']}, {'type_name': ['悬疑'], 'type': ['10'], 'interval_id': ['100:90']}, {'type_name': ['惊悚'], 'type': ['19'], 'interval_id': ['100:90']}, {'type_name': ['恐怖'], 'type': ['20'], 'interval_id': ['100:90']}, {'type_name': ['纪录片'], 'type': ['1'], 'interval_id': ['100:90']}, {'type_name': ['短片'], 'type': ['23'], 'interval_id': ['100:90']}, {'type_name': ['情色'], 'type': ['6'], 'interval_id': ['100:90']}, {'type_name': ['音乐'], 'type': ['14'], 'interval_id': ['100:90']}, {'type_name': ['歌舞'], 'type': ['7'], 'interval_id': ['100:90']}, {'type_name': ['家庭'], 'type': ['28'], 'interval_id': ['100:90']}, {'type_name': ['儿童'], 'type': ['8'], 'interval_id': ['100:90']}, {'type_name': ['传记'], 'type': ['2'], 'interval_id': ['100:90']}, {'type_name': ['历史'], 'type': ['4'], 'interval_id': ['100:90']}, {'type_name': ['战争'], 'type': ['22'], 'interval_id': ['100:90']}, {'type_name': ['犯罪'], 'type': ['3'], 'interval_id': ['100:90']}, {'type_name': ['西部'], 'type': ['27'], 'interval_id': ['100:90']}, {'type_name': ['奇幻'], 'type': ['16'], 'interval_id': ['100:90']}, {'type_name': ['冒险'], 'type': ['15'], 'interval_id': ['100:90']}, {'type_name': ['灾难'], 'type': ['12'], 'interval_id': ['100:90']}, {'type_name': ['武侠'], 'type': ['29'], 'interval_id': ['100:90']}, {'type_name': ['古装'], 'type': ['30'], 'interval_id': ['100:90']}, {'type_name': ['运动'], 'type': ['18'], 'interval_id': ['100:90']}, {'type_name': ['黑色电影'], 'type': ['31'], 'interval_id': ['100:90']}]

2. 获取排行榜中不同分类的电影数据

在获取分类链接中的参数代码实现的基础上，编写代码来获取不同类型的电影数据，并保存为json文件。

import json
from pathlib import Path
from urllib.parse import urlparse, parse_qs

import requests
from bs4 import BeautifulSoup

# 设置请求头，模拟浏览器访问，避免被服务器识别为爬虫
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36'
}


# 获取网页文本内容，获取失败返回None
def fetch_page_content(url, headers):
    try:
        response = requests.get(url=url, headers=headers)
        if response.status_code == 200:
            return response.text
        else:
            print(f"检索页面失败，状态码：{response.status_code}")
            return None
    except Exception as e:
        print(f"发生错误：{e}")
        return None


# 使用BeautifulSoup解析HTML文档
def parse_html(content):
    return BeautifulSoup(content, features='lxml')


# 提取电影类型链接，包含<a>标签的列表
def extract_movie_type_links(soup):
    return soup.select('.types > span > a')


# 从给定的URL中提取查询参数，并返回一个字典。
def get_url_parameters(url):
    parsed_url = urlparse(url)
    query_params = parse_qs(parsed_url.query)
    return query_params


# 处理电影类型链接，构建完整的URL并提取查询参数，返回每个电影类型的查询参数字典列表。
def process_movie_types(base_url, a_list):
    type_dicts = []
    for a in a_list:
        href = a.attrs.get('href')
        full_url = base_url + href if href.startswith('/') else href  # 确保链接是完整的URL
        type_dict = get_url_parameters(full_url)
        type_dicts.append(type_dict)
    return type_dicts


# 获取电影类型参数列表
def get_type_params_list():
    chart_base_url = 'https://movie.douban.com/chart'
    # 获取页面内容
    content = fetch_page_content(chart_base_url, headers)
    if not content:
        return
    # 解析HTML文档
    soup = parse_html(content)
    # 提取电影类型链接
    a_list = extract_movie_type_links(soup)
    # 处理电影类型链接并提取查询参数
    return process_movie_types(chart_base_url, a_list)


# 获取电影数据，为json格式，每次获取50条，start为开始位置，count为每次获取多少条
def get_content(type_id, interval_id, start, count):
    top_list_base_url = 'https://movie.douban.com/j/chart/top_list'
    payload = {
        'type': type_id,
        'interval_id': interval_id,
        'action': '',
        'start': start,
        'limit': count
    }
    try:
        response = requests.get(url=top_list_base_url, params=payload, headers=headers)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"检索页面失败，状态码：{response.status_code}")
            return None
    except Exception as e:
        print(f"发生错误：{e}")
        return None


# 把数据保存为json文件
def save_movie_data_to_file(file_name, json_data):
    # 定义保存目录路径
    save_dir = '../douban_movie_data/'
    dir_path = Path(save_dir)
    # 确保保存目录存在，如果不存在则创建所有必要的父级目录
    dir_path.mkdir(parents=True, exist_ok=True)
    # 使用 'with' 语句打开文件以确保正确关闭文件流
    with open(save_dir + file_name, 'w', encoding='utf-8') as fp:
        print(f"{save_dir + file_name} 文件已保存")
        fp.write(str(json_data))


if __name__ == '__main__':
    # 获取类型参数列表，这个函数应该返回一个包含多个字典的列表，
    type_params_list = get_type_params_list()
    # 遍历类型参数列表，针对每一种类型的数据进行处理
    for type_params in type_params_list:
        # 从当前类型参数中提取出 interval_id, type_name 和 type_id，
        interval_id = type_params.get('interval_id')[0]
        type_name = type_params.get('type_name')[0]
        type_id = type_params.get('type')[0]
        # 初始化起始位置和每次请求的数据数量
        start = 0
        count = 50
        # 使用 while 循环不断获取并保存数据，直到没有更多数据为止
        while True:
            # 构造文件名，格式为 original_movie_data_{type_name}_{type_id}_{start + 1}_{end}.json
            file_name = f"original_movie_data_{type_name}_{type_id}_{start + 1}_{start + count}.json"
            # 调用 get_content 函数获取指定范围内的电影数据
            movie_data = get_content(type_id, interval_id, start, count)
            # 如果没有获取到任何数据，则认为该类别的所有数据都已获取完毕，跳出循环
            if not movie_data:
                print(f"================分类 {type_name} 的数据获取完成================")
                break
            # 将获取到的电影数据保存到文件中，确保中文字符不被转义
            save_movie_data_to_file(file_name, json.dumps(movie_data, ensure_ascii=False))
            # 更新起始位置，准备下一轮的数据获取
            start += count

执行过程中打印的部分信息如下图所示：

数据文件和单个文件部分内容如下图所示：

四、数据处理与存储

先把爬取到的原始json数据保存到json文件中；然后根 json文件中的数据的键和值的类型设计对应的MySQL原始数据表，把json文件中的原始数据转存到MySQL的原始数据表中。

1. 数据表设计与实现

1.1 原始数据表设计

• 表名：original_movie_data
• 描述：用于存储未经处理的数据，按照json文件中的键作为字段来保存。

根据保存的JSON数据结构来设计一个MySQL表。

1.2 原始数据表实现

创建数据库douban_movie。

create database douban_movie;

切换到数据库douban_movie。

use douban_movie;

创建原始数据表original_movie_data。

CREATE TABLE original_movie_data (
    id VARCHAR(20) PRIMARY KEY,
    title VARCHAR(255) NOT NULL,
    url VARCHAR(255),
    cover_url VARCHAR(255),
    release_date DATE,
    score DECIMAL(3,1),
    vote_count INT,
    actor_count INT,
    `rank` INT,
    is_playable BOOLEAN,
    is_watched BOOLEAN,
    rating JSON,
    types JSON,
    regions JSON,
    actors JSON
);

字段类型说明：

• VARCHAR：对于字符串类型的字段，使用VARCHAR并指定最大长度。
• DECIMAL(3,1)：用于存储评分，其中3表示总位数，1表示小数点后的位数。
• DATE：用于存储日期格式的数据。
• INT：用于整数类型的字段。
• BOOLEAN：用于存储布尔值（真/假）。
• JSON：对于包含多个值或复杂结构的字段（如评分、类型、地区和演员），可以使用MySQL的JSON数据类型来存储这些信息。

2. 数据处理与存储

将json文件中数据处理后，保存到MySQL中。

2.1 安装相关库

pip install pandas sqlalchemy mysql-connector-python -i https://mirrors.aliyun.com/pypi/simple/

2.2 代码实现

从目录..\douban_movie_data中读取所有.json文件，将这些json文件的内容合并成一个统一的 DataFrame，并对数据进行清理和转换，包括日期格式化、评分数值化、布尔值标准化以及列表字段转换为 JSON 字符串。接着，它删除电影 id 相同的重复记录，并将最终的数据写入 MySQL 数据库中的original_movie_data表，确保中文字符不被转码。整个过程实现了从本地 JSON 文件到数据库表的自动化数据导入和处理。

import json
import os
import re

import pandas as pd
from sqlalchemy import create_engine

# 设置数据库连接信息
DB_USER = 'root'
DB_PASSWORD = 'zxcvbq'
DB_HOST = '127.0.0.1'  # 或者你的数据库主机地址
DB_PORT = '3306'  # MySQL默认端口是3306
DB_NAME = 'douban_movie'

# 创建数据库引擎
engine = create_engine(f'mysql+mysqlconnector://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}')


# 定义函数：读取JSON文件并转换为DataFrame
def read_json_to_df(file_path):
    with open(file_path, 'r', encoding='utf-8') as file:
        data = json.load(file)
        df = pd.DataFrame(data)
        return df


def preprocess_date(date_str):
    if isinstance(date_str, str) and re.match(r'^\d{4}$', date_str):  # 只有四位数字（年）
        return f"{date_str}-01-01"
    else:
        return date_str


# 定义函数：清理和转换数据格式
def clean_and_transform(df):
    # 转换日期格式
    df['release_date'] = df['release_date'].apply(preprocess_date)
    df['release_date'] = pd.to_datetime(df['release_date'], errors='coerce').dt.date

    # 将字符串评分转换为浮点数，保留一位小数
    df['score'] = df['score'].astype(float).round(1)

    # 确保布尔值字段正确
    df['is_playable'] = df['is_playable'].astype(bool)
    df['is_watched'] = df['is_watched'].astype(bool)

    # 将列表类型的字段转换为JSON字符串
    for col in ['rating', 'types', 'regions', 'actors']:
        # df[col] = df[col].apply(json.dumps)
        df[col] = df[col].apply(lambda x: json.dumps(x, ensure_ascii=False))

    return df


if __name__ == '__main__':
    # 获取目录下所有JSON文件路径
    directory = r'..\douban_movie_data'
    json_files = [os.path.join(directory, f) for f in os.listdir(directory) if f.endswith('.json')]

    # 初始化一个空的DataFrame用于存储所有电影数据
    all_movies_df = pd.DataFrame()

    # 遍历每个JSON文件，读取内容并追加到总的DataFrame中
    for file in json_files:
        movie_df = read_json_to_df(file)
        all_movies_df = pd.concat([all_movies_df, movie_df], ignore_index=True)

    # 清理和转换数据
    cleaned_df = clean_and_transform(all_movies_df)

    # 删除完全重复数据
    cleaned_df = cleaned_df.drop_duplicates(subset=['id'])

    # 将DataFrame写入MySQL数据库
    cleaned_df.to_sql(name='original_movie_data3', con=engine, if_exists='append', index=False)

    print("所有JSON文件的数据已成功导入MySQL数据库")

查看original_movie_data表中的电影数据：

select * from original_movie_data limit 10;