如何使用Python requests库批量下载USDA FIA站点的xlsm格式数据文件
批量下载USDA FIA Excel文件可行方案
方案1:基于州缩写列表循环下载
如果你确认站点所有文件严格遵循「州缩写.xlsm」的命名规则,可以直接预定义美国50个州+关联领地的缩写列表,通过循环批量下载,代码如下:
import requests from time import sleep # 美国州及常用领地缩写列表 state_codes = [ 'AL', 'AK', 'AZ', 'AR', 'CA', 'CO', 'CT', 'DE', 'FL', 'GA', 'HI', 'ID', 'IL', 'IN', 'IA', 'KS', 'KY', 'LA', 'ME', 'MD', 'MA', 'MI', 'MN', 'MS', 'MO', 'MT', 'NE', 'NV', 'NH', 'NJ', 'NM', 'NY', 'NC', 'ND', 'OH', 'OK', 'OR', 'PA', 'RI', 'SC', 'SD', 'TN', 'TX', 'UT', 'VT', 'VA', 'WA', 'WV', 'WI', 'WY', 'PR', 'VI' # 领地可根据实际需求增删 ] base_url = 'https://apps.fs.usda.gov/fia/datamart/Workbooks/{}.xlsm' headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'} for code in state_codes: file_url = base_url.format(code) file_name = f"{code}.xlsm" try: # 加超时避免请求卡住,加UA避免被反爬拦截 resp = requests.get(file_url, headers=headers, timeout=30) resp.raise_for_status() # 状态码非200直接抛出异常 with open(file_name, 'wb') as f: f.write(resp.content) print(f"{file_name} 下载完成") sleep(1) # 加延时避免请求频率过高被站点拦截 except Exception as e: print(f"{file_name} 下载失败,错误信息:{str(e)}")
方案2:爬取目标页面提取链接后下载(更稳妥)
如果不确定缩写列表是否完整,或者站点有额外的文件,直接从你提供的目标页面提取所有符合要求的xlsm下载链接,避免漏下/错下,需要先安装依赖:pip install requests beautifulsoup4,代码如下:
import requests from bs4 import BeautifulSoup from time import sleep from urllib.parse import urljoin # 目标页面地址 list_page_url = 'https://apps.fs.usda.gov/fia/datamart/datamart_excel.html' headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36'} try: # 先请求列表页获取所有下载链接 resp = requests.get(list_page_url, headers=headers, timeout=30) resp.raise_for_status() soup = BeautifulSoup(resp.text, 'html.parser') # 筛选所有后缀为xlsm的链接 xlsm_links = [] for a_tag in soup.find_all('a', href=True): href = a_tag['href'] if href.endswith('.xlsm'): # 拼接完整下载链接 full_url = urljoin(list_page_url, href) xlsm_links.append(full_url) print(f"共找到{len(xlsm_links)}个待下载文件") # 循环下载 for file_url in xlsm_links: file_name = file_url.split('/')[-1] try: resp = requests.get(file_url, headers=headers, timeout=30) resp.raise_for_status() with open(file_name, 'wb') as f: f.write(resp.content) print(f"{file_name} 下载完成") sleep(1) except Exception as e: print(f"{file_name} 下载失败,错误信息:{str(e)}") except Exception as e: print(f"列表页解析失败,错误信息:{str(e)}")
可选优化点
- 如果文件体积较大,可以改用
requests.get(..., stream=True)流式下载,避免内存占用过高 - 可以增加文件存在校验,已下载完成的文件直接跳过,支持断点续下
- 可以引入多线程/协程提升下载速度,注意控制并发量避免触发站点反爬策略
内容的提问来源于stack exchange,提问作者dswebber7373
相关产品推荐
相关产品推荐

