ai-agent-admin/backend-fastapi/core/database_manager/ddl_builder.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
DDL 生成器：canonical 字段类型 → 多方言 CREATE TABLE / SCHEMA / INDEX / COMMENT

与前端 web/apps/web-ele/src/utils/database-types.ts 的 mapToDbType 规则对齐。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

from core.database_manager.sql_utils import quote_identifier, quote_table

# 表单工作流系统字段（设计节点与 DDL 生成共用）
SYSTEM_FIELDS: List[Dict[str, Any]] = [
    {
        'name': 'id',
        'type': 'varchar',
        'maxLength': 36,
        'comment': '主键ID',
        'nullable': False,
        'isPrimaryKey': True,
    },
    {
        'name': 'sys_create_datetime',
        'type': 'datetime',
        'comment': '创建时间',
        'nullable': True,
        'isPrimaryKey': False,
    },
    {
        'name': 'sys_update_datetime',
        'type': 'datetime',
        'comment': '更新时间',
        'nullable': True,
        'isPrimaryKey': False,
    },
    {
        'name': 'sys_creator_id',
        'type': 'varchar',
        'maxLength': 36,
        'comment': '创建人ID',
        'nullable': True,
        'isPrimaryKey': False,
    },
    {
        'name': 'sys_modifier_id',
        'type': 'varchar',
        'maxLength': 36,
        'comment': '修改人ID',
        'nullable': True,
        'isPrimaryKey': False,
    },
    {
        'name': 'sys_dept_id',
        'type': 'varchar',
        'maxLength': 36,
        'comment': '部门ID',
        'nullable': True,
        'isPrimaryKey': False,
    },
    {
        'name': 'is_deleted',
        'type': 'boolean',
        'comment': '是否删除',
        'nullable': False,
        'isPrimaryKey': False,
    },
    {
        'name': 'sort',
        'type': 'int',
        'comment': '排序',
        'nullable': False,
        'isPrimaryKey': False,
    },
]

SYSTEM_INDEX_FIELDS = frozenset({
    'sys_update_datetime',
    'sys_create_datetime',
    'sys_creator_id',
    'sys_dept_id',
    'is_deleted',
})

CANONICAL_TYPE_MAPPING: Dict[str, str] = {
    'string': 'varchar',
    'str': 'varchar',
    'varchar': 'varchar',
    'char': 'char',
    'text': 'text',
    'int': 'int',
    'integer': 'int',
    'bigint': 'bigint',
    'smallint': 'smallint',
    'decimal': 'decimal',
    'numeric': 'decimal',
    'float': 'float',
    'double': 'double',
    'datetime': 'datetime',
    'timestamp': 'datetime',
    'date': 'date',
    'time': 'time',
    'boolean': 'boolean',
    'bool': 'boolean',
    'json': 'json',
    'jsonb': 'jsonb',
}

