IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

如何编写抓取表格数据的脚本

wen 实用脚本 2

本文目录导读：

如何编写抓取表格数据的脚本

  1. Python + BeautifulSoup（静态网页）
  2. Python + Pandas（快速方法）
  3. Selenium（动态网页）
  4. 处理分页表格
  5. 错误处理和重试机制
  6. 保存数据到Excel
  7. 完整示例：综合脚本
  8. 注意事项
  9. 常见问题解决

Python + BeautifulSoup（静态网页）

基础示例

import requests
from bs4 import BeautifulSoup
import pandas as pd
# 发送请求
url = 'https://example.com/table'
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(url, headers=headers)
response.encoding = 'utf-8'
# 解析HTML
soup = BeautifulSoup(response.text, 'html.parser')
# 找到表格
table = soup.find('table')  # 或 soup.find_all('table')
# 提取数据
data = []
for row in table.find_all('tr'):
    row_data = []
    for cell in row.find_all(['td', 'th']):
        row_data.append(cell.text.strip())
    if row_data:
        data.append(row_data)
# 转换为DataFrame
df = pd.DataFrame(data)

更高级的表格处理

def extract_table_data(soup, table_index=0):
    """提取表格数据的通用函数"""
    tables = soup.find_all('table')
    if table_index >= len(tables):
        return None
    table = tables[table_index]
    rows = table.find_all('tr')
    headers = []
    data = []
    # 提取表头
    header_row = rows[0]
    headers = [th.text.strip() for th in header_row.find_all(['th', 'td'])]
    # 提取数据行
    for row in rows[1:]:
        row_data = []
        for cell in row.find_all(['td', 'th']):
            # 处理 colspan 和 rowspan
            rowspan = int(cell.get('rowspan', 1))
            colspan = int(cell.get('colspan', 1))
            cell_text = cell.text.strip()
            # 复制单元格以处理跨行跨列
            for i in range(colspan):
                row_data.append(cell_text)
        if row_data:
            data.append(row_data)
    # 处理不完整的行（补齐空值）
    max_length = max(len(row) for row in data) if data else 0
    for row in data:
        while len(row) < max_length:
            row.append('')
    return headers, data

Python + Pandas（快速方法）

最简单的表格提取

import pandas as pd
# 方法1：直接从网页读取所有表格
tables = pd.read_html('https://example.com/page')
df = tables[0]  # 第一个表格
# 方法2：指定表格
df = pd.read_html(url, match='表格名称')[0]
# 方法3：带header参数
df = pd.read_html(url, header=0)[0]
# 方法4：处理多个表格
for i, table in enumerate(tables):
    print(f"表格{i}: {table.shape}")
    table.to_csv(f'table_{i}.csv', index=False)

Selenium（动态网页）

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import pandas as pd
import time
def scrape_dynamic_table():
    # 初始化驱动
    driver = webdriver.Chrome()
    driver.get('https://example.com/dynamic-table')
    # 等待表格加载（等待指定元素出现）
    wait = WebDriverWait(driver, 10)
    table = wait.until(EC.presence_of_element_located((By.TAG_NAME, 'table')))
    # 滚动加载更多数据的场景
    last_height = driver.execute_script("return document.body.scrollHeight")
    while True:
        driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        time.sleep(2)
        new_height = driver.execute_script("return document.body.scrollHeight")
        if new_height == last_height:
            break
        last_height = new_height
    # 提取表格数据
    rows = driver.find_elements(By.CSS_SELECTOR, 'table tbody tr')
    data = []
    for row in rows:
        cols = row.find_elements(By.TAG_NAME, 'td')
        row_data = [col.text.strip() for col in cols]
        data.append(row_data)
    # 提取表头
    headers = [th.text.strip() for th in driver.find_elements(By.CSS_SELECTOR, 'table thead th')]
    df = pd.DataFrame(data, columns=headers)
    driver.quit()
    return df

处理分页表格

def scrape_paginated_table(base_url, total_pages=10):
    all_data = []
    for page in range(1, total_pages + 1):
        # 构建分页URL
        url = f"{base_url}?page={page}"
        # 使用requests或selenium获取页面
        response = requests.get(url)
        soup = BeautifulSoup(response.text, 'html.parser')
        # 提取当前页数据
        table_data = extract_table_data(soup)
        if table_data:
            headers, data = table_data
            all_data.extend(data)
        # 暂停避免过快请求
        time.sleep(1)
    return pd.DataFrame(all_data, columns=headers)

错误处理和重试机制

import time
from functools import wraps
def retry(max_attempts=3, delay=1):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise e
                    time.sleep(delay * (attempt + 1))
            return None
        return wrapper
    return decorator
@retry(max_attempts=3, delay=2)
def scrape_data_safely(url):
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.5',
        'Connection': 'keep-alive',
    }
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()
    return response.text

保存数据到Excel

def save_to_excel(df, filename, sheet_name='Sheet1'):
    """保存数据到Excel"""
    with pd.ExcelWriter(filename, engine='openpyxl', mode='w') as writer:
        df.to_excel(writer, sheet_name=sheet_name, index=False)
        # 调整列宽
        worksheet = writer.sheets[sheet_name]
        for column in df:
            column_width = max(df[column].astype(str).map(len).max(), len(column))
            col_idx = df.columns.get_loc(column)
            worksheet.column_dimensions[chr(65 + col_idx)].width = column_width + 2
    print(f"数据已保存到 {filename}")
