从MySQL转换到PostgreSQL的迁移过程

-- mysql
`time_stamp_` timestamp not null default current_timestamp(3)

错误: 语法错误 在 "null" 或附近的
line 8: `time_stamp_` null not null default current_timestamp(3)

import re
def fix_timestamp_issues(content):
    # 1. 移除 current_timestamp 的精度参数
    content = re.sub(r'current_timestamp\(\d+\)', 'current_timestamp', content)
    # 2. 修复字段名中的 null 关键字冲突（用双引号包裹）
    content = re.sub(r'(\w+_null)\s+', r'"\1" ', content)
    return content

def fix_auto_increment(content):
    # 方案一：使用 identity（postgresql 10+ 推荐）
    content = re.sub(
        r'bigint not null auto_increment',
        'bigint not null generated by default as identity',
        content
    )
    # 方案二：使用 serial 简写（兼容旧版本）
    # content = re.sub(r'int\(11\) not null auto_increment', 'serial', content)
    return content

def fix_data_types(content):
    type_mappings = {
        r'\btinyint\b': 'int2',
        r'\bsmallint\b': 'int2',
        r'\bmediumint\b': 'int4',
        r'\bint\b(?!4|8|eger)': 'int4',  # 避免匹配 int4/int8/integer
        r'\bbigint\b': 'int8',
        r'\bdatetime\b': 'timestamp',
        r'\bblob\b': 'bytea',
        r'\btinyblob\b': 'bytea',
        r'\bmediumblob\b': 'bytea',
        r'\blongblob\b': 'bytea',
        r'\bfloat\b': 'float4',
        r'\bdouble\b': 'float8',
    }
    for mysql_type, pg_type in type_mappings.items():
        content = re.sub(mysql_type, pg_type, content, flags=re.ignorecase)
    return content

-- 错误：转换后的 sql
drop "table" "if" exists dual;
-- 正确：应该是
drop table if exists dual;

def fix_reserved_words(content):
    # 修复被错误引用的 sql 关键字
    fixes = [
        (r'drop "table"', 'drop table'),
        (r'drop "if"', 'drop if'),
        (r'"if" exists', 'if exists'),
        (r'create "table"', 'create table'),
        (r'"primary" key', 'primary key'),
        (r'"foreign" key', 'foreign key'),
        (r'"not" null', 'not null'),
        (r'"default"', 'default'),
    ]
    for pattern, replacement in fixes:
        content = re.sub(pattern, replacement, content, flags=re.ignorecase)
    return content

def add_missing_columns(content):
    missing_columns_config = {
        'sys_coffee': [
            'deleted int2 default 0',
            'tenant_id int4'
        ],
        'erp_product': [
            'brand_id bigint',
            'skus text'
        ],
        'erp_product_brand': [
            'brand_category varchar(255)'
        ],
        'erp_product_reagent': [
            'skus text'
        ],
    }
    for table_name, columns in missing_columns_config.items():
        # 匹配 create table 语句的主键结束位置
        pattern = rf'(create table {table_name} \([\s\s]*?)(primary key \(id\));'
        columns_def = '\n  ' + ',\n  '.join(columns) + ','
        replacement = rf'\1{columns_def}\n\2;'
        content = re.sub(pattern, replacement, content)
    return content

def fix_boolean_literals(content):
    # 替换二进制字面量
    content = re.sub(r"b'0'", "'0'", content)
    content = re.sub(r"b'1'", "'1'", content)
    # 或者转换为布尔值（如果字段类型已是 boolean）
    # content = re.sub(r"b'0'", "false", content)
    # content = re.sub(r"b'1'", "true", content)
    return content

def fix_mysql_specific_syntax(content):
    # 删除 engine、auto_increment、row_format 等
    content = re.sub(
        r'engine\s*=\s*\w+.*?row_format\s*=\s*\w+;',
        ';',
        content,
        flags=re.ignorecase
    )
    # 删除 character set / collate 子句
    content = re.sub(
        r'character\s+set\s+\w+\s+collate\s+\w+',
        '',
        content,
        flags=re.ignorecase
    )
    # 处理表注释（postgresql 需要单独处理）
    # mysql: comment 'xxx' → 需提取并转换为 comment on table
    return content

