\r \r

🏗️ 建设工程数据模型设计器\r

\r

面向中国建设工程项目的数据模型设计工具\r 基于 GB/T 50500-2024《建设工程工程量清单计价标准》\r 分类：工程\r \r ---\r \r

一、业务需求\r

\r

问题痛点\r

\r | 问题 | 影响 |\r |------|------|\r | 📦 数据分散在各系统中 | 信息孤岛，难以整合 |\r | 🔀 数据结构不统一 | 各参与方数据无法互通 |\r | 🔗 实体关系缺失 | 无法追溯数据血缘 |\r | 📊 集成困难 | BIM、造价、进度系统割裂 |\r \r

解决方案\r

\r 系统性设计建设工程数据模型，定义实体、关系、字段，实现：\r

• 项目全生命周期数据管理\r
• 多方数据互通互联\r
• 数据血缘可追溯\r \r ---\r \r

二、技术实现\r

\r

核心类库\r

\r

from typing import Dict, Any, List, Optional\r
from dataclasses import dataclass, field\r
from enum import Enum\r
import json\r
\r
\r
class DataType(Enum):\r
    """数据类型枚举"""\r
    STRING = "string"      # 字符串\r
    INTEGER = "integer"   # 整数\r
    FLOAT = "float"       # 浮点数\r
    BOOLEAN = "boolean"  # 布尔值\r
    DATE = "date"         # 日期\r
    DATETIME = "datetime" # 日期时间\r
    TEXT = "text"         # 长文本\r
    DECIMAL = "decimal"   # 精确小数（用于金额）\r
\r
\r
class RelationType(Enum):\r
    """关系类型枚举"""\r
    ONE_TO_ONE = "1:1"    # 一对一\r
    ONE_TO_MANY = "1:N"  # 一对多\r
    MANY_TO_MANY = "N:M" # 多对多\r
\r
\r
class ConstraintType(Enum):\r
    """约束类型枚举"""\r
    PRIMARY_KEY = "primary_key"  # 主键\r
    FOREIGN_KEY = "foreign_key"  # 外键\r
    UNIQUE = "unique"           # 唯一\r
    NOT_NULL = "not_null"       # 非空\r
\r
\r
@dataclass\r
class Field:\r
    """字段定义"""\r
    name: str              # 字段名\r
    data_type: DataType   # 数据类型\r
    nullable: bool = True\r
    default: Any = None\r
    description: str = ""  # 中文描述\r
    constraints: List[ConstraintType] = field(default_factory=list)\r
\r
\r
@dataclass\r
class Entity:\r
    """实体定义"""\r
    name: str              # 英文表名\r
    description: str       # 中文描述\r
    fields: List[Field] = field(default_factory=list)\r
    primary_key: str = "id"\r
\r
\r
@dataclass\r
class Relationship:\r
    """关系定义"""\r
    name: str\r
    from_entity: str\r
    to_entity: str\r
    relation_type: RelationType\r
    from_field: str\r
    to_field: str\r
\r
\r
class CNConstructionDataModel:\r
    """中国建设工程数据模型设计器"""\r
\r
    def __init__(self, project_name: str):\r
        self.project_name = project_name\r
        self.entities: Dict[str, Entity] = {}\r
        self.relationships: List[Relationship] = []\r
\r
    def add_entity(self, entity: Entity):\r
        """添加实体"""\r
        self.entities[entity.name] = entity\r
\r
    def add_relationship(self, relationship: Relationship):\r
        """添加关系"""\r
        self.relationships.append(relationship)\r
\r
    def create_entity(self, name: str, description: str,\r
                      fields: List[Dict[str, Any]]) -> Entity:\r
        """从字段定义创建实体"""\r
        entity_fields = [\r
            Field(\r
                name=f['name'],\r
                data_type=DataType(f.get('type', 'string')),\r
                nullable=f.get('nullable', True),\r
                default=f.get('default'),\r
                description=f.get('description', ''),\r
                constraints=[ConstraintType(c) for c in f.get('constraints', [])]\r
            )\r
            for f in fields\r
        ]\r
        entity = Entity(name=name, description=description, fields=entity_fields)\r
        self.add_entity(entity)\r
        return entity\r
\r
    def create_relationship(self, from_entity: str, to_entity: str,\r
                           relation_type: str = "1:N",\r
                           from_field: str = None) -> Relationship:\r
        """创建实体间关系"""\r
        rel = Relationship(\r
            name=f"{from_entity}_{to_entity}",\r
            from_entity=from_entity,\r
            to_entity=to_entity,\r
            relation_type=RelationType(relation_type),\r
            from_field=from_field or f"{to_entity.lower()}_id",\r
            to_field="id"\r
        )\r
        self.add_relationship(rel)\r
        return rel\r
