无论是科研需要还是毕业论文数据采集空气质量，都可仔细读一读该文章，本篇文章介绍了如何采集该网页的数据。

空气质量数据采集与分析系统开发实践

系统概述

本文将介绍一个完整的空气质量数据采集与分析系统的Python实现，该系统能够从"aqistudy.cn"网站抓取历史空气质量数据，并按城市分类存储到Excel文件中。以下是完整的代码实现及详细解析。

核心功能模块

1. 基础配置与日志设置

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

# 配置日志系统
logging.basicConfig(
    level=logging.INFO,  # 设置日志级别为INFO
    format='%(asctime)s - %(levelname)s - %(message)s',  # 定义日志格式
    handlers=[
        logging.FileHandler('air_quality.log'),  # 输出到文件
        logging.StreamHandler()  # 输出到控制台
    ]
)

功能说明：

• 配置了双日志输出（文件+控制台）
• 包含时间戳、日志级别和消息内容
• 便于系统运行监控和问题排查

2. 数据结构定义

@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)

设计要点：

• 使用Python数据类清晰定义数据结构
• 实现了字段过滤功能，确保数据纯净性
• 类型注解提高代码可读性和IDE支持

3. 主采集器类实现

class AQIStudyDataFetcher:
    """空气质量数据采集器核心类"""
    
    # 类常量定义
    BASE_URL = 'https://www.aqistudy.cn/historydata/api/historyapi.php'
    COLUMNS = ['时间', 'aqi', 'max_aqi', 'min_aqi', 'pm2_5', 'pm10', 
              'co', 'no2', 'o3', 'so2', 'quality']
    
    def __init__(self, js_file: str = 'gethistory.js', 
                 excel_file: str = 'air_quality_data.xlsx'):
        """初始化采集器
        Args:
            js_file: JS解密文件路径
            excel_file: 数据存储Excel文件路径
        """
        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)...',
            # 其他必要头信息...
        }
        
        # Cookie配置(需替换为有效值)
        self.cookies = {
            'Hm_lpvt_6088e7f72f5a363447d4bafe03026db8': str(int(time.time())),
        }

架构设计：

• 采用面向对象设计，封装核心功能
• 配置与业务逻辑分离
• 支持依赖注入（JS文件路径、输出文件路径）

4. 核心方法实现

数据获取方法

    def fetch_city_data(self, city: str) -> Optional[List[Dict]]:
        """获取指定城市空气质量数据
        Args:
            city: 城市名称(如"北京")
        Returns:
            成功返回数据列表，失败返回None
        """
        try:
            # 1. 调用JS生成加密查询参数
            query = self.env.call("gethistory", city)
            
            # 2. 发送POST请求获取数据
            response = requests.post(
                self.BASE_URL,
                cookies=self.cookies,
                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

技术要点：

• 集成JS加密/解密流程
• 完善的错误处理和日志记录
• 严格的HTTP状态检查

数据存储方法

    def save_to_excel(self, city: str, data: List[Dict]) -> bool:
        """保存城市数据到Excel
        Args:
            city: 城市名称(作为Sheet名)
            data: 待保存的数据列表
        Returns:
            保存成功返回True，失败返回False
        """
        try:
            # 1. 获取或创建工作表
            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)
            
            # 2. 写入表头
            sheet.append(self.COLUMNS)
            
            # 3. 写入数据行
            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,
                    # 其他字段...
                ])
            
            return True
        except Exception as e:
            logging.error(f"保存{city}数据失败: {e}")
            return False

存储策略：

• 每个城市独立Sheet存储
• 支持数据覆盖更新
• 结构化数据转换确保一致性

5. 批量处理与省份管理

    def process_cities(self, cities: List[str]) -> None:
        """批量处理多个城市数据
        Args:
            cities: 城市名称列表
        """
        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}")

    def process_by_province(self, cities_by_province: Dict[str, List[str]]) -> None:
        """按省份分类处理城市数据
        Args:
            cities_by_province: 省份到城市列表的映射字典
        """
        for province, cities in cities_by_province.items():
            logging.info(f"开始处理省份: {province}")
            self.process_cities(cities)
            logging.info(f"省份{province}处理完成")

批量处理优势：

• 支持两种处理模式：直接城市列表或按省份分类
• 进度统计和日志记录
• 统一资源管理

系统使用示例

if __name__ == '__main__':
    # 省份-城市映射配置
    CITIES_BY_PROVINCE = {
        "陕西省": ["西安", "宝鸡", "咸阳", "铜川"],
        "甘肃省": ["兰州", "定西", "嘉峪关"],
        # 其他省份配置...
    }

    try:
        # 1. 创建采集器实例
        fetcher = AQIStudyDataFetcher()
        
        # 2. 按省份处理数据
        fetcher.process_by_province(CITIES_BY_PROVINCE)
        
    except Exception as e:
        logging.error(f"系统运行异常: {e}", exc_info=True)

技术总结

