Python爬虫实战：Flask+Requests+Pandas构建猫眼电影数据分析系统-开发者社区

在数据驱动的时代，如何高效获取并分析网络公开数据成为开发者必备技能。近期在完成一个影视数据分析项目时，需要采集猫眼电影平台的票房、评分、演员等信息，但直接手动收集效率低下且容易出错。通过结合 Flask、requests 和 Pandas 这三个 Python 利器，我整理出一套完整的爬虫+可视化解决方案，从环境搭建到数据呈现全程可复现，特别适合有一定 Python 基础想要实战爬虫和数据分析的开发者。

本文将手把手带你实现猫眼电影数据的爬取、清洗、存储和可视化分析，包含完整的源码和可视化图表。无论你是学生做课程设计，还是开发者需要快速获取市场数据，都能直接复用这套方案。

1. 项目背景与技术选型

1.1 为什么选择猫眼电影数据

猫眼电影作为国内领先的影视信息平台，包含了丰富的电影数据：实时票房、评分详情、演员阵容、上映时间等。这些数据对于影视行业分析、市场调研、竞品分析都具有重要价值。相比其他平台，猫眼数据的接口相对规范，适合作为爬虫入门到进阶的实战项目。

1.2 技术栈优势分析

Flask：轻量级 Web 框架，用于构建简单的数据展示界面，将爬取的数据通过 Web 页面可视化呈现。

requests：Python 中最常用的 HTTP 库，简洁易用，能够高效处理网络请求和响应。

Pandas：数据分析核心库，提供强大的数据清洗、转换和分析能力，特别适合处理结构化数据。

三者的协同作用：requests 负责数据获取，Pandas 进行数据处理，Flask 实现结果展示，形成一个完整的数据流水线。

1.3 项目目标与预期成果

通过本项目，你将掌握：

网页数据爬取的基本原理和实战技巧
反爬虫机制的应对策略
数据清洗和存储的最佳实践
使用 Pandas 进行多维度数据分析
利用 Flask 构建简单数据可视化界面

最终实现一个能够自动爬取猫眼电影数据、生成多种可视化图表的数据分析系统。

2. 环境准备与工具配置

2.1 Python 环境要求

本项目需要 Python 3.8+ 环境，建议使用 Python 3.9 或 3.10 以获得更好的兼容性。可以通过以下命令检查 Python 版本：

python --version # 或 python3 --version

如果尚未安装 Python，可以从官网下载安装包，或者使用 Miniconda/Anaconda 进行环境管理。

2.2 必要库安装

创建并激活虚拟环境后，安装项目依赖：

# 创建虚拟环境（可选但推荐） python -m venv movie_env source movie_env/bin/activate # Linux/Mac # movie_env\Scripts\activate # Windows # 安装核心依赖 pip install flask requests pandas matplotlib seaborn beautifulsoup4

2.3 开发工具推荐

VS Code：配置 Python 扩展，提供代码提示和调试功能。Jupyter Notebook：适合数据分析和调试阶段使用。Postman：用于测试 API 接口和请求头设置。

3. 爬虫核心原理与实战技巧

3.1 网页结构分析与数据定位

猫眼电影的数据主要通过 API 接口返回 JSON 格式数据，这比解析 HTML 页面更高效。通过浏览器开发者工具（F12）可以找到数据接口：

打开猫眼电影官网
按 F12 打开开发者工具
切换到 Network 标签页
刷新页面并筛选 XHR 请求
查找包含电影数据的接口

3.2 requests 库的高级用法

import requests import json from time import sleep class MaoyanSpider: def __init__(self): self.session = requests.Session() # 设置请求头，模拟浏览器行为 self.headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': 'https://maoyan.com/', 'Accept': 'application/json, text/plain, */*' } def get_movie_list(self, offset=0, limit=30): """获取电影列表数据""" url = f'https://api.maoyan.com/ajax/movieOnInfoList' params = { 'token': '', 'offset': offset, 'limit': limit } try: response = self.session.get(url, headers=self.headers, params=params, timeout=10) response.raise_for_status() # 检查请求是否成功 return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") return None

3.3 应对反爬虫策略

猫眼平台有一定的反爬虫机制，需要采取以下策略：

请求频率控制：在连续请求之间添加随机延时，避免触发频率限制。