# 使用示例
save_to_excel(df, 'output.xlsx', '数据表')

完整示例：综合脚本

import requests
from bs4 import BeautifulSoup
import pandas as pd
import time
import os
class TableScraper:
    def __init__(self, base_url, headers=None):
        self.base_url = base_url
        self.headers = headers or {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'
        }
        self.data = []
        self.headers_list = []
    def fetch_page(self, url):
        """获取页面内容"""
        try:
            response = requests.get(url, headers=self.headers, timeout=10)
            response.raise_for_status()
            response.encoding = 'utf-8'
            return response.text
        except Exception as e:
            print(f"请求失败: {e}")
            return None
    def extract_tables(self, html):
        """从HTML中提取表格"""
        soup = BeautifulSoup(html, 'html.parser')
        tables = soup.find_all('table')
        results = []
        for table in tables:
            # 提取表格标题（如果有）
            caption = table.find('caption')
            caption_text = caption.text.strip() if caption else "无标题"
            # 提取表头和数据
            headers, data = self.parse_table(table)
            results.append({
                'caption': caption_text,
                'headers': headers,
                'data': data
            })
        return results
    def parse_table(self, table):
        """解析单个表格"""
        rows = table.find_all('tr')
        headers = []
        data = []
        # 处理表头
        header_row = rows[0] if rows else None
        if header_row:
            headers = [th.text.strip() for th in header_row.find_all(['th', 'td'])]
        # 处理数据行
        for row in rows[1:]:
            row_data = []
            for cell in row.find_all(['td', 'th']):
                row_data.append(cell.text.strip())
            # 验证行长度
            if len(row_data) == len(headers):
                data.append(row_data)
            elif len(row_data) > len(headers):
                data.append(row_data[:len(headers)])
            elif len(row_data) < len(headers):
                # 补齐缺失数据
                row_data.extend([''] * (len(headers) - len(row_data)))
                data.append(row_data)
        return headers, data
    def save_to_file(self, format_type='csv', filename='output'):
        """保存数据"""
        if not self.data:
            print("没有数据可保存")
            return
        if format_type == 'csv':
            self.save_csv(filename)
        elif format_type == 'excel':
            self.save_excel(filename)
        elif format_type == 'json':
            self.save_json(filename)
    def save_csv(self, filename):
        """保存为CSV"""
        os.makedirs(filename, exist_ok=True)
        for i, (headers, data) in enumerate(zip(self.headers_list, self.data)):
            df = pd.DataFrame(data, columns=headers)
            df.to_csv(f"{filename}/table_{i+1}.csv", index=False, encoding='utf-8-sig')
    def save_excel(self, filename):
        """保存为Excel"""
        with pd.ExcelWriter(f'{filename}.xlsx', engine='openpyxl') as writer:
            for i, (headers, data) in enumerate(zip(self.headers_list, self.data)):
                df = pd.DataFrame(data, columns=headers)
                df.to_excel(writer, sheet_name=f'Table_{i+1}', index=False)
    def save_json(self, filename):
        """保存为JSON"""
        import json
        output = {'tables': []}
        for i, (headers, data) in enumerate(zip(self.headers_list, self.data)):
            output['tables'].append({
                'index': i,
                'headers': headers,
                'data': data
            })
        with open(f'{filename}.json', 'w', encoding='utf-8') as f:
            json.dump(output, f, ensure_ascii=False, indent=2)
    def run(self, url=None):
        """主执行函数"""
        target_url = url or self.base_url
        html = self.fetch_page(target_url)
        if html:
            tables = self.extract_tables(html)
            if tables:
                self.headers_list = [t['headers'] for t in tables]
                self.data = [t['data'] for t in tables]
                for i, table_info in enumerate(tables):
                    print(f"表格 {i+1}: {table_info['caption']}")
                    print(f"  列数: {len(table_info['headers'])}")
                    print(f"  行数: {len(table_info['data'])}")
                return tables
            else:
                print("在页面中未找到表格")
        return []
# 使用示例
if __name__ == "__main__":
    scraper = TableScraper('https://example.com/page-with-tables')
    tables = scraper.run()
    if tables:
        # 保存所有格式
        scraper.save_to_file('csv', 'scraped_data')
        scraper.save_to_file('excel', 'scraped_data')
        scraper.save_to_file('json', 'scraped_data')

注意事项

  1. 遵守robots.txt：抓取前检查网站是否允许爬取
  2. 设置请求头：模拟真实浏览器访问
  3. 控制频率：添加延迟避免对服务器造成压力
  4. 异常处理：处理超时、连接错误等异常情况
  5. 数据验证：检查和验证提取的数据
  6. 编码问题：正确处理中文字符编码（如utf-8）

常见问题解决

  • 动态加载数据：使用Selenium或Playwright
  • 反爬机制：使用代理IP、添加延迟、模拟真人操作
  • 复杂的表格结构：使用find_all + 条件判断
  • 多种表格格式：设计通用函数或使用模式匹配

选择哪种方法取决于你的具体需求：静态页面用BeautifulSoup，需要JavaScript渲染的用Selenium,想要快速实现且页面简单的用Pandas的read_html。

上一篇怎么用脚本获取硬盘序列号

下一篇怎么用脚本延时启动程序

相关文章

抱歉，评论功能暂时关闭!