豆瓣Top250电影数据爬取实战：手把手教你用Python+Xpath搞定数据清洗与CSV保存

豆瓣Top250电影数据清洗实战：Python+Xpath高效处理脏数据

当你第一次成功爬取到豆瓣Top250电影的原始HTML时，那种成就感很快会被眼前的"数据泥潭"冲淡——导演和主演信息粘连在一起、字段中夹杂着\xa0乱码、某些电影缺少主演信息、年份和国籍被斜杠分隔……作为爬虫进阶者，你需要的不再是基础爬取教程，而是一套系统化的数据清洗方法论。本文将带你深入Xpath与Python字符串处理的配合技巧，解决以下真实痛点：

• 如何处理\xa0这类HTML特殊字符？
• 当导演和主演信息合并在一个字段时如何智能拆分？
• 遇到没有主演的电影该怎么避免索引越界错误？
• 怎样把"1994/美国/犯罪"这样的复合字段拆分成结构化数据？

1. 解析前的准备工作：建立数据清洗框架

1.1 分析原始数据结构

豆瓣Top250页面中，每部电影的信息区块都遵循相似结构。通过Chrome开发者工具检查，我们发现关键数据分布在几个核心节点：

# 基础Xpath选择器框架 movie_list = tree.xpath('//ol[@class="grid_view"]/li') for movie in movie_list: title_cn = movie.xpath('.//span[@class="title"][1]/text()') info_line1 = movie.xpath('.//div[@class="info"]/div[@class="bd"]/p[1]/text()') info_line2 = movie.xpath('.//div[@class="info"]/div[@class="bd"]/p[1]/text()[2]')

典型的问题数据示例：

导演: 弗兰克·德拉邦特\xa0\xa0\xa0主演: 蒂姆·罗宾斯 / 摩根·弗里曼 1994\xa0/\xa0美国\xa0/\xa0犯罪

1.2 设计数据清洗流水线

建立分阶段处理流程：

1. 原始提取：用Xpath获取原始文本
2. 初级清洗：去除空白字符、特殊编码
3. 结构化拆分：分离导演/主演、年份/国家/类型
4. 异常处理：处理缺失字段
5. 最终格式化：统一数据格式

def clean_movie_data(raw_data): # 初级清洗 cleaned = remove_special_chars(raw_data) # 结构化处理 structured = split_combined_fields(cleaned) # 容错处理 validated = handle_missing_values(structured) return validated

2. 核心清洗技术实战

2.1 处理特殊字符与空白

HTML中常见的\xa0是不间断空格，需要特殊处理：

import re def clean_whitespace(text): """统一处理各种空白字符""" if not text: return "" # 替换各种空白字符为普通空格 text = re.sub(r'[\xa0\u3000]+', ' ', str(text)) # 去除首尾空格 return text.strip()

注意：豆瓣的年份信息中常出现\xa0/\xa0，需要特别处理

2.2 导演与主演信息分离

当遇到"导演: 张三\xa0\xa0\xa0主演: 李四/王五"这样的合并字段时：

def split_director_actors(info_str): """分离导演和主演信息""" info_str = clean_whitespace(info_str) # 初始化默认值 director = None actors = [] if "导演:" in info_str: parts = re.split(r'导演:\s*|\s*主演:\s*', info_str) parts = [p for p in parts if p] if len(parts) >= 1: director = parts[0] if len(parts) >= 2: actors = [a.strip() for a in parts[1].split('/') if a.strip()] return director, actors

处理边界情况：

• 只有导演没有主演的电影（如纪录片）
• 导演信息缺失的异常情况

2.3 复合字段的智能拆分

对于"1994/美国/犯罪"这样的复合字段：

def split_movie_info(info_str): """拆分年份、国家、类型信息""" if not info_str: return None, None, None parts = [p.strip() for p in info_str.split('/')] parts = [p for p in parts if p] # 处理不同长度的分割结果 year = parts[0] if len(parts) > 0 else None country = parts[1] if len(parts) > 1 else None genre = parts[2] if len(parts) > 2 else None return year, country, genre

3. 构建健壮的清洗系统

