如何批量请求多个 URL 并合并结果为 CSV 文件

13次阅读

如何批量请求多个 URL 并合并结果为 CSV 文件

本文介绍如何将单 url 的 python 请求脚本扩展为批量处理 url 列表,自动采集每个响应中的结构化数据,逐次追加至 pandas dataframe,并最终导出为 csv 文件。

在实际数据采集或地址验证等场景中,往往需要对一批 URL(例如不同邮政编码的查询接口)发起请求,而非仅执行一次。原始脚本仅支持单 URL 处理,缺乏可扩展性。下面我们将重构代码,实现循环请求 + 增量构建 DataFrame + 一键导出 csv 的完整工作流。

✅ 核心改进点

  • 将硬编码 url 替换为 urls 列表,支持任意数量目标;
  • 使用 for url in urls: 显式遍历,避免重复逻辑;
  • 每次响应解析后生成临时 DataFrame,并通过 pd.concat()(推荐)或 .append()(已弃用,但兼容旧版)合并到主 DataFrame;
  • 最终调用 .to_csv() 保存结果,禁用索引以保证 CSV 清洁可读。

? 完整可运行代码(优化版)

import requests import pandas as pd  # ✅ 定义待请求的 URL 列表(可扩展至数百/数千条) urls = [     'https://ws1.postescanada-canadapost.ca/AddressComplete/Interactive/Find/v2.10/json3ex.ws?Key=TF36-KU97-AB94-ZC85&Country=CAN&SearchTerm=L2A%200A2&LanguagePreference=en&LastId=CA%7CCP%7CENG%7C0A2-L2A&SearchFor=Everything&OrderBy=UserLocation&$block=true&$cache=true',     'https://ws1.postescanada-canadapost.ca/AddressComplete/Interactive/Find/v2.10/json3ex.ws?Key=TF36-KU97-AB94-ZC85&Country=CAN&SearchTerm=M5V%203L9&LanguagePreference=en&LastId=CA%7CCP%7CENG%7C3L9-M5V&SearchFor=Everything&OrderBy=UserLocation&$block=true&$cache=true',     # ? 可继续添加更多 URL ]  # ✅ 统一请求头(保持与原脚本一致,确保服务端识别合法来源) headers = {     'authority': 'ws1.postescanada-canadapost.ca',     'accept': '*/*',     'accept-language': 'en-US,en;q=0.9',     'cache-control': 'no-cache',     'origin': 'https://www.canadapost-postescanada.ca',     'pragma': 'no-cache',     'referer': 'https://www.canadapost-postescanada.ca/ac/',     'sec-ch-ua': '"Not A(Brand";v="99", "google Chrome";v="121", "Chromium";v="121"',     'sec-ch-ua-mobile': '?0',     'sec-ch-ua-platform': '"windows"',     'sec-fetch-dest': 'empty',     'sec-fetch-mode': 'cors',     'sec-fetch-site': 'cross-site',     'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) appleWebKit/537.36 (Khtml, like Gecko) Chrome/121.0.0.0 safari/537.36', }  # ✅ 初始化空 DataFrame(列名需提前定义,确保类型一致) final_df = pd.DataFrame(columns=['results_subset_alpha', 'results_subset_beta'])  # ✅ 主循环:逐个请求、解析、追加 for i, url in enumerate(urls, 1):     print(f"[{i}/{len(urls)}] Requesting: {url[:60]}...")     try:         response = requests.get(url, headers=headers, timeout=10)         response.raise_for_status()  # 抛出 HTTP 错误(如 404/500)         data = response.json()         items = data.get('Items', [])          # ✅ 使用列表推导式高效提取字段(更简洁、更 pythonic)         texts = [item['Text'] for item in items]         descriptions = [item['Description'] for item in items]          # ✅ 构建当前批次 DataFrame 并合并(推荐使用 concat 替代已弃用的 append)         batch_df = pd.DataFrame({             'results_subset_alpha': texts,             'results_subset_beta': descriptions         })         final_df = pd.concat([final_df, batch_df], ignore_index=True)      except requests.exceptions.RequestException as e:         print(f"❌ Request failed for {url}: {e}")         continue     except (KeyError, ValueError, TypeError) as e:         print(f"❌ JSON parsing error for {url}: {e}")         continue  # ✅ 导出结果(安全路径 + 中文兼容编码可选) output_file = 'address_results.csv' final_df.to_csv(output_file, index=False, encoding='utf-8-sig') print(f"✅ Successfully saved {len(final_df)} rows to '{output_file}'")

⚠️ 注意事项与最佳实践

  • 错误处理必加:网络请求易失败(超时、连接中断、返回非 JSON),务必用 try/except 包裹核心逻辑;
  • 避免 .append():Pandas 2.0+ 已弃用该方法;应统一使用 pd.concat([df1, df2], ignore_index=True);
  • 响应校验不可少:调用 response.raise_for_status() 可捕获 HTTP 异常;检查 data.get(‘Items’) 防止 KeyError;
  • 性能提示:若 URL 数量极大(>100),建议加入 time.sleep(0.1) 或使用 concurrent.futures.ThreadPoolExecutor 实现并发(注意 API 限流);
  • CSV 编码:encoding=’utf-8-sig’ 可确保 excel 正确显示中文/特殊字符。

通过以上改造,你的脚本即可稳健、可维护地完成多 URL 批量采集任务——从单点调试迈向生产就绪。

text=ZqhQzanResources