def create_sequences_for_tables(content):
    # 提取所有 create table 的表名
    table_names = re.findall(r'create table (\w+)', content, re.ignorecase)
    # 排除系统表或特定前缀的表
    exclude_prefixes = ('act_', 'flw_', 'qrtz_')
    sequence_statements = []
    for table_name in table_names:
        if not table_name.startswith(exclude_prefixes):
            sequence_statements.append(f"""
-- 为表 {table_name} 创建序列
create sequence if not exists {table_name}_id_seq;
alter table {table_name} alter column id set default nextval('{table_name}_id_seq');
alter sequence {table_name}_id_seq owned by {table_name}.id;
""")
    # 将序列语句追加到文件末尾
    content += "\n\n-- 自动生成的序列定义\n" + "\n".join(sequence_statements)
    return content

def fix_index_syntax(content):
    # 1. 移除 fulltext 索引（需替换为 postgresql 的 gin 索引）
    # mysql: fulltext index idx_name (col1, col2)
    # postgresql: create index idx_name on table_name using gin (to_tsvector('english', col1 || ' ' || col2))
    content = re.sub(
        r'fulltext\s+index\s+\w+\s*\([^)]+\)',
        '-- fulltext index removed, needs manual conversion to gin',
        content,
        flags=re.ignorecase
    )
    # 2. 处理前缀索引（postgresql 不支持）
    # mysql: index idx_name (col(10))
    # 需要移除前缀长度或替换为表达式索引
    return content

mysql_to_pg_migration/
├── fixers/
│   ├── __init__.py
│   ├── timestamp_fixer.py      # 时间戳修复
│   ├── type_fixer.py           # 数据类型映射
│   ├── auto_increment_fixer.py # 自增主键修复
│   ├── reserved_words_fixer.py # 保留字修复
│   ├── column_fixer.py         # 缺失列补充
│   ├── boolean_fixer.py        # 布尔值修复
│   ├── mysql_syntax_fixer.py   # mysql 特有语法清理
│   ├── sequence_fixer.py       # 序列管理
│   └── index_fixer.py          # 索引兼容性处理
├── complete_fix.py             # 主控脚本
├── verify_fix.py               # 验证脚本
└── config.yaml                 # 配置文件