3.1 防御式编程实践

from typing import Optional, List class MovieDataCleaner: def __init__(self): self.default_values = { 'director': '未知', 'actors': [], 'year': '0000', 'country': '未知', 'genre': '其他' } def clean(self, raw_data: dict) -> dict: """整体清洗入口""" try: # 导演演员处理 director, actors = self._process_people(raw_data.get('people_info')) # 基础信息处理 year, country, genre = self._process_basic_info(raw_data.get('basic_info')) return { 'title': raw_data.get('title', '').strip(), 'director': director or self.default_values['director'], 'actors': actors or self.default_values['actors'], 'year': year or self.default_values['year'], 'country': country or self.default_values['country'], 'genre': genre or self.default_values['genre'], 'rating': float(raw_data.get('rating', 0)) } except Exception as e: print(f"清洗数据时出错: {e}") return self.default_values def _process_people(self, info_str: Optional[str]) -> tuple: """处理人员信息""" # 实现细节同上... def _process_basic_info(self, info_str: Optional[str]) -> tuple: """处理基础信息""" # 实现细节同上...

3.2 数据验证与日志记录

import logging logging.basicConfig(filename='data_cleaning.log', level=logging.INFO) def validate_movie_data(data): """验证清洗后的数据完整性""" required_fields = ['title', 'director', 'year'] for field in required_fields: if not data.get(field): logging.warning(f"缺失必要字段 {field}: {data}") return False if not isinstance(data.get('rating'), (int, float)): logging.warning(f"评分格式错误: {data}") return False return True

4. 完整数据处理流程示例

4.1 从爬取到保存的端到端流程

import csv from lxml import etree import requests def scrape_and_clean(): headers = {'User-Agent': 'Mozilla/5.0'} cleaner = MovieDataCleaner() with open('douban_top250_cleaned.csv', 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=[ 'title', 'director', 'actors', 'year', 'country', 'genre', 'rating', 'review_count' ]) writer.writeheader() for start in range(0, 250, 25): url = f'https://movie.douban.com/top250?start={start}' response = requests.get(url, headers=headers) tree = etree.HTML(response.text) for movie in tree.xpath('//ol[@class="grid_view"]/li'): raw_data = extract_raw_data(movie) cleaned_data = cleaner.clean(raw_data) if validate_movie_data(cleaned_data): writer.writerow(cleaned_data) def extract_raw_data(movie_element): """从HTML元素提取原始数据""" return { 'title': movie_element.xpath('.//span[@class="title"][1]/text()')[0], 'people_info': ''.join(movie_element.xpath('.//div[@class="bd"]/p[1]/text()')), 'basic_info': movie_element.xpath('.//div[@class="bd"]/p[1]/text()[2]')[0], 'rating': movie_element.xpath('.//span[@class="rating_num"]/text()')[0], 'review_count': movie_element.xpath('.//div[@class="star"]/span[4]/text()')[0] }

4.2 常见问题解决方案

5. 高级技巧与性能优化

5.1 多进程清洗加速

from multiprocessing import Pool def parallel_clean(raw_data_list): """并行处理大量数据""" with Pool(processes=4) as pool: results = pool.map(MovieDataCleaner().clean, raw_data_list) return [r for r in results if validate_movie_data(r)]

5.2 使用pandas进行批量处理

import pandas as pd def pandas_clean(df): """利用pandas向量化操作批量清洗""" # 处理特殊字符 df['info'] = df['raw_info'].str.replace(r'[\xa0]+', ' ', regex=True) # 拆分导演演员 df[['director', 'actors']] = df['info'].apply( lambda x: pd.Series(split_director_actors(x)) ) # 拆分年份国家类型 df[['year', 'country', 'genre']] = df['basic_info'].apply( lambda x: pd.Series(split_movie_info(x)) ) return df.drop(columns=['raw_info'])

在实际项目中，数据清洗往往比数据采集耗费更多时间。一个健壮的清洗系统应该能处理各种边界情况，同时保持代码的可读性和可维护性。建议将清洗规则模块化，并为每种异常情况编写单元测试，确保长期维护时不会引入新的问题。