导入Excel逐行解析校验

本文目录导读：

1. 方案一：使用pandas + 自定义校验
2. 方案二：使用openpyxl逐行处理
3. 安装依赖
4. 自定义校验规则示例
5. 批量处理示例

我来为您提供一个完整的Excel逐行解析校验的Python实现方案,包含数据验证和错误处理。

使用pandas + 自定义校验

import pandas as pd
import numpy as np
from typing import List, Dict, Tuple
import re
from datetime import datetime
class ExcelValidator:
    def __init__(self, excel_path: str):
        self.excel_path = excel_path
        self.errors = []
        self.valid_data = []
    def validate_row(self, row_data: Dict, row_num: int) -> Tuple[bool, List[str]]:
        """
        对单行数据进行校验
        返回：(是否通过, 错误列表)
        """
        row_errors = []
        # 示例校验规则 - 根据实际需求调整
        required_fields = ['姓名', '年龄', '邮箱', '手机号']
        # 1. 检查必填字段
        for field in required_fields:
            if field not in row_data or pd.isna(row_data[field]):
                row_errors.append(f"第{row_num}行: 缺少必填字段'{field}'")
        # 2. 年龄校验
        if '年龄' in row_data and not pd.isna(row_data['年龄']):
            age = row_data['年龄']
            if not isinstance(age, (int, float)) or age < 0 or age > 150:
                row_errors.append(f"第{row_num}行: 年龄格式错误 ({age})")
        # 3. 邮箱格式校验
        if '邮箱' in row_data and not pd.isna(row_data['邮箱']):
            email = str(row_data['邮箱'])
            email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            if not re.match(email_pattern, email):
                row_errors.append(f"第{row_num}行: 邮箱格式错误 ({email})")
        # 4. 手机号校验
        if '手机号' in row_data and not pd.isna(row_data['手机号']):
            phone = str(row_data['手机号'])
            phone_pattern = r'^1[3-9]\d{9}$'
            if not re.match(phone_pattern, phone):
                row_errors.append(f"第{row_num}行: 手机号格式错误 ({phone})")
        # 5. 数值范围校验
        if '分数' in row_data and not pd.isna(row_data['分数']):
            score = float(row_data['分数'])
            if score < 0 or score > 100:
                row_errors.append(f"第{row_num}行: 分数超出范围 (0-100)")
        return len(row_errors) == 0, row_errors
    def process_excel(self) -> Dict:
        """
        处理Excel文件，逐行校验
        """
        try:
            # 读取Excel文件
            df = pd.read_excel(self.excel_path, dtype=str)  # 全部读为字符串避免类型转换问题
            print(f"读取到 {len(df)} 行数据")
            valid_rows = []
            invalid_rows = []
            # 逐行处理
            for index, row in df.iterrows():
                row_data = row.to_dict()
                row_num = index + 2  # 加2因为excel从1开始，且包含表头
                is_valid, row_errors = self.validate_row(row_data, row_num)
                if is_valid:
                    # 转换数据类型
                    processed_row = self.convert_types(row_data)
                    valid_rows.append(processed_row)
                else:
                    invalid_rows.append({
                        'row_num': row_num,
                        'data': row_data,
                        'errors': row_errors
                    })
            return {
                'total_rows': len(df),
                'valid_count': len(valid_rows),
                'invalid_count': len(invalid_rows),
                'valid_data': valid_rows,
                'errors': invalid_rows
            }
        except Exception as e:
            print(f"处理Excel文件时出错: {str(e)}")
            return None
    def convert_types(self, row_data: Dict) -> Dict:
        """
        转换数据类型
        """
        converted = row_data.copy()
        # 示例类型转换
        if '年龄' in converted and converted['年龄']:
            try:
                converted['年龄'] = int(float(converted['年龄']))
            except:
                pass
        if '分数' in converted and converted['分数']:
            try:
                converted['分数'] = float(converted['分数'])
            except:
                pass
        if '出生日期' in converted and converted['出生日期']:
            try:
                converted['出生日期'] = pd.to_datetime(converted['出生日期'])
            except:
                pass
        return converted
# 使用示例
def main():
    validator = ExcelValidator('data.xlsx')
    result = validator.process_excel()
    if result:
        print(f"总行数: {result['total_rows']}")
        print(f"有效行数: {result['valid_count']}")
        print(f"无效行数: {result['invalid_count']}")
        # 输出错误信息
        if result['errors']:
            print("\n=== 错误详情 ===")
            for error in result['errors']:
                print(f"行 {error['row_num']}:")
                for err in error['errors']:
                    print(f"  - {err}")
        # 处理有效数据
        if result['valid_data']:
            valid_df = pd.DataFrame(result['valid_data'])
            valid_df.to_excel('valid_data.xlsx', index=False)
            print(f"\n有效数据已保存到 valid_data.xlsx")
