# Python爬虫实战：5分钟搞定全国空气质量数据批量采集（附完整代码） 空气质量数据对于环境研究、健康分析和城市规划等领域至关重要。想象一下，你正在准备一份关于中国城市空气质量的报告，或者需要分析过去几年PM2.5的变化趋势——手动收集这些数据不仅耗时耗力，而且容易出错。这就是Python爬虫技术大显身手的时候了。 本文将带你快速构建一个高效的空气质量数据采集系统，从零开始实现数据的自动化获取、解析和存储。不同于简单的单页爬虫，我们的目标是建立一个可以批量处理全国多个城市数据的完整解决方案，并提供详细的代码注释和实用技巧，即使你是Python初学者也能轻松上手。 ## 1. 环境准备与基础配置 在开始编写爬虫之前，我们需要确保开发环境配置正确。这个项目需要Python 3.6或更高版本，以及几个关键的第三方库。 首先，使用pip安装必要的依赖包： ```bash pip install requests execjs openpyxl ``` 这些库的作用分别是： - **requests**：用于发送HTTP请求获取网页数据 - **execjs**：执行JavaScript代码，处理网站的反爬机制 - **openpyxl**：将采集的数据保存到Excel文件中 接下来，我们创建一个新的Python文件（如`air_quality_crawler.py`），并导入所需的模块： ```python import execjs import openpyxl import requests from openpyxl import Workbook, load_workbook from typing import Dict, List, Optional import time from dataclasses import dataclass import logging ``` > 提示：使用类型注解（如`List[str]`）虽然不是必须的，但能显著提高代码的可读性和IDE的智能提示能力。 配置日志系统是开发爬虫时的一个重要步骤，它能帮助我们在程序运行时监控状态和排查问题： ```python # 配置日志系统 logging.basicConfig( level=logging.INFO, # 设置日志级别为INFO format='%(asctime)s - %(levelname)s - %(message)s', # 定义日志格式 handlers=[ logging.FileHandler('air_quality.log'), # 输出到文件 logging.StreamHandler() # 输出到控制台 ] ) ``` ## 2. 核心数据结构与类设计 良好的数据结构设计是构建稳定爬虫的基础。我们将使用Python的数据类（dataclass）来定义空气质量数据的结构。 ```python @dataclass class CityData: """城市空气质量数据类""" time_point: str # 时间点(年月) aqi: int # 空气质量指数 max_aqi: str # 最高AQI min_aqi: str # 最低AQI pm2_5: int # PM2.5浓度 pm10: int # PM10浓度 co: float # 一氧化碳浓度 no2: int # 二氧化氮浓度 o3: int # 臭氧浓度 so2: int # 二氧化硫浓度 quality: str # 空气质量等级 @classmethod def from_dict(cls, data: Dict): """从字典创建CityData实例，过滤不需要的字段""" valid_fields = { 'time_point', 'aqi', 'max_aqi', 'min_aqi', 'pm2_5', 'pm10', 'co', 'no2', 'o3', 'so2', 'quality' } # 只保留需要的字段 filtered_data = {k: v for k, v in data.items() if k in valid_fields} return cls(**filtered_data) ``` 接下来，我们设计主采集器类`AQIStudyDataFetcher`，它将封装所有的爬取逻辑： ```python class AQIStudyDataFetcher: """空气质量数据采集器核心类""" # 类常量定义 BASE_URL = 'https://www.aqistudy.cn/historydata/api/historyapi.php' COLUMNS = ['时间', 'AQI', '最高AQI', '最低AQI', 'PM2.5', 'PM10', 'CO', 'NO2', 'O3', 'SO2', '质量等级'] def __init__(self, js_file: str = 'gethistory.js', excel_file: str = 'air_quality_data.xlsx'): """初始化采集器""" self.js_file = js_file self.excel_file = excel_file self.env = self._init_js_env() # 初始化JS执行环境 self.workbook = self._init_workbook() # 初始化Excel工作簿 # HTTP请求头配置 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', 'Referer': 'https://www.aqistudy.cn/historydata/', 'Origin': 'https://www.aqistudy.cn' } ``` ## 3. 实现数据获取与处理逻辑 现代网站通常会有各种反爬虫机制，我们的目标网站使用了JavaScript加密参数。我们需要先准备好JS执行环境： ```python def _init_js_env(self) -> execjs.ExternalRuntime.Context: """初始化JS执行环境""" try: with open(self.js_file, 'r', encoding='utf-8') as f: js_code = f.read() return execjs.compile(js_code) except Exception as e: logging.error(f"初始化JS环境失败: {e}") raise ``` 数据获取是爬虫的核心功能，我们需要处理加密参数和响应数据： ```python def fetch_city_data(self, city: str) -> Optional[List[Dict]]: """获取指定城市空气质量数据""" try: # 1. 调用JS生成加密查询参数 query = self.env.call("gethistory", city) # 2. 发送POST请求获取数据 response = requests.post( self.BASE_URL, headers=self.headers, data={'hA4Nse2cT': query}, timeout=10 ) response.raise_for_status() # 检查HTTP状态码 # 3. 使用JS解密返回数据 encrypted_data = response.text result = self.env.call("get_data", encrypted_data) return result['result']['data']['items'] except Exception as e: logging.error(f"获取{city}数据失败: {e}") return None ``` 获取到数据后，我们需要将其保存到Excel文件中，每个城市使用单独的工作表： ```python def save_to_excel(self, city: str, data: List[Dict]) -> bool: """保存城市数据到Excel""" try: # 获取或创建工作表 if city in self.workbook.sheetnames: sheet = self.workbook[city] sheet.delete_rows(1, sheet.max_row) # 清空现有数据 else: sheet = self.workbook.create_sheet(title=city) # 写入表头 sheet.append(self.COLUMNS) # 写入数据行 for item in data: row_data = CityData.from_dict(item) # 使用过滤后的数据 sheet.append([ row_data.time_point, row_data.aqi, row_data.max_aqi, row_data.min_aqi, row_data.pm2_5, row_data.pm10, row_data.co, row_data.no2, row_data.o3, row_data.so2, row_data.quality ]) return True except Exception as e: logging.error(f"保存{city}数据失败: {e}") return False ``` ## 4. 批量处理与实战应用 现在我们已经实现了单城市数据的获取和保存，接下来扩展为批量处理多个城市： ```python def process_cities(self, cities: List[str]) -> None: """批量处理多个城市数据""" success_count = 0 for city in cities: logging.info(f"开始处理: {city}") city_data = self.fetch_city_data(city) if city_data and self.save_to_excel(city, city_data): success_count += 1 logging.info(f"{city}处理完成") # 统一保存工作簿(减少IO操作) try: self.workbook.save(self.excel_file) logging.info(f"处理完成: {success_count}/{len(cities)}") except Exception as e: logging.error(f"保存工作簿失败: {e}") ``` 为了更方便地按省份处理数据，我们可以添加一个省份管理方法： ```python def process_by_province(self, cities_by_province: Dict[str, List[str]]) -> None: """按省份分类处理城市数据""" for province, cities in cities_by_province.items(): logging.info(f"开始处理省份: {province}") self.process_cities(cities) logging.info(f"省份{province}处理完成") ``` 最后，我们创建一个主程序入口来运行整个采集系统： ```python if __name__ == '__main__': # 省份-城市映射配置 CITIES_BY_PROVINCE = { "北京市": ["北京"], "上海市": ["上海"], "广东省": ["广州", "深圳", "珠海"], "江苏省": ["南京", "苏州", "无锡"], "浙江省": ["杭州", "宁波", "温州"] # 可以继续添加更多省份和城市 } try: # 1. 创建采集器实例 fetcher = AQIStudyDataFetcher() # 2. 按省份处理数据 fetcher.process_by_province(CITIES_BY_PROVINCE) # 或者直接处理所有城市 # all_cities = [city for cities in CITIES_BY_PROVINCE.values() for city in cities] # fetcher.process_cities(all_cities) except Exception as e: logging.error(f"系统运行异常: {e}", exc_info=True) ``` ## 5. 高级技巧与优化建议 ### 5.1 处理反爬机制 现代网站通常会有多种反爬虫措施，我们的代码已经处理了参数加密，但还可以进一步优化： - **随机延迟**：在请求之间添加随机延迟，模拟人类操作 - **代理IP池**：使用代理IP避免单个IP被封锁 - **请求头轮换**：定期更换User-Agent等请求头信息 ```python import random import time def fetch_city_data(self, city: str) -> Optional[List[Dict]]: """获取指定城市空气质量数据（带随机延迟）""" try: # 添加随机延迟（1-3秒） time.sleep(random.uniform(1, 3)) # 原有获取逻辑... except Exception as e: logging.error(f"获取{city}数据失败: {e}") return None ``` ### 5.2 数据验证与清洗 采集到的数据可能存在缺失或异常值，我们需要进行验证和清洗： ```python def validate_data(self, data: List[Dict]) -> List[Dict]: """验证和清洗数据""" valid_data = [] for item in data: # 检查必要字段是否存在 if not all(key in item for key in ['time_point', 'aqi', 'pm2_5']): continue # 检查AQI值是否合理 if not 0 <= int(item.get('aqi', -1)) <= 500: continue valid_data.append(item) return valid_data ``` ### 5.3 性能优化 当处理大量城市数据时，可以考虑以下优化措施： - **多线程/异步请求**：使用`concurrent.futures`或`asyncio`提高采集效率 - **内存优化**：分批处理数据，避免一次性加载过多数据到内存 - **断点续传**：记录已处理的城市，程序中断后可以从上次位置继续 ```python from concurrent.futures import ThreadPoolExecutor, as_completed def process_cities_parallel(self, cities: List[str], max_workers: int = 3) -> None: """多线程处理多个城市数据""" success_count = 0 with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { executor.submit(self.fetch_and_save, city): city for city in cities } for future in as_completed(futures): city = futures[future] try: if future.result(): success_count += 1 except Exception as e: logging.error(f"处理{city}时出错: {e}") # 统一保存工作簿 try: self.workbook.save(self.excel_file) logging.info(f"处理完成: {success_count}/{len(cities)}") except Exception as e: logging.error(f"保存工作簿失败: {e}") def fetch_and_save(self, city: str) -> bool: """获取并保存单个城市数据""" city_data = self.fetch_city_data(city) if not city_data: return False return self.save_to_excel(city, city_data) ``` ### 5.4 扩展功能 根据实际需求，可以考虑添加以下扩展功能： - **数据可视化**：使用matplotlib或pyecharts生成空气质量趋势图 - **定时任务**：设置定期自动采集最新数据 - **数据库存储**：将数据保存到MySQL或MongoDB等数据库 - **API接口**：构建Flask或FastAPI服务提供数据查询 ```python # 示例：简单的数据可视化 import matplotlib.pyplot as plt import pandas as pd def plot_aqi_trend(city: str, excel_file: str = 'air_quality_data.xlsx'): """绘制城市AQI趋势图""" df = pd.read_excel(excel_file, sheet_name=city) df['时间'] = pd.to_datetime(df['时间']) df.sort_values('时间', inplace=True) plt.figure(figsize=(12, 6)) plt.plot(df['时间'], df['AQI'], label='AQI') plt.plot(df['时间'], df['PM2.5'], label='PM2.5') plt.title(f'{city}空气质量趋势') plt.xlabel('日期') plt.ylabel('数值') plt.legend() plt.grid() plt.savefig(f'{city}_air_quality.png') plt.close() ``` ## 6. 常见问题与解决方案 在实际使用爬虫过程中，可能会遇到各种问题。以下是几个常见问题及其解决方法： **问题1：JS执行出错，无法生成加密参数** *可能原因*： - 缺少必要的JS文件 - JS代码执行环境不兼容 - 网站JS代码已更新 *解决方案*： - 确保`gethistory.js`文件存在且内容完整 - 尝试安装Node.js作为JS执行后端 - 检查网站是否有更新，可能需要重新分析加密逻辑 **问题2：请求返回403 Forbidden错误** *可能原因*： - 请求头信息不完整 - IP被网站封禁 - Cookie失效 *解决方案*： - 完善请求头，特别是User-Agent和Referer - 使用代理IP轮换 - 更新有效的Cookie值 **问题3：数据保存到Excel时出错** *可能原因*： - Excel文件被其他程序占用 - 工作表名称包含非法字符 - 数据格式不正确 *解决方案*： - 确保Excel文件未被打开 - 清理城市名称中的特殊字符 - 验证数据格式后再保存 **问题4：程序运行速度慢** *可能原因*： - 网络延迟 - 单线程顺序处理 - 反爬延迟设置过长 *解决方案*： - 使用多线程或异步IO - 适当减少延迟时间 - 考虑分布式采集架构 > 注意：在实际开发中，建议先在小规模数据上测试爬虫的各个功能模块，确认无误后再进行大规模数据采集。同时，要尊重网站的robots.txt协议，合理控制请求频率，避免对目标网站造成过大负担。