From 5e2bd765bfc12e5e66b681c8d0a63829d8847cf7 Mon Sep 17 00:00:00 2001 From: hoteas <925970985@qq.com> Date: Fri, 20 Mar 2026 11:54:42 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E6=9B=B4=E6=96=B0=20.gitignore=20?= =?UTF-8?q?=E6=96=87=E4=BB=B6=EF=BC=8C=E6=B7=BB=E5=8A=A0=E5=AF=B9=20.py=20?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E7=9A=84=E5=BF=BD=E7=95=A5=EF=BC=9B=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E4=B8=8D=E5=86=8D=E4=BD=BF=E7=94=A8=E7=9A=84=20mysql2?= =?UTF-8?q?dm.py=20=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + docs/mysql2dm.py | 382 ----------------------------------------------- 2 files changed, 1 insertion(+), 382 deletions(-) delete mode 100644 docs/mysql2dm.py diff --git a/.gitignore b/.gitignore index 5dfad93..26c2e52 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ /example/config /.cursor/*.log *.sql +*.py diff --git a/docs/mysql2dm.py b/docs/mysql2dm.py deleted file mode 100644 index 793ed7e..0000000 --- a/docs/mysql2dm.py +++ /dev/null @@ -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()