#!/usr/bin/env python3
"""
mysql 到 postgresql sql 文件完整修复工具
用法: python complete_fix.py input.sql output.sql
"""
import sys
import re
import argparse
from pathlib import path
# 导入各个修复模块
from fixers import (
    fix_timestamp_issues,
    fix_auto_increment,
    fix_data_types,
    fix_reserved_words,
    add_missing_columns,
    fix_boolean_literals,
    fix_mysql_specific_syntax,
    create_sequences_for_tables,
    fix_index_syntax
)
class sqlfixer:
    def __init__(self, input_path, output_path):
        self.input_path = path(input_path)
        self.output_path = path(output_path)
        self.stats = {
            'fixes_applied': 0,
            'tables_found': 0,
            'sequences_created': 0
        }
    def read_sql(self):
        with open(self.input_path, 'r', encoding='utf-8') as f:
            return f.read()
    def write_sql(self, content):
        with open(self.output_path, 'w', encoding='utf-8') as f:
            f.write(content)
        print(f"✅ 修复完成，输出文件: {self.output_path}")
    def apply_fixes(self, content):
        """按顺序应用所有修复规则"""
        print("🔧 开始修复 sql 文件...")
        # 1. 基础语法修复
        content = fix_timestamp_issues(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 时间戳语法修复完成")
        content = fix_auto_increment(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 自增主键修复完成")
        content = fix_data_types(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 数据类型映射完成")
        content = fix_reserved_words(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 保留字冲突修复完成")
        content = add_missing_columns(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 缺失列补充完成")
        content = fix_boolean_literals(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 布尔值语法修复完成")
        content = fix_mysql_specific_syntax(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ mysql 特有语法清理完成")
        content = fix_index_syntax(content)
        self.stats['fixes_applied'] += 1
        print("  ✓ 索引语法兼容性处理完成")
        # 2. 序列管理（在表结构之后）
        content = create_sequences_for_tables(content)
        self.stats['fixes_applied'] += 1
        self.stats['sequences_created'] = content.count('create sequence')
        print(f"  ✓ 序列创建完成（共 {self.stats['sequences_created']} 个）")
        return content
    def run(self):
        print(f"📂 读取文件: {self.input_path}")
        content = self.read_sql()
        original_size = len(content)
        content = self.apply_fixes(content)
        final_size = len(content)
        self.write_sql(content)
        # 打印统计信息
        print("\n📊 修复统计:")
        print(f"  - 原始文件大小: {original_size / 1024:.2f} kb")
        print(f"  - 最终文件大小: {final_size / 1024:.2f} kb")
        print(f"  - 应用修复规则: {self.stats['fixes_applied']}")
        print(f"  - 创建序列数量: {self.stats['sequences_created']}")
def main():
    parser = argparse.argumentparser(description='mysql 到 postgresql sql 文件修复工具')
    parser.add_argument('input', help='输入 sql 文件路径')
    parser.add_argument('output', help='输出 sql 文件路径')
    args = parser.parse_args()
    fixer = sqlfixer(args.input, args.output)
    fixer.run()
if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
验证修复后的 sql 文件，检测潜在问题
"""
import re
import sys
from pathlib import path
class sqlvalidator:
    def __init__(self, sql_path):
        self.sql_path = path(sql_path)
        self.content = self.read_sql()
        self.issues = []
    def read_sql(self):
        with open(self.sql_path, 'r', encoding='utf-8') as f:
            return f.read()
    def check_auto_increment(self):
        """检查是否还有未转换的 auto_increment"""
        matches = re.findall(r'auto_increment', self.content, re.ignorecase)
        if matches:
            self.issues.append(f"⚠️ 发现 {len(matches)} 处未转换的 auto_increment")
    def check_timestamp_precision(self):
        """检查是否还有带精度的时间戳"""
        matches = re.findall(r'current_timestamp\(\d+\)', self.content)
        if matches:
            self.issues.append(f"⚠️ 发现 {len(matches)} 处带精度的时间戳")
    def check_mysql_types(self):
        """检查是否还有 mysql 特有类型"""
        mysql_types = ['tinyint', 'mediumint', 'datetime', 'blob']
        for typ in mysql_types:
            matches = re.findall(rf'\b{typ}\b', self.content, re.ignorecase)
            if matches:
                self.issues.append(f"⚠️ 发现 {len(matches)} 处未转换的类型 '{typ}'")
    def check_engine_clause(self):
        """检查是否还有 engine 子句"""
        matches = re.findall(r'engine\s*=', self.content, re.ignorecase)
        if matches:
            self.issues.append(f"⚠️ 发现 {len(matches)} 处未清理的 engine 子句")
    def check_binary_literals(self):
        """检查是否还有 b'0'/b'1' 字面量"""
        matches = re.findall(r"b'[01]'", self.content)
        if matches:
            self.issues.append(f"⚠️ 发现 {len(matches)} 处未转换的二进制字面量")
    def run_all_checks(self):
        self.check_auto_increment()
        self.check_timestamp_precision()
        self.check_mysql_types()
        self.check_engine_clause()
        self.check_binary_literals()
    def report(self):
        print(f"\n🔍 验证文件: {self.sql_path}")
        print("=" * 50)
        if not self.issues:
            print("✅ 未发现明显问题，文件可以提交测试")
        else:
            print(f"⚠️ 发现 {len(self.issues)} 个潜在问题:\n")
            for issue in self.issues:
                print(f"  {issue}")
        return len(self.issues) == 0
def main():
    if len(sys.argv) != 2:
        print("用法: python verify_fix.py <sql_file>")
        sys.exit(1)
    validator = sqlvalidator(sys.argv[1])
    validator.run_all_checks()
    success = validator.report()
    sys.exit(0 if success else 1)
if __name__ == '__main__':
    main()

# 1. 克隆或创建脚本目录
mkdir mysql_to_pg_migration
cd mysql_to_pg_migration
# 2. 运行完整修复
python complete_fix.py original.sql fixed.sql
# 3. 验证修复结果
python verify_fix.py fixed.sql
# 4. 在 postgresql 中执行
psql -d target_db -f fixed.sql

# 逐步应用修复，便于定位问题
python fix_timestamp.py input.sql step1.sql
python fix_type.py step1.sql step2.sql
python fix_auto_increment.py step2.sql step3.sql
python verify_fix.py step3.sql

-- 1. 确认序列存在
select * from information_schema.sequences where sequence_name like '%_id_seq';
-- 2. 确认默认值已设置
\d table_name
-- 3. 手动设置序列值（如果需要同步现有数据）
select setval('table_name_id_seq', (select max(id) from table_name));