归档表和业务表表结构对齐脚本
# mysql数据库内归档表和业务表表结构对齐脚本。业务库名称以"wms-prod"结尾，业务表与归档表存放与业务库中，归档表已存在，表名以"archive_"开头；
# 现需要根据业务表的表结构生成修改归档表的DDL语句，使归档表表结构与业务表一致。
# 脚本要求：
# 1. 脚本需要接收业务库名称作为参数，例如：python script.py wms-prod
# 2. 脚本需要连接到业务库，获取业务表的表结构和归档表的表结构
# 4. 脚本需要判断业务表和归档表的表结构是否一致，如果一致，则不生成修改归档表的DDL语句，输出"表结构一致，无需修改"
# 5. 脚本需要判断业务表和归档表的表结构是否不一致，如果不一致，则生成修改归档表的DDL语句 例如：ALTER TABLE wms-prod.archive_shipment_header MODIFY COLUMN name VARCHAR(255);
import sys
import mysql.connector
from mysql.connector import Error
def get_table_structure(cursor, database, table_name):
"""获取表结构信息"""
cursor.execute(f"""
SELECT
COLUMN_NAME,
COLUMN_TYPE,
IS_NULLABLE,
COLUMN_DEFAULT,
EXTRA
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = '{database}'
AND TABLE_NAME = '{table_name}'
ORDER BY ORDINAL_POSITION
""")
return cursor.fetchall()
def generate_alter_statements(business_table, archive_table, business_structure, archive_structure):
"""生成ALTER TABLE语句"""
alter_statements = []
# 创建字典以便快速查找
archive_columns = {col[0]: col for col in archive_structure}
business_columns = {col[0]: col for col in business_structure}
# 检查需要修改的列
for col_name, col_info in business_columns.items():
if col_name in archive_columns:
archive_col = archive_columns[col_name]
if col_info != archive_col:
alter_statements.append(
f"ALTER TABLE {archive_table} MODIFY COLUMN {col_name} {col_info[1]} "
f"{'NULL' if col_info[2] == 'YES' else 'NOT NULL'} "
f"{f'DEFAULT {col_info[3]}' if col_info[3] is not None else ''} "
f"{col_info[4]}"
)
else:
# 如果归档表中不存在该列，添加该列
alter_statements.append(
f"ALTER TABLE {archive_table} ADD COLUMN {col_name} {col_info[1]} "
f"{'NULL' if col_info[2] == 'YES' else 'NOT NULL'} "
f"{f'DEFAULT {col_info[3]}' if col_info[3] is not None else ''} "
f"{col_info[4]}"
)
# 检查需要删除的列
for col_name in archive_columns:
if col_name not in business_columns:
alter_statements.append(f"ALTER TABLE {archive_table} DROP COLUMN {col_name}")
return alter_statements
def main():
if len(sys.argv) != 2:
print("使用方法: python script.py <数据库名称>")
sys.exit(1)
database = sys.argv[1]
if not database.endswith("wms-prod"):
print("错误：数据库名称必须以'wms-prod'结尾")
sys.exit(1)
try:
# 连接数据库
conn = mysql.connector.connect(
host='192.168.95.15',
port=3306,
user='usera',
password='password'
)
cursor = conn.cursor()
# 获取所有业务表
cursor.execute(f"""
SELECT TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = '{database}'
AND TABLE_NAME NOT LIKE 'archive_%'
""")
business_tables = cursor.fetchall()
for business_table in business_tables:
business_table = business_table[0]
archive_table = f"archive_{business_table}"
# 检查归档表是否存在
cursor.execute(f"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA = '{database}'
AND TABLE_NAME = '{archive_table}'
""")
if cursor.fetchone()[0] == 0:
print(f"警告：归档表 {archive_table} 不存在，跳过")
continue
# 获取表结构
business_structure = get_table_structure(cursor, database, business_table)
archive_structure = get_table_structure(cursor, database, archive_table)
# 比较表结构
if business_structure == archive_structure:
print(f"表 {business_table} 和 {archive_table} 结构一致，无需修改")
else:
print(f"\n表 {business_table} 和 {archive_table} 结构不一致，生成修改语句：")
alter_statements = generate_alter_statements(
business_table,
archive_table,
business_structure,
archive_structure
)
for statement in alter_statements:
print(statement)
except Error as e:
print(f"数据库错误: {e}")
finally:
if 'conn' in locals() and conn.is_connected():
cursor.close()
conn.close()
if __name__ == "__main__":
main()
