解决KeyError列不存在异常 实现Python程序异常后持续运行

我编写了一段包含爬虫、数据处理的Python代码，运行时并非每次都会报错，但可能在第100次运行时触发KeyError: "None of [Index([....]) are in the [columns]"错误，推测问题出在读取CSV文件环节。我希望通过异常处理实现程序遇到异常时不终止、继续运行，但目前的异常处理未达到预期效果，代码如下：

import json
from datetime import datetime
import pandas as pd
import requests
from bs4 import BeautifulSoup
import time
import os
import urllib.request
import urllib.parse
import smtplib
from requests.exceptions import ConnectionError
from requests.packages.urllib3.exceptions import MaxRetryError
from requests.packages.urllib3.exceptions import ProxyError as urllib3_ProxyError


def calc():
    json_url = "https://www.example.com/........"
    proxies = {
        'http': '......',
        'https': '.........',
    }
    headers = requests.utils.default_headers()
    headers.update(
        {
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36'
            })

    try:
        s = requests.Session()
        response = s.get(json_url, proxies=proxies, headers=headers, timeout=15)
        soup = BeautifulSoup(response.content, 'html.parser')

        try:
            json.loads(soup.text)
            result = json.loads(soup.text)
            result = pd.json_normalize(result["listings"])
            result = result[['itemId', 'title', 'attributes', 'vipUrl',
                             'priceInfo.priceCents', 'priceInfo.priceType']]

            result = result.join(pd.json_normalize(result["attributes"]))

            result = result[['itemId', 'title', 'vipUrl', 'priceInfo.priceCents',
                             'priceInfo.priceType', 'constructionYear', 'mileage', 'fuel',
                             'transmission']]

            try:
                df = pd.read_csv(r'filecsv.csv')   #Maybe that's where the problem lies.
                df = df.rename(columns={'title': 'title_1', 'vipUrl': 'vipUrl_1', 'priceInfo.priceCents': 'pp',
                                        'priceInfo.priceType': 'dd',
                                        'constructionYear': 'Year_1', 'mileage': 'mileage_1', 'fuel': 'fuel_1',
                                        'transmission': 'transmission_1'})
                filter_df = result.merge(df, how='outer', on='itemId')
                filter_df = filter_df[filter_df['dd'].isnull()]
                filter_df = filter_df[['itemId', 'title', 'vipUrl', 'priceInfo.priceCents', 'priceInfo.priceType',
                                       'constructionYear', 'mileage', 'fuel', 'transmission']]

                Max_ID = df['itemId'].str.extract('(\d+)').astype('int').max()[0]
                Max_ID_Filter = filter_df['itemId'].str.extract('(\d+)').astype('int').max()[0]
                print(Max_ID_Filter, Max_ID)
                if (Max_ID_Filter > Max_ID):
                    print("Yes")
                    for index, row in filter_df.iterrows():
                        apiToken = '63691.....'
                        chatID =[630.......] 
                        for i in chatID:
                            apiURL = f'https://api.telegram.org/bot{apiToken}/sendMessage'

                            try:
                                msgg = 'Ok'
                                response = requests.post(apiURL, json={'chat_id':i, 'text':msgg})
                                
                            except Exception as e:
                                print(e)
                else:
                    print("No")

                df = df.rename(columns={'title_1': 'title', 'vipUrl_1': 'vipUrl', 'pp': 'priceInfo.priceCents',
                                        'dd': 'priceInfo.priceType',
                                        'Year_1': 'constructionYear', 'mileage_1': 'mileage', 'fuel_1': 'fuel',
                                        'transmission_1': 'transmission'})
                df = pd.concat([df, filter_df])
                df.to_csv("filecsv.csv", encoding='utf-8-sig', index=False)
            except KeyError:
                pass
            except:
                result.to_csv("filecsv.csv", encoding='utf-8-sig', index=False)

        except json.JSONDecodeError as e:
            print(e.msg, e)
            print(repr(soup))
            print(len(soup))
    except ConnectionError as ce:
        if (isinstance(ce.args[0], MaxRetryError) and
                isinstance(ce.args[0].reason, urllib3_ProxyError)):
            pass
    except requests.exceptions.Timeout:
        pass
    except requests.exceptions.ChunkedEncodingError:
        pass
    time.sleep(6)



stop = 1
while stop > 0:
    #stop = stop - 1
    calc()

问题分析 #

触发KeyError的场景不止读取CSV，还可能是：

• 读取CSV后重命名列时，原CSV文件缺失指定的列
• 处理result数据框时，选取的列不存在（比如接口返回的JSON结构变更）
• 提取Max_ID时，itemId格式不符合预期导致提取失败

当前异常处理的缺陷：

1. 内层try块仅覆盖了CSV读取后的逻辑，前面的result列选取、JSON解析后的数据处理未被覆盖，若这些环节出错会直接终止程序
2. 通用except会覆盖KeyError，且处理逻辑（直接将result写入CSV）可能破坏原有数据
3. 异常发生时无详细日志，无法定位具体出错位置

修复方案 #

1. 扩大异常捕获范围并记录错误 #

将整个数据处理逻辑（从JSON解析到CSV写入）都纳入一个try-except块，捕获异常时打印错误类型和详情，方便后续排查：

2. 优化CSV操作的安全性 #

• 读取CSV后检查必要列是否存在，避免重命名或选取列时出错
• 选取列时使用df.reindex(columns=所需列)替代直接索引，避免列不存在触发KeyError

修复后的代码示例 #

