如何编写数据格式批量校验脚本

本文目录导读：

1. 基础校验框架
2. 特定格式校验实现
3. 批量校验脚本
4. 配置驱动的校验器
5. 命令行工具
6. 使用示例

我来介绍如何编写批量数据格式校验脚本,包含多种常见场景。

基础校验框架

Python实现的基础校验器

import json
import csv
import re
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Optional
class DataValidator:
    """通用数据格式校验器"""
    def __init__(self):
        self.errors = []
        self.warnings = []
    def validate_file(self, file_path: str, file_type: str = 'auto') -> bool:
        """校验单个文件"""
        file_path = Path(file_path)
        if not file_path.exists():
            self.errors.append(f"文件不存在: {file_path}")
            return False
        # 自动检测文件类型
        if file_type == 'auto':
            file_type = file_path.suffix.lower().lstrip('.')
        # 根据文件类型选择校验方法
        validators = {
            'json': self.validate_json,
            'csv': self.validate_csv,
            'xml': self.validate_xml,
            'yaml': self.validate_yaml,
            'txt': self.validate_txt
        }
        validator = validators.get(file_type)
        if not validator:
            self.errors.append(f"不支持的文件类型: {file_type}")
            return False
        return validator(file_path)
    def validate_json(self, file_path: Path) -> bool:
        """校验JSON文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            # 基本结构校验
            if isinstance(data, list):
                return self.validate_json_array(data)
            elif isinstance(data, dict):
                return self.validate_json_object(data)
            else:
                self.errors.append(f"JSON根元素必须是对象或数组")
                return False
        except json.JSONDecodeError as e:
            self.errors.append(f"JSON解析错误: {e}")
            return False
        except Exception as e:
            self.errors.append(f"读取文件错误: {e}")
            return False
    def validate_csv(self, file_path: Path) -> bool:
        """校验CSV文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                reader = csv.reader(f)
                # 读取表头
                headers = next(reader, None)
                if not headers:
                    self.errors.append("CSV文件为空")
                    return False
                # 校验每一行
                row_num = 1
                valid = True
                for row in reader:
                    row_num += 1
                    if len(row) != len(headers):
                        self.errors.append(f"第{row_num}行列数不匹配: {len(row)} != {len(headers)}")
                        valid = False
                return valid
        except Exception as e:
            self.errors.append(f"CSV读取错误: {e}")
            return False
    def validate_json_array(self, data: List) -> bool:
        """校验JSON数组"""
        if not data:
            self.warnings.append("JSON数组为空")
            return True
        valid = True
        for i, item in enumerate(data):
            if not self.validate_record(item, i):
                valid = False
        return valid
    def validate_record(self, record: Dict, index: int) -> bool:
        """校验单条记录"""
        # 子类重写此方法实现具体业务校验
        return True

特定格式校验实现

电子邮件格式校验

class EmailValidator(DataValidator):
    """邮箱格式校验器"""
    def __init__(self):
        super().__init__()
        self.email_pattern = re.compile(
            r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        )
        self.domain_list = ['gmail.com', '163.com', 'qq.com', 'outlook.com']
    def validate_email(self, email: str) -> bool:
        """校验单个邮箱"""
        if not self.email_pattern.match(email):
            self.errors.append(f"邮箱格式无效: {email}")
            return False
        # 校验域名
        domain = email.split('@')[1]
        if self.domain_list and domain not in self.domain_list:
            self.errors.append(f"不支持的邮箱域名: {domain}")
            return False
        return True
    def validate_record(self, record: Dict, index: int) -> bool:
        """校验包含邮箱的记录"""
        email = record.get('email', '')
        if not email:
            self.errors.append(f"第{index}条记录缺少email字段")
            return False
        return self.validate_email(email)
# 使用示例
validator = EmailValidator()
validator.validate_file('emails.json')
print(validator.errors)

日期格式校验