COMMON_TYPE_TO_DB_TYPE: Dict[str, Dict[str, str]] = {
    'postgresql': {
        'int': 'INTEGER',
        'bigint': 'BIGINT',
        'smallint': 'SMALLINT',
        'float': 'REAL',
        'double': 'DOUBLE PRECISION',
        'datetime': 'TIMESTAMP',
        'boolean': 'BOOLEAN',
        'json': 'JSON',
        'jsonb': 'JSONB',
        'varchar': 'VARCHAR',
        'char': 'CHAR',
        'text': 'TEXT',
        'decimal': 'DECIMAL',
        'numeric': 'NUMERIC',
        'date': 'DATE',
        'time': 'TIME',
    },
    'mysql': {
        'int': 'INT',
        'bigint': 'BIGINT',
        'smallint': 'SMALLINT',
        'float': 'FLOAT',
        'double': 'DOUBLE',
        'datetime': 'DATETIME',
        'boolean': 'TINYINT(1)',
        'json': 'JSON',
        'jsonb': 'JSON',
        'varchar': 'VARCHAR',
        'char': 'CHAR',
        'text': 'TEXT',
        'decimal': 'DECIMAL',
        'numeric': 'DECIMAL',
        'date': 'DATE',
        'time': 'TIME',
    },
    'sqlserver': {
        'int': 'INT',
        'bigint': 'BIGINT',
        'smallint': 'SMALLINT',
        'float': 'FLOAT',
        'double': 'FLOAT',
        'datetime': 'DATETIME2',
        'boolean': 'BIT',
        'json': 'NVARCHAR(MAX)',
        'jsonb': 'NVARCHAR(MAX)',
        'varchar': 'NVARCHAR',
        'char': 'NCHAR',
        'text': 'NVARCHAR(MAX)',
        'decimal': 'DECIMAL',
        'numeric': 'NUMERIC',
        'date': 'DATE',
        'time': 'TIME',
    },
    'oracle': {
        'int': 'NUMBER',
        'bigint': 'NUMBER',
        'smallint': 'NUMBER',
        'float': 'BINARY_FLOAT',
        'double': 'BINARY_DOUBLE',
        'datetime': 'TIMESTAMP',
        'boolean': 'NUMBER(1)',
        'json': 'CLOB',
        'jsonb': 'CLOB',
        'varchar': 'VARCHAR2',
        'char': 'CHAR',
        'text': 'CLOB',
        'decimal': 'NUMBER',
        'numeric': 'NUMBER',
        'date': 'DATE',
        'time': 'TIMESTAMP',
    },
}


def normalize_db_type(db_type: str) -> str:
    db = (db_type or 'postgresql').lower().strip()
    if db in ('mssql', 'sql server'):
        return 'sqlserver'
    if db in ('postgres', 'psql'):
        return 'postgresql'
    return db


def normalize_canonical_type(field_type: str) -> str:
    if not field_type:
        return 'varchar'
    return CANONICAL_TYPE_MAPPING.get(field_type.lower().strip(), field_type.lower().strip())


def process_canonical_field(field: dict) -> dict:
    """标准化单个 canonical 字段配置（设计节点使用）。"""
    field_name = field.get('name', '')
    standard_type = normalize_canonical_type(field.get('type', 'varchar'))

    processed: Dict[str, Any] = {
        'name': field_name,
        'type': standard_type,
        'comment': field.get('comment', field_name),
        'nullable': field.get('nullable', True),
        'isPrimaryKey': field.get('isPrimaryKey', False),
    }

    if standard_type in ('varchar', 'char'):
        processed['maxLength'] = field.get('maxLength', 255)
    elif standard_type in ('decimal', 'numeric'):
        processed['precision'] = field.get('precision', 10)
        processed['scale'] = field.get('scale', 2)

    return processed


def map_to_dialect_type(field: dict, db_type: str) -> str:
    """将 canonical 字段映射为带长度/精度的方言类型 SQL 片段。"""
    db = normalize_db_type(db_type)
    field_type = normalize_canonical_type(field.get('type', 'varchar'))
    type_map = COMMON_TYPE_TO_DB_TYPE.get(db, COMMON_TYPE_TO_DB_TYPE['postgresql'])
    base_type = type_map.get(field_type, field_type.upper())

    if field_type == 'varchar' and base_type in ('VARCHAR', 'NVARCHAR', 'VARCHAR2'):
        length = field.get('maxLength', 255)
        return f'{base_type}({length})'
    if field_type == 'char' and base_type in ('CHAR', 'NCHAR'):
        length = field.get('maxLength', 10)
        return f'{base_type}({length})'
    if field_type in ('decimal', 'numeric') and base_type in ('DECIMAL', 'NUMERIC', 'NUMBER'):
        precision = field.get('precision', 10)
        scale = field.get('scale', 2)
        if db == 'oracle':
            return f'NUMBER({precision}, {scale})'
        return f'{base_type}({precision}, {scale})'

    return base_type


def quote_table_name(table_name: str, schema: str, db_type: str) -> str:
    db = normalize_db_type(db_type)
    if schema and db in ('postgresql', 'sqlserver', 'oracle'):
        return quote_table(schema, table_name, db)
    return quote_identifier(table_name, db)


