Pandas read_html 与 PyExecJS 双方案对比：解析 aqistudy.cn 加密数据的 2 种实战路径

Just do it

358人浏览 · 2026-07-05 15:44:24

Just do it · 2026-07-05 15:44:24 发布

Pandas read_html 与 PyExecJS 双方案对比：解析 aqistudy.cn 加密数据的 2 种实战路径

在数据采集领域，面对动态渲染和加密数据时，技术选型往往决定了项目的成败。本文将深入探讨两种截然不同的技术路线——基于 Pandas 的快速表格提取与基于 PyExecJS 的 JavaScript 解密执行，帮助开发者根据实际场景选择最优解。

1. 技术方案选型背景与目标

空气质量数据作为环境研究的重要指标，其采集过程常面临三个核心挑战：动态渲染的表格数据、加密的 Ajax 响应以及反爬机制的应对。传统 Selenium 方案虽然通用性强，但存在性能瓶颈和维护成本高的缺陷。

我们针对 aqistudy.cn 的实际情况，提炼出两个典型场景：

场景A ：可直接访问的未加密历史数据页面（如月度统计）
场景B ：需要解密处理的动态接口数据（如实时监测）

以下为两种技术路线的适用性对比：

特性	Pandas read_html	PyExecJS
开发效率	★★★★★	★★★☆☆
执行性能	★★★★☆	★★☆☆☆
反爬绕过能力	★★☆☆☆	★★★★★
加密数据处理	不支持	原生支持
浏览器依赖	可选	无需

2. Pandas read_html 极简实现方案

对于可直接访问的表格数据，Pandas 提供了开箱即用的解决方案。以下是通过历史数据页面获取北京2023年空气质量数据的完整示例：

import pandas as pd
from selenium import webdriver

def get_monthly_data(city, year_month):
    url = f"https://www.aqistudy.cn/historydata/daydata.php?city={city}&month={year_month}"
    driver = webdriver.Chrome()
    driver.get(url)
    
    # 关键步骤：提取页面中的第一个表格
    df = pd.read_html(driver.page_source, header=0)[0]
    driver.quit()
    
    # 数据清洗
    df['城市'] = city
    df['日期'] = pd.to_datetime(df['日期'])
    return df.dropna()

# 示例：获取北京2023年1月数据
beijing_jan = get_monthly_data('北京', '2023-01')
print(beijing_jan.head())

提示：当遇到动态加载延迟时，可添加 WebDriverWait 确保表格完全渲染：

from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

wait = WebDriverWait(driver, 10)
wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'table')))

该方案的三大优势：

代码精简 ：核心逻辑仅需3行代码
自动解析 ：智能识别表头和多级索引
数据就绪 ：直接返回DataFrame对象

但需要注意两个常见问题：

部分页面需要模拟点击展开表格
表格结构变化时需要调整索引位置

3. PyExecJS 解密核心方案

当面对加密接口时，我们需要深入分析前端加密逻辑。以 aqistudy.cn 的实时数据接口为例，其加密流程如下：

graph TD
    A[前端请求] --> B[参数加密]
    B --> C[发送加密请求]
    C --> D[接收加密响应]
    D --> E[数据解密]
    E --> F[渲染展示]

3.1 JavaScript 逆向分析

通过浏览器开发者工具，可定位到核心加密函数：

// 示例加密函数（已简化）
function getEncryptedParams(method, params) {
    const key = CryptoJS.enc.Utf8.parse('密钥字符串');
    const iv = CryptoJS.enc.Utf8.parse('偏移量');
    const encrypted = CryptoJS.AES.encrypt(
        JSON.stringify(params), 
        key, 
        { iv: iv, mode: CryptoJS.mode.CBC }
    );
    return encrypted.toString();
}

3.2 Python 实现方案

安装必要依赖：

pip install PyExecJS requests

完整解密代码实现：

import execjs
import requests

# 1. 加载JS解密文件
with open('aqi_crypto.js', 'r', encoding='utf-8') as f:
    js_code = f.read()

# 2. 创建JS运行时环境
ctx = execjs.compile(js_code)

def get_real_time_aqi(city):
    # 3. 调用JS函数加密参数
    encrypted_params = ctx.call('getEncryptedParams', 
                               'GETCITYWEATHER', 
                               {'city': city, 'type': 'HOUR'})
    
    # 4. 发送加密请求
    api_url = 'https://www.aqistudy.cn/apinew/aqistudyapi.php'
    response = requests.post(api_url, data={'d': encrypted_params})
    
    # 5. 解密响应数据
    decrypted_data = ctx.call('decodeData', response.text)
    return decrypted_data

# 示例：获取北京实时数据
data = get_real_time_aqi('北京')
print(data)

关键点说明：

aqi_crypto.js 需包含完整的加密/解密函数
注意处理时区差异导致的时间戳问题
建议添加异常重试机制应对网络波动

4. 混合方案实战应用

结合两种方案的优点，我们可以构建更健壮的数据采集系统：

class AQIDataFetcher:
    def __init__(self):
        self.js_runtime = self._init_js_runtime()
        self.session = requests.Session()
    
    def _init_js_runtime(self):
        with open('crypto.js') as f:
            return execjs.compile(f.read())
    
    def fetch_data(self, city, date_range):
        results = []
        for month in pd.date_range(*date_range, freq='MS'):
            try:
                # 优先尝试直接获取
                df = self._try_direct_fetch(city, month)
            except Exception:
                # 降级到加密接口
                df = self._fetch_via_api(city, month)
            results.append(df)
        return pd.concat(results)
    
    def _try_direct_fetch(self, city, month):
        url = f"https://www.aqistudy.cn/historydata/daydata.php?city={city}&month={month.strftime('%Y-%m')}"
        df = pd.read_html(url, header=0)[0]
        df['采集方式'] = 'direct'
        return df
    
    def _fetch_via_api(self, city, month):
        params = {
            'city': city,
            'start': month.strftime('%Y-%m-01'),
            'end': (month + pd.offsets.MonthEnd()).strftime('%Y-%m-%d')
        }
        encrypted = self.js_runtime.call('encrypt', params)
        response = self.session.post(API_URL, data={'d': encrypted})
        data = self.js_runtime.call('decrypt', response.text)
        df = pd.DataFrame(data['result']['rows'])
        df['采集方式'] = 'api'
        return df

该混合方案实现了：

自动降级机制保障数据完整性
统一的数据输出格式
可追溯的数据来源标记

5. 性能优化与异常处理

针对大规模采集需求，我们还需要关注以下关键点：

内存优化技巧 ：

# 使用迭代方式处理大数据
def batch_fetch(cities, chunk_size=10):
    for i in range(0, len(cities), chunk_size):
        chunk = cities[i:i + chunk_size]
        with ThreadPoolExecutor() as executor:
            yield from executor.map(fetch_city_data, chunk)

智能重试机制 ：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, min=4, max=10))
def fetch_with_retry(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response

日志监控方案 ：

import logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('aqi_monitor.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def fetch_data():
    try:
        # 数据采集逻辑
        logger.info(f"开始采集{city}数据")
    except Exception as e:
        logger.error(f"采集失败: {str(e)}", exc_info=True)

在实际项目中，建议根据具体需求选择合适的技术路线。对于定期更新的历史数据采集，Pandas 方案更加高效；而对于需要突破反爬限制的实时数据获取，PyExecJS 方案则展现出不可替代的优势。