Python高效获取DataFrame中URL状态码：替代for循环方案

嘿，你的问题我太熟了——用串行for循环处理一堆URL请求简直是效率杀手，毕竟每个请求都要等服务器响应，大把时间都浪费在等待上了。下面给你两个高效的解决方案，分分钟把状态码批量塞进DataFrame的新列里，重点覆盖你关心的200、403、404状态码，还能处理各类异常情况～

方案一：异步IO（强推！超大量URL首选） #

HTTP请求是典型的IO密集型任务，异步IO能让所有请求同时发起，不用等前一个完成再处理下一个，效率直接拉满。这里用aiohttp库实现，上手难度也不高。

步骤走起来： #

1. 先安装依赖：

pip install aiohttp pandas

2. 完整代码直接复用：

import aiohttp
import asyncio
import pandas as pd

async def fetch_single_status(session, url, timeout=10):
    """异步获取单个URL的状态码"""
    try:
        # 加User-Agent避免被网站当成爬虫拦截
        async with session.get(url, timeout=timeout, headers={"User-Agent": "Mozilla/5.0"}) as response:
            return response.status
    except Exception as e:
        # 超时、DNS解析失败等异常，返回None或自定义标记
        print(f"请求{url}出错: {str(e)}")
        return None

async def batch_fetch_statuses(url_list):
    """批量异步处理所有URL"""
    async with aiohttp.ClientSession() as session:
        # 创建所有请求任务
        tasks = [fetch_single_status(session, url) for url in url_list]
        # 并发执行所有任务并收集结果
        status_codes = await asyncio.gather(*tasks)
        return status_codes

def add_status_column(df, url_column_name):
    """给DataFrame新增状态码列"""
    url_list = df[url_column_name].tolist()
    # 运行异步任务（Jupyter环境直接用await batch_fetch_statuses(url_list)即可，无需run）
    status_results = asyncio.run(batch_fetch_statuses(url_list))
    df['status_code_url'] = status_results
    return df

# 示例用法
if __name__ == "__main__":
    # 模拟你的目标DataFrame
    test_df = pd.DataFrame({'target_url': [
        'https://www.example.com',
        'https://www.google.com',
        'https://this-is-a-404-url.com'
    ]})
    test_df = add_status_column(test_df, 'target_url')
    print(test_df)

小提醒： #

• 务必加User-Agent！很多网站会直接拦截无标识的请求；
• 超时时间设置要合理，避免单个慢请求拖垮整个任务；
• 若URL数量过万，建议拆分成批次处理，一次性发起太多请求容易被网站拉黑IP；
• 异常处理可按需调整，比如把超时请求标记为-1，方便后续筛选。

方案二：多线程（上手快，中等规模URL友好） #

要是你对异步不太熟悉，多线程方案也能大幅提速，代码逻辑和你原来的循环更接近，学习成本极低。用Python自带的concurrent.futures.ThreadPoolExecutor即可，无需额外装库（除了pandas和requests）。

直接看代码： #

import requests
from concurrent.futures import ThreadPoolExecutor
import pandas as pd

def fetch_single_status(url, timeout=10):
    """获取单个URL状态码，附带重试机制"""
    session = requests.Session()
    # 遇到5xx服务器错误自动重试3次，避免偶然故障导致结果不准
    retry_strategy = requests.adapters.Retry(
        total=3,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503, 504]
    )
    adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    
    try:
        response = session.get(url, timeout=timeout, headers={"User-Agent": "Mozilla/5.0"})
        return response.status
    except Exception as e:
        print(f"请求{url}出问题了: {str(e)}")
        return None

def add_status_column(df, url_column_name, max_workers=10):
    """给DataFrame添加状态码列"""
    url_list = df[url_column_name].tolist()
    # 用线程池并行处理，max_workers别设太大，避免被网站判定为恶意请求
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        status_results = list(executor.map(fetch_single_status, url_list))
    df['status_code_url'] = status_results
    return df

# 示例用法
if __name__ == "__main__":
    test_df = pd.DataFrame({'target_url': [
        'https://www.example.com',
        'https://www.google.com',
        'https://this-is-a-404-url.com'
    ]})
    test_df = add_status_column(test_df, 'target_url')
    print(test_df)

小提醒： #

• max_workers建议设为10-20之间，过大容易触发网站的反爬机制；
• 内置的重试机制能提升结果准确性，避免服务器临时故障导致误判；
• 同样要带User-Agent，别做“裸奔”的请求。

两种方案怎么选？ #

• 若你有几千甚至上万条URL，优先选异步IO，效率比多线程高出不少；
• 要是URL数量在几百到几千条，多线程足够用，代码更省心；
• 不管用哪种方案，都先拿几十条URL测试验证，确认没问题再批量执行，避免踩坑。

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