\r
    def generate_sql_schema(self, dialect: str = "mysql") -> str:\r
        """生成SQL DDL语句"""\r
        sql = []\r
        type_map = {\r
            DataType.STRING: "VARCHAR(255)",\r
            DataType.INTEGER: "INT",\r
            DataType.FLOAT: "DECIMAL(15,2)",\r
            DataType.BOOLEAN: "TINYINT(1)",\r
            DataType.DATE: "DATE",\r
            DataType.DATETIME: "DATETIME",\r
            DataType.TEXT: "TEXT",\r
            DataType.DECIMAL: "DECIMAL(18,2)"\r
        }\r
\r
        for name, entity in self.entities.items():\r
            columns = []\r
            for fld in entity.fields:\r
                col = f"    `{fld.name}` {type_map.get(fld.data_type, 'VARCHAR(255)')}"\r
                if not fld.nullable:\r
                    col += " NOT NULL"\r
                if ConstraintType.PRIMARY_KEY in fld.constraints:\r
                    col += " PRIMARY KEY"\r
                if fld.default is not None:\r
                    col += f" DEFAULT {fld.default}"\r
                if fld.description:\r
                    col += f" COMMENT '{fld.description}'"\r
                columns.append(col)\r
\r
            sql.append(f"CREATE TABLE `{name}` (\
" + ",\
".join(columns) + "\
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='" + entity.description + "';")\r
\r
        for rel in self.relationships:\r
            sql.append(f"""ALTER TABLE `{rel.from_entity}`\r
ADD CONSTRAINT `fk_{rel.name}`\r
FOREIGN KEY (`{rel.from_field}`) REFERENCES `{rel.to_entity}`(`{rel.to_field}`);""")\r
\r
        return "\
\
".join(sql)\r
\r
    def generate_json_schema(self) -> Dict[str, Any]:\r
        """生成JSON Schema"""\r
        schemas = {}\r
        for name, entity in self.entities.items():\r
            properties = {}\r
            required = []\r
\r
            for fld in entity.fields:\r
                prop = {"description": fld.description}\r
                if fld.data_type == DataType.STRING:\r
                    prop["type"] = "string"\r
                elif fld.data_type == DataType.INTEGER:\r
                    prop["type"] = "integer"\r
                elif fld.data_type == DataType.FLOAT:\r
                    prop["type"] = "number"\r
                elif fld.data_type == DataType.BOOLEAN:\r
                    prop["type"] = "boolean"\r
                else:\r
                    prop["type"] = "string"\r
\r
                properties[fld.name] = prop\r
                if not fld.nullable:\r
                    required.append(fld.name)\r
\r
            schemas[name] = {\r
                "type": "object",\r
                "title": entity.description,\r
                "properties": properties,\r
                "required": required\r
            }\r
        return schemas\r
\r
    def generate_er_diagram(self) -> str:\r
        """生成Mermaid ER图"""\r
        lines = ["erDiagram"]\r
        for name, entity in self.entities.items():\r
            lines.append(f"    {name} {{")\r
            for fld in entity.fields[:6]:\r
                ftype = {\r
                    DataType.STRING: "VARCHAR",\r
                    DataType.INTEGER: "INT",\r
                    DataType.FLOAT: "DEC",\r
                    DataType.DECIMAL: "DEC",\r
                    DataType.DATE: "DATE",\r
                    DataType.DATETIME: "DT"\r
                }.get(fld.data_type, "STR")\r
                lines.append(f"        {ftype} {fld.name}")\r
            lines.append("    }")\r
\r
        for rel in self.relationships:\r
            rel_symbol = {\r
                RelationType.ONE_TO_ONE: "||--||",\r
                RelationType.ONE_TO_MANY: "||--o{",\r
                RelationType.MANY_TO_MANY: "}o--o{"\r
            }.get(rel.relation_type, "||--o{")\r
            lines.append(f"    {rel.from_entity} {rel_symbol} {rel.to_entity} : \"{rel.name}\"")\r
\r
        return "\
".join(lines)\r
\r
    def validate_model(self) -> List[str]:\r
        """验证数据模型"""\r
        issues = []\r
        for rel in self.relationships:\r
            if rel.from_entity not in self.entities:\r
                issues.append(f"❌ 缺少实体: {rel.from_entity}")\r
            if rel.to_entity not in self.entities:\r
                issues.append(f"❌ 缺少实体: {rel.to_entity}")\r
