如何为pd.read_html设置5秒超时？解决无内置timeout属性问题

我现在批量处理URL，流程是用requests获取网页响应后，执行dfs = pd.read_html(str(response.text))解析表格。现在需要限制这个解析操作的运行时长：如果超过5秒就终止当前URL的处理，切换到下一个URL。但pd.read_html没有内置timeout属性，请问该怎么实现？

我的代码如下：

from bs4 import BeautifulSoup
import re
import requests
import os
import time
from pandas import DataFrame
import pandas as pd
from urllib.request import urlopen

headers = {'User-Agent': 'regsre@jh.edu'}

urls={'https://www.sec.gov/Archives/edgar/data/1058307/0001493152-21-003451.txt', 'https://www.sec.gov/Archives/edgar/data/1064722/0001760319-21-000006.txt'}


for url in urls:
  response = requests.get(url, headers = headers)
  response.raise_for_status()
  time.sleep(0.1)
  dfs = pd.read_html(str(response.text))
  print(url)
  for item in dfs:
    try:
      Operation=(item[0].apply(str).str.contains('Revenue') | item[0].apply(str).str.contains('profit'))
      if Operation.empty:
        pass
      if Operation.any():
        Operation_sheet=item
      if not Operation.any():
        CashFlows=(item[0].apply(str).str.contains('income') | item[0].apply(str).str.contains('loss'))
        if CashFlows.any():
          Operation_sheet=item
        if not CashFlows.any():
          pass

方法一：用timeout-decorator库快速实现（最简单） #

这个第三方库可以直接给函数套超时限制，步骤很简单：

1. 先安装库：

pip install timeout-decorator

2. 修改代码，把解析表格的逻辑封装成带超时装饰器的函数：

from bs4 import BeautifulSoup
import re
import requests
import os
import time
from pandas import DataFrame
import pandas as pd
from urllib.request import urlopen
import timeout_decorator

headers = {'User-Agent': 'regsre@jh.edu'}

urls={'https://www.sec.gov/Archives/edgar/data/1058307/0001493152-21-003451.txt', 'https://www.sec.gov/Archives/edgar/data/1064722/0001760319-21-000006.txt'}

@timeout_decorator.timeout(5)  # 设置5秒超时
def parse_html_table(html_text):
    return pd.read_html(html_text)

for url in urls:
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        time.sleep(0.1)
        dfs = parse_html_table(str(response.text))
        print(url)
        for item in dfs:
            try:
                Operation = (item[0].apply(str).str.contains('Revenue') | item[0].apply(str).str.contains('profit'))
                if Operation.empty:
                    pass
                if Operation.any():
                    Operation_sheet = item
                if not Operation.any():
                    CashFlows = (item[0].apply(str).str.contains('income') | item[0].apply(str).str.contains('loss'))
                    if CashFlows.any():
                        Operation_sheet = item
                    if not CashFlows.any():
                        pass
            except Exception as e:
                print(f"处理表格时出错: {e}")
                continue
    except timeout_decorator.TimeoutError:
        print(f"URL {url} 解析超时，跳过")
    except Exception as e:
        print(f"处理URL {url} 时出错: {e}")

方法二：用内置threading模块手动实现（无需额外依赖） #

如果不想装第三方库，用Python自带的线程模块也能实现超时控制：

from bs4 import BeautifulSoup
import re
import requests
import os
import time
from pandas import DataFrame
import pandas as pd
from urllib.request import urlopen
import threading

headers = {'User-Agent': 'regsre@jh.edu'}

urls={'https://www.sec.gov/Archives/edgar/data/1058307/0001493152-21-003451.txt', 'https://www.sec.gov/Archives/edgar/data/1064722/0001760319-21-000006.txt'}

def parse_with_timeout(html_text, result):
    try:
        result['dfs'] = pd.read_html(html_text)
    except Exception as e:
        result['error'] = e

for url in urls:
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        time.sleep(0.1)
        
        # 用字典存储线程执行结果（线程无法直接返回值）
        result = {}
        parse_thread = threading.Thread(target=parse_with_timeout, args=(str(response.text), result))
        parse_thread.start()
        parse_thread.join(timeout=5)  # 等待5秒
        
        if parse_thread.is_alive():
            print(f"URL {url} 解析超时，跳过")
            continue
        
        if 'error' in result:
            print(f"URL {url} 解析出错: {result['error']}")
            continue
        
        dfs = result['dfs']
        print(url)
        for item in dfs:
            try:
                Operation = (item[0].apply(str).str.contains('Revenue') | item[0].apply(str).str.contains('profit'))
                if Operation.empty:
                    pass
                if Operation.any():
                    Operation_sheet = item
                if not Operation.any():
                    CashFlows = (item[0].apply(str).str.contains('income') | item[0].apply(str).str.contains('loss'))
                    if CashFlows.any():
                        Operation_sheet = item
                    if not CashFlows.any():
                        pass
            except Exception as e:
                print(f"处理表格时出错: {e}")
                continue
    except Exception as e:
        print(f"处理URL {url} 时出错: {e}")

注意事项 #

• 超时限制针对的是pd.read_html的整个执行过程，包括HTML解析和表格提取的全部时间。
• 线程方法中超时后，后台线程可能仍在运行，但我们已经跳过当前URL的处理，不会影响后续任务。
• 建议给所有IO和解析操作加异常捕获，避免单个URL的错误导致整个批量任务中断。

内容的提问来源于Stack Exchange，提问作者Candice LE