diff --git a/sql/tools/convertor.py b/sql/tools/convertor.py
index d9ed76b..b2e804a 100644
--- a/sql/tools/convertor.py
+++ b/sql/tools/convertor.py
@@ -23,7 +23,7 @@ import re
 import sys
 import time
 from abc import ABC, abstractmethod
-from typing import Dict, Generator, Optional, Tuple, Union
+from typing import Dict, Generator, List, Optional, Tuple, Union
 
 from simple_ddl_parser import DDLParser
 
@@ -43,6 +43,52 @@ PREAMBLE = """/*
 """
 
 
+# 依次匹配：引号字符串（支持反斜杠转义、双写引号）、注释、语句分隔符；未闭合的引号吞掉剩余内容
+SQL_TOKEN_PATTERN = re.compile(
+    r"'(?:\\[\s\S]|''|[^'\\])*'"
+    r'|"(?:\\[\s\S]|""|[^"\\])*"'
+    r"|`(?:``|[^`])*`"
+    r"|--(?=\s|\Z)[^\n]*|#[^\n]*|/\*[\s\S]*?(?:\*/|\Z)"
+    r"|['\"`][\s\S]*"
+    r"|;"
+)
+# 语句开头的空白、注释
+SQL_LEADING_PATTERN = re.compile(r"(?:\s|--(?=\s|\Z)[^\n]*|#[^\n]*|/\*[\s\S]*?(?:\*/|\Z))*")
+CREATE_TABLE_PATTERN = re.compile(r"CREATE\s+TABLE\s", flags=re.IGNORECASE)
+
+
+def split_statements(content: str) -> List[str]:
+    """按分号拆分 SQL 语句，忽略引号、注释内的分号
+
+    仅用于定位顶层语句，不是完整的 SQL 解析器。返回的语句保留原文，去掉开头的空白、注释，以分号结尾
+
+    Args:
+        content (str): sql 内容
+
+    Returns:
+        List[str]: 语句列表
+    """
+    statements = []
+    start = 0
+    for match in SQL_TOKEN_PATTERN.finditer(content):
+        if match.group() != ";":
+            continue
+        statement = content[start:match.end()]
+        start = match.end()
+        statement = statement[SQL_LEADING_PATTERN.match(statement).end():]
+        if statement != ";":
+            statements.append(statement)
+    return statements
+
+
+def table_scripts(content: str) -> List[str]:
+    """获得真实的 CREATE TABLE 语句，不包括 INSERT 数据、注释中的建表文本
+
+    相关 issue：https://github.com/YunaiV/ruoyi-vue-pro/issues/1273
+    """
+    return [statement for statement in split_statements(content) if CREATE_TABLE_PATTERN.match(statement)]
+
+
 def load_and_clean(sql_file: str) -> str:
     """加载源 SQL 文件，并清理内容方便下一步 ddl 解析
 
@@ -52,6 +98,19 @@ def load_and_clean(sql_file: str) -> str:
     Returns:
         str: 清理后的sql文件内容
     """
+    # utf-8-sig：只去掉文件开头的 BOM，避免首条语句无法识别
+    return clean(open(sql_file, encoding="utf-8-sig").read())
+
+
+def clean(content: str) -> str:
+    """清理 SQL 内容方便下一步 ddl 解析
+
+    Args:
+        content (str): sql 内容
+
+    Returns:
+        str: 清理后的sql内容
+    """
     REPLACE_PAIR_LIST = (
         (")\nVALUES ", ") VALUES "),
         (" CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci ", " "),
@@ -62,7 +121,6 @@ def load_and_clean(sql_file: str) -> str:
         ("b'1'", "'1'"),
     )
 
-    content = open(sql_file, encoding="utf-8").read()
     for replace_pair in REPLACE_PAIR_LIST:
         content = content.replace(*replace_pair)
     # 移除所有 CHARACTER SET / COLLATE 变体 (utf8mb3、utf8 等)
@@ -88,10 +146,14 @@ class Convertor(ABC):
         self.db_type = db_type
         self.content = load_and_clean(self.src)
         # original_content 保留原始 COMMENT 信息，用于注释提取