import json
from datetime import datetime
import pandas as pd
import requests
from bs4 import BeautifulSoup
import time
import os
import urllib.request
import urllib.parse
import smtplib
from requests.exceptions import ConnectionError
from requests.packages.urllib3.exceptions import MaxRetryError
from requests.packages.urllib3.exceptions import ProxyError as urllib3_ProxyError


def calc():
    json_url = "https://www.example.com/........"
    proxies = {
        'http': '......',
        'https': '.........',
    }
    headers = requests.utils.default_headers()
    headers.update(
        {
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36'
            })

    try:
        s = requests.Session()
        response = s.get(json_url, proxies=proxies, headers=headers, timeout=15)
        soup = BeautifulSoup(response.content, 'html.parser')

        # 将所有数据处理逻辑纳入统一的异常捕获
        try:
            json_data = json.loads(soup.text)
            result = pd.json_normalize(json_data["listings"])
            
            # 安全选取列：用reindex避免KeyError，缺失列会填充NaN
            required_result_cols = ['itemId', 'title', 'attributes', 'vipUrl',
                                   'priceInfo.priceCents', 'priceInfo.priceType']
            result = result.reindex(columns=required_result_cols)
            
            # 处理attributes字段
            if 'attributes' in result.columns:
                result = result.join(pd.json_normalize(result["attributes"]))
            
            # 再次安全选取最终需要的列
            final_result_cols = ['itemId', 'title', 'vipUrl', 'priceInfo.priceCents',
                                'priceInfo.priceType', 'constructionYear', 'mileage', 'fuel',
                                'transmission']
            result = result.reindex(columns=final_result_cols)
            
            # 处理CSV逻辑
            try:
                df = pd.read_csv(r'filecsv.csv')
                # 检查CSV是否包含重命名所需的列
                rename_cols = {'title': 'title_1', 'vipUrl': 'vipUrl_1', 
                              'priceInfo.priceCents': 'pp', 'priceInfo.priceType': 'dd',
                              'constructionYear': 'Year_1', 'mileage': 'mileage_1', 
                              'fuel': 'fuel_1', 'transmission': 'transmission_1'}
                # 只重命名存在的列
                existing_rename_cols = {k:v for k,v in rename_cols.items() if k in df.columns}
                df = df.rename(columns=existing_rename_cols)
                
                filter_df = result.merge(df, how='outer', on='itemId')
                # 检查dd列是否存在再过滤
                if 'dd' in filter_df.columns:
                    filter_df = filter_df[filter_df['dd'].isnull()]
                # 安全选取filter_df的列
                filter_df = filter_df.reindex(columns=final_result_cols)
                
                # 安全提取Max_ID
                if 'itemId' in df.columns:
                    df_id_extract = df['itemId'].str.extract('(\d+)').dropna()
                    Max_ID = df_id_extract.astype('int').max()[0] if not df_id_extract.empty else 0
                else:
                    Max_ID = 0
                
                if 'itemId' in filter_df.columns:
                    filter_id_extract = filter_df['itemId'].str.extract('(\d+)').dropna()
                    Max_ID_Filter = filter_id_extract.astype('int').max()[0] if not filter_id_extract.empty else 0
                else:
                    Max_ID_Filter = 0
                
                print(Max_ID_Filter, Max_ID)
                if Max_ID_Filter > Max_ID:
                    print("Yes")
                    apiToken = '63691.....'
                    chatID = [630.......] 
                    for i in chatID:
                        apiURL = f'https://api.telegram.org/bot{apiToken}/sendMessage'
                        try:
                            msgg = 'Ok'
                            response = requests.post(apiURL, json={'chat_id':i, 'text':msgg})
                        except Exception as e:
                            print(f"Telegram发送失败: {e}")
                else:
                    print("No")
                
                # 还原列名
                reverse_rename = {v:k for k,v in existing_rename_cols.items()}
                df = df.rename(columns=reverse_rename)
                df = pd.concat([df, filter_df]).drop_duplicates(subset='itemId')
                df.to_csv("filecsv.csv", encoding='utf-8-sig', index=False)
            except FileNotFoundError:
                # 若CSV不存在，直接写入初始数据
                result.to_csv("filecsv.csv", encoding='utf-8-sig', index=False)
            except Exception as e:
                print(f"CSV处理出错: {type(e).__name__}: {e}")
                # 出错时不覆盖原有CSV，仅打印日志
                pass

        except json.JSONDecodeError as e:
            print(f"JSON解析错误: {e.msg}, {e}")
            print(f"响应内容长度: {len(soup)}, 内容预览: {repr(soup.text[:200])}")
        except Exception as e:
            print(f"数据处理出错: {type(e).__name__}: {e}")
            # 数据处理出错，不终止程序，继续循环
            
    except ConnectionError as ce:
        if (isinstance(ce.args[0], MaxRetryError) and
                isinstance(ce.args[0].reason, urllib3_ProxyError)):
            print("代理连接失败，跳过本次请求")
    except requests.exceptions.Timeout:
        print("请求超时，跳过本次请求")
    except requests.exceptions.ChunkedEncodingError:
        print("响应编码错误，跳过本次请求")
    except Exception as e:
        print(f"请求环节出错: {type(e).__name__}: {e}")
    time.sleep(6)


stop = 1
while stop > 0:
    calc()

关键修改点 #

• 使用reindex替代直接列索引，避免列不存在触发KeyError
• 检查列是否存在后再进行重命名、过滤等操作
• 扩大异常捕获范围，覆盖所有可能出错的环节
• 打印详细异常信息，方便定位问题
• 出错时不随意覆盖原有CSV文件，避免数据丢失

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