class DateValidator(DataValidator):
    """日期格式校验器"""
    def __init__(self, date_formats: List[str] = None):
        super().__init__()
        self.date_formats = date_formats or [
            '%Y-%m-%d',
            '%Y/%m/%d',
            '%d-%m-%Y',
            '%Y%m%d'
        ]
    def validate_date(self, date_str: str) -> bool:
        """校验日期字符串"""
        for fmt in self.date_formats:
            try:
                datetime.strptime(date_str, fmt)
                return True
            except ValueError:
                continue
        self.errors.append(f"日期格式无效: {date_str}")
        return False
    def validate_record(self, record: Dict, index: int) -> bool:
        """校验包含日期的记录"""
        valid = True
        # 校验单个日期字段
        if 'date' in record:
            valid &= self.validate_date(record['date'])
        # 校验日期范围
        if 'start_date' in record and 'end_date' in record:
            start = record['start_date']
            end = record['end_date']
            if self.validate_date(start) and self.validate_date(end):
                if start > end:
                    self.errors.append(f"开始日期不能晚于结束日期: {start} > {end}")
                    valid = False
        return valid

批量校验脚本

目录批量校验

import os
from concurrent.futures import ThreadPoolExecutor, as_completed
class BatchValidator:
    """批量文件校验器"""
    def __init__(self, validators: Dict[str, DataValidator] = None):
        self.validators = validators or {
            '.json': DataValidator(),
            '.csv': DataValidator(),
        }
        self.results = {}
    def validate_directory(self, directory: str, recursive: bool = True) -> Dict:
        """校验目录下所有文件"""
        directory = Path(directory)
        if not directory.exists():
            return {"error": f"目录不存在: {directory}"}
        # 获取所有文件
        if recursive:
            files = list(directory.rglob('*'))
        else:
            files = list(directory.glob('*'))
        # 只处理支持的文件类型
        valid_files = [
            f for f in files 
            if f.is_file() and f.suffix.lower() in self.validators
        ]
        # 多线程并行校验
        with ThreadPoolExecutor(max_workers=4) as executor:
            future_to_file = {
                executor.submit(self.validate_single, f): f 
                for f in valid_files
            }
            for future in as_completed(future_to_file):
                file_path = future_to_file[future]
                try:
                    result = future.result()
                    self.results[str(file_path)] = result
                except Exception as e:
                    self.results[str(file_path)] = {"error": str(e)}
        return self.results
    def validate_single(self, file_path: Path) -> Dict:
        """校验单个文件"""
        file_type = file_path.suffix.lower()
        validator = self.validators.get(file_type)
        if not validator:
            return {"error": f"不支持的文件类型: {file_type}"}
        valid = validator.validate_file(file_path)
        return {
            "valid": valid,
            "errors": validator.errors,
            "warnings": validator.warnings
        }

配置驱动的校验器

import yaml
class ConfigDrivenValidator(DataValidator):
    """配置驱动的校验器"""
    def __init__(self, config_path: str):
        super().__init__()
        with open(config_path, 'r', encoding='utf-8') as f:
            self.config = yaml.safe_load(f)
    def validate_record(self, record: Dict, index: int) -> bool:
        """根据配置校验记录"""
        valid = True
        for field, rules in self.config.get('fields', {}).items():
            value = record.get(field)
            # 必填校验
            if rules.get('required', False) and not value:
                self.errors.append(f"第{index}条记录缺少必填字段: {field}")
                valid = False
                continue
            if value is None:
                continue
            # 类型校验
            expected_type = rules.get('type')
            if expected_type:
                type_map = {
                    'string': str,
                    'number': (int, float),
                    'integer': int,
                    'boolean': bool,
                }
                expected = type_map.get(expected_type)
                if expected and not isinstance(value, expected):
                    self.errors.append(
                        f"第{index}条记录字段{field}类型错误: "
                        f"期望{expected_type}, 实际{type(value).__name__}"
                    )
                    valid = False
            # 正则校验
            pattern = rules.get('pattern')
            if pattern and not re.match(pattern, str(value)):
                self.errors.append(
                    f"第{index}条记录字段{field}格式不符合要求: {value}"
                )
                valid = False
            # 长度/范围校验
            if 'min_length' in rules and len(str(value)) < rules['min_length']:
                self.errors.append(
                    f"第{index}条记录字段{field}长度不足: {value}"
                )
                valid = False
            if 'max_length' in rules and len(str(value)) > rules['max_length']:
                self.errors.append(
                    f"第{index}条记录字段{field}长度超限: {value}"
                )
                valid = False
            # 枚举值校验
            if 'enum' in rules and value not in rules['enum']:
                self.errors.append(
                    f"第{index}条记录字段{field}值不在允许范围内: {value}"
                )
                valid = False
        return valid
