从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));