import random import time def safe_request(url, headers, max_retries=3): """安全的请求函数，包含重试机制""" for attempt in range(max_retries): try: response = requests.get(url, headers=headers, timeout=10) if response.status_code == 200: return response elif response.status_code == 429: # 请求过于频繁 wait_time = (2 ** attempt) + random.random() print(f"请求频繁，等待 {wait_time:.2f} 秒后重试") time.sleep(wait_time) else: print(f"HTTP错误: {response.status_code}") break except Exception as e: print(f"请求异常: {e}") if attempt == max_retries - 1: return None time.sleep(1) return None

IP 代理轮换：如果需要大规模爬取，建议使用代理 IP 池。

4. 完整项目实战：从爬取到可视化

4.1 项目结构设计

创建以下项目目录结构：

movie_analysis/ ├── app.py # Flask 主应用 ├── spider/ │ ├── __init__.py │ ├── maoyan_spider.py # 爬虫核心逻辑 │ └── data_processor.py # 数据处理 ├── static/ │ ├── css/ │ └── js/ ├── templates/ │ └── index.html ├── data/ │ └── movies.csv # 存储爬取的数据 └── requirements.txt

4.2 爬虫模块完整实现

maoyan_spider.py：

import requests import pandas as pd import time import random import json from datetime import datetime class MaoyanMovieSpider: def __init__(self): self.base_url = "https://api.maoyan.com" self.session = requests.Session() self.set_headers() self.movies_data = [] def set_headers(self): """设置请求头""" self.headers = { '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', 'Referer': 'https://maoyan.com/films', 'Origin': 'https://maoyan.com', 'Connection': 'keep-alive', } self.session.headers.update(self.headers) def crawl_movie_list(self, movie_count=100): """爬取电影列表""" print("开始爬取猫眼电影数据...") offset = 0 limit = 30 while len(self.movies_data) < movie_count: url = f"{self.base_url}/ajax/movieOnInfoList" params = { 'token': '', 'offset': offset, 'limit': limit } try: response = self.session.get(url, params=params, timeout=15) if response.status_code == 200: data = response.json() movies = data.get('movieList', []) if not movies: print("没有更多数据") break for movie in movies: movie_detail = self.extract_movie_info(movie) if movie_detail: self.movies_data.append(movie_detail) print(f"已爬取 {len(self.movies_data)} 部电影数据") offset += len(movies) # 随机延时，避免请求过快 time.sleep(random.uniform(1, 3)) else: print(f"请求失败，状态码: {response.status_code}") break except Exception as e: print(f"爬取过程中出现错误: {e}") break print(f"爬取完成，共获取 {len(self.movies_data)} 部电影数据") return self.movies_data def extract_movie_info(self, movie): """提取电影信息""" try: movie_info = { 'movie_id': movie.get('id'), 'name': movie.get('nm'), # 电影名称 'english_name': movie.get('enm', ''), 'score': movie.get('sc', 0), # 评分 'score_count': movie.get('snum', 0), # 评分人数 'release_date': movie.get('rt'), # 上映日期 'duration': movie.get('dur', 0), # 时长（分钟） 'type': movie.get('cat', ''), # 类型 'actors': '|'.join([actor.get('name') for actor in movie.get('star', [])]), # 主演 'director': movie.get('dir', ''), # 导演 'area': movie.get('fra', ''), # 地区 'description': movie.get('dra', ''), # 简介 'img_url': movie.get('img', ''), # 海报URL 'wish_count': movie.get('wish', 0), # 想看人数 'crawl_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } return movie_info except Exception as e: print(f"提取电影信息错误: {e}") return None def save_to_csv(self, filename='data/movies.csv'): """保存数据到CSV文件""" if not self.movies_data: print("没有数据可保存") return False df = pd.DataFrame(self.movies_data) df.to_csv(filename, index=False, encoding='utf-8-sig') print(f"数据已保存到 {filename}") return True

4.3 数据处理与分析模块

data_processor.py：

