You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何优化ASPX动态弹窗网站的网页爬虫效率?

问题:优化ASPX网站表格爬虫的方法

我需要抓取一个ASPX网站(地址:https://www.osfi-bsif.gc.ca/en/data-forms/financial-data/financial-data-banks)的表格数据:该网站包含表单,填写表单后会弹出动态生成的窗口,窗口内有需抓取的HTML表格。弹窗的URL是带UUID的临时路径(类似www.xyz.com/something/something/"Temp"FinancialData.aspx),会随表单中月份切换而变化,通过页面内window.open脚本触发。

我已编写Python+Selenium代码完成数据提取,但运行速度极慢,且未找到网站的隐藏API,希望能优化现有代码或获取其他可行的实现方案。现有代码如下:

def extract_table_data(driver, date_label, bank_code):
  
    tables = driver.find_elements(By.CSS_SELECTOR, 'table.w100.borderspace-0.table.table-lined')
    #checking to see if two table exist
    if len(tables) < 2:
        print("Not enough tables found on the page.")
        return pd.DataFrame()
    
    #looking at second table
    target_table = tables[1]
    
   
    descriptions = ['(a) Federal and Provincial', '(b) Municipal or School Corporations',
                    '(c) Deposit-taking institutions', '(i) Tax sheltered', '(ii) Other', '(e) Other'] 
    data = []
    description_index = 0
    capture = False

    # Start iterating over rows in the target table
    for row in target_table.find_elements(By.TAG_NAME, 'tr'):
        row_text = row.text.strip()  # Get row text once to minimize repeated calls

        # Capturing when we find start of table
        if "1. Demand and notice deposits" in row_text:
            capture = True  # Enable data capturing
        
        if capture:
            cols = row.find_elements(By.TAG_NAME, 'td')
            if len(cols) > 1 and description_index < len(descriptions):
                # Append data for 'Total' column only (assuming it's in the second cell)
                data.append((date_label, descriptions[description_index], bank_code, cols[1].text.strip()))
                description_index += 1

            if description_index >= len(descriptions) or "2. Fixed-term deposits" in row_text:
                break  

    return pd.DataFrame(data, columns=['Date', 'Category', 'Bank Code', 'Total'])

# Main scraping function
def scrape_data(driver, bank_code, date_option):
    try:
        # Load website
        driver.get("https://ws1ext.osfi-bsif.gc.ca/WebApps/FINDAT/DTIBanks.aspx?T=0&LANG=E")
        WebDriverWait(driver, 5).until(EC.presence_of_element_located((By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_institutionTypeCriteria_type1RadioButton"))).click()

        # Select bank and date
        Select(driver.find_element(By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_institutionTypeCriteria_institutionsDropDownList")).select_by_value(bank_code)
        WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_dtiReportCriteria_monthlyRadioButton"))).click()
        date_dropdown = Select(WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_dtiReportCriteria_monthlyDatesDropDownList"))))
        date_dropdown.select_by_value(date_option)

        # Extract the formatted date for labeling
        date_label = date_dropdown.first_selected_option.text.split()
        date_label = f"{date_label[0][:3]}-{date_label[-1]}"

        # Submit form
        driver.find_element(By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_submitButton").click()
        time.sleep(2)

        # Switch to new window for data extraction
        main_window = driver.current_window_handle
        for handle in driver.window_handles:
            if handle != main_window:
                driver.switch_to.window(handle)
                break

        # Wait for tables to load and capture data
        WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.TAG_NAME, "table")))
        return extract_table_data(driver, date_label, bank_code)

    except Exception as e:
        print(f"Error processing bank {bank_code} and date {date_option}: {e}")
        return None

    finally:
        driver.close()
        driver.switch_to.window(main_window)

def main():
    date_options = generate_date_options(2024, 8, 2024, 8)
    bank_codes = ["Z005"]

    final_data, final_summary = [], []
    driver = initialize_driver()

    for bank_code in bank_codes:
        for date_option in date_options:
            print(f"Processing bank: {bank_code}, date: {date_option}")
            df = scrape_data(driver, bank_code, date_option)
            if not df.empty:
                # Accumulate data without immediately processing each DataFrame
                final_data.append(df)

    # Code for generating two tables, one detailed and one with only total
    if final_data:
        final_df = pd.concat(final_data, ignore_index=True)
        final_df['Total'] = pd.to_numeric(final_df['Total'].str.replace(',', ''), errors='coerce')  

        summary_df = final_df.groupby(['Date', 'Bank Code', 'Category'])['Total'].sum().reset_index()
        summary_df.to_csv('demand_and_notice_deposits_summary.csv', index=False)
        final_df.to_csv('demand_and_notice_deposits_detailed.csv', index=False)
        print("Scraping complete!")
    else:
        print("No data was found for any of the banks or dates.")

    driver.quit()
    return final_data, final_summary

final_data, final_summary = main()
优化方案

一、Selenium代码本身的性能优化

1. 减少页面重复加载

当前scrape_data函数每次都重新打开主页面,完全可以把主页面的初始化(点击单选框、加载下拉选项)放到main函数里执行一次,之后只需要切换银行和日期,不用重复加载页面:

def main():
    date_options = generate_date_options(2024, 8, 2024, 8)
    bank_codes = ["Z005"]

    final_data, final_summary = [], []
    driver = initialize_driver()
    # 初始化主页面,只执行一次
    driver.get("https://ws1ext.osfi-bsif.gc.ca/WebApps/FINDAT/DTIBanks.aspx?T=0&LANG=E")
    WebDriverWait(driver, 5).until(EC.presence_of_element_located((By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_institutionTypeCriteria_type1RadioButton"))).click()
    WebDriverWait(driver, 2).until(EC.presence_of_element_located((By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_dtiReportCriteria_monthlyRadioButton"))).click()
    # 缓存下拉元素
    bank_dropdown = Select(driver.find_element(By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_institutionTypeCriteria_institutionsDropDownList"))
    date_dropdown = Select(driver.find_element(By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_dtiReportCriteria_monthlyDatesDropDownList"))

    for bank_code in bank_codes:
        bank_dropdown.select_by_value(bank_code)
        for date_option in date_options:
            print(f"Processing bank: {bank_code}, date: {date_option}")
            date_dropdown.select_by_value(date_option)
            # 提取日期标签
            date_label = date_dropdown.first_selected_option.text.split()
            date_label = f"{date_label[0][:3]}-{date_label[-1]}"
            # 提交表单
            driver.find_element(By.ID, "DTIWebPartManager_gwpDTIBankControl1_DTIBankControl1_submitButton").click()
            # 等待新窗口出现
            WebDriverWait(driver, 5).until(lambda d: len(d.window_handles) > 1)
            # 切换窗口并提取数据
            main_window = driver.current_window_handle
            for handle in driver.window_handles:
                if handle != main_window:
                    driver.switch_to.window(handle)
                    break
            WebDriverWait(driver, 5).until(EC.presence_of_element_located((By.CSS_SELECTOR, 'table.w100.borderspace-0.table.table-lined')))
            df = extract_table_data(driver, date_label, bank_code)
            if not df.empty:
                final_data.append(df)
            # 关闭弹窗并切回主窗口
            driver.close()
            driver.switch_to.window(main_window)

    # 后续数据保存逻辑不变
    if final_data:
        final_df = pd.concat(final_data, ignore_index=True)
        final_df['Total'] = pd.to_numeric(final_df['Total'].str.replace(',', ''), errors='coerce')  

        summary_df = final_df.groupby(['Date', 'Bank Code', 'Category'])['Total'].sum().reset_index()
        summary_df.to_csv('demand_and_notice_deposits_summary.csv', index=False)
        final_df.to_csv('demand_and_notice_deposits_detailed.csv', index=False)
        print("Scraping complete!")
    else:
        print("No data was found for any of the banks or dates.")

    driver.quit()
    return final_data, final_summary

2. 替换time.sleep为显式等待

time.sleep(2)是固定等待,会浪费时间,改成等待新窗口出现或目标元素加载完成,避免不必要的等待。

3. 优化表格提取逻辑

直接用pd.read_html读取页面中的表格,比手动遍历tr和td快得多:

def extract_table_data(driver, date_label, bank_code):
    # 直接用pandas读取页面所有表格
    tables = pd.read_html(driver.page_source)
    if len(tables) < 2:
        print("Not enough tables found on the page.")
        return pd.DataFrame()
    target_table = tables[1]
    # 定位目标行范围
    start_mask = target_table.apply(lambda row: "1. Demand and notice deposits" in str(row), axis=1)
    end_mask = target_table.apply(lambda row: "2. Fixed-term deposits" in str(row), axis=1)
    if not start_mask.any() or not end_mask.any():
        return pd.DataFrame()
    start_idx = start_mask[start_mask].index[0]
    end_idx = end_mask[end_mask].index[0]
    # 提取Total列数据
    target_values = target_table.iloc[start_idx+1:end_idx, 1].dropna().astype(str).str.strip().tolist()
    descriptions = ['(a) Federal and Provincial', '(b) Municipal or School Corporations',
                    '(c) Deposit-taking institutions', '(i) Tax sheltered', '(ii) Other', '(e) Other']
    # 匹配数据长度,避免索引越界
    valid_length = min(len(target_values), len(descriptions))
    data = list(zip([date_label]*valid_length, descriptions[:valid_length], [bank_code]*valid_length, target_values[:valid_length]))
    return pd.DataFrame(data, columns=['Date', 'Category', 'Bank Code', 'Total'])

4. 使用无头浏览器模式

初始化driver时启用无头模式,减少浏览器渲染开销:

def initialize_driver():
    options = webdriver.ChromeOptions()
    options.add_argument('--headless=new')
    options.add_argument('--disable-gpu')
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-images')  # 禁用图片加载进一步提速
    return webdriver.Chrome(options=options)

二、绕过Selenium,直接请求临时页面(大幅提速)

Selenium慢的核心是模拟浏览器,直接用requests库构造请求获取临时页面的HTML,效率会提升数倍:

1. 抓包获取关键参数

打开浏览器开发者工具(F12),填写表单提交后:

  • 记录主页面的Cookie、__VIEWSTATE、__VIEWSTATEGENERATOR、__EVENTVALIDATION等ASPX表单必备参数;
  • 解析提交表单后返回的响应,提取window.open中的临时页面URL。

2. 构造请求示例

import requests
from bs4 import BeautifulSoup
import pandas as pd

session = requests.Session()
main_url = "https://ws1ext.osfi-bsif.gc.ca/WebApps/FINDAT/DTIBanks.aspx?T=0&LANG=E"

# 第一步:获取表单初始化参数
response = session.get(main_url)
soup = BeautifulSoup(response.text, 'html.parser')
form_params = {
    '__VIEWSTATE': soup.find('input', {'id': '__VIEWSTATE'})['value'],
    '__VIEWSTATEGENERATOR': soup.find('input', {'id': '__VIEWSTATEGENERATOR'})['value'],
    '__EVENTVALIDATION': soup.find('input', {'id': '__EVENTVALIDATION'})['value'],
    'DTIWebPartManager$gwpDTIBankControl1$DTIBankControl1$institutionTypeCriteria$type1RadioButton': 'on',
    'DTIWebPartManager$gwpDTIBankControl1$DTIBankControl1$dtiReportCriteria$monthlyRadioButton': 'on',
    'DTIWebPartManager$gwpDTIBankControl1$DTIBankControl1$submitButton': 'Submit'
}

# 第二步:遍历银行和日期,批量请求
bank_codes = ["Z005"]
date_options = generate_date_options(2024, 8, 2024, 8)
final_data = []

for bank_code in bank_codes:
    for date_option in date_options:
        print(f"Processing bank: {bank_code}, date: {date_option}")
        # 更新表单参数中的银行和日期
        form_params['DTIWebPartManager$gwpDTIBankControl1$DTIBankControl1$institutionTypeCriteria$institutionsDropDownList'] = bank_code
        form_params['DTIWebPartManager$gwpDTIBankControl1$DTIBankControl1$dtiReportCriteria$monthlyDatesDropDownList'] = date_option
        
        # 提交表单
        submit_response = session.post(main_url, data=form_params)
        # 提取临时页面URL
        submit_soup = BeautifulSoup(submit_response.text, 'html.parser')
        script_text = submit_soup.find('script', text=lambda t: 'window.open' in str(t)).text
        popup_url = script_text.split("'")[1]
        
        # 请求临时页面并提取表格
        popup_response = session.get(popup_url)
        tables = pd.read_html(popup_response.text)
        if len(tables) < 2:
            print("Not enough tables found on the page.")
            continue
        # 处理日期标签
        date_label = date_option[:4] + "-" + date_option[4:]  # 假设date_option是YYYYMM格式
        # 提取数据(复用之前的extract_table_data逻辑,改造成处理HTML文本)
        target_table = tables[1]
        start_mask = target_table.apply(lambda row: "1. Demand and notice deposits" in str(row), axis=1)
        end_mask = target_table.apply(lambda row: "2. Fixed-term deposits" in str(row), axis=1)
        if not start_mask.any() or not end_mask.any():
            continue
        start_idx = start_mask[start_mask].index[0]
        end_idx = end_mask[end_mask].index[0]
        target_values = target_table.iloc[start_idx+1:end_idx, 1].dropna().astype(str).str.strip().tolist()
        descriptions = ['(a) Federal and Provincial', '(b) Municipal or School Corporations',
                        '(c) Deposit-taking institutions', '(i) Tax sheltered', '(ii) Other', '(e) Other']
        valid_length = min(len(target_values), len(descriptions))
        data = list(zip([date_label]*valid_length, descriptions[:valid_length], [bank_code]*valid_length, target_values[:valid_length]))
        final_data.append(pd.DataFrame(data, columns=['Date', 'Category', 'Bank Code', 'Total']))

# 保存数据
if final_data:
    final_df = pd.concat(final_data, ignore_index=True)
    final_df['Total'] = pd.to_numeric(final_df['Total'].str.replace(',', ''), errors='coerce')  
    summary_df = final_df.groupby(['Date', 'Bank Code', 'Category'])['Total'].sum().reset_index()
    summary_df.to_csv('demand_and_notice_deposits_summary.csv', index=False)
    final_df.to_csv('demand_and_notice_deposits_detailed.csv', index=False)
    print("Scraping complete!")
else:
    print("No data was found for any of the banks or dates.
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.16 23:09:44