校验规则构建器

业务场景

问题陈述

建筑工程数据质量面临的挑战：

• 命名规范不统一
• 成本编码（cost code）和工作分解结构（WBS）无效
• 数据缺失或格式异常
• BIM 元素不符合规范要求

解决方案

基于规则的校验引擎，采用正则表达式（RegEx）与逻辑规则，确保跨建筑工程系统中的数据质量。

技术实现

import re
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from datetime import date


class RuleType(Enum):
    REGEX = "regex"
    RANGE = "range"
    ENUM = "enum"
    CUSTOM = "custom"
    REQUIRED = "required"
    DATE = "date"
    REFERENCE = "reference"


class Severity(Enum):
    ERROR = "error"
    WARNING = "warning"
    INFO = "info"


@dataclass
class ValidationResult:
    field: str
    is_valid: bool
    message: str
    severity: Severity
    value: Any = None


@dataclass
class ValidationRule:
    name: str
    field: str
    rule_type: RuleType
    pattern: str = ""
    min_value: float = None
    max_value: float = None
    allowed_values: List[Any] = field(default_factory=list)
    custom_func: Callable = None
    severity: Severity = Severity.ERROR
    message: str = ""
    enabled: bool = True


class ValidationRulesBuilder:
    """构建并执行建筑工程数据的校验规则。"""

    # 工程建设数据预定义的正则模式
    PATTERNS = {
        'wbs_code': r'^[0-9]{2}\.[0-9]{2}\.[0-9]{2}(\.[0-9]{2})?$',
        'cost_code': r'^[A-Z]{1,3}-[0-9]{3,6}$',
        'activity_id': r'^[A-Z]{1,3}[0-9]{4,6}$',
        'drawing_number': r'^[A-Z]{1,2}-[0-9]{3}-[A-Z0-9]{2,4}$',
        'specification_section': r'^[0-9]{2}s?[0-9]{2}s?[0-9]{2}(\.[0-9]{2})?$',
        'level_name': r'^(Level|L|FL)s?[-_]?s?([0-9]{1,3}|B[0-9]|R|G|M)$',
        'grid_line': r'^[A-Z].?[0-9]?$|^[0-9]{1,2}.?[A-Z]?$',
        'revision': r'^[A-Z]$|^[0-9]{1,2}$|^Rev\.?s?[A-Z0-9]+$',
        'date_iso': r'^\d{4}-\d{2}-\d{2}$',
        'email': r'^[\w.-]+@[\w.-]+\.\w+$',
        'phone': r'^\+?[0-9]{1,3}[-.\s]?[0-9]{3,4}[-.\s]?[0-9]{4}$',
    }

    def __init__(self):
        self.rules: List[ValidationRule] = []
        self.custom_patterns: Dict[str, str] = {}

    def add_regex_rule(self,
                       name: str,
                       field: str,
                       pattern: str,
                       message: str = "",
                       severity: Severity = Severity.ERROR) -> 'ValidationRulesBuilder':
        """添加正则表达式校验规则。"""

        self.rules.append(ValidationRule(
            name=name,
            field=field,
            rule_type=RuleType.REGEX,
            pattern=pattern,
            message=message or f"字段 '{field}' 不符合正则模式",
            severity=severity
        ))
        return self

    def add_range_rule(self,
                       name: str,
                       field: str,
                       min_value: float = None,
                       max_value: float = None,
                       message: str = "",
                       severity: Severity = Severity.ERROR) -> 'ValidationRulesBuilder':
        """添加数值范围校验规则。"""

        self.rules.append(ValidationRule(
            name=name,
            field=field,
            rule_type=RuleType.RANGE,
            min_value=min_value,
            max_value=max_value,
            message=message or f"字段 '{field}' 超出范围 [{min_value}, {max_value}]",
            severity=severity
        ))
        return self

    def add_enum_rule(self,
                      name: str,
                      field: str,
                      allowed_values: List[Any],
                      message: str = "",
                      severity: Severity = Severity.ERROR) -> 'ValidationRulesBuilder':
        """添加枚举值校验规则。"""

        self.rules.append(ValidationRule(
            name=name,
            field=field,
            rule_type=RuleType.ENUM,
            allowed_values=allowed_values,
            message=message or f"字段 '{field}' 必须为以下值之一：{allowed_values}",
            severity=severity
        ))
        return self

    def add_required_rule(self,
                          name: str,
                          field: str,
                          message: str = "",
                          severity: Severity = Severity.ERROR) -> 'ValidationRulesBuilder':
        """添加必填字段校验规则。"""

        self.rules.append(ValidationRule(
            name=name,
            field=field,
            rule_type=RuleType.REQUIRED,
            message=message or f"字段 '{field}' 为必填项",
            severity=severity
        ))
        return self

    def add_custom_rule(self,
                        name: str,
                        field: str,
                        func: Callable[[Any], bool],
                        message: str = "",
                        severity: Severity = Severity.ERROR) -> 'ValidationRulesBuilder':
        """添加自定义校验函数。"""

        self.rules.append(ValidationRule(
            name=name,
            field=field,
            rule_type=RuleType.CUSTOM,
            custom_func=func,
            message=message or f"字段 '{field}' 自定义校验失败",
            severity=severity
        ))
        return self

    def add_pattern(self, name: str, pattern: str):
        """添加可复用的自定义正则模式。"""
        self.custom_patterns[name] = pattern

    def use_pattern(self,
                    rule_name: str,
                    field: str,
                    pattern_name: str,
                    message: str = "",
                    severity: Severity = Severity.ERROR) -> 'ValidationRulesBuilder':
        """使用预定义或自定义正则模式。"""

        pattern = self.custom_patterns.get(pattern_name) or self.PATTERNS.get(pattern_name)
        if not pattern:
            raise ValueError(f"未找到正则模式 '{pattern_name}'")

        return self.add_regex_rule(rule_name, field, pattern, message, severity)

    def validate_record(self, record: Dict[str, Any]) -> List[ValidationResult]:
        """对单条记录应用全部校验规则。"""

        results = []

        for rule in self.rules:
            if not rule.enabled:
                continue

            value = record.get(rule.field)
            result = self._apply_rule(rule, value)
            results.append(result)

        return results

    def validate_records(self, records: List[Dict[str, Any]]) -> Dict[str, Any]:
        """批量校验多条记录，并返回汇总结果。"""

        all_results = []
        error_count = 0
        warning_count = 0

        for i, record in enumerate(records):
            record_results = self.validate_record(record)
            for result in record_results:
                if not result.is_valid:
                    result_dict = {
                        'record_index': i,
                        'field': result.field,
                        'message': result.message,
                        'severity': result.severity.value,
                        'value': result.value
                    }
                    all_results.append(result_dict)

                    if result.severity == Severity.ERROR:
                        error_count += 1
                    elif result.severity == Severity.WARNING:
                        warning_count += 1

        return {
            'total_records': len(records),
            'valid_records': len(records) - len(set(r['record_index'] for r in all_results if r['severity'] == 'error')),
            'error_count': error_count,
            'warning_count': warning_count,
            'issues': all_results
        }

    def _apply_rule(self, rule: ValidationRule, value: Any) -> ValidationResult:
        """应用单条校验规则。"""

        if rule.rule_type == RuleType.REQUIRED:
            is_valid = value is not None and value != "" and value != []
            return ValidationResult(
                field=rule.field,
                is_valid=is_valid,
                message="" if is_valid else rule.message,
                severity=rule.severity,
                value=value
            )

        # 若值为空（None/空字符串），跳过其余校验
        if value is None or value == "":
            return ValidationResult(
                field=rule.field,
                is_valid=True,
                message="",
                severity=rule.severity,
                value=value
            )

        if rule.rule_type == RuleType.REGEX:
            is_valid = bool(re.match(rule.pattern, str(value)))

        elif rule.rule_type == RuleType.RANGE:
            try:
                num_value = float(value)
                is_valid = True
                if rule.min_value is not None and num_value < rule.min_value:
                    is_valid = False
                if rule.max_value is not None and num_value > rule.max_value:
                    is_valid = False
            except (ValueError, TypeError):
                is_valid = False

        elif rule.rule_type == RuleType.ENUM:
            is_valid = value in rule.allowed_values

        elif rule.rule_type == RuleType.CUSTOM:
            try:
                is_valid = rule.custom_func(value)
            except Exception:
                is_valid = False

        else:
            is_valid = True

        return ValidationResult(
            field=rule.field,
            is_valid=is_valid,
            message="" if is_valid else rule.message,
            severity=rule.severity,
            value=value
        )

    def get_rules_summary(self) -> List[Dict]:
        """获取所有校验规则的摘要信息。"""

        return [{
            'name': r.name,
            'field': r.field,
            'type': r.rule_type.value,
            'severity': r.severity.value,
            'enabled': r.enabled
        } for r in self.rules]


