猫眼电影数据爬取与可视化分析：从数据采集到Flask展示全流程

你有没有试过想看最近哪部电影值得去影院结果翻遍猫眼、豆瓣、淘票票最后还是被各种评分和评论搞得一头雾水或者想分析某个导演的作品票房规律却发现数据散落在不同页面手动整理费时费力上个月我帮朋友做一个小型影视数据分析就遇到了类似问题。原本以为猫眼这类平台的数据应该很好抓真正动手才发现单次抓取几页数据不难难的是稳定获取批量数据、处理反爬机制、把零散信息整理成可分析的结构化数据最后还能直观呈现出来。这篇文章我就以猫眼电影数据爬取与可视化分析为例带你走完从单次请求到完整项目的全过程。但我要先提醒你这个项目的价值不在于“能抓到数据”而在于理解一套可复用的数据采集-清洗-分析-展示工作流。真正重要的不是代码本身而是背后的问题拆解思路和工程化意识。1. 先别急着写爬虫搞清楚你要什么和能要什么很多人一上来就打开编辑器写 requests 代码这是最容易踩坑的开始。猫眼这样的平台对数据抓取有明确的技术限制和合规边界先理解规则比直接冲更重要。1.1 明确数据边界你能获取什么应该获取什么猫眼电影页面公开显示的信息通常包括电影基本信息名称、评分、主演、上映时间、片长票房数据实时票房、累计票房部分页面评论数据用户评分、短评内容但这里有个关键区别公开可见的数据不等于可以无限批量抓取的数据。从技术伦理和平台规则角度你应该遵循只获取公开显示的非个人化数据控制请求频率避免对服务器造成压力不获取需要登录才能访问的敏感信息不将抓取数据用于商业用途在实际代码中这意味着你需要设置合理的请求间隔、使用正式的 HTTP 头部信息、避免触发反爬机制。1.2 技术方案选型为什么是 Flask requests Pandas这个组合看起来简单但每个组件都有明确的分工requests负责 HTTP 请求处理比 urllib 更简洁的 API自动处理连接复用和持久化内置会话管理和 Cookie 保持Pandas负责数据清洗和结构化处理缺失值处理、数据类型转换表格化数据操作和筛选为后续分析和可视化做准备Flask提供轻量级 Web 界面快速展示分析结果可扩展为数据查询接口学习成本低于 Django关键是这个组合覆盖了从数据获取到展示的全链路而且每个环节都可以独立优化。1.3 环境准备别在依赖版本上踩坑Python 环境建议使用 3.8主要依赖库版本兼容性# 基础请求和解析 requests2.25.1 beautifulsoup44.9.3 # 数据处理 pandas1.3.0 numpy1.20.0 # 可视化 matplotlib3.3.0 seaborn0.11.0 # Web 框架 flask2.0.0安装时最容易出问题的是环境隔离。建议使用虚拟环境python -m venv movie_analysis source movie_analysis/bin/activate # Linux/Mac movie_analysis\Scripts\activate # Windows pip install -r requirements.txt2. 爬虫核心从单页请求到稳定批量采集写爬虫最怕的就是代码看起来能跑一上量就各种报错。关键在于理解请求逻辑和异常处理。2.1 构建稳健的单页请求函数先看一个基础的请求函数重点不在功能完整而在错误处理import requests import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session_with_retries(): 创建带重试机制的会话 session requests.Session() # 重试策略 retry_strategy Retry( total3, # 最大重试次数 backoff_factor1, # 重试等待时间因子 status_forcelist[429, 500, 502, 503, 504], # 需要重试的状态码 ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 设置合理的请求头 session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Connection: keep-alive }) return session def get_movie_page(url, delay1): 获取单页数据包含错误处理 session create_session_with_retries() try: time.sleep(delay) # 请求间隔 response session.get(url, timeout10) response.raise_for_status() # 检查HTTP状态码 # 检查是否被反爬 if 验证 in response.text or access denied in response.text.lower(): print(可能触发了反爬机制需要调整策略) return None return response.text except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None这个函数有几个关键设计使用会话对象保持连接复用重试机制处理临时网络问题超时设置避免长时间等待反爬检测提前发现问题请求间隔避免频率过高2.2 解析逻辑用 BeautifulSoup 提取结构化数据猫眼页面结构相对规整但还是要处理各种边界情况from bs4 import BeautifulSoup import re def parse_movie_list(html): 解析电影列表页 if not html: return [] soup BeautifulSoup(html, html.parser) movies [] # 根据实际页面结构调整选择器 movie_items soup.select(.movie-item) # 示例选择器 for item in movie_items: try: movie {} # 电影名称 name_elem item.select_one(.movie-name) movie[name] name_elem.text.strip() if name_elem else 未知 # 评分处理 score_elem item.select_one(.score) if score_elem: score_text score_elem.text.strip() movie[score] float(score_text) if score_text.replace(., ).isdigit() else 0.0 else: movie[score] 0.0 # 上映时间解析 date_elem item.select_one(.release-date) if date_elem: date_text date_elem.text.strip() # 多种日期格式处理 movie[release_date] parse_date(date_text) else: movie[release_date] None # 演员信息处理多演员情况 actors_elems item.select(.actors) movie[actors] [actor.text.strip() for actor in actors_elems] if actors_elems else [] movies.append(movie) except Exception as e: print(f解析电影信息失败: {e}) continue # 单条记录失败不影响整体 return movies def parse_date(date_text): 处理多种日期格式 patterns [ r\d{4}-\d{2}-\d{2}, r\d{4}年\d{1,2}月\d{1,2}日, r\d{4}\.\d{1,2}\.\d{1,2} ] for pattern in patterns: match re.search(pattern, date_text) if match: return match.group() return date_text # 无法解析时返回原文本解析时最容易忽略的是数据清洗文本前后的空白字符数字类型的转换验证空值和缺失值处理日期格式的统一2.3 批量采集策略平衡效率与稳定性单页解析完成后批量采集要考虑分页逻辑和频率控制def crawl_maoyan_movies(base_url, max_pages10, start_page0): 批量爬取多页电影数据 all_movies [] session create_session_with_retries() for page in range(start_page, start_page max_pages): print(f正在爬取第 {page 1} 页...) # 构造分页URL根据实际网站结构调整 url f{base_url}?offset{page * 30} # 示例分页参数 html get_movie_page(url, delay1.5) # 增加延迟 if html: movies parse_movie_list(html) all_movies.extend(movies) print(f第 {page 1} 页获取到 {len(movies)} 部电影) else: print(f第 {page 1} 页获取失败停止爬取) break # 每5页增加一次较长延迟 if (page 1) % 5 0: time.sleep(3) return all_movies批量爬取的关键平衡点请求频率太快容易被封太慢效率低错误处理单页失败不应影响整体流程进度保存长时间运行需要断点续传资源释放及时关闭连接避免内存泄漏3. 数据清洗从原始文本到分析就绪的结构化数据爬取到的原始数据往往包含各种不一致和缺失直接分析会得到错误结论。3.1 使用 Pandas 进行数据质量检查import pandas as pd from datetime import datetime def clean_movie_data(movies_list): 清洗电影数据 df pd.DataFrame(movies_list) print(原始数据概况:) print(f总记录数: {len(df)}) print(\n各字段缺失值情况:) print(df.isnull().sum()) # 数据类型转换 if score in df.columns: df[score] pd.to_numeric(df[score], errorscoerce) # 处理重复数据 duplicate_count df.duplicated(subset[name]).sum() if duplicate_count 0: print(f发现 {duplicate_count} 条重复记录进行去重) df df.drop_duplicates(subset[name], keepfirst) # 日期字段标准化 if release_date in df.columns: df[release_date] pd.to_datetime(df[release_date], errorscoerce) # 提取年份月份等信息 df[release_year] df[release_date].dt.year df[release_month] df[release_date].dt.month # 演员列表处理 if actors in df.columns: # 统计演员数量 df[actor_count] df[actors].apply(lambda x: len(x) if isinstance(x, list) else 0) # 提取主要演员第一个 df[main_actor] df[actors].apply( lambda x: x[0] if isinstance(x, list) and len(x) 0 else 未知 ) return df def analyze_data_quality(df): 分析数据质量 print(\n 数据质量分析 ) # 数值型字段统计 numeric_cols df.select_dtypes(include[number]).columns for col in numeric_cols: print(f\n{col} 字段统计:) print(f 有效值数量: {df[col].notna().sum()}) print(f 平均值: {df[col].mean():.2f}) print(f 中位数: {df[col].median():.2f}) print(f 标准差: {df[col].std():.2f}) # 文本型字段分析 text_cols df.select_dtypes(include[object]).columns for col in text_cols: unique_count df[col].nunique() print(f\n{col} 字段: {unique_count} 个唯一值) # 显示最常见的几个值 top_values df[col].value_counts().head(3) for value, count in top_values.items(): print(f {value}: {count} 次)3.2 处理常见数据问题实际数据清洗中会遇到的各种问题def handle_common_issues(df): 处理常见数据问题 # 1. 评分异常值处理 if score in df.columns: # 假设合理评分范围是 0-10 score_mask (df[score] 0) (df[score] 10) outlier_count (~score_mask).sum() if outlier_count 0: print(f发现 {outlier_count} 条异常评分记录) df df[score_mask] # 直接过滤异常值 # 2. 电影名称规范化 if name in df.columns: df[name] df[name].str.strip() # 去除多余的空格和特殊字符 df[name] df[name].str.replace(r\s, , regexTrue) # 3. 年代久远电影处理 if release_year in df.columns: current_year datetime.now().year # 标记近5年的电影 df[is_recent] df[release_year] (current_year - 5) # 4. 演员信息展开为分析做准备 if actors in df.columns: # 创建演员出现频次统计 from collections import Counter all_actors [] for actors_list in df[actors].dropna(): if isinstance(actors_list, list): all_actors.extend(actors_list) actor_stats Counter(all_actors) df[actor_popularity] df[main_actor].map( lambda x: actor_stats.get(x, 0) if pd.notna(x) else 0 ) return df数据清洗的核心原则保持数据真实性不随意修改原始值记录所有处理步骤确保可追溯区分数据修复和数据增强为后续分析优化数据结构4. 可视化分析用图表讲好数据故事清洗后的数据需要合适的可视化来发现规律。重点不是图表数量而是每个图表要回答一个明确问题。4.1 评分分布分析import matplotlib.pyplot as plt import seaborn as sns def plot_rating_distribution(df): 绘制评分分布图 plt.figure(figsize(12, 8)) # 设置中文字体根据系统调整 plt.rcParams[font.sans-serif] [SimHei, DejaVu Sans] plt.rcParams[axes.unicode_minus] False # 1. 评分分布直方图 plt.subplot(2, 2, 1) plt.hist(df[score].dropna(), bins20, alpha0.7, colorskyblue, edgecolorblack) plt.xlabel(评分) plt.ylabel(电影数量) plt.title(电影评分分布) plt.grid(True, alpha0.3) # 2. 评分箱线图 plt.subplot(2, 2, 2) df[score].dropna().plot(kindbox, vertFalse) plt.title(评分箱线图) plt.xlabel(评分) # 3. 年度平均评分趋势 plt.subplot(2, 2, 3) yearly_avg df.groupby(release_year)[score].mean() yearly_count df.groupby(release_year).size() # 只显示有足够数据的年份 valid_years yearly_count[yearly_count 5].index yearly_avg_filtered yearly_avg[valid_years] plt.plot(yearly_avg_filtered.index, yearly_avg_filtered.values, markero, linewidth2, markersize4) plt.xlabel(年份) plt.ylabel(平均评分) plt.title(年度平均评分趋势) plt.grid(True, alpha0.3) plt.xticks(rotation45) # 4. 评分与演员数量关系 plt.subplot(2, 2, 4) if actor_count in df.columns: plt.scatter(df[actor_count], df[score], alpha0.5) plt.xlabel(演员数量) plt.ylabel(评分) plt.title(评分与演员数量关系) plt.grid(True, alpha0.3) plt.tight_layout() return plt4.2 时间维度分析def analyze_time_trends(df): 时间趋势分析 fig, axes plt.subplots(2, 2, figsize(14, 10)) # 1. 年度电影产量 yearly_production df[release_year].value_counts().sort_index() axes[0, 0].bar(yearly_production.index, yearly_production.values, alpha0.7) axes[0, 0].set_xlabel(年份) axes[0, 0].set_ylabel(电影数量) axes[0, 0].set_title(年度电影产量) axes[0, 0].tick_params(axisx, rotation45) # 2. 月度分布忽略年份 monthly_dist df[release_month].value_counts().sort_index() axes[0, 1].bar(monthly_dist.index, monthly_dist.values, colororange, alpha0.7) axes[0, 1].set_xlabel(月份) axes[0, 1].set_ylabel(电影数量) axes[0, 1].set_title(电影上映月份分布) axes[0, 1].set_xticks(range(1, 13)) # 3. 热门演员分析 if main_actor in df.columns: top_actors df[main_actor].value_counts().head(10) axes[1, 0].barh(range(len(top_actors)), top_actors.values) axes[1, 0].set_yticks(range(len(top_actors))) axes[1, 0].set_yticklabels(top_actors.index) axes[1, 0].set_xlabel(电影数量) axes[1, 0].set_title(热门演员作品数量) # 4. 评分区间统计 rating_bins [0, 3, 6, 8, 10] rating_labels [3分及以下, 3-6分, 6-8分, 8分以上] df[rating_category] pd.cut(df[score], binsrating_bins, labelsrating_labels) rating_dist df[rating_category].value_counts() axes[1, 1].pie(rating_dist.values, labelsrating_dist.index, autopct%1.1f%%) axes[1, 1].set_title(评分区间分布) plt.tight_layout() return fig4.3 高级分析相关性探索def advanced_analysis(df): 高级分析探索变量间关系 # 选择数值型列进行相关性分析 numeric_df df.select_dtypes(include[number]) if len(numeric_df.columns) 1: plt.figure(figsize(10, 8)) # 相关性热力图 correlation_matrix numeric_df.corr() sns.heatmap(correlation_matrix, annotTrue, cmapcoolwarm, center0, squareTrue, linewidths0.5) plt.title(变量相关性热力图) plt.tight_layout() # 保存图表 plt.savefig(correlation_heatmap.png, dpi300, bbox_inchestight) return correlation_matrix if correlation_matrix in locals() else None可视化分析的关键洞察评分分布是否呈现特定模式如正态分布是否存在明显的年度趋势或季节性规律演员影响力与电影评分的关系数据质量对分析结论的影响5. Flask 集成从数据分析到交互展示数据分析结果需要合适的展示方式Flask 提供了轻量级的 Web 界面方案。5.1 基础 Flask 应用结构from flask import Flask, render_template, request, jsonify import os app Flask(__name__) app.route(/) def index(): 主页面展示分析概览 # 基础统计信息 stats { total_movies: len(df), avg_rating: df[score].mean(), recent_movies: df[is_recent].sum() if is_recent in df.columns else 0, top_actor: df[main_actor].mode()[0] if main_actor in df.columns else 未知 } return render_template(index.html, statsstats) app.route(/api/movies) def get_movies(): 电影数据API接口 page request.args.get(page, 1, typeint) per_page request.args.get(per_page, 20, typeint) start_idx (page - 1) * per_page end_idx start_idx per_page movies_data df.iloc[start_idx:end_idx].to_dict(records) return jsonify({ movies: movies_data, total: len(df), page: page, per_page: per_page }) app.route(/analysis/analysis_type) def show_analysis(analysis_type): 展示特定分析结果 analysis_types { rating: 评分分析, time: 时间趋势, actor: 演员分析 } if analysis_type not in analysis_types: return 分析类型不存在, 404 # 生成对应的分析图表 if analysis_type rating: chart_path generate_rating_chart() elif analysis_type time: chart_path generate_time_chart() else: chart_path generate_actor_chart() return render_template(analysis.html, analysis_typeanalysis_types[analysis_type], chart_pathchart_path) def generate_rating_chart(): 生成评分分析图表 plt plot_rating_distribution(df) chart_path static/images/rating_analysis.png plt.savefig(chart_path, dpi300, bbox_inchestight) plt.close() return chart_path if __name__ __main__: # 确保静态文件目录存在 os.makedirs(static/images, exist_okTrue) app.run(debugTrue, host0.0.0.0, port5000)5.2 模板设计简洁有效的信息展示创建templates/index.html!DOCTYPE html html head title猫眼电影数据分析/title meta charsetutf-8 style .stats-container { display: flex; justify-content: space-around; margin: 20px 0; } .stat-card { background: #f5f5f5; padding: 20px; border-radius: 8px; text-align: center; } .nav-links { text-align: center; margin: 30px 0; } .nav-links a { margin: 0 15px; text-decoration: none; color: #007bff; } /style /head body h1 styletext-align: center;猫眼电影数据分析平台/h1 div classstats-container div classstat-card h3总电影数量/h3 p stylefont-size: 24px; font-weight: bold;{{ stats.total_movies }}/p /div div classstat-card h3平均评分/h3 p stylefont-size: 24px; font-weight: bold;{{ %.1f|format(stats.avg_rating) }}/p /div div classstat-card h3近5年电影/h3 p stylefont-size: 24px; font-weight: bold;{{ stats.recent_movies }}/p /div div classstat-card h3最活跃演员/h3 p stylefont-size: 24px; font-weight: bold;{{ stats.top_actor }}/p /div /div div classnav-links a href/analysis/rating评分分析/a a href/analysis/time时间趋势/a a href/analysis/actor演员分析/a a href/api/movies数据接口/a /div /body /html5.3 数据导出功能app.route(/export/format_type) def export_data(format_type): 数据导出功能 if format_type csv: from io import StringIO output StringIO() df.to_csv(output, indexFalse, encodingutf-8) output.seek(0) return app.response_class( output.getvalue(), mimetypetext/csv, headers{Content-Disposition: attachment;filenamemaoyan_movies.csv} ) elif format_type json: return jsonify(df.to_dict(records)) else: return 不支持的导出格式, 400Flask 集成的核心价值快速验证分析结果提供数据查询接口支持结果导出和分享为更复杂应用提供基础框架6. 项目优化与生产级考量这个基础版本能跑通流程但要用于真实场景还需要考虑更多工程化问题。6.1 性能优化策略# 1. 数据缓存机制 import pickle import hashlib def get_data_cache_key(url): 生成缓存键 return hashlib.md5(url.encode()).hexdigest() def cached_request(url, cache_dircache, expire_hours24): 带缓存的请求函数 os.makedirs(cache_dir, exist_okTrue) cache_key get_data_cache_key(url) cache_file os.path.join(cache_dir, f{cache_key}.pkl) # 检查缓存是否有效 if os.path.exists(cache_file): file_age time.time() - os.path.getmtime(cache_file) if file_age expire_hours * 3600: with open(cache_file, rb) as f: return pickle.load(f) # 执行新请求 result get_movie_page(url) # 保存缓存 if result: with open(cache_file, wb) as f: pickle.dump(result, f) return result # 2. 数据库集成可选 import sqlite3 def save_to_database(df, db_pathmovies.db): 保存到SQLite数据库 conn sqlite3.connect(db_path) # 创建表 df.to_sql(movies, conn, if_existsreplace, indexFalse) conn.close() def load_from_database(db_pathmovies.db): 从数据库加载数据 conn sqlite3.connect(db_path) df pd.read_sql(SELECT * FROM movies, conn) conn.close() return df6.2 错误处理与日志记录import logging from logging.handlers import RotatingFileHandler def setup_logging(): 配置日志系统 logger logging.getLogger(movie_analyzer) logger.setLevel(logging.INFO) # 文件日志滚动记录 file_handler RotatingFileHandler( movie_analysis.log, maxBytes10*1024*1024, # 10MB backupCount5 ) file_handler.setFormatter(logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s )) # 控制台日志 console_handler logging.StreamHandler() console_handler.setFormatter(logging.Formatter( %(levelname)s - %(message)s )) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 在关键函数中添加日志记录 logger setup_logging() def robust_crawl_maoyan_movies(base_url, max_pages10): 增强版的爬取函数 try: logger.info(f开始爬取猫眼电影数据最大页数: {max_pages}) result crawl_maoyan_movies(base_url, max_pages) logger.info(f爬取完成共获取 {len(result)} 条记录) return result except Exception as e: logger.error(f爬取过程出现异常: {e}) return []6.3 配置化管理创建config.pyimport os from datetime import timedelta class Config: # 爬虫配置 REQUEST_DELAY 1.5 # 请求延迟秒 MAX_RETRIES 3 TIMEOUT 10 # Flask配置 SECRET_KEY os.environ.get(SECRET_KEY) or dev-key-please-change DEBUG os.environ.get(DEBUG, False).lower() true # 数据存储 CACHE_DIR data/cache DATABASE_PATH data/movies.db EXPORT_DIR data/exports # 分析配置 RECENT_YEARS 5 # 近几年的定义 MIN_MOVIES_PER_YEAR 5 # 年度分析的最小电影数量 classmethod def init_directories(cls): 初始化所需目录 os.makedirs(cls.CACHE_DIR, exist_okTrue) os.makedirs(cls.EXPORT_DIR, exist_okTrue) os.makedirs(os.path.dirname(cls.DATABASE_PATH), exist_okTrue)6.4 生产部署建议对于实际部署需要考虑环境配置使用环境变量管理敏感信息配置正式数据库MySQL/PostgreSQL设置反向代理Nginx安全考虑限制访问IP范围添加身份验证如果需要防止SQL注入等常见攻击监控维护添加健康检查接口设置日志轮转和监控定期备份数据这个项目真正的价值不在于代码本身而在于展示了一个完整的数据处理流程从需求分析、技术选型、数据获取、清洗处理、分析可视化到最终展示。每个环节都有其独特的技术考量和最佳实践。最重要的是这套方法论可以复用到其他类似的数据分析项目中。下次当你需要分析电商商品、社交媒体内容或其他公开数据时只需要调整数据获取和解析逻辑核心的数据处理和分析框架都可以直接复用。