使用Python从CSV文件提取特定表格的问题排查

我在Jupyter Notebook中使用Python从报表CSV提取特定表格并合并为单个表格，但目前代码找到目标列名SUBSCRIBER NAME后，会复制下方所有数据，其中包含机器不可读的无效内容，需要输出标准的表格格式。

原代码如下：

import csv
import pandas as pd

# Function to extract relevant data from the CSV file
def extract_data_from_csv(file_path):
    data_start = False
    headers = []
    rows = []

    with open(file_path, newline='') as csvfile:
        csv_reader = csv.reader(csvfile)
        
        for row in csv_reader:
            if not data_start:
                if row and row[0] == "SUBSCRIBER NAME":
                    headers = row
                    data_start = True
            elif data_start and row:
                if row[0].strip().upper() == "PRODUCT TOTAL":
                    break
                rows.append(row)

    return headers, rows

# Specify the path to your CSV file
csv_file_path = 'Testing/sample report.csv'

# Extract data
headers, rows = extract_data_from_csv(csv_file_path)

# Create a DataFrame using pandas
df = pd.DataFrame(rows, columns=headers)

# Display the DataFrame as a table
display(df)

# Save the DataFrame to a new CSV file
output_csv_file_path = 'mark_table_1.csv'
df.to_csv(output_csv_file_path, index=False)

原代码仅通过PRODUCT TOTAL作为终止条件，未对中间的无效行做过滤。以下是增强过滤逻辑的优化代码，确保只保留有效数据行：

import csv
import pandas as pd

def extract_data_from_csv(file_path):
    data_start = False
    headers = []
    rows = []

    with open(file_path, newline='') as csvfile:
        csv_reader = csv.reader(csvfile)
        
        for row in csv_reader:
            # 跳过完全空的行
            if not row:
                continue
                
            if not data_start:
                if row[0].strip() == "SUBSCRIBER NAME":
                    headers = row
                    header_len = len(headers)
                    data_start = True
            else:
                # 终止条件：遇到PRODUCT TOTAL行
                if row[0].strip().upper() == "PRODUCT TOTAL":
                    break
                # 过滤条件1：行长度和表头一致，避免格式错乱的行
                if len(row) != header_len:
                    continue
                # 过滤条件2：关键列（SUBSCRIBER NAME）不为空，排除无效汇总/空数据行
                if not row[0].strip():
                    continue
                rows.append(row)

    return headers, rows

# 路径配置
csv_file_path = 'Testing/sample report.csv'
output_csv_file_path = 'mark_table_1.csv'

# 提取并生成表格
headers, rows = extract_data_from_csv(csv_file_path)
df = pd.DataFrame(rows, columns=headers)

# 在Jupyter中显示表格
display(df)
# 保存结果
df.to_csv(output_csv_file_path, index=False)

• 跳过空行：直接跳过完全无内容的行，避免无效数据
• 行长度校验：确保数据行的列数和表头一致，过滤掉格式混乱的行（比如合并单元格导致的列数不匹配）
• 关键列非空校验：检查SUBSCRIBER NAME列不为空，排除中间的汇总行或无效数据行
• 保留原终止逻辑：依然在遇到PRODUCT TOTAL时停止提取，确保不包含后续的汇总内容

内容的提问来源于stack exchange，提问作者Mark M.