if __name__ == "__main__":
    main()

使用openpyxl逐行处理

from openpyxl import load_workbook
from openpyxl.utils import get_column_letter
import re
from typing import List, Dict
class SimpleExcelValidator:
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.wb = load_workbook(file_path, data_only=True)
        self.ws = self.wb.active
    def get_column_mapping(self) -> Dict:
        """获取列名到列索引的映射"""
        mapping = {}
        for col_idx, cell in enumerate(self.ws[1], 1):
            if cell.value:
                mapping[cell.value] = col_idx
        return mapping
    def validate_all_rows(self) -> List[Dict]:
        """逐行校验所有数据"""
        column_mapping = self.get_column_mapping()
        results = []
        errors = []
        # 从第2行开始（跳过表头）
        for row_idx in range(2, self.ws.max_row + 1):
            row_data = {}
            row_errors = []
            # 读取当前行数据
            for col_name, col_idx in column_mapping.items():
                cell = self.ws.cell(row=row_idx, column=col_idx)
                row_data[col_name] = cell.value
            # 校验规则
            row_num = row_idx
            # 示例校验
            name = row_data.get('姓名')
            age = row_data.get('年龄')
            email = row_data.get('邮箱')
            # 姓名非空校验
            if not name or str(name).strip() == '':
                row_errors.append(f"第{row_num}行: 姓名为空")
            # 年龄校验
            if age is not None:
                try:
                    age_val = int(age)
                    if age_val < 0 or age_val > 150:
                        row_errors.append(f"第{row_num}行: 年龄无效 ({age})")
                except (ValueError, TypeError):
                    row_errors.append(f"第{row_num}行: 年龄格式错误 ({age})")
            # 邮箱校验
            if email and str(email).strip():
                email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
                if not re.match(email_pattern, str(email)):
                    row_errors.append(f"第{row_num}行: 邮箱格式错误 ({email})")
            # 记录结果
            if row_errors:
                errors.append({
                    'row': row_num,
                    'data': row_data,
                    'errors': row_errors
                })
            else:
                results.append(row_data)
        return {
            'valid_rows': results,
            'error_rows': errors,
            'total_rows': len(results) + len(errors)
        }
# 使用示例
def process_excel():
    validator = SimpleExcelValidator('data.xlsx')
    result = validator.validate_all_rows()
    print(f"总行数: {result['total_rows']}")
    print(f"有效行数: {len(result['valid_rows'])}")
    print(f"错误行数: {len(result['error_rows'])}")
    # 处理错误
    for error in result['error_rows']:
        print(f"\n第{error['row']}行错误:")
        for e in error['errors']:
            print(f"  - {e}")

安装依赖

pip install pandas openpyxl xlrd numpy

自定义校验规则示例

class CustomValidator:
    """自定义校验规则集合"""
    @staticmethod
    def validate_phone(phone: str) -> bool:
        """手机号校验"""
        if not phone:
            return False
        return bool(re.match(r'^1[3-9]\d{9}$', str(phone)))
    @staticmethod
    def validate_id_card(id_card: str) -> bool:
        """身份证号校验"""
        if not id_card:
            return False
        # 18位身份证校验
        pattern = r'^[1-9]\d{5}(19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]$'
        return bool(re.match(pattern, str(id_card)))
    @staticmethod
    def validate_date(date_str: str) -> bool:
        """日期校验"""
        try:
            datetime.strptime(str(date_str), '%Y-%m-%d')
            return True
        except:
            return False
    @staticmethod
    def validate_amount(amount) -> bool:
        """金额校验"""
        try:
            amount = float(amount)
            return amount >= 0 and amount <= 999999999.99
        except:
            return False

批量处理示例

def batch_process_excel(files: List[str]):
    """批量处理多个Excel文件"""
    all_results = {}
    for file_path in files:
        print(f"\n处理文件: {file_path}")
        validator = ExcelValidator(file_path)
        result = validator.process_excel()
        all_results[file_path] = result
        # 汇总错误信息
        if result and result['errors']:
            error_df = pd.DataFrame(result['errors'])
            error_file = file_path.replace('.xlsx', '_errors.xlsx')
            error_df.to_excel(error_file, index=False)
            print(f"错误信息已保存到: {error_file}")
    return all_results

这个方案提供了完整的Excel逐行解析校验功能,您可以根据实际需求：

1. 调整校验规则
2. 添加新的数据验证逻辑
3. 自定义错误处理方式
4. 导出验证结果

需要我为您定制特定的校验规则或优化代码吗？