-        self.original_content = open(src, encoding="utf-8").read()
-        # 剥离列级 COMMENT 以避免 COMMENT 值内的分号截断 CREATE TABLE 正则
-        content_no_comment = re.sub(r" COMMENT '(?:[^'\\]|\\.)*'", "", self.content)
-        self.table_script_list = re.findall(r"CREATE TABLE [^;]*;", content_no_comment)
+        self.original_content = open(src, encoding="utf-8-sig").read()
+        # 在原始内容上扫描建表语句，避免全局清理破坏引号边界；每个建表语句再单独清理
+        self.original_table_script_list = table_scripts(self.original_content)
+        # 剥离列级 COMMENT，交给 DDLParser 解析
+        self.table_script_list = [
+            re.sub(r" COMMENT '(?:[^'\\]|\\.)*'", "", clean(table_sql))
+            for table_sql in self.original_table_script_list
+        ]
 
     @abstractmethod
     def translate_type(self, type: str, size: Optional[Union[int, Tuple[int]]]) -> str:
@@ -330,7 +392,7 @@ class Convertor(ABC):
             )
 
         error_scripts = []
-        for table_sql in self.table_script_list:
+        for table_sql, orig_table_sql in zip(self.table_script_list, self.original_table_script_list):
             # 剥离 COMMENT 子句避免 DDLParser 无法处理中文等特殊字符
             table_sql_for_parse = re.sub(r"\s+COMMENT\s+'[^']*'", "", table_sql)
             ddl = DDLParser(table_sql_for_parse.replace("`", "")).run()
@@ -354,14 +416,7 @@ class Convertor(ABC):
                 continue
 
             # 解析注释
-            # 从原始 SQL 提取注释（支持中文等 DDLParser 不能处理的内容）
-            # 按表名定位完整建表段（以「换行 + )」起始的 ENGINE 行作为终止），避免被列 COMMENT 内的分号截断
-            orig_match = re.search(
-                rf"CREATE TABLE\s+`{re.escape(table_name)}`[\s\S]*?\n\)[^;]*;",
-                self.original_content,
-                flags=re.IGNORECASE,
-            )
-            orig_table_sql = orig_match.group() if orig_match else table_sql
+            # 从原始建表语句提取注释（支持中文等 DDLParser 不能处理的内容），避免 INSERT 数据中的同名建表文本干扰
             comments_dict = dict(Convertor.filed_comments(orig_table_sql))
             for column in table_ddl["columns"]:
                 column["comment"] = comments_dict.get(column["name"], "")
