Data Model Designer — 数据模型设计师

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

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

解决方案 系统性设计建设工程数据模型，定义实体、关系、字段，实现： 项目全生命周期数据管理 多方数据互通互联 数据血缘可追溯

二、技术实现 核心类库

from typing import Dict, Any, List, Optional 
from dataclasses import dataclass, field 
from enum import Enum 
import json 
class DataType(Enum): 
    """数据类型枚举""" 
    STRING = "string"  # 字符串 
    INTEGER = "integer"  # 整数 
    FLOAT = "float"  # 浮点数 
    BOOLEAN = "boolean"  # 布尔值 
    DATE = "date"  # 日期 
    DATETIME = "datetime"  # 日期时间 
    TEXT = "text"  # 长文本 
    DECIMAL = "decimal"  # 精确小数（用于金额） 
class RelationType(Enum): 
    """关系类型枚举""" 
    ONE_TO_ONE = "1:1"  # 一对一 
    ONE_TO_MANY = "1:N"  # 一对多 
    MANY_TO_MANY = "N:M"  # 多对多 
class ConstraintType(Enum): 
    """约束类型枚举""" 
    PRIMARY_KEY = "primary_key"  # 主键 
    FOREIGN_KEY = "foreign_key"  # 外键 
    UNIQUE = "unique"  # 唯一 
    NOT_NULL = "not_null"  # 非空 
@dataclass 
class Field: 
    """字段定义""" 
    name: str  # 字段名 
    data_type: DataType  # 数据类型 
    nullable: bool = True 
    default: Any = None 
    description: str = ""  # 中文描述 
    constraints: List[ConstraintType] = field(default_factory=list) 
@dataclass 
class Entity: 
    """实体定义""" 
    name: str  # 英文表名 
    description: str  # 中文描述 
    fields: List[Field] = field(default_factory=list) 
    primary_key: str = "id" 
@dataclass 
class Relationship: 
    """关系定义""" 
    name: str 
    from_entity: str 
    to_entity: str 
    relation_type: RelationType 
    from_field: str 
    to_field: str 
class CNConstructionDataModel: 
    """中国建设工程数据模型设计器""" 
    def __init__(self, project_name: str): 
        self.project_name = project_name 
        self.entities: Dict[str, Entity] = {} 
        self.relationships: List[Relationship] = [] 
    def add_entity(self, entity: Entity): 
        """添加实体""" 
        self.entities[entity.name] = entity 
    def add_relationship(self, relationship: Relationship): 
        """添加关系""" 
        self.relationships.append(relationship) 
    def create_entity(self, name: str, description: str, fields: List[Dict[str, Any]]) -> Entity: 
        """从字段定义创建实体""" 
        entity_fields = [ 
            Field( 
                name=f['name'], 
                data_type=DataType(f.get('type', 'string')), 
                nullable=f.get('nullable', True), 
                default=f.get('default'), 
                description=f.get('description', ''), 
                constraints=[ConstraintType(c) for c in f.get('constraints', [])] 
            ) for f in fields 
        ] 
        entity = Entity(name=name, description=description, fields=entity_fields) 
        self.add_entity(entity) 
        return entity 
    def create_relationship(self, from_entity: str, to_entity: str, relation_type: str = "1:N", from_field: str = None) -> Relationship: 
        """创建实体间关系""" 
        rel = Relationship( 
            name=f"{from_entity}_{to_entity}", 
            from_entity=from_entity, 
            to_entity=to_entity, 
            relation_type=RelationType(relation_type), 
            from_field=from_field or f"{to_entity.lower()}_id", 
            to_field="id" 
        ) 
        self.add_relationship(rel) 
        return rel 
    def generate_sql_schema(self, dialect: str = "mysql") -> str: 
        """生成SQL DDL语句""" 
        sql = [] 
        type_map = { 
            DataType.STRING: "VARCHAR(255)", 
            DataType.INTEGER: "INT", 
            DataType.FLOAT: "DECIMAL(15,2)", 
            DataType.BOOLEAN: "TINYINT(1)", 
            DataType.DATE: "DATE", 
            DataType.DATETIME: "DATETIME", 
            DataType.TEXT: "TEXT", 
            DataType.DECIMAL: "DECIMAL(18,2)" 
        } 
        for name, entity in self.entities.items(): 
            columns = [] 
            for fld in entity.fields: 
                col = f" {fld.name} {type_map.get(fld.data_type, 'VARCHAR(255)')}" 
                if not fld.nullable: 
                    col += " NOT NULL" 
                if ConstraintType.PRIMARY_KEY in fld.constraints: 
                    col += " PRIMARY KEY" 
                if fld.default is not None: 
                    col += f" DEFAULT {fld.default}" 
                if fld.description: 
                    col += f" COMMENT '{fld.description}'" 
                columns.append(col) 
            sql.append(f"CREATE TABLE {name} (\n" + ",\n".join(columns) + "\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='" + entity.description + "';") 
        for rel in self.relationships: 
            sql.append(f"""ALTER TABLE {rel.from_entity} ADD CONSTRAINT fk_{rel.name} FOREIGN KEY ({rel.from_field}) REFERENCES {rel.to_entity}({rel.to_field});""") 
        return "\n\n".join(sql) 
    def generate_json_schema(self) -> Dict[str, Any]: 
        """生成JSON Schema""" 
        schemas = {} 
        for name, entity in self.entities.items(): 
            # ...