如何将BeautifulSoup爬取内容转DataFrame并实现全量增量采集

实现逻辑说明 #

• 你当前调用的表格接口直接返回逗号分隔的CSV格式纯文本，不需要用BeautifulSoup做HTML解析，可直接转换为Pandas DataFrame，首行会自动识别为你需要的MapDate,FIPS,County,State,Nothing,D0,D1,D2,D3,D4,ValidStart,ValidEnd列名
• 该平台干旱监测数据为每周二发布，全量采集不需要遍历所有自然日，仅需生成目标时段内所有周二的日期列表构造请求即可，大幅减少请求量
• 增量采集通过读取本地已存CSV的最新日期，仅爬取该日期之后的新发布数据，去重后合并存储即可

具体实现代码 #

依赖安装 #

先安装需要的第三方库：

pip install pandas requests python-dateutil

完整可运行代码 #

import os
import time
import requests
import pandas as pd
from io import StringIO
from datetime import datetime, timedelta

# 基础配置
BASE_URL = "https://droughtmonitor.unl.edu/DmData/GISData.aspx/?mode=table&aoi=county&date="
SAVE_PATH = "us_drought_county_data.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"
}

def get_tuesdays(start_date: datetime, end_date: datetime) -> list:
    """生成指定时间范围内所有周二的日期列表，格式为%Y-%m-%d"""
    tuesdays = []
    current = start_date
    # 找到区间内第一个周二
    while current.weekday() != 1:
        current += timedelta(days=1)
    while current <= end_date:
        tuesdays.append(current.strftime("%Y-%m-%d"))
        current += timedelta(days=7)
    return tuesdays

def fetch_single_date(date_str: str) -> pd.DataFrame:
    """爬取单个日期的数据，转换为DataFrame"""
    url = BASE_URL + date_str
    try:
        resp = requests.get(url, headers=HEADERS, timeout=10)
        resp.raise_for_status()
        # 直接读取CSV文本，自动识别首行列名
        df = pd.read_csv(StringIO(resp.text))
        # 校验返回数据格式是否符合预期
        required_cols = {"MapDate","FIPS","County","State","Nothing","D0","D1","D2","D3","D4","ValidStart","ValidEnd"}
        if not required_cols.issubset(set(df.columns)):
            print(f"日期{date_str}返回数据格式异常，跳过")
            return pd.DataFrame()
        return df
    except Exception as e:
        print(f"爬取日期{date_str}失败：{str(e)}，跳过")
        return pd.DataFrame()

def main():
    all_dfs = []
    # 判断本地是否已有存量数据，确定爬取起始点
    if os.path.exists(SAVE_PATH):
        print("检测到本地已有数据文件，启动增量采集模式")
        exist_df = pd.read_csv(SAVE_PATH, dtype={"FIPS": str})
        all_dfs.append(exist_df)
        # 找到已采集的最新数据截止日期
        latest_date = pd.to_datetime(exist_df["ValidEnd"]).max()
        start_date = latest_date + timedelta(days=1)
        end_date = datetime.today()
    else:
        print("未检测到本地数据文件，启动2000-2022年全量采集模式")
        start_date = datetime(2000, 1, 1)
        end_date = datetime(2022, 12, 31)
    
    # 生成待爬取的日期列表
    crawl_dates = get_tuesdays(start_date, end_date)
    print(f"本次需爬取{len(crawl_dates)}个日期的数据")
    
    for idx, date in enumerate(crawl_dates):
        print(f"正在爬取{date}，进度{idx+1}/{len(crawl_dates)}")
        single_df = fetch_single_date(date)
        if not single_df.empty:
            # FIPS字段存为5位字符串，避免前导0丢失
            single_df["FIPS"] = single_df["FIPS"].astype(str).str.zfill(5)
            all_dfs.append(single_df)
        # 加延时避免请求过于频繁被拦截
        time.sleep(0.8)
    
    # 合并所有数据，按日期+区县编码去重
    final_df = pd.concat(all_dfs, ignore_index=True).drop_duplicates(subset=["MapDate", "FIPS"], keep="last")
    # 保存为CSV文件，用utf-8-sig编码避免Excel打开乱码
    final_df.to_csv(SAVE_PATH, index=False, encoding="utf-8-sig")
    print(f"数据采集完成，共保存{len(final_df)}条记录到{SAVE_PATH}")

if __name__ == "__main__":
    main()

使用说明 #

• 首次运行代码会自动爬取2000年到2022年的所有县级干旱数据，保存为脚本同目录下的us_drought_county_data.csv文件
• 后续需要更新数据时，直接再次运行脚本即可，程序会自动识别本地已存数据的最新日期，爬取之后新发布的所有数据，自动合并去重，不需要手动修改参数
• 数值类干旱占比字段会自动识别为浮点型，可直接用于后续统计分析

内容的提问来源于stack exchange，提问作者Lalit Joshi