diff --git a/sql/tools/test_convertor.py b/sql/tools/test_convertor.py
new file mode 100644
index 0000000..8090c7a
--- /dev/null
+++ b/sql/tools/test_convertor.py
@@ -0,0 +1,314 @@
+# encoding=utf8
+"""convertor.py 的单元测试
+
+cd sql/tools
+uv run --with simple-ddl-parser python -m unittest test_convertor -v
+"""
+
+import contextlib
+import io
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from convertor import (
+    DM8Convertor,
+    HighGoConvertor,
+    KingbaseConvertor,
+    OpengaussConvertor,
+    OracleConvertor,
+    PostgreSQLConvertor,
+    SQLServerConvertor,
+    split_statements,
+)
+
+TOOLS_DIR = os.path.dirname(os.path.abspath(__file__))
+MYSQL_DIR = os.path.join(TOOLS_DIR, "..", "mysql")
+
+HEADER = """/*
+ Navicat Premium Data Transfer
+
+ Source Server Type    : MySQL
+*/
+
+SET NAMES utf8mb4;
+SET FOREIGN_KEY_CHECKS = 0;
+"""
+
+DEMO_TABLE = """
+-- ----------------------------
+-- Table structure for demo
+-- ----------------------------
+DROP TABLE IF EXISTS `demo`;
+CREATE TABLE `demo`  (
+  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号',
+  `name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '名称;别名',
+  PRIMARY KEY (`id`) USING BTREE
+) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '演示表';
+
+-- ----------------------------
+-- Records of demo
+-- ----------------------------
+BEGIN;
+INSERT INTO `demo` (`id`, `name`) VALUES (1, 'a;b');
+COMMIT;
+"""
+
+DEMO_DOC_TABLE = """
+DROP TABLE IF EXISTS `demo_doc`;
+CREATE TABLE `demo_doc`  (
+  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号',
+  `content` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL COMMENT '内容',
+  PRIMARY KEY (`id`) USING BTREE
+) ENGINE = InnoDB AUTO_INCREMENT = 3 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '文档表';
+"""
+
+# INSERT 数据中的建表文本：转义换行（issue #1273 的实际场景）
+ESCAPED_NEWLINE_INSERT = r"""
+INSERT INTO `demo_doc` (`id`, `content`) VALUES (1, 'Seata:\nCREATE TABLE `undo_log` (\n  `id` bigint NOT NULL,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB;\n');
+"""
+
+# INSERT 数据中的建表文本：真实换行
+LITERAL_NEWLINE_INSERT = """
+INSERT INTO `demo_doc` (`id`, `content`) VALUES (2, 'line1
+CREATE TABLE `ghost` (
+  `id` bigint NOT NULL
+);
+line3');
+"""
+
+# 注释中的建表文本
+COMMENTED_DDL = """
+/* CREATE TABLE `ghost_block` (`id` bigint NOT NULL); */
+-- CREATE TABLE `ghost_line` (`id` bigint NOT NULL);
+# CREATE TABLE `ghost_hash` (`id` bigint NOT NULL);
+"""
+
+
+def table_names(table_script_list):
+    return [re.match(r"CREATE\s+TABLE\s+`([^`]+)`", script).group(1) for script in table_script_list]
+
+
+def convert(convertor):
+    """执行转换，返回 stdout、stderr 的内容"""
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        convertor.print()
+    return out.getvalue(), err.getvalue()
+
+
+class SplitStatementsTest(unittest.TestCase):
+
+    def test_semicolon_in_quotes(self):
+        sql = r"""INSERT INTO `t` VALUES ('a;b', "c;d", 'it\'s;', 'x''y;', "q\";", 'C:\\');
+INSERT INTO `we;ird` VALUES (1);"""
+        self.assertEqual(sql.split("\n"), split_statements(sql))
+
+    def test_semicolon_in_comments(self):
+        sql = """-- comment; here
+# hash; it's
+/* block; here */ SELECT 1;
+CREATE TABLE `c` (
+  `id` bigint -- 编号; 注释
+);"""
+        self.assertEqual(["SELECT 1;", "CREATE TABLE `c` (\n  `id` bigint -- 编号; 注释\n);"], split_statements(sql))
+
+    def test_double_dash_without_space(self):
+        # MySQL 中 -- 后面必须跟空白，才是注释
+        self.assertEqual(["SELECT 5--2;"], split_statements("SELECT 5--2;"))
+
+    def test_multiple_statements_one_line(self):
+        self.assertEqual(["SET NAMES utf8mb4;", "SET FOREIGN_KEY_CHECKS = 0;"],
+                         split_statements("SET NAMES utf8mb4; SET FOREIGN_KEY_CHECKS = 0;"))
+
+    def test_unterminated(self):
+        # 未闭合的引号、缺少分号的结尾，都不作为语句
+        self.assertEqual(["SELECT 1;"], split_statements("SELECT 1; INSERT INTO `t` VALUES ('a;"))
+        self.assertEqual(["SELECT 1;"], split_statements("SELECT 1;\nSELECT 2"))
+
+
+class ConvertorTableScanTest(unittest.TestCase):
+
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.directory = directory.name
+
+    def write_sql(self, content):
+        path = os.path.join(self.directory, "dump.sql")
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(content)
+        return path
+
+    def test_create_table_in_insert_ignored(self):
+        path = self.write_sql(HEADER + DEMO_DOC_TABLE + ESCAPED_NEWLINE_INSERT + LITERAL_NEWLINE_INSERT
+                              + COMMENTED_DDL + DEMO_TABLE)
+        for convertor_class in (PostgreSQLConvertor, KingbaseConvertor, OpengaussConvertor, HighGoConvertor,
+                                OracleConvertor, SQLServerConvertor, DM8Convertor):
+            with self.subTest(convertor_class.__name__):
+                convertor = convertor_class(path)
+                self.assertEqual(["demo_doc", "demo"], table_names(convertor.table_script_list))
+
+                out, err = convert(convertor)
+                self.assertNotIn("无法正常解析", err)
+                self.assertIn("Table structure for demo_doc", out)
+                self.assertIn("Table structure for demo\n", out)
+                for ghost in ("undo_log", "ghost", "ghost_block", "ghost_line", "ghost_hash"):
+                    self.assertNotIn(f"Table structure for {ghost}\n", out)
+
+    def test_comment_from_real_table(self):
+        # 前面 INSERT 数据中存在同名的建表文本，不能影响真实建表语句的注释
+        fake_insert = """
+INSERT INTO `demo_doc` (`id`, `content`) VALUES (1, 'CREATE TABLE `demo`  (
+  `name` varchar(64) NOT NULL COMMENT \\'假名称\\',
+  PRIMARY KEY (`id`)
+) ENGINE = InnoDB COMMENT = \\'假表\\';
+');
+"""
+        path = self.write_sql(HEADER + DEMO_DOC_TABLE + fake_insert + DEMO_TABLE)
+        convertor = PostgreSQLConvertor(path)
+        self.assertEqual(["demo_doc", "demo"], table_names(convertor.table_script_list))
+        # 保留原始建表语句
+        self.assertIn("COMMENT '名称;别名'", convertor.original_table_script_list[1])
+
+        out, err = convert(convertor)
+        self.assertNotIn("无法正常解析", err)
+        self.assertIn("COMMENT ON COLUMN demo.id IS '编号';", out)
+        self.assertIn("COMMENT ON COLUMN demo.name IS '名称;别名';", out)
+        self.assertIn("COMMENT ON TABLE demo IS '演示表';", out)
+        self.assertIn("COMMENT ON TABLE demo_doc IS '文档表';", out)
+        comment_lines = [line for line in out.split("\n") if line.startswith("COMMENT ON")]
+        self.assertFalse([line for line in comment_lines if "假" in line])
+
+    def test_indented_and_same_line_create_table(self):
+        sql = HEADER + """
+    CREATE TABLE `indented`  (
+      `id` bigint NOT NULL COMMENT '编号',
+      PRIMARY KEY (`id`) USING BTREE
+    ) ENGINE = InnoDB COMMENT = '缩进表';
+CREATE TABLE `one` (`id` bigint NOT NULL, PRIMARY KEY (`id`)) ENGINE = InnoDB; CREATE TABLE `two` (`id` bigint NOT NULL, PRIMARY KEY (`id`)) ENGINE = InnoDB;
+"""
+        convertor = PostgreSQLConvertor(self.write_sql(sql))
+        self.assertEqual(["indented", "one", "two"], table_names(convertor.table_script_list))
+
+        out, err = convert(convertor)
+        self.assertNotIn("无法正常解析", err)
+        self.assertIn("COMMENT ON COLUMN indented.id IS '编号';", out)
+        self.assertIn("COMMENT ON TABLE indented IS '缩进表';", out)
+        self.assertIn("CREATE TABLE one (", out)
+        self.assertIn("CREATE TABLE two (", out)
+
+    def test_multiple_tables_with_inserts(self):
+        demo_item = """
+DROP TABLE IF EXISTS `demo_item`;
+CREATE TABLE `demo_item`  (
+  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号',
+  `demo_id` bigint NOT NULL COMMENT '演示编号',
+  PRIMARY KEY (`id`) USING BTREE
+) ENGINE = InnoDB COMMENT = '演示项';
+
+BEGIN;
+INSERT INTO `demo_item` (`id`, `demo_id`) VALUES (7, 1);
+INSERT INTO `demo_item` (`id`, `demo_id`) VALUES (8, 1);
+COMMIT;
+"""
+        convertor = PostgreSQLConvertor(self.write_sql(HEADER + DEMO_TABLE + demo_item))
+        out, err = convert(convertor)
+        self.assertNotIn("无法正常解析", err)
+        self.assertIn("INSERT INTO demo (id, name) VALUES (1, 'a;b');", out)
+        self.assertIn("INSERT INTO demo_item (id, demo_id) VALUES (7, 1);\n"
+                      "INSERT INTO demo_item (id, demo_id) VALUES (8, 1);", out)
+        self.assertIn("CREATE SEQUENCE demo_seq\n    START 2;", out)
+        self.assertIn("CREATE SEQUENCE demo_item_seq\n    START 9;", out)
+
+    def test_postgres_char(self):
+        sql = HEADER + """
+CREATE TABLE `demo_char`  (
+  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号',
+  `flag` char NOT NULL DEFAULT 'N' COMMENT '标记',
+  `code` char(2) NULL DEFAULT NULL COMMENT '编码',
+  PRIMARY KEY (`id`) USING BTREE
+) ENGINE = InnoDB COMMENT = '字符表';
+"""
+        out, err = convert(PostgreSQLConvertor(self.write_sql(sql)))
+        self.assertNotIn("无法正常解析", err)
+        self.assertIn("flag varchar(1) NOT NULL", out)
+        self.assertIn("code varchar(2) NULL", out)
+
+    def test_utf8_bom(self):
+        # 文件开头的 BOM 紧挨着首条 CREATE TABLE；字符串中的 BOM 需原样保留
+        sql = "\ufeff" + """CREATE TABLE `demo_bom`  (
+  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '编号',
+  `name` varchar(64) NOT NULL DEFAULT '' COMMENT '名称',
+  PRIMARY KEY (`id`) USING BTREE
+) ENGINE = InnoDB COMMENT = 'BOM 表';
+INSERT INTO `demo_bom` (`id`, `name`) VALUES (1, 'a\ufeffb');
+"""
+        path = self.write_sql(sql)
+        with open(path, "rb") as f:
+            self.assertTrue(f.read().startswith(b"\xef\xbb\xbf"))
+
+        convertor = PostgreSQLConvertor(path)
+        self.assertEqual(["demo_bom"], table_names(convertor.table_script_list))
+        self.assertTrue(convertor.original_table_script_list[0].startswith("CREATE TABLE `demo_bom`"))
+
+        out, err = convert(convertor)
+        self.assertNotIn("无法正常解析", err)
+        self.assertIn("CREATE TABLE demo_bom (", out)
+        self.assertIn("name varchar(64) NOT NULL", out)
+        self.assertIn("COMMENT ON COLUMN demo_bom.name IS '名称';", out)
+        self.assertIn("COMMENT ON TABLE demo_bom IS 'BOM 表';", out)
+        self.assertIn("INSERT INTO demo_bom (id, name) VALUES (1, 'a\ufeffb');", out)
+
+    def test_cli(self):
+        # 等价于：python convertor.py <type> dump.sql
+        sql = HEADER + DEMO_DOC_TABLE + ESCAPED_NEWLINE_INSERT + LITERAL_NEWLINE_INSERT + DEMO_TABLE
+        path = self.write_sql(sql)
+
+        # OceanBase：原样输出
+        result = subprocess.run([sys.executable, "convertor.py", "oceanbase", path],
+                                cwd=TOOLS_DIR, capture_output=True)
+        self.assertEqual(0, result.returncode, result.stderr.decode("utf-8"))
+        self.assertEqual(sql, result.stdout.decode("utf-8"))
+
+        # PostgreSQL：不误识别 INSERT 数据中的建表文本
+        result = subprocess.run([sys.executable, "convertor.py", "postgres", path],
+                                cwd=TOOLS_DIR, capture_output=True)
+        out, err = result.stdout.decode("utf-8"), result.stderr.decode("utf-8")
+        self.assertEqual(0, result.returncode, err)
+        self.assertNotIn("无法正常解析", err)
+        self.assertIn("Table structure for demo_doc", out)
+        self.assertIn("Table structure for demo\n", out)
+        self.assertNotIn("Table structure for undo_log", out)
+        self.assertNotIn("Table structure for ghost", out)
+
+
+class ConvertorOfficialDumpTest(unittest.TestCase):
+
+    def test_ruoyi_vue_pro(self):
+        convertor = PostgreSQLConvertor(os.path.join(MYSQL_DIR, "ruoyi-vue-pro.sql"))
+        names = table_names(convertor.table_script_list)
+        self.assertEqual(48, len(names))
+        self.assertEqual(len(names), len(set(names)))
+
+        out, err = convert(convertor)
+        self.assertNotIn("无法正常解析", err)
+        # 48 张表，加上 dual 表
+        self.assertEqual(49, out.count("-- Table structure for "))
+
+    def test_quartz(self):
+        convertor = PostgreSQLConvertor(os.path.join(MYSQL_DIR, "quartz.sql"))
+        names = table_names(convertor.table_script_list)
+        self.assertEqual(11, len(names))
+        self.assertTrue(all(name.startswith("QRTZ_") for name in names))
+
+        # quartz 的表不转换
+        out, _ = convert(convertor)
+        self.assertNotIn("qrtz", out.lower())
+
+
+if __name__ == "__main__":
+    unittest.main()