import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from datetime import datetime import json class MovieDataAnalyzer: def __init__(self, data_file='data/movies.csv'): self.df = pd.read_csv(data_file, encoding='utf-8-sig') self.clean_data() def clean_data(self): """数据清洗""" print("开始数据清洗...") # 处理缺失值 self.df['score'] = self.df['score'].fillna(0) self.df['score_count'] = self.df['score_count'].fillna(0) self.df['duration'] = self.df['duration'].fillna(0) # 转换数据类型 self.df['score'] = pd.to_numeric(self.df['score'], errors='coerce') self.df['score_count'] = pd.to_numeric(self.df['score_count'], errors='coerce') self.df['duration'] = pd.to_numeric(self.df['duration'], errors='coerce') self.df['wish_count'] = pd.to_numeric(self.df['wish_count'], errors='coerce') # 提取年份信息 self.df['release_year'] = self.df['release_date'].str.extract(r'(\d{4})') self.df['release_year'] = pd.to_numeric(self.df['release_year'], errors='coerce') print(f"数据清洗完成，共 {len(self.df)} 条记录") def analyze_score_distribution(self): """分析评分分布""" plt.figure(figsize=(12, 8)) # 评分分布直方图 plt.subplot(2, 2, 1) scores = self.df[self.df['score'] > 0]['score'] plt.hist(scores, bins=20, alpha=0.7, color='skyblue', edgecolor='black') plt.xlabel('评分') plt.ylabel('电影数量') plt.title('电影评分分布') # 评分与评分人数关系 plt.subplot(2, 2, 2) plt.scatter(self.df['score_count'], self.df['score'], alpha=0.6) plt.xlabel('评分人数') plt.ylabel('评分') plt.title('评分与评分人数关系') # 类型分布 plt.subplot(2, 2, 3) type_counts = self.df['type'].str.split(',').explode().value_counts().head(10) type_counts.plot(kind='bar', color='lightcoral') plt.xlabel('电影类型') plt.ylabel('数量') plt.title('电影类型分布') plt.xticks(rotation=45) # 年度电影数量 plt.subplot(2, 2, 4) year_counts = self.df['release_year'].value_counts().sort_index().tail(10) year_counts.plot(kind='line', marker='o', color='green') plt.xlabel('年份') plt.ylabel('电影数量') plt.title('年度电影数量趋势') plt.tight_layout() plt.savefig('static/score_analysis.png', dpi=300, bbox_inches='tight') plt.close() def analyze_duration_trend(self): """分析电影时长趋势""" plt.figure(figsize=(10, 6)) # 过滤有效时长数据 duration_data = self.df[self.df['duration'] > 0] # 时长分布 plt.hist(duration_data['duration'], bins=30, alpha=0.7, color='orange', edgecolor='black') plt.xlabel('时长（分钟）') plt.ylabel('电影数量') plt.title('电影时长分布') plt.axvline(duration_data['duration'].mean(), color='red', linestyle='--', label=f'平均时长: {duration_data["duration"].mean():.1f}分钟') plt.legend() plt.tight_layout() plt.savefig('static/duration_analysis.png', dpi=300, bbox_inches='tight') plt.close() def generate_summary_report(self): """生成数据摘要报告""" summary = { 'total_movies': len(self.df), 'avg_score': self.df['score'].mean(), 'max_score': self.df['score'].max(), 'min_score': self.df['score'].min(), 'avg_duration': self.df[self.df['duration'] > 0]['duration'].mean(), 'most_common_type': self.df['type'].str.split(',').explode().mode().iloc[0] if not self.df['type'].empty else '无', 'recent_year_count': self.df['release_year'].max() } return summary

4.4 Flask Web 应用集成

app.py：

from flask import Flask, render_template, jsonify import os from spider.maoyan_spider import MaoyanMovieSpider from spider.data_processor import MovieDataAnalyzer app = Flask(__name__) @app.route('/') def index(): """首页显示数据分析结果""" return render_template('index.html') @app.route('/crawl') def crawl_data(): """触发数据爬取""" spider = MaoyanMovieSpider() movies_data = spider.crawl_movie_list(movie_count=50) if spider.save_to_csv(): return jsonify({'status': 'success', 'message': f'成功爬取 {len(movies_data)} 条数据'}) else: return jsonify({'status': 'error', 'message': '爬取失败'}) @app.route('/analyze') def analyze_data(): """执行数据分析并生成图表""" if not os.path.exists('data/movies.csv'): return jsonify({'status': 'error', 'message': '请先爬取数据'}) analyzer = MovieDataAnalyzer() analyzer.analyze_score_distribution() analyzer.analyze_duration_trend() summary = analyzer.generate_summary_report() return jsonify({ 'status': 'success', 'message': '分析完成', 'summary': summary }) @app.route('/data') def show_data(): """显示原始数据""" if not os.path.exists('data/movies.csv'): return jsonify({'status': 'error', 'message': '数据文件不存在'}) import pandas as pd df = pd.read_csv('data/movies.csv', encoding='utf-8-sig') return jsonify({ 'status': 'success', 'data': df.head(20).to_dict('records') }) if __name__ == '__main__': # 确保数据目录存在 os.makedirs('data', exist_ok=True) os.makedirs('static', exist_ok=True) app.run(debug=True, host='0.0.0.0', port=5000)