# 工程建设领域专用校验器
class ConstructionValidators:
    """面向工程建设数据的预置校验器。"""

    @staticmethod
    def wbs_validator() -> ValidationRulesBuilder:
        """WBS 编码校验器。"""

        return (ValidationRulesBuilder()
            .add_required_rule("wbs_required", "wbs_code")
            .use_pattern("wbs_format", "wbs_code", "wbs_code", "WBS 格式无效（应为 XX.XX.XX）")
        )

    @staticmethod
    def cost_item_validator() -> ValidationRulesBuilder:
        """成本项校验器。"""

        return (ValidationRulesBuilder()
            .add_required_rule("code_required", "cost_code")
            .add_required_rule("desc_required", "description")
            .use_pattern("code_format", "cost_code", "cost_code")
            .add_range_rule("quantity_positive", "quantity", min_value=0)
            .add_range_rule("unit_cost_positive", "unit_cost", min_value=0)
            .add_enum_rule("unit_valid", "unit", ["EA", "LF", "SF", "CY", "TON", "HR", "LS"])
        )

    @staticmethod
    def schedule_activity_validator() -> ValidationRulesBuilder:
        """进度计划活动校验器。"""

        def dates_valid(record):
            start = record.get('start_date')
            end = record.get('end_date')
            if start and end:
                return start <= end
            return True

        return (ValidationRulesBuilder()
            .add_required_rule("id_required", "activity_id")
            .add_required_rule("name_required", "activity_name")
            .use_pattern("id_format", "activity_id", "activity_id")
            .add_range_rule("duration_positive", "duration", min_value=0)
            .add_range_rule("progress_range", "percent_complete", min_value=0, max_value=100)
        )

    @staticmethod
    def bim_element_validator() -> ValidationRulesBuilder:
        """BIM 元素校验器。"""

        return (ValidationRulesBuilder()
            .add_required_rule("guid_required", "element_guid")
            .add_required_rule("type_required", "element_type")
            .add_required_rule("level_required", "level")
            .use_pattern("level_format", "level", "level_name", severity=Severity.WARNING)
            .add_enum_rule("status_valid", "status",
                          ["New", "Existing", "Demolished", "Temporary"])
        )