1. 反爬虫应对：通过集成JS加密/解密处理复杂的数据获取流程
2. 数据完整性：使用强类型数据类确保数据结构一致性
3. 健壮性设计：全面的异常处理和日志记录
4. 可扩展性：模块化设计方便功能扩展
5. 用户体验：清晰的进度反馈和结果统计

扩展建议

1. 添加定时任务功能实现定期数据采集
2. 集成数据可视化模块生成分析图表
3. 增加数据库存储支持长期数据保存
4. 实现异常自动重试机制
5. 添加邮件通知功能

这个系统为空气质量研究提供了可靠的数据采集方案，架构设计合理，可直接用于生产环境或作为二次开发的基础框架。

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

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('air_quality.log'),
        logging.StreamHandler()
    ]
)


@dataclass
class CityData:
    time_point: str
    aqi: int
    max_aqi: str
    min_aqi: str
    pm2_5: int
    pm10: int
    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)


class AQIStudyDataFetcher:
    """空气质量数据采集器"""

    # 类常量
    BASE_URL = 'https://www.aqistudy.cn/historydata/api/historyapi.php'
    COLUMNS = ['时间', 'aqi', 'max_aqi', 'min_aqi', 'pm2_5', 'pm10', 'co', 'no2', 'o3', 'so2', 'quality']

    def __init__(self, js_file: str = 'gethistory.js', excel_file: str = 'air_quality_data.xlsx'):
        """初始化采集器

        Args:
            js_file: JS解密文件路径
            excel_file: Excel输出文件路径
        """
        self.js_file = js_file
        self.excel_file = excel_file
        self.env = self._init_js_env()
        self.workbook = self._init_workbook()

        # 请求配置
        self.headers = {
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
            'Accept-Language': 'zh-CN,zh;q=0.9',
            'Connection': 'keep-alive',
            'Referer': 'https://www.aqistudy.cn/historydata/',
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Upgrade-Insecure-Requests': '1',
        }

        self.cookies = {
            'Hm_lpvt_6088e7f72f5a363447d4bafe03026db8': str(int(time.time())),
        }

    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

    def _init_workbook(self) -> Workbook:
        """初始化Excel工作簿"""
        try:
            workbook = load_workbook(self.excel_file)
            logging.info(f"加载现有工作簿: {self.excel_file}")
        except FileNotFoundError:
            workbook = Workbook()
            workbook.remove(workbook.active)  # 删除默认Sheet
            logging.info(f"创建新工作簿: {self.excel_file}")
        return workbook

    def fetch_city_data(self, city: str) -> Optional[List[Dict]]:
        """获取单个城市数据

        Args:
            city: 城市名称

        Returns:
            城市数据列表或None(如果获取失败)
        """
        try:
            # 生成查询参数
            query = self.env.call("gethistory", city)
            data = {'hA4Nse2cT': query}

            # 发送请求
            response = requests.post(
                self.BASE_URL,
                cookies=self.cookies,
                headers=self.headers,
                data=data,
                timeout=10
            )
            response.raise_for_status()

            # 解密数据
            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

    def save_to_excel(self, city: str, data: List[Dict]) -> bool:
        """保存城市数据到Excel

        Args:
            city: 城市名称
            data: 城市数据列表

        Returns:
            是否保存成功
        """
        try:
            # 获取或创建Sheet
            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}数据到Excel失败: {e}")
            return False

    def process_cities(self, cities: List[str]) -> None:
        """处理多个城市数据

        Args:
            cities: 城市名称列表
        """
        success_count = 0
        for city in cities:
            logging.info(f"开始处理城市: {city}")

            # 获取数据
            city_data = self.fetch_city_data(city)
            if not city_data:
                continue

            # 保存数据
            if self.save_to_excel(city, city_data):
                success_count += 1
                logging.info(f"城市{city}数据处理完成")
            else:
                logging.warning(f"城市{city}数据保存失败")

        # 保存工作簿
        try:
            self.workbook.save(self.excel_file)
            logging.info(f"成功处理{success_count}/{len(cities)}个城市数据，已保存到{self.excel_file}")
        except Exception as e:
            logging.error(f"保存工作簿失败: {e}")

    def process_by_province(self, cities_by_province: Dict[str, List[str]]) -> None:
        """按省份处理城市数据

        Args:
            cities_by_province: 按省份分类的城市字典
        """
        for province, cities in cities_by_province.items():
            logging.info(f"开始处理省份: {province}")
            self.process_cities(cities)
            logging.info(f"省份{province}处理完成")


if __name__ == '__main__':
    # 配置要采集的城市
    CITIES_BY_PROVINCE = {
        "陕西省": ["西安", "宝鸡", "咸阳", "铜川"],
        "宁夏回族自治区": ["银川", "固原"],
        "新疆维吾尔自治区": ["乌鲁木齐", "石河子", "吐鲁番", "哈密"],
        "甘肃省": ["兰州", "定西", "嘉峪关"],
        "青海省": ["西宁", "海东", "海北"]
    }

    # 创建采集器实例
    try:
        fetcher = AQIStudyDataFetcher()

        # 按省份处理
        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)