def _escape_sql_literal(value: str) -> str:
    return (value or '').replace("'", "''")


@dataclass
class CreateTableDdlResult:
    create_sql: str = ''
    comment_sqls: List[str] = field(default_factory=list)
    skipped: bool = False


def build_create_schema_sql(schema: str, db_type: str) -> str:
    db = normalize_db_type(db_type)
    if not schema:
        return ''

    if db == 'postgresql':
        quoted = quote_identifier(schema, db)
        return f'CREATE SCHEMA IF NOT EXISTS {quoted};'

    if db == 'sqlserver':
        safe_schema = schema.replace("'", "''")
        return f"""
IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '{safe_schema}')
BEGIN
    EXEC('CREATE SCHEMA [{schema}]')
END;
""".strip()

    return ''


def _generate_field_definition(field: dict, db_type: str) -> str:
    field_name = field.get('name', '')
    if not field_name:
        return ''

    db = normalize_db_type(db_type)
    field_type = normalize_canonical_type(field.get('type', 'varchar'))
    nullable = field.get('nullable', True)
    is_primary = field.get('isPrimaryKey', False)

    quoted_name = quote_identifier(field_name, db)
    sql_type = map_to_dialect_type(field, db)
    parts = [quoted_name, sql_type]

    if is_primary:
        parts.append('PRIMARY KEY')

    if not nullable and not is_primary:
        parts.append('NOT NULL')

    if field_type == 'boolean' and not nullable:
        if db == 'postgresql':
            parts.append('DEFAULT FALSE')
        else:
            parts.append('DEFAULT 0')

    if field_name == 'sort' and field_type in ('int', 'integer', 'bigint', 'smallint'):
        parts.append('DEFAULT 0')

    return ' '.join(parts)


def _generate_create_index_sql(
    idx_name: str,
    full_table_name: str,
    field_name: str,
    table_name: str,
    db_type: str,
) -> str:
    db = normalize_db_type(db_type)
    quoted_idx = quote_identifier(idx_name, db)
    quoted_field = quote_identifier(field_name, db)

    if db == 'postgresql':
        return f'CREATE INDEX IF NOT EXISTS {quoted_idx} ON {full_table_name} ({quoted_field});'

    if db == 'mysql':
        safe_table = table_name.replace("'", "''")
        safe_idx = idx_name.replace("'", "''")
        return (
            f"SET @exist := (SELECT COUNT(*) FROM information_schema.statistics "
            f"WHERE table_schema=DATABASE() AND table_name='{safe_table}' "
            f"AND index_name='{safe_idx}');\n"
            f"SET @sqlstmt := IF(@exist > 0, 'SELECT ''index exists''', "
            f"'CREATE INDEX `{idx_name}` ON {full_table_name} (`{field_name}`)');\n"
            f'PREPARE stmt FROM @sqlstmt;\nEXECUTE stmt;\nDEALLOCATE PREPARE stmt;'
        )

    if db == 'oracle':
        return f'CREATE INDEX {quoted_idx} ON {full_table_name} ({quoted_field});'

    if db == 'sqlserver':
        safe_idx = idx_name.replace("'", "''")
        return (
            f"IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = '{safe_idx}')\n"
            f'CREATE INDEX {quoted_idx} ON {full_table_name} ({quoted_field});'
        )

    return ''


def _build_column_comments_sql(
    fields: List[dict],
    full_table_name: str,
    db_type: str,
) -> List[str]:
    db = normalize_db_type(db_type)
    if db not in ('postgresql', 'oracle'):
        return []

    comment_sqls: List[str] = []
    for fld in fields:
        comment = fld.get('comment', '')
        field_name = fld.get('name', '')
        if not comment or not field_name:
            continue
        safe_comment = _escape_sql_literal(comment)
        quoted_col = quote_identifier(field_name, db)
        comment_sqls.append(
            f"COMMENT ON COLUMN {full_table_name}.{quoted_col} IS '{safe_comment}';"
        )
    return comment_sqls