# 配置文件示例 (validate_config.yaml)
"""
fields:
  name:
    type: string
    required: true
    min_length: 2
    max_length: 50
  age:
    type: integer
    required: true
    min: 0
    max: 150
  email:
    type: string
    pattern: '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
  gender:
    type: string
    enum: ['male', 'female', 'other']
"""
# 使用示例
validator = ConfigDrivenValidator('validate_config.yaml')
validator.validate_file('data.json')
print(f"错误数: {len(validator.errors)}")
for error in validator.errors:
    print(f"  - {error}")

命令行工具

#!/usr/bin/env python3
import argparse
import sys
from pathlib import Path
def main():
    parser = argparse.ArgumentParser(description='批量数据格式校验工具')
    parser.add_argument('path', help='要校验的文件或目录路径')
    parser.add_argument('-t', '--type', choices=['json', 'csv', 'yaml', 'auto'],
                       default='auto', help='文件类型')
    parser.add_argument('-r', '--recursive', action='store_true',
                       help='递归处理子目录')
    parser.add_argument('-o', '--output', help='输出错误报告到文件')
    parser.add_argument('-v', '--verbose', action='store_true',
                       help='详细输出')
    args = parser.parse_args()
    # 创建校验器
    validator = DataValidator()
    batch_validator = BatchValidator()
    path = Path(args.path)
    if path.is_file():
        # 校验单个文件
        result = validator.validate_file(path, args.type)
        if result:
            print(f"[✓] 校验通过: {path}")
        else:
            print(f"[✗] 校验失败: {path}")
            if args.verbose:
                for error in validator.errors:
                    print(f"  Error: {error}")
                for warning in validator.warnings:
                    print(f"  Warning: {warning}")
    elif path.is_dir():
        # 批量校验目录
        results = batch_validator.validate_directory(path, args.recursive)
        # 统计结果
        total = len(results)
        passed = sum(1 for r in results.values() if r.get('valid'))
        failed = total - passed
        print(f"\n校验结果统计:")
        print(f"  总数: {total}")
        print(f"  通过: {passed}")
        print(f"  失败: {failed}")
        if args.verbose:
            for file_path, result in results.items():
                status = "✓" if result.get('valid') else "✗"
                print(f"\n[{status}] {file_path}")
                for error in result.get('errors', []):
                    print(f"  Error: {error}")
    # 输出到文件
    if args.output:
        with open(args.output, 'w', encoding='utf-8') as f:
            json.dump({
                'path': str(path),
                'results': results if path.is_dir() else [{
                    'errors': validator.errors,
                    'warnings': validator.warnings
                }]
            }, f, ensure_ascii=False, indent=2)
        print(f"\n报告已保存到: {args.output}")
if __name__ == '__main__':
    main()

使用示例

# 校验单个文件
python validator.py data.json
# 校验目录下所有文件
python validator.py /path/to/data -r
# 指定文件类型并输出详细错误
python validator.py data.csv -t csv -v
# 输出错误报告
python validator.py /path/to/data -o report.json

这些代码提供了一个完整的批量数据格式校验框架,可以根据实际需求进行扩展和定制。