chore: 更新 .gitignore 文件，添加对 .py 文件的忽略；删除不再使用的 mysql2dm.py 文件
This commit is contained in:
@@ -5,3 +5,4 @@
|
||||
/example/config
|
||||
/.cursor/*.log
|
||||
*.sql
|
||||
*.py
|
||||
|
||||
@@ -1,382 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
MySQL 8.0 dump 转达梦 DM8 SQL 转换脚本
|
||||
兼容 HoTime 框架达梦适配规范
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
INPUT_FILE = os.path.join(os.path.dirname(__file__), "zwbg_2026-03-20_11-00-59_mysql_data_zhqCt.sql")
|
||||
OUTPUT_FILE = os.path.join(os.path.dirname(__file__), "zwbg_dm8.sql")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# 工具函数
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def backtick_to_doublequote(s: str) -> str:
|
||||
"""把反引号标识符改为双引号，并强制小写"""
|
||||
def replacer(m):
|
||||
return '"' + m.group(1).lower() + '"'
|
||||
return re.sub(r'`([^`]+)`', replacer, s)
|
||||
|
||||
|
||||
def convert_col_type(coldef: str) -> str:
|
||||
"""转换列的数据类型（不含 COMMENT）"""
|
||||
# tinyint -> INT（先处理，否则会被 int 规则误匹配）
|
||||
coldef = re.sub(r'\btinyint(?:\s*\(\s*\d+\s*\))?\b', 'INT', coldef, flags=re.I)
|
||||
# bigint -> BIGINT
|
||||
coldef = re.sub(r'\bbigint(?:\s*\(\s*\d+\s*\))?\b', 'BIGINT', coldef, flags=re.I)
|
||||
# int unsigned / int(N) -> INT
|
||||
coldef = re.sub(r'\bint\s+unsigned(?:\s*\(\s*\d+\s*\))?\b', 'INT', coldef, flags=re.I)
|
||||
coldef = re.sub(r'\bint(?:\s*\(\s*\d+\s*\))?\b', 'INT', coldef, flags=re.I)
|
||||
# datetime -> TIMESTAMP
|
||||
coldef = re.sub(r'\bdatetime\b', 'TIMESTAMP', coldef, flags=re.I)
|
||||
# text / longtext / mediumtext -> CLOB
|
||||
coldef = re.sub(r'\b(?:long|medium)?text\b', 'CLOB', coldef, flags=re.I)
|
||||
# 去掉 CHARACTER SET ... / COLLATE ... 修饰
|
||||
coldef = re.sub(r'\bCHARACTER\s+SET\s+\S+', '', coldef, flags=re.I)
|
||||
coldef = re.sub(r'\bCOLLATE\s+\S+', '', coldef, flags=re.I)
|
||||
# 去掉列级 AUTO_INCREMENT（由外层统一处理自增列声明）
|
||||
coldef = re.sub(r'\bAUTO_INCREMENT\b', '', coldef, flags=re.I)
|
||||
# 压缩多余空格
|
||||
coldef = re.sub(r' +', ' ', coldef).strip()
|
||||
return coldef
|
||||
|
||||
|
||||
def extract_comment(line: str) -> tuple[str, Optional[str]]:
|
||||
"""提取行尾 COMMENT '...' 并返回 (去掉comment的行, comment文本 or None)"""
|
||||
m = re.search(r"\bCOMMENT\s+'((?:[^'\\]|\\.)*)'\s*,?\s*$", line, re.I)
|
||||
if m:
|
||||
comment_text = m.group(1)
|
||||
# 转义单引号
|
||||
comment_text = comment_text.replace("''", "'")
|
||||
line_clean = line[:m.start()].rstrip().rstrip(',')
|
||||
return line_clean, comment_text
|
||||
return line, None
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# 主解析器状态机
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
class Table:
|
||||
def __init__(self, name: str, table_comment: Optional[str]):
|
||||
self.name = name # 原始小写名
|
||||
self.table_comment = table_comment
|
||||
self.col_defs: list[str] = [] # 已转换的列定义（不含索引）
|
||||
self.col_comments: dict[str, str] = {} # col_name -> comment
|
||||
self.pk_cols: list[str] = [] # PRIMARY KEY 列
|
||||
self.unique_keys: list[tuple[str, list[str]]] = [] # (key_name, cols)
|
||||
self.normal_keys: list[tuple[str, list[str]]] = [] # (key_name, cols)
|
||||
self.identity_col: Optional[str] = None # 自增列名
|
||||
self.auto_increment_start: int = 1 # 表级 AUTO_INCREMENT 值
|
||||
self.inserts: list[str] = [] # INSERT 语句列表
|
||||
|
||||
def qname(self) -> str:
|
||||
"""双引号包裹小写表名"""
|
||||
return f'"{self.name}"'
|
||||
|
||||
|
||||
def parse_key_cols(col_str: str) -> list[str]:
|
||||
"""解析索引列列表，如 `col1`,`col2` -> ['col1','col2']"""
|
||||
cols = []
|
||||
for part in col_str.split(','):
|
||||
part = part.strip()
|
||||
# 去掉长度修饰 `col`(10)
|
||||
part = re.sub(r'\(\d+\)$', '', part.strip())
|
||||
m = re.match(r'[`"\']?([^`"\'()\s]+)[`"\']?', part)
|
||||
if m:
|
||||
cols.append(m.group(1).lower())
|
||||
return cols
|
||||
|
||||
|
||||
def parse_mysql_dump(sql_text: str) -> list[Table]:
|
||||
"""解析 MySQL dump，提取所有表的结构和数据"""
|
||||
tables: list[Table] = []
|
||||
current_table: Optional[Table] = None
|
||||
in_create = False
|
||||
|
||||
lines = sql_text.splitlines()
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
raw = lines[i]
|
||||
stripped = raw.strip()
|
||||
|
||||
# 跳过 MySQL 条件注释行和空行
|
||||
if re.match(r'^/\*!', stripped) or stripped == '' or stripped.startswith('-- '):
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# DROP TABLE IF EXISTS
|
||||
m = re.match(r'DROP TABLE IF EXISTS `([^`]+)`', stripped, re.I)
|
||||
if m:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# CREATE TABLE 开始
|
||||
m = re.match(r'CREATE TABLE `([^`]+)`\s*\(', stripped, re.I)
|
||||
if m:
|
||||
tname = m.group(1).lower()
|
||||
current_table = Table(tname, None)
|
||||
in_create = True
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# CREATE TABLE 结束行：) ENGINE=... COMMENT='...'
|
||||
if in_create and stripped.startswith(')'):
|
||||
in_create = False
|
||||
# 提取表注释
|
||||
tc = re.search(r"COMMENT\s*=\s*'((?:[^'\\]|\\.)*)'", stripped, re.I)
|
||||
if tc:
|
||||
current_table.table_comment = tc.group(1)
|
||||
# 提取 AUTO_INCREMENT=N
|
||||
ai = re.search(r'AUTO_INCREMENT=(\d+)', stripped, re.I)
|
||||
if ai:
|
||||
current_table.auto_increment_start = int(ai.group(1))
|
||||
tables.append(current_table)
|
||||
current_table = None
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# CREATE TABLE 内部行
|
||||
if in_create and current_table is not None:
|
||||
# 去掉行尾逗号暂时，用于判断
|
||||
col_line = stripped
|
||||
|
||||
# PRIMARY KEY
|
||||
m = re.match(r'PRIMARY KEY\s*\((.+)\)', col_line, re.I)
|
||||
if m:
|
||||
current_table.pk_cols = parse_key_cols(m.group(1))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# UNIQUE KEY
|
||||
m = re.match(r'UNIQUE KEY\s*`([^`]+)`\s*\((.+?)\)', col_line, re.I)
|
||||
if m:
|
||||
kname = m.group(1).lower()
|
||||
kcols = parse_key_cols(m.group(2))
|
||||
current_table.unique_keys.append((kname, kcols))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# KEY (普通索引)
|
||||
m = re.match(r'KEY\s*`([^`]+)`\s*\((.+?)\)', col_line, re.I)
|
||||
if m:
|
||||
kname = m.group(1).lower()
|
||||
kcols = parse_key_cols(m.group(2))
|
||||
current_table.normal_keys.append((kname, kcols))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# CONSTRAINT ... PRIMARY KEY (复合pk in some versions)
|
||||
m = re.match(r'CONSTRAINT\s*`[^`]+`\s*PRIMARY KEY\s*\((.+)\)', col_line, re.I)
|
||||
if m:
|
||||
current_table.pk_cols = parse_key_cols(m.group(1))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# CONSTRAINT ... UNIQUE KEY
|
||||
m = re.match(r'CONSTRAINT\s*`([^`]+)`\s*UNIQUE\s*(?:KEY)?\s*\((.+?)\)', col_line, re.I)
|
||||
if m:
|
||||
kname = m.group(1).lower()
|
||||
kcols = parse_key_cols(m.group(2))
|
||||
current_table.unique_keys.append((kname, kcols))
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# 列定义
|
||||
m = re.match(r'`([^`]+)`\s+(.+)', col_line)
|
||||
if m:
|
||||
col_name = m.group(1).lower()
|
||||
col_rest = m.group(2)
|
||||
|
||||
# 提取 COMMENT
|
||||
comment_m = re.search(r"\bCOMMENT\s+'((?:[^'\\]|\\.)*)'\s*,?\s*$", col_rest, re.I)
|
||||
comment_text = None
|
||||
if comment_m:
|
||||
comment_text = comment_m.group(1)
|
||||
col_rest = col_rest[:comment_m.start()].rstrip().rstrip(',')
|
||||
|
||||
# 检测是否是自增列
|
||||
is_identity = bool(re.search(r'\bAUTO_INCREMENT\b', col_rest, re.I))
|
||||
|
||||
# 类型转换
|
||||
col_rest = convert_col_type(col_rest)
|
||||
|
||||
# 去掉行尾逗号
|
||||
col_rest = col_rest.rstrip(',').strip()
|
||||
|
||||
if is_identity:
|
||||
# 替换成 IDENTITY(1,1)，保留 NOT NULL
|
||||
# 先提取基础类型
|
||||
type_m = re.match(r'((?:BIG)?INT)\s+(.*)', col_rest, re.I)
|
||||
if type_m:
|
||||
base_type = type_m.group(1).upper()
|
||||
rest_attrs = type_m.group(2).strip()
|
||||
# 去掉 NOT NULL（IDENTITY 列隐含 NOT NULL）
|
||||
rest_attrs = re.sub(r'\bNOT\s+NULL\b', '', rest_attrs, flags=re.I).strip()
|
||||
col_rest = f"{base_type} IDENTITY(1,1) {rest_attrs}".strip()
|
||||
else:
|
||||
col_rest = re.sub(r'\bNOT\s+NULL\b', '', col_rest, flags=re.I).strip()
|
||||
col_rest = f"INT IDENTITY(1,1) {col_rest}".strip()
|
||||
current_table.identity_col = col_name
|
||||
|
||||
col_def = f' "{col_name}" {col_rest}'
|
||||
current_table.col_defs.append(col_def)
|
||||
if comment_text is not None:
|
||||
current_table.col_comments[col_name] = comment_text
|
||||
i += 1
|
||||
continue
|
||||
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# INSERT INTO `tablename` VALUES (...)
|
||||
m = re.match(r'INSERT INTO `([^`]+)`\s+VALUES\s+', stripped, re.I)
|
||||
if m:
|
||||
tname = m.group(1).lower()
|
||||
# 找到对应的 table 对象
|
||||
tbl = next((t for t in tables if t.name == tname), None)
|
||||
if tbl is not None:
|
||||
# 转换反引号 -> 双引号
|
||||
insert_line = re.sub(r'`([^`]+)`', lambda mm: f'"{mm.group(1).lower()}"', stripped)
|
||||
# 去掉行尾分号后的 /*!...*/
|
||||
insert_line = re.sub(r'\s*/\*!.*?\*/\s*', ' ', insert_line)
|
||||
insert_line = insert_line.strip().rstrip(';') + ';'
|
||||
tbl.inserts.append(insert_line)
|
||||
i += 1
|
||||
continue
|
||||
|
||||
i += 1
|
||||
|
||||
return tables
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# 输出生成
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def gen_drop(tbl: Table) -> str:
|
||||
return f'DROP TABLE IF EXISTS {tbl.qname()};\n'
|
||||
|
||||
|
||||
def gen_create(tbl: Table) -> str:
|
||||
lines = []
|
||||
col_parts = list(tbl.col_defs)
|
||||
|
||||
# PRIMARY KEY 约束
|
||||
if tbl.pk_cols:
|
||||
pk_str = ', '.join(f'"{c}"' for c in tbl.pk_cols)
|
||||
col_parts.append(f' PRIMARY KEY ({pk_str})')
|
||||
|
||||
create = f'CREATE TABLE {tbl.qname()} (\n'
|
||||
create += ',\n'.join(col_parts)
|
||||
create += '\n);\n'
|
||||
return create
|
||||
|
||||
|
||||
def gen_comments(tbl: Table) -> str:
|
||||
out = ''
|
||||
if tbl.table_comment:
|
||||
tc = tbl.table_comment.replace("'", "''")
|
||||
out += f"COMMENT ON TABLE {tbl.qname()} IS '{tc}';\n"
|
||||
for col, comment in tbl.col_comments.items():
|
||||
cc = comment.replace("'", "''")
|
||||
out += f"COMMENT ON COLUMN {tbl.qname()}.\"{ col}\" IS '{cc}';\n"
|
||||
return out
|
||||
|
||||
|
||||
def gen_indexes(tbl: Table) -> str:
|
||||
out = ''
|
||||
for kname, kcols in tbl.unique_keys:
|
||||
cols_str = ', '.join(f'"{c}"' for c in kcols)
|
||||
out += f'CREATE UNIQUE INDEX "{kname}" ON {tbl.qname()} ({cols_str});\n'
|
||||
for kname, kcols in tbl.normal_keys:
|
||||
cols_str = ', '.join(f'"{c}"' for c in kcols)
|
||||
out += f'CREATE INDEX "{kname}" ON {tbl.qname()} ({cols_str});\n'
|
||||
return out
|
||||
|
||||
|
||||
def gen_inserts(tbl: Table) -> str:
|
||||
if not tbl.inserts:
|
||||
return ''
|
||||
out = ''
|
||||
if tbl.identity_col:
|
||||
out += f'SET IDENTITY_INSERT {tbl.qname()} ON;\n'
|
||||
for ins in tbl.inserts:
|
||||
out += ins + '\n'
|
||||
if tbl.identity_col:
|
||||
out += f'SET IDENTITY_INSERT {tbl.qname()} OFF;\n'
|
||||
# 重置自增起始值，取 auto_increment_start（MySQL 导出时已包含下一个可用值）
|
||||
next_val = tbl.auto_increment_start
|
||||
if next_val > 1:
|
||||
out += (
|
||||
f'ALTER TABLE {tbl.qname()} ALTER COLUMN "{tbl.identity_col}" '
|
||||
f'SET GENERATED BY DEFAULT AS IDENTITY (START WITH {next_val});\n'
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def generate_dm8_sql(tables: list[Table]) -> str:
|
||||
header = """\
|
||||
-- =============================================================================
|
||||
-- 达梦 DM8 数据库初始化脚本
|
||||
-- 源文件: zwbg_2026-03-20_11-00-59_mysql_data_zhqCt.sql (MySQL 8.0)
|
||||
-- 目标数据库: 达梦 DM8
|
||||
-- 兼容框架: HoTime (标识符统一小写双引号, IDENTITY 自增, TIMESTAMP 时间)
|
||||
-- 生成时间: 2026-03-20
|
||||
-- =============================================================================
|
||||
|
||||
"""
|
||||
footer = "\n-- 脚本执行完毕\n"
|
||||
|
||||
parts = [header]
|
||||
for tbl in tables:
|
||||
parts.append(f'-- ---- 表: {tbl.name} ----\n')
|
||||
parts.append(gen_drop(tbl))
|
||||
parts.append(gen_create(tbl))
|
||||
comments = gen_comments(tbl)
|
||||
if comments:
|
||||
parts.append(comments)
|
||||
indexes = gen_indexes(tbl)
|
||||
if indexes:
|
||||
parts.append(indexes)
|
||||
inserts = gen_inserts(tbl)
|
||||
if inserts:
|
||||
parts.append(inserts)
|
||||
parts.append('\n')
|
||||
parts.append(footer)
|
||||
return ''.join(parts)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# 入口
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
print(f"读取源文件: {INPUT_FILE}")
|
||||
with open(INPUT_FILE, 'r', encoding='utf-8') as f:
|
||||
sql_text = f.read()
|
||||
|
||||
print("解析 MySQL dump...")
|
||||
tables = parse_mysql_dump(sql_text)
|
||||
print(f" 共解析到 {len(tables)} 张表: {', '.join(t.name for t in tables)}")
|
||||
|
||||
print("生成达梦 DM8 SQL...")
|
||||
dm8_sql = generate_dm8_sql(tables)
|
||||
|
||||
print(f"写出目标文件: {OUTPUT_FILE}")
|
||||
with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
|
||||
f.write(dm8_sql)
|
||||
|
||||
size_kb = os.path.getsize(OUTPUT_FILE) / 1024
|
||||
print(f"完成! 文件大小: {size_kb:.1f} KB")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user