-- 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 的精度參數(shù)
    content = re.sub(r'CURRENT_TIMESTAMP\(\d+\)', 'CURRENT_TIMESTAMP', content)
    # 2. 修復字段名中的 NULL 關(guān)鍵字沖突（用雙引號包裹）
    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

-- 錯誤：轉(zhuǎn)換后的 SQL
DROP "table" "if" EXISTS dual;
-- 正確：應該是
DROP TABLE IF EXISTS dual;

def fix_reserved_words(content):
    # 修復被錯誤引用的 SQL 關(guān)鍵字
    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 語句的主鍵結(jié)束位置
        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)
    # 或者轉(zhuǎn)換為布爾值（如果字段類型已是 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' → 需提取并轉(zhuǎn)換為 COMMENT ON TABLE
    return content

def create_sequences_for_tables(content):
    # 提取所有 CREATE TABLE 的表名
    table_names = re.findall(r'CREATE TABLE (\w+)', content, re.IGNORECASE)
    # 排除系統(tǒng)表或特定前綴的表
    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} 創(chuàng)建序列
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           # 數(shù)據(jù)類型映射
│   ├── 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):
        """按順序應用所有修復規(guī)則"""
        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("  ? 數(shù)據(jù)類型映射完成")
        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. 序列管理（在表結(jié)構(gòu)之后）
        content = create_sequences_for_tables(content)
        self.stats['fixes_applied'] += 1
        self.stats['sequences_created'] = content.count('CREATE SEQUENCE')
        print(f"  ? 序列創(chuàng)建完成（共 {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)
        # 打印統(tǒng)計信息
        print("\n?? 修復統(tǒng)計:")
        print(f"  - 原始文件大小: {original_size / 1024:.2f} KB")
        print(f"  - 最終文件大小: {final_size / 1024:.2f} KB")
        print(f"  - 應用修復規(guī)則: {self.stats['fixes_applied']}")
        print(f"  - 創(chuàng)建序列數(shù)量: {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):
        """檢查是否還有未轉(zhuǎn)換的 AUTO_INCREMENT"""
        matches = re.findall(r'AUTO_INCREMENT', self.content, re.IGNORECASE)
        if matches:
            self.issues.append(f"?? 發(fā)現(xiàn) {len(matches)} 處未轉(zhuǎn)換的 AUTO_INCREMENT")
    def check_timestamp_precision(self):
        """檢查是否還有帶精度的時間戳"""
        matches = re.findall(r'CURRENT_TIMESTAMP\(\d+\)', self.content)
        if matches:
            self.issues.append(f"?? 發(fā)現(xiàn) {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"?? 發(fā)現(xiàn) {len(matches)} 處未轉(zhuǎn)換的類型 '{typ}'")
    def check_engine_clause(self):
        """檢查是否還有 ENGINE 子句"""
        matches = re.findall(r'ENGINE\s*=', self.content, re.IGNORECASE)
        if matches:
            self.issues.append(f"?? 發(fā)現(xiàn) {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"?? 發(fā)現(xiàn) {len(matches)} 處未轉(zhuǎn)換的二進制字面量")
    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("? 未發(fā)現(xiàn)明顯問題，文件可以提交測試")
        else:
            print(f"?? 發(fā)現(xiàn) {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. 克隆或創(chuàng)建腳本目錄
mkdir mysql_to_pg_migration
cd mysql_to_pg_migration
# 2. 運行完整修復
python complete_fix.py original.sql fixed.sql
# 3. 驗證修復結(jié)果
python verify_fix.py fixed.sql
# 4. 在 PostgreSQL 中執(zhí)行
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. 手動設置序列值（如果需要同步現(xiàn)有數(shù)據(jù)）
SELECT setval('table_name_id_seq', (SELECT MAX(id) FROM table_name));