4.5 前端界面模板

templates/index.html：

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>猫眼电影数据分析系统</title> <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet"> <style> .chart-container { background: white; border-radius: 10px; padding: 20px; margin-bottom: 20px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); } .summary-card { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; border-radius: 10px; padding: 20px; margin: 10px; } </style> </head> <body> <div class="container-fluid"> <div class="row bg-dark text-white p-3 mb-4"> <div class="col"> <h1>🐱 猫眼电影数据分析系统</h1> <p class="mb-0">基于 Flask + Requests + Pandas 的完整数据爬取与可视化解决方案</p> </div> </div> <div class="row mb-4"> <div class="col-md-12"> <div class="card"> <div class="card-body"> <h5 class="card-title">操作面板</h5> <button class="btn btn-primary" onclick="crawlData()">开始爬取数据</button> <button class="btn btn-success" onclick="analyzeData()">执行数据分析</button> <button class="btn btn-info" onclick="showRawData()">查看原始数据</button> <span id="statusMessage" class="ms-3"></span> </div> </div> </div> </div> <div id="summarySection" class="row mb-4" style="display: none;"> <div class="col-md-12"> <h4>数据摘要</h4> <div class="row" id="summaryCards"></div> </div> </div> <div class="row"> <div class="col-md-6"> <div class="chart-container"> <h4>评分分布分析</h4> <img id="scoreChart" src="" alt="评分分析图表" class="img-fluid" style="display: none;"> </div> </div> <div class="col-md-6"> <div class="chart-container"> <h4>时长分布分析</h4> <img id="durationChart" src="" alt="时长分析图表" class="img-fluid" style="display: none;"> </div> </div> </div> <div class="row mt-4"> <div class="col-md-12"> <div class="chart-container"> <h4>原始数据预览</h4> <div id="rawDataTable"></div> </div> </div> </div> </div> <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/js/bootstrap.bundle.min.js"></script> <script> function updateStatus(message, isError = false) { const statusElement = document.getElementById('statusMessage'); statusElement.textContent = message; statusElement.className = isError ? 'ms-3 text-danger' : 'ms-3 text-success'; } async function crawlData() { updateStatus('正在爬取数据...'); try { const response = await fetch('/crawl'); const result = await response.json(); updateStatus(result.message, result.status === 'error'); } catch (error) { updateStatus('请求失败: ' + error.message, true); } } async function analyzeData() { updateStatus('正在分析数据...'); try { const response = await fetch('/analyze'); const result = await response.json(); if (result.status === 'success') { updateStatus('分析完成'); displaySummary(result.summary); // 显示图表（需要确保图表已生成） document.getElementById('scoreChart').src = '/static/score_analysis.png?' + new Date().getTime(); document.getElementById('scoreChart').style.display = 'block'; document.getElementById('durationChart').src = '/static/duration_analysis.png?' + new Date().getTime(); document.getElementById('durationChart').style.display = 'block'; } else { updateStatus(result.message, true); } } catch (error) { updateStatus('分析失败: ' + error.message, true); } } function displaySummary(summary) { document.getElementById('summarySection').style.display = 'block'; const summaryCards = document.getElementById('summaryCards'); summaryCards.innerHTML = ` <div class="col-md-3"> <div class="summary-card"> <h5>总电影数</h5> <h2>${summary.total_movies}</h2> </div> </div> <div class="col-md-3"> <div class="summary-card"> <h5>平均评分</h5> <h2>${summary.avg_score.toFixed(1)}</h2> </div> </div> <div class="col-md-3"> <div class="summary-card"> <h5>平均时长</h5> <h2>${summary.avg_duration.toFixed(0)}分钟</h2> </div> </div> <div class="col-md-3"> <div class="summary-card"> <h5>主要类型</h5> <h4>${summary.most_common_type}</h4> </div> </div> `; } async function showRawData() { try { const response = await fetch('/data'); const result = await response.json(); if (result.status === 'success') { const table = document.getElementById('rawDataTable'); if (result.data.length > 0) { let html = '<table class="table table-striped"><thead><tr>'; // 表头 Object.keys(result.data[0]).forEach(key => { html += `<th>${key}</th>`; }); html += '</tr></thead><tbody>'; // 数据行 result.data.forEach(row => { html += '<tr>'; Object.values(row).forEach(value => { html += `<td>${value}</td>`; }); html += '</tr>'; }); html += '</tbody></table>'; table.innerHTML = html; } } } catch (error) { console.error('获取数据失败:', error); } } </script> </body> </html>

5. 运行与结果验证

5.1 启动应用

在项目根目录下执行：

python app.py

访问 http://localhost:5000 即可看到系统界面。

5.2 操作流程

数据爬取：点击"开始爬取数据"按钮，系统将自动从猫眼平台获取电影数据
数据分析：爬取完成后点击"执行数据分析"，生成可视化图表
结果查看：系统将显示评分分布、时长分析、数据摘要等信息

5.3 预期输出结果

成功运行后，你将看到：

电影评分分布直方图
评分与评分人数关系散点图
电影类型分布条形图
年度电影数量趋势图
电影时长分布分析
数据摘要统计信息

6. 常见问题与解决方案

6.1 爬虫相关问题

问题1：请求被拒绝，返回 403 或 429 状态码

原因分析：触发了网站的反爬虫机制，可能是请求频率过高或缺少必要的请求头。

解决方案：

# 增加随机延时 time.sleep(random.uniform(2, 5)) # 完善请求头信息 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': 'https://maoyan.com/', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', }

问题2：数据解析错误，JSON 解码失败

原因分析：接口返回的数据格式发生变化或包含异常字符。

解决方案：

try: data = response.json() except json.JSONDecodeError: # 尝试手动清理数据 text = response.text.strip() if text.startswith('(') and text.endswith(')'): text = text[1:-1] data = json.loads(text)

6.2 数据处理问题

问题3：Pandas 读取 CSV 文件乱码

解决方案：指定正确的编码格式

df = pd.read_csv('data/movies.csv', encoding='utf-8-sig') # 或尝试其他编码 # df = pd.read_csv('data/movies.csv', encoding='gbk')

问题4：数据类型转换错误

解决方案：使用安全的类型转换方法

# 不安全的转换 # df['score'] = df['score'].astype(float) # 安全的转换 df['score'] = pd.to_numeric(df['score'], errors='coerce')

6.3 Flask 应用问题

问题5：静态文件无法加载

解决方案：确保静态文件目录结构正确，并检查文件路径

# 在模板中正确引用静态文件 <img src="{{ url_for('static', filename='score_analysis.png') }}" alt="评分分析">

问题6：端口被占用

解决方案：更换端口或终止占用进程

if __name__ == '__main__': app.run(debug=True, port=5001) # 使用其他端口

7. 项目优化与扩展建议

7.1 性能优化方案

数据库存储：对于大量数据，建议使用 SQLite 或 MySQL 替代 CSV 文件

import sqlite3 def save_to_sqlite(data): conn = sqlite3.connect('movies.db') df = pd.DataFrame(data) df.to_sql('movies', conn, if_exists='replace', index=False) conn.close()

异步爬取：使用 aiohttp 实现异步请求，提高爬取效率

import aiohttp import asyncio async def fetch_movie(session, url): async with session.get(url) as response: return await response.json()

7.2 功能扩展方向

多平台数据整合：增加豆瓣、时光网等平台的数据爬取实时数据更新：设置定时任务，自动更新电影数据高级分析功能：增加票房预测、演员影响力分析等高级功能用户交互功能：允许用户筛选、排序、自定义分析维度

7.3 生产环境部署

使用 Gunicorn 部署 Flask 应用：

pip install gunicorn gunicorn -w 4 -b 0.0.0.0:5000 app:app

添加 Nginx 反向代理，提高应用性能和安全性。

8. 法律与道德注意事项

8.1 合规爬取原则

遵守 robots.txt：检查目标网站的爬虫协议
控制请求频率：避免对目标网站造成压力
尊重数据版权：仅用于学习研究，不用于商业用途
用户隐私保护：不爬取个人隐私信息

8.2 数据使用规范

本项目获取的数据应仅用于：

学习和技术研究
个人数据分析练习
学术研究目的

禁止将爬取的数据用于：

商业盈利活动
不正当竞争
侵犯他人权益的行为

这个完整的 Flask + requests + Pandas 电影数据爬取与可视化项目，涵盖了从环境搭建到生产部署的全流程，每个模块都提供了可运行的代码示例和详细解释。通过这个项目，你不仅能掌握爬虫技术，还能学会如何将爬取的数据进行有效的分析和可视化展示，为后续更复杂的数据分析项目打下坚实基础。

Python爬虫实战：Flask+Requests+Pandas构建猫眼电影数据分析系统