Python实现Web scraping自动合并网页CSV为单个DataFrame求助

依赖安装 #

首先确保安装所需依赖：

pip install requests beautifulsoup4 pandas

完整实现代码 #

import requests
from bs4 import BeautifulSoup
import pandas as pd
from io import StringIO
from urllib.parse import urljoin

def merge_csv_from_web(page_url: str) -> pd.DataFrame:
    # 1. 请求目标页面提取所有.csv后缀的文件链接
    headers = {"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"}
    page_resp = requests.get(page_url, headers=headers)
    page_resp.raise_for_status()
    
    soup = BeautifulSoup(page_resp.text, "html.parser")
    csv_links = []
    for a_tag in soup.find_all("a", href=True):
        href = a_tag["href"]
        if href.lower().endswith(".csv"):
            # 处理相对路径链接
            if not href.startswith(("http://", "https://")):
                href = urljoin(page_url, href)
            csv_links.append(href)
    
    if not csv_links:
        raise ValueError("当前页面未匹配到任何.csv格式的文件")
    
    # 2. 批量读取csv并按列名分组
    df_groups = {}
    for link in csv_links:
        try:
            csv_resp = requests.get(link, headers=headers)
            csv_resp.raise_for_status()
            current_df = pd.read_csv(StringIO(csv_resp.text))
            # 以列名元组作为分组标识，只有列名完全一致才会分到同一组
            group_key = tuple(current_df.columns)
            if group_key not in df_groups:
                df_groups[group_key] = []
            df_groups[group_key].append(current_df)
        except Exception as e:
            print(f"文件{link}读取失败，已跳过，错误信息：{str(e)}")
    
    # 3. 合并同组DataFrame，默认返回行数最多的合并结果，可按需修改返回所有组
    merged_result = [pd.concat(group, ignore_index=True) for group in df_groups.values()]
    return max(merged_result, key=lambda x: len(x))

# 调用示例
if __name__ == "__main__":
    target_page_url = "替换为你的目标页面地址"
    final_df = merge_csv_from_web(target_page_url)
    # 导出到本地验证结果
    final_df.to_csv("merged_output.csv", index=False, encoding="utf-8-sig")

代码说明 #

• 自动处理相对路径的csv链接，无需手动拼接域名
• 加入异常捕获逻辑，单个文件读取失败不会中断整个流程
• 默认严格按照列名+列顺序分组，只有列名完全一致且顺序相同的文件才会合并，如果不需要校验列顺序，可将group_key = tuple(current_df.columns)替换为group_key = tuple(sorted(current_df.columns))
• 所有文件直接在内存中处理，无需下载到本地，运行效率更高

内容的提问来源于stack exchange，提问作者Will Pratt