如何实现CSV每行对应URL文件中X个连续URL进行POST请求

核心思路 #

先一次性加载所有URL和CSV数据到内存，再按每行CSV对应X个连续URL的逻辑批量处理，避免重复打开文件的低效操作。

修改后的完整代码示例 #

import csv
import requests

# 配置参数
X = 10  # 每行CSV对应的连续URL数量
URLS_FILE_PATH = 'C:\\urls.txt'
CSV_FILE_PATH = 'mycsv.csv'

def main():
    # 1. 加载所有URL到列表，清理空行和换行符
    with open(URLS_FILE_PATH, 'r', encoding="utf-8") as urls_file:
        urls = [url.strip() for url in urls_file if url.strip()]
    
    # 2. 加载CSV数据（跳过表头）
    csv_rows = []
    with open(CSV_FILE_PATH, "r", encoding="utf-8") as csvfile:
        csv_reader = csv.reader(csvfile)
        next(csv_reader)  # 跳过表头
        csv_rows = [row for row in csv_reader]
    
    # 3. 按逻辑映射并发送POST请求
    for row_idx, row in enumerate(csv_rows):
        # 计算当前行对应的URL区间：第row_idx行对应 [row_idx*X : (row_idx+1)*X] 的URL
        start_idx = row_idx * X
        end_idx = (row_idx + 1) * X
        target_urls = urls[start_idx:end_idx]
        
        if not target_urls:
            print(f"第{row_idx+1}行CSV没有对应的URL可发送")
            continue
        
        # 将CSV行转为字典（如果需要按表头键值对发送，可注释掉下面两行用字典）
        # payload = dict(zip(header, row))
        payload = row
        
        for url in target_urls:
            try:
                response = requests.post(url, data=payload)
                response.raise_for_status()  # 触发HTTP错误
                print(f"第{row_idx+1}行CSV -> URL {url} 发送成功，状态码: {response.status_code}")
            except requests.exceptions.RequestException as e:
                print(f"第{row_idx+1}行CSV -> URL {url} 发送失败: {str(e)}")

if __name__ == "__main__":
    main()

关键说明 #

• 提前加载数据：把URL和CSV数据一次性读入内存，避免原代码中循环每个URL都重新打开CSV的冗余操作
• URL分组逻辑：用row_idx * X和(row_idx+1)*X计算区间，通过列表切片urls[start_idx:end_idx]获取当前行对应的X个连续URL
• 错误处理：捕获请求异常，打印失败信息，避免单个请求失败导致程序终止
• 数据格式可选：如果需要按表头键值对发送POST数据，可取消注释payload = dict(zip(header, row))，同时需要保存表头变量（修改CSV加载部分：header = next(csv_reader)）

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