Python解析结构不一致的PDF赛事报告表格如何避免代码重复？

问题背景 #

有100多份结构不统一的赛事报告PDF，需提取数据存入DataFrame用于后续分析。使用pdfplumber读取后，表格行长度不一致，当前实现的提取代码包含大量重复try/except块，不同行类型的索引逻辑无规律，导致代码冗余、可读性差且难以调试。

原代码片段 #

PDF表格提取代码 #

import os
import pdfplumber

directory = os.fsencode('')
matchs_raw = {}
for file in os.listdir(directory):
    
    filename = os.fsdecode(file)
    if '.pdf' not in filename:
        continue
    matchs_raw[filename] = []

    with pdfplumber.open(f'\\{filename}') as pdf:
        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                for i in table:
                    matchs_raw[filename].append(i)

DataFrame提取核心代码 #

import re
import numpy as np
from tqdm import tqdm

columns = ['file_name','minute','role','numero','nom','recevant_ou_visiteur','equipe',
'description','score_1_ponctuel','score_2_ponctuel']

data = {col: [] for col in columns}

for match in tqdm(matchs_raw):
    if matchs_raw[match][0][0]== "Organisateur":
        for l in range(len(matchs_raw[match])):
            if matchs_raw[match][l][0]=="Déroulé du Match" or matchs_raw[match][l][0]=="DérouléduMatch":
                starting_row = l+3
                break
        for j in matchs_raw[match][starting_row:]:
            
            if len(j)!=13:
                ## LEFT COLUMN
                
                data['file_name'].append(matchs_raw[match][0][20])
                
                try:
                    data['minute'].append(j[0])
                except:
                    data['minute'].append(np.nan)
                try:
                    data['role'].append(re.findall("JR|JV|OR|OV",j[3])[0][0])  
                except:
                    data['role'].append(np.nan)
                
                try:
                    if re.findall("JR|JV|OR|OV",j[3])[0][0] == "J":
                        try:
                            data['numero'].append(re.findall("N[^\x00-\x7F]+\d*",j[3])[0])
                        except:
                            data['numero'].append(np.nan)
                        try:
                            data['nom'].append(re.findall("N[^\x00-\x7F]+\d*(\D+)",j[3])[0])
                        except:
                            data['nom'].append(np.nan)
                        try:
                            data['description'].append(re.findall("(.+?)(JR|JV|OR|OV)N[^\x00-\x7F]",j[3])[0][0])
                        except:
                            data['description'].append(np.nan)
                    
                    else:
                        try:
                            data['numero'].append("Officiel")
                        except:
                            data['numero'].append(np.nan)
                        try:
                            data['nom'].append(re.findall("^(.+?)(OV|OR)(.+)",j[3])[0][2].strip())
                        except:
                            data['nom'].append(np.nan)
                        try:
                            data['description'].append(re.findall("^(.+?)(OV|OR)",j[3])[0][0].strip())
                        except:
                            data['description'].append(np.nan)

优化方案 #

1. 封装通用异常处理工具函数 #

将重复的try/except逻辑封装为工具函数，消除代码冗余：

def safe_extract(extract_func, default=np.nan):
    """安全执行提取逻辑，捕获常见异常并返回默认值"""
    try:
        return extract_func()
    except (IndexError, ValueError, TypeError, AttributeError):
        return default

2. 按行类型封装提取处理器 #

针对不同行长度、结构的行，分别封装独立的提取处理器，将提取逻辑与主循环解耦，便于扩展和调试：

class LeftColumnRowProcessor:
    """处理长度非13的左侧列行数据"""
    def __init__(self, row, file_name):
        self.row = row
        self.file_name = file_name
        self.role = safe_extract(lambda: re.findall(r"JR|JV|OR|OV", self.row[3])[0][0])
    
    def extract(self):
        return {
            'file_name': self.file_name,
            'minute': safe_extract(lambda: self.row[0]),
            'role': self.role,
            'numero': self._extract_numero(),
            'nom': self._extract_nom(),
            'description': self._extract_description(),
            'recevant_ou_visiteur': np.nan,
            'equipe': np.nan,
            'score_1_ponctuel': np.nan,
            'score_2_ponctuel': np.nan
        }
    
    def _extract_numero(self):
        if self.role == "J":
            return safe_extract(lambda: re.findall(r"N[^\x00-\x7F]+\d*", self.row[3])[0])
        return "Officiel"
    
    def _extract_nom(self):
        if self.role == "J":
            return safe_extract(lambda: re.findall(r"N[^\x00-\x7F]+\d*(\D+)", self.row[3])[0])
        return safe_extract(lambda: re.findall(r"^(.+?)(OV|OR)(.+)", self.row[3])[0][2].strip())
    
    def _extract_description(self):
        if self.role == "J":
            return safe_extract(lambda: re.findall(r"(.+?)(JR|JV|OR|OV)N[^\x00-\x7F]", self.row[3])[0][0])
        return safe_extract(lambda: re.findall(r"^(.+?)(OV|OR)", self.row[3])[0][0].strip())

# 可新增其他行类型处理器，如Length13RowProcessor、RightColumnRowProcessor等

3. 统一数据收集流程 #

改用列表收集每行的字典数据，最后一次性转换为DataFrame，比逐列append更高效且易维护：

import pandas as pd

columns = ['file_name','minute','role','numero','nom','recevant_ou_visiteur','equipe',
'description','score_1_ponctuel','score_2_ponctuel']

data_list = []

for match in tqdm(matchs_raw):
    match_data = matchs_raw[match]
    if match_data[0][0] != "Organisateur":
        continue
    
    # 定位数据起始行
    starting_row = None
    for l in range(len(match_data)):
        if match_data[l][0] in ("Déroulé du Match", "DérouléduMatch"):
            starting_row = l + 3
            break
    if starting_row is None:
        continue
    
    file_name = match_data[0][20]
    for row in match_data[starting_row:]:
        # 根据行长度选择对应处理器
        if len(row) != 13:
            processor = LeftColumnRowProcessor(row, file_name)
            data_list.append(processor.extract())
        # 新增其他行类型的判断与处理
        # elif len(row) == 13:
        #     processor = Length13RowProcessor(row, file_name)
        #     data_list.append(processor.extract())

# 转换为DataFrame
df = pd.DataFrame(data_list, columns=columns)

4. 优化PDF原始数据提取逻辑 #

在提取PDF表格时过滤空行或无效行，减少后续处理的无效数据量：

directory = os.fsencode('')
matchs_raw = {}
for file in os.listdir(directory):
    filename = os.fsdecode(file)
    if not filename.endswith('.pdf'):
        continue
    matchs_raw[filename] = []

    with pdfplumber.open(f'\\{filename}') as pdf:
        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                # 过滤空行或全为空的行
                valid_rows = [row for row in table if any(cell and cell.strip() for cell in row)]
                matchs_raw[filename].extend(valid_rows)

内容的提问来源于stack exchange，提问作者Pablo Martín Calvo