def _wrap_sqlserver_create_if_not_exists(
    schema: str,
    table_name: str,
    inner_sql: str,
) -> str:
    schema_part = schema.replace("'", "''")
    table_part = table_name.replace("'", "''")
    return f"""
IF NOT EXISTS (
    SELECT 1 FROM sys.tables t
    INNER JOIN sys.schemas s ON t.schema_id = s.schema_id
    WHERE s.name = '{schema_part}' AND t.name = '{table_part}'
)
BEGIN
{inner_sql}
END;
""".strip()


def _wrap_oracle_create_if_not_exists(table_name: str, inner_sql: str) -> str:
    safe_table = table_name.upper().replace("'", "''")
    single_line = ' '.join(inner_sql.split())
    escaped_sql = single_line.replace("'", "''")
    return f"""
DECLARE
  v_count NUMBER;
BEGIN
  SELECT COUNT(*) INTO v_count FROM user_tables WHERE table_name = '{safe_table}';
  IF v_count = 0 THEN
    EXECUTE IMMEDIATE '{escaped_sql}';
  END IF;
END;
""".strip()


def build_create_table_ddl(
    table: dict,
    *,
    db_type: str,
    if_exists: str = 'skip',
    effective_schema: str = '',
) -> CreateTableDdlResult:
    """
    根据 canonical 表定义生成 CREATE TABLE DDL。

    Args:
        table: 含 tableName、fields、meta
        db_type: postgresql / mysql / sqlserver / oracle
        if_exists: skip | error | replace
        effective_schema: 优先使用的 schema
    """
    db = normalize_db_type(db_type)
    table_name = table.get('tableName', '')
    fields = table.get('fields', [])
    meta = table.get('meta', {})
    schema = effective_schema or meta.get('schema', '')

    if not table_name or not fields:
        raise ValueError(f'表 {table_name or "(unknown)"} 缺少必要配置')

    full_table_name = quote_table_name(table_name, schema, db)
    sql_parts: List[str] = []

    if if_exists == 'replace':
        if db == 'postgresql':
            sql_parts.append(f'DROP TABLE IF EXISTS {full_table_name} CASCADE;')
        elif db == 'mysql':
            sql_parts.append(f'DROP TABLE IF EXISTS {full_table_name};')
        elif db == 'sqlserver':
            sql_parts.append(
                f'IF OBJECT_ID(N\'{full_table_name.replace("[", "").replace("]", "")}\', N\'U\') IS NOT NULL '
                f'DROP TABLE {full_table_name};'
            )
        elif db == 'oracle':
            sql_parts.append(f'BEGIN EXECUTE IMMEDIATE \'DROP TABLE {full_table_name}\'; EXCEPTION WHEN OTHERS THEN NULL; END;')

    if db in ('postgresql', 'mysql'):
        create_clause = 'CREATE TABLE IF NOT EXISTS' if if_exists == 'skip' else 'CREATE TABLE'
    else:
        create_clause = 'CREATE TABLE'

    field_defs = []
    for fld in fields:
        field_def = _generate_field_definition(fld, db)
        if field_def:
            field_defs.append(f'  {field_def}')

    create_body = f'{create_clause} {full_table_name} (\n' + ',\n'.join(field_defs) + '\n);'

    if db == 'sqlserver' and if_exists == 'skip' and schema:
        create_body = _wrap_sqlserver_create_if_not_exists(schema, table_name, create_body)
    elif db == 'oracle' and if_exists == 'skip':
        create_body = _wrap_oracle_create_if_not_exists(table_name, create_body)

    sql_parts.append(create_body)

    field_names = {f.get('name', '') for f in fields}
    for field_name in sorted(field_names & SYSTEM_INDEX_FIELDS):
        idx_name = f'idx_{table_name}_{field_name}'
        index_sql = _generate_create_index_sql(
            idx_name, full_table_name, field_name, table_name, db,
        )
        if index_sql:
            sql_parts.append(index_sql)

    comment_sqls = _build_column_comments_sql(fields, full_table_name, db)

    return CreateTableDdlResult(
        create_sql='\n'.join(sql_parts),
        comment_sqls=comment_sqls,
        skipped=False,
    )