如何编写数据校验脚本

本文目录导读：

1. Python 基础数据校验
2. JSON/CSV 文件数据校验
3. 数据库数据校验
4. Web API 请求数据校验
5. 批量数据校验工具类
6. 命令行工具
7. 最佳实践建议

我来介绍几种常见的数据校验脚本编写方法,针对不同场景：

Python 基础数据校验

import re
from typing import Any, Dict, List, Optional
class DataValidator:
    """数据校验器"""
    @staticmethod
    def validate_email(email: str) -> bool:
        """验证邮箱格式"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email))
    @staticmethod
    def validate_phone(phone: str) -> bool:
        """验证手机号（中国）"""
        pattern = r'^1[3-9]\d{9}$'
        return bool(re.match(pattern, phone))
    @staticmethod
    def validate_id_card(id_card: str) -> bool:
        """验证身份证号"""
        pattern = r'^\d{17}[\dXx]$'
        if not re.match(pattern, id_card):
            return False
        # 校验码验证
        weights = [7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2]
        check_codes = ['1', '0', 'X', '9', '8', '7', '6', '5', '4', '3', '2']
        total = sum(int(id_card[i]) * weights[i] for i in range(17))
        return check_codes[total % 11] == id_card[-1].upper()
# 使用示例
validator = DataValidator()
print(validator.validate_email("test@example.com"))  # True
print(validator.validate_phone("13800138000"))      # True

JSON/CSV 文件数据校验

import json
import csv
from typing import Dict, List
def validate_json_data(file_path: str) -> List[Dict]:
    """校验JSON文件数据"""
    valid_data = []
    errors = []
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    for idx, record in enumerate(data):
        try:
            # 定义校验规则
            required_fields = ['name', 'age', 'email']
            # 检查必填字段
            for field in required_fields:
                if field not in record:
                    raise ValueError(f"缺少必填字段: {field}")
            # 字段类型检查
            if not isinstance(record['age'], int) or record['age'] < 0:
                raise ValueError("年龄必须是正整数")
            if not validate_email(record['email']):
                raise ValueError("邮箱格式错误")
            valid_data.append(record)
        except (ValueError, TypeError) as e:
            errors.append({
                'index': idx,
                'error': str(e),
                'data': record
            })
    return valid_data, errors
# CSV文件校验
def validate_csv_data(file_path: str) -> tuple:
    """校验CSV文件数据"""
    valid_rows = []
    error_rows = []
    with open(file_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row_num, row in enumerate(reader, start=2):  # 从第2行开始
            try:
                # 转换为合适的数据类型
                row['age'] = int(row['age'])
                row['salary'] = float(row['salary'])
                # 业务规则校验
                if row['age'] < 18:
                    raise ValueError("年龄必须大于18岁")
                if row['salary'] <= 0:
                    raise ValueError("薪资必须大于0")
                valid_rows.append(row)
            except (ValueError, KeyError) as e:
                error_rows.append({
                    'row': row_num,
                    'error': str(e),
                    'data': row
                })
    return valid_rows, error_rows

数据库数据校验

import sqlite3
from datetime import datetime
def validate_database_data(db_path: str) -> List[str]:
    """校验数据库数据完整性"""
    issues = []
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    # 检查空值
    cursor.execute("""
        SELECT id, name FROM users 
        WHERE email IS NULL OR name IS NULL
    """)
    null_records = cursor.fetchall()
    for record in null_records:
        issues.append(f"用户 {record[0]} 存在空值字段")
    # 检查重复数据
    cursor.execute("""
        SELECT email, COUNT(*) as count 
        FROM users 
        GROUP BY email 
        HAVING count > 1
    """)
    duplicates = cursor.fetchall()
    for email, count in duplicates:
        issues.append(f"邮箱 {email} 重复 {count} 次")
    # 检查数据一致性
    cursor.execute("""
        SELECT u.id, u.name, o.id
        FROM users u
        LEFT JOIN orders o ON u.id = o.user_id
        WHERE u.status = 'active' AND o.id IS NULL
    """)
    inconsistent = cursor.fetchall()
    for user_id, name, _ in inconsistent:
        issues.append(f"用户 {name}({user_id}) 状态异常")
    conn.close()
    return issues

Web API 请求数据校验

from flask import Flask, request, jsonify
from marshmallow import Schema, fields, validate, ValidationError
app = Flask(__name__)
class UserSchema(Schema):
    """用户数据校验模式"""
    username = fields.String(required=True, validate=validate.Length(min=3, max=50))
    email = fields.Email(required=True)
    age = fields.Integer(validate=validate.Range(min=0, max=150))
    phone = fields.String(validate=validate.Regexp(r'^1[3-9]\d{9}$'))
@app.route('/api/user', methods=['POST'])
def create_user():
    """创建用户API"""
    json_data = request.get_json()
    if not json_data:
        return jsonify({'error': '请提供JSON数据'}), 400
    schema = UserSchema()
    try:
        # 校验并解析数据
        validated_data = schema.load(json_data)
        # 业务逻辑处理
        # save_to_database(validated_data)
        return jsonify({
            'message': '用户创建成功',
            'data': validated_data
        }), 201
    except ValidationError as e:
        return jsonify({
            'error': '数据校验失败',
            'details': e.messages
        }), 400
if __name__ == '__main__':
    app.run(debug=True)

批量数据校验工具类

from typing import Any, Callable, List, Tuple
from dataclasses import dataclass
@dataclass
class ValidationRule:
    """校验规则定义"""
    field: str
    validator: Callable[[Any], bool]
    message: str
class BatchValidator:
    """批量数据校验器"""
    def __init__(self, rules: List[ValidationRule] = None):
        self.rules = rules or []
    def add_rule(self, rule: ValidationRule):
        """添加校验规则"""
        self.rules.append(rule)
    def validate(self, data: dict) -> Tuple[bool, List[str]]:
        """执行校验"""
        errors = []
        for rule in self.rules:
            value = data.get(rule.field)
            if value is None or not rule.validator(value):
                errors.append(rule.message)
        return len(errors) == 0, errors
    def validate_batch(self, dataset: List[dict]) -> Tuple[List[dict], List[dict]]:
        """批量校验数据"""
        valid_data = []
        invalid_data = []
        for item in dataset:
            is_valid, errors = self.validate(item)
            if is_valid:
                valid_data.append(item)
            else:
                invalid_data.append({
                    'data': item,
                    'errors': errors
                })
        return valid_data, invalid_data
# 使用示例
validator = BatchValidator([
    ValidationRule('name', 
                  lambda x: len(x) >= 2,
                  '姓名至少2个字符'),
    ValidationRule('age',
                  lambda x: isinstance(x, int) and 0 <= x <= 150,
                  '年龄必须在0-150之间'),
])
data = {'name': '张三', 'age': 25}
is_valid, errors = validator.validate(data)
print(is_valid, errors)  # True []

命令行工具

#!/usr/bin/env python3
import argparse
import sys
def main():
    """命令行数据校验工具"""
    parser = argparse.ArgumentParser(description='数据校验工具')
    parser.add_argument('file', help='要校验的数据文件')
    parser.add_argument('--type', choices=['csv', 'json', 'txt'],
                        default='csv', help='文件类型')
    parser.add_argument('--rules', help='校验规则配置文件')
    args = parser.parse_args()
    try:
        if args.type == 'csv':
            valid_data, errors = validate_csv_data(args.file)
        elif args.type == 'json':
            valid_data, errors = validate_json_data(args.file)
        print(f"校验完成:")
        print(f"- 有效数据: {len(valid_data)} 条")
        print(f"- 错误数据: {len(errors)} 条")
        if errors:
            print("\n错误详情:")
            for error in errors:
                print(f"  行 {error.get('row', '?')}: {error['error']}")
    except Exception as e:
        print(f"校验失败: {e}", file=sys.stderr)
        sys.exit(1)
if __name__ == '__main__':
    main()

最佳实践建议

1. 分层校验：

   • 语法层：检查数据格式
   • 逻辑层：检查业务规则
   • 完整性层：检查数据依赖

2. 错误处理：

   • 收集所有错误，而非第一个错误就停止
   • 提供清晰的错误信息

3. 性能优化：

   • 大数据量使用异步处理
   • 使用批量操作

4. 可配置性：

   • 将校验规则外部化到配置文件
   • 支持动态添加规则

需要我针对特定场景提供更详细的实现吗？