""" 表名转换工具 处理 SQL Server 和 MySQL 之间的表名格式转换 """ import re from typing import List class TableNameConverter: """表名转换工具类""" # 匹配 SQL Server 表名格式:[schema].[tablename] 或 [schema].[table name] SQLSERVER_PATTERN = re.compile(r"\[([^\]]+)\]\.\[([^\]]+)\]") @staticmethod def to_mysql(table_name: str) -> str: """ 将 SQL Server 表名格式转换为 MySQL 格式 SQL Server: [schema].[tablename] → MySQL: schema_tablename SQL Server: tablename → MySQL: dbo_tablename (默认 dbo) Args: table_name: SQL Server 格式的表名 Returns: MySQL 格式的表名 Examples: >>> TableNameConverter.to_mysql('[dbo].[BIPUsers]') 'dbo_BIPUsers' >>> TableNameConverter.to_mysql('DiscreteMaterialPlanData') 'dbo_DiscreteMaterialPlanData' >>> TableNameConverter.to_mysql('[productionContractData].[26年压力表合同数据]') 'productionContractData_26年压力表合同数据' """ # 尝试匹配 [schema].[tablename] 格式 match = TableNameConverter.SQLSERVER_PATTERN.match(table_name.strip()) if match: schema = match.group(1) table = match.group(2) return f"{schema}_{table}" # 如果没有匹配到,使用默认 schema dbo return f"dbo_{table_name}" @staticmethod def to_sqlserver(table_name: str) -> str: """ 将 MySQL 表名格式转换为 SQL Server 格式 MySQL: schema_tablename → SQL Server: [schema].[tablename] Args: table_name: MySQL 格式的表名 Returns: SQL Server 格式的表名 Examples: >>> TableNameConverter.to_sqlserver('dbo_BIPUsers') '[dbo].[BIPUsers]' >>> TableNameConverter.to_sqlserver('productionContractData_26年压力表合同数据') '[productionContractData].[26年压力表合同数据]' """ # 分割第一个下划线 parts = table_name.split("_", 1) if len(parts) == 2: schema = parts[0] table = parts[1] return f"[{schema}].[{table}]" # 如果没有下划线,使用默认 schema dbo return f"[dbo].[{table_name}]" @staticmethod def convert_sql(sql: str, db_type: str) -> str: """ 批量转换 SQL 语句中的表名 Args: sql: SQL 语句 db_type: 目标数据库类型 ('sqlserver' 或 'mysql') Returns: 转换后的 SQL 语句 Examples: >>> sql = "SELECT * FROM [dbo].[BIPUsers] WHERE ID = ?" >>> TableNameConverter.convert_sql(sql, 'mysql') 'SELECT * FROM dbo_BIPUsers WHERE ID = ?' """ if db_type == "mysql": # SQL Server → MySQL def replace_to_mysql(match): schema = match.group(1) table = match.group(2) return f"{schema}_{table}" result = TableNameConverter.SQLSERVER_PATTERN.sub(replace_to_mysql, sql) return result elif db_type == "sqlserver": # MySQL → SQL Server # 首先查找可能的 MySQL 格式表名(schema_table 格式) # 这是一个简化版本,可能无法处理所有边缘情况 result = sql # 查找单词字符_单词字符 的模式(可能是表名) mysql_pattern = re.compile( r"\b([a-zA-Z_][a-zA-Z0-9_]*)_([a-zA-Z0-9_\u4e00-\u9fff]+)\b" ) matches = mysql_pattern.findall(result) for schema, table in set(matches): mysql_name = f"{schema}_{table}" sqlserver_name = f"[{schema}].[{table}]" result = result.replace(mysql_name, sqlserver_name) return result return sql @staticmethod def extract_table_names(sql: str) -> List[str]: """ 从 SQL 语句中提取所有表名 Args: sql: SQL 语句 Returns: 表名列表 """ tables = [] # 查找 SQL Server 格式 sqlserver_matches = TableNameConverter.SQLSERVER_PATTERN.findall(sql) for schema, table in sqlserver_matches: tables.append(f"{schema}_{table}") # 查找可能的 MySQL 格式 mysql_pattern = re.compile(r"\b[a-zA-Z_][a-zA-Z0-9_]*_[a-zA-Z0-9_]+\b") mysql_matches = mysql_pattern.findall(sql) tables.extend(mysql_matches) return list(set(tables))