快速上手

# 创建校验器实例
validator = ValidationRulesBuilder()

# 添加校验规则
validator.add_required_rule("id_required", "item_id")
validator.use_pattern("wbs_valid", "wbs_code", "wbs_code")
validator.add_range_rule("cost_range", "total_cost", min_value=0, max_value=10000000)
validator.add_enum_rule("status_valid", "status", ["Active", "Completed", "Cancelled"])

# 批量校验记录
records = [
    {"item_id": "001", "wbs_code": "01.02.03", "total_cost": 50000, "status": "Active"},
    {"item_id": "", "wbs_code": "invalid", "total_cost": -100, "status": "Unknown"}
]

results = validator.validate_records(records)
print(f"有效记录数：{results['valid_records']}/{results['total_records']}")
print(f"错误数：{results['error_count']}，警告数：{results['warning_count']}")

常见使用场景

1. 成本数据校验

cost_validator = ConstructionValidators.cost_item_validator()
results = cost_validator.validate_records(cost_items)

2. 进度计划校验

schedule_validator = ConstructionValidators.schedule_activity_validator()
results = schedule_validator.validate_records(activities)

3. BIM 元素校验

bim_validator = ConstructionValidators.bim_element_validator()
results = bim_validator.validate_records(elements)

参考资料

• DDC Book：第 2.6 章 — 数据质量要求
• 官网：https://datadrivenconstruction.io