为建筑数据构建验证规则。针对BIM元素、成本编码及进度数据创建正则表达式与基于逻辑的验证。
验证规则 Builder.Business Case.Problem Results. 构建数据质量挑战:1. 不一致的命名惯例.无效的成本代码和WBS. 缺失
功能概述
验证规则 Builder.Business Case.Problem Results. 构建数据质量挑战:1. 不一致的命名惯例.无效的成本代码和WBS. 缺失是一项面向实际任务的技能。它将相关步骤、工具调用和结果整理方式集中到统一流程中，帮助使用者更快完成目标并减少重复操作。
核心要点
- 使用时应结合输入条件选择合适的执行方式，核对必要参数、依赖环境与输出内容，并按原始要求处理异常情况。
- 该技能适合需要稳定复用相关能力的场景，可作为自动化工作流的一部分，也便于后续检查、调整和扩展。
- 从功能定位来看，该技能强调把分散的操作要求整理成清晰、可复用的处理流程，使用户能够围绕既定目标快速准备输入、选择执行方式并获得结构化结果。
使用与执行
实际使用前应先确认任务范围、数据来源、运行环境、必要权限和关键参数，再依据技能说明逐步执行；若输入条件不完整，应先补齐信息或采用保守配置，避免因错误假设导致结果偏离需求。
结果检查与注意事项
执行过程中需要关注工具调用是否成功、接口或依赖是否可用、输出格式是否符合预期，并对异常提示、缺失字段和边界情况进行处理；涉及批量任务时，还应保存进度，避免中断后重复操作。
校验规则构建器
业务场景
问题陈述
建筑工程数据质量面临的挑战：
- 命名规范不统一
- 成本编码（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
目录
热门AI工具
相关专题
Gradle插件开发入门教程
本专题介绍Gradle插件开发的基本流程，涵盖插件类编写、自定义任务注册、扩展参数设计和插件应用方法，帮助你理解插件如何组织构建逻辑，减少多个项目之间的重复配置。
0
2026.10.22
Gradle Wrapper配置教程
本专题介绍Gradle Wrapper的基本作用，涵盖文件组成、生成方法、版本配置和任务调用，帮助你理解项目如何使用指定的Gradle版本，减少不同开发环境带来的构建差异。
0
2026.10.22
Golangci-lint检查器使用教程
本专题整理Golangci-lint常用Linters的用途与配置方法，覆盖错误处理、无效代码、安全、复杂度、命名和依赖检查，帮助项目合理选规则。
0
2026.10.22
Gradle怎么构建Java项目
本专题介绍Gradle Java项目构建方法，涵盖Java插件配置、依赖声明、源码编译、测试执行和打包检查，帮助你理解各阶段的任务关系，逐步完成项目构建。
0
2026.10.22
PHP时间戳漏洞修复教程
本专题汇总PHP常见时间戳漏洞，包括2038年溢出、时区漂移、时间戳伪造、重放攻击、时间盲注与日志审计失效等，并给出升级64位环境、使用 DateTimeImmutable、统一UTC、严格超时校验、预处理绑定、hash_equals 比对等修复方案，帮助开发者快速排查和加固时间相关代码。
0
2026.10.22
Golangci-lint配置文件与规则设置
本专题聚焦Golangci-lint YAML配置文件，讲清运行参数、规则启停、目录排除、问题过滤及严重级别设置，便于团队统一检查标准。
0
2026.10.22
Golangci-lint安装与入门教程
本专题讲解Golangci-lint的安装、初始化、基础命令和首次代码检查，覆盖本地环境、Docker及多平台用法，帮助Go开发者快速上手。
0
2026.10.22
PHP2Shell高危漏洞修复教程
PHP2Shell漏洞是什么，哪些PHP版本和运行环境会受到影响，服务器应该如何检测与修复？本专题提供版本自查脚本、Linux与Docker升级命令、PHP-FPM和Apache排查步骤，以及修复验证、临时缓解和应急响应方案。
0
2026.10.22
热门下载
相关下载
精品课程
热门推荐
/
最新课程
JavaScript ES5基础线上课程教学
共6课时 | 12.1万人学习
最新ThinkPHP 5.1全球首发视频教程(60天成就PHP大牛线上培训班课)
共79课时 | 161.5万人学习
phpStudy极速入门视频教程
共6课时 | 54.7万人学习