\r
        for name, entity in self.entities.items():\r
            has_pk = any(ConstraintType.PRIMARY_KEY in f.constraints for f in entity.fields)\r
            if not has_pk:\r
                issues.append(f"⚠️ 实体 '{name}' 缺少主键")\r
\r
        if not issues:\r
            issues.append("✅ 数据模型验证通过")\r
        return issues\r
\r
\r
class CNConstructionEntities:\r
    """中国建设工程标准实体库"""\r
\r
    @staticmethod\r
    def project_entity() -> Entity:\r
        """工程项目实体"""\r
        return Entity(\r
            name="engineering_projects",\r
            description="工程项目主表",\r
            fields=[\r
                Field("id", DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),\r
                Field("project_code", DataType.STRING, False, description="项目编码", constraints=[ConstraintType.UNIQUE]),\r
                Field("project_name", DataType.STRING, False, description="项目名称"),\r
                Field("project_type", DataType.STRING, False, description="项目类型（房屋建筑/市政/公路等）"),\r
                Field("construction_address", DataType.STRING, description="建设地点"),\r
                Field("total_investment", DataType.DECIMAL, description="总投资额（元）"),\r
                Field("planned_area", DataType.FLOAT, description="建筑面积（㎡）"),\r
                Field("structure_type", DataType.STRING, description="结构类型"),\r
                Field("building_height", DataType.FLOAT, description="建筑高度（m）"),\r
                Field("basement_area", DataType.FLOAT, description="地下室面积（㎡）"),\r
                Field("construction_period", DataType.INTEGER, description="工期（天）"),\r
                Field("start_date", DataType.DATE, description="开工日期"),\r
                Field("completion_date", DataType.DATE, description="竣工日期"),\r
                Field("status", DataType.STRING, description="项目状态"),\r
                Field("construction_unit", DataType.STRING, description="建设单位"),\r
                Field("design_unit", DataType.STRING, description="设计单位"),\r
                Field("supervision_unit", DataType.STRING, description="监理单位"),\r
                Field("create_time", DataType.DATETIME, description="创建时间"),\r
            ]\r
        )\r
\r
    @staticmethod\r
    def bill_item_entity() -> Entity:\r
        """工程量清单实体"""\r
        return Entity(\r
            name="bill_items",\r
            description="工程量清单项目",\r
            fields=[\r
                Field("id", DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),\r
                Field("project_id", DataType.INTEGER, False),\r
                Field("bill_no", DataType.STRING, False, description="清单编码"),\r
                Field("item_code", DataType.STRING, description="项目编码（GB/T标准）"),\r
                Field("item_name", DataType.STRING, False, description="项目名称"),\r
                Field("project_unit", DataType.STRING, description="计量单位"),\r
                Field("quantity", DataType.DECIMAL, description="工程量"),\r
                Field("unit_price", DataType.DECIMAL, description="综合单价（元）"),\r
                Field("total_price", DataType.DECIMAL, description="合价（元）"),\r
                Field("tax_rate", DataType.FLOAT, description="税率"),\r
                Field("remarks", DataType.TEXT, description="备注"),\r
            ]\r
        )\r
\r
    @staticmethod\r
    def cost_control_entity() -> Entity:\r
        """造价管控实体"""\r
        return Entity(\r
            name="cost_control",\r
            description="造价管控记录",\r
            fields=[\r
                Field("id", DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),\r
                Field("project_id", DataType.INTEGER, False),\r
                Field("cost_stage", DataType.STRING, False, description="阶段（估算/概算/预算/结算）"),\r
                Field("target_cost", DataType.DECIMAL, description="目标成本（元）"),\r
                Field("actual_cost", DataType.DECIMAL, description="实际成本（元）"),\r
                Field("budget_cost", DataType.DECIMAL, description="预算成本（元）"),\r
                Field("change_amount", DataType.DECIMAL, description="变更金额（元）"),\r
                Field("cost_diff", DataType.DECIMAL, description="成本偏差（元）"),\r
                Field("cost_ratio", DataType.FLOAT, description="成本偏差率（%）"),\r
                Field("record_date", DataType.DATE, description="记录日期"),\r
            ]\r
        )\r
