如何优化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.
相关产品推荐
相关产品推荐

