3b7c00377f
Apply Black formatter to the entire codebase for consistent code style. Co-Authored-By: Claude (glm-5) <noreply@anthropic.com>
144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
"""
|
||
表名转换工具
|
||
|
||
处理 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))
|