hoteas
5ad4e2e11b
- 更新 .gitignore 文件，添加对 .sql 文件的忽略 - 移除 cache_db.go 中的调试日志相关代码，简化缓存操作逻辑 - 在多个文档中添加达梦 DM8 数据库的支持信息，包括配置、使用注意事项及示例 - 更新 QUICKSTART.md，明确支持的数据库类型及配置示例 - 在 ROADMAP.md 中记录达梦数据库完整支持的进展
383 lines
15 KiB
Python
383 lines
15 KiB
Python
#!/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()
|