\r
    @staticmethod\r
    def schedule_activity_entity() -> Entity:\r
        """进度计划实体"""\r
        return Entity(\r
            name="schedule_activities",\r
            description="进度计划活动",\r
            fields=[\r
                Field("id", DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),\r
                Field("project_id", DataType.INTEGER, False),\r
                Field("wbs_code", DataType.STRING, description="WBS编码"),\r
                Field("activity_name", DataType.STRING, False, description="活动名称"),\r
                Field("planned_start", DataType.DATE, description="计划开始日期"),\r
                Field("planned_end", DataType.DATE, description="计划结束日期"),\r
                Field("actual_start", DataType.DATE, description="实际开始日期"),\r
                Field("actual_end", DataType.DATE, description="实际结束日期"),\r
                Field("progress_percent", DataType.FLOAT, description="完成百分比（%）"),\r
                Field("前置活动", DataType.STRING, description="前置活动ID"),\r
                Field("responsible_party", DataType.STRING, description="责任单位"),\r
            ]\r
        )\r
\r
    @staticmethod\r
    def change_order_entity() -> Entity:\r
        """工程变更实体"""\r
        return Entity(\r
            name="change_orders",\r
            description="工程变更单",\r
            fields=[\r
                Field("id", DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),\r
                Field("project_id", DataType.INTEGER, False),\r
                Field("change_no", DataType.STRING, description="变更编号"),\r
                Field("change_type", DataType.STRING, description="变更类型"),\r
                Field("change_reason", DataType.TEXT, description="变更原因"),\r
                Field("original_amount", DataType.DECIMAL, description="原金额（元）"),\r
                Field("change_amount", DataType.DECIMAL, description="变更金额（元）"),\r
                Field("new_amount", DataType.DECIMAL, description="新金额（元）"),\r
                Field("change_date", DataType.DATE, description="变更日期"),\r
                Field("approval_status", DataType.STRING, description="审批状态"),\r
                Field("approver", DataType.STRING, description="审批人"),\r
            ]\r
        )\r
```\r
\r
---\r
\r
## 三、快速开始\r
\r
### 示例代码\r
\r
```python\r
# 创建模型\r
model = CNConstructionDataModel("某学校建设项目")\r
\r
# 添加标准实体\r
model.add_entity(CNConstructionEntities.project_entity())\r
model.add_entity(CNConstructionEntities.bill_item_entity())\r
model.add_entity(CNConstructionEntities.cost_control_entity())\r
\r
# 添加关系\r
model.create_relationship("bill_items", "engineering_projects")\r
model.create_relationship("cost_control", "engineering_projects")\r
\r
# 生成SQL\r
sql = model.generate_sql_schema("mysql")\r
print(sql)\r
```\r
\r
---\r
\r
## 四、常用场景\r
\r
### 场景1：自定义清单实体\r
\r
```python\r
model.create_entity(\r
    name="safety_inspection",\r
    description="安全检查记录",\r
    fields=[\r
        {"name": "id", "type": "integer", "nullable": False, "constraints": ["primary_key"]},\r
        {"name": "project_id", "type": "integer", "nullable": False},\r
        {"name": "inspection_date", "type": "date"},\r
        {"name": "inspector", "type": "string"},\r
        {"name": "issue_level", "type": "string", "description": "问题等级"},\r
        {"name": "description", "type": "text", "description": "问题描述"},\r
        {"name": "rectification_deadline", "type": "date", "description": "整改期限"},\r
        {"name": "rectification_status", "type": "string", "description": "整改状态"}\r
    ]\r
)\r
```\r
\r
### 场景2：生成ER图\r
\r
```python\r
er_diagram = model.generate_er_diagram()\r
print(er_diagram)\r
# 可复制到 https://mermaid.live 生成可视化图形\r
```\r
\r
### 场景3：模型验证\r
\r
```python\r
issues = model.validate_model()\r
for issue in issues:\r
    print(issue)\r
```\r
\r
---\r
\r
## 五、支持的数据库\r
\r
| 数据库 | SQL方言 | 特点 |\r
|--------|---------|------|\r
| MySQL | `mysql` | 默认，推荐国内项目 |\r
| PostgreSQL | `postgresql` | 适合大数据量 |\r
| SQL Server | `mssql` | 适合Windows环境 |\r
| SQLite | `sqlite` | 适合单机应用 |\r
\r
---\r
\r
## 六、符合的标准规范\r
\r
- GB/T 50500-2024《建设工程工程量清单计价标准》\r
- GB/T 51262-2017《建筑工程施工质量验收统一标准》\r
- GB/T 50328-2019《建设工程文件归档规范》\r
- DBJ/T 15-xxx 广东省标准系列\r
\r
---\r
\r
*本工具由度量衡智库出品，专为建设工程数字化管理设计。*\r