本文介绍一种基于 ast.literal_eval 的稳健方法，从含复杂字段（如嵌套逗号、引号、换行）的 SQL INSERT 语句中准确提取列名与对应值，生成标准字典列表，避免正则或简单字符串分割导致的解析错误。
本文介绍一种基于 `ast.literal_eval` 的稳健方法，从含复杂字段（如嵌套逗号、引号、换行）的 sql insert 语句中准确提取列名与对应值，生成标准字典列表，避免正则或简单字符串分割导致的解析错误。
在数据库迁移或 ETL 场景中，常需从原始 .sql 文件（含 CREATE TABLE 和 INSERT INTO ... VALUES (...)）中提取插入数据，并映射到目标表结构。但直接使用 split(',') 或正则极易出错——因为字段值本身可能包含逗号（如 'Sport,Food,Creature'）、单引号、括号甚至多行文本。Python 标准库中的 ast.literal_eval() 提供了安全、语法感知的字面量解析能力，恰好适用于将 SQL 中格式良好的括号表达式转换为原生 Python 数据结构。
以下是一个完整、可复用的解析方案：
✅ 核心思路
-
分离 SQL 语句结构：首行提取列名（
INSERT INTO table (col1, col2, ...) VALUES），后续行为值元组； -
标准化 SQL 字段名语法：将反引号
`col`替换为单引号'col'，使其符合 Python 字面量格式； -
安全解析列名与数据：用
ast.literal_eval()将清洗后的字符串转为list[str]（列名）和list[tuple]（每行值）； -
组合为字典列表：通过
zip(fieldnames, row)构建键值对，最终生成[dict, dict, ...]。
? 示例代码（支持多行 INSERT）
import ast
import re
def parse_sql_inserts(sql_content: str) -> list[dict]:
# 按行分割，过滤空行和注释
lines = [line.strip() for line in sql_content.splitlines() if line.strip() and not line.strip().startswith('--')]
# 定位 INSERT 块（支持跨多行的 VALUES 部分）
insert_start = None
for i, line in enumerate(lines):
if line.upper().startswith('INSERT INTO'):
insert_start = i
break
if insert_start is None:
raise ValueError("No INSERT INTO statement found")
# 合并从 INSERT 开始直到下一个非值行（如分号或另一条语句）的所有内容
values_lines = []
i = insert_start
while i < len(lines):
line = lines[i]
# 截断至 VALUES 关键字后，并收集所有值行（含括号内的多行）
if 'VALUES' in line.upper():
# 取 VALUES 后的内容，去除末尾分号
rest = line.upper().split('VALUES', 1)[-1].strip().rstrip(';')
if rest:
values_lines.append(rest)
i += 1
while i < len(lines) and not lines[i].upper().startswith(('INSERT INTO', 'CREATE', 'UPDATE', 'DELETE')):
cleaned_line = lines[i].strip().rstrip(';')
if cleaned_line:
values_lines.append(cleaned_line)
i += 1
break
i += 1
if not values_lines:
raise ValueError("No VALUES clause found")
# 提取列名：从第一行中提取 `(col1, col2, ...)` 部分
first_line = lines[insert_start]
# 匹配括号内字段（支持反引号包裹）
cols_match = re.search(r'\(([^()]*)\)\s+VALUES', first_line, re.IGNORECASE | re.DOTALL)
if not cols_match:
raise ValueError("Failed to extract column names from INSERT statement")
# 清洗列名：将 `col` → 'col'，保留原始顺序
cols_raw = cols_match.group(1)
# 替换反引号为单引号，并确保每个字段被正确引用
cols_clean = cols_raw.replace('`', "'")
# 补齐缺失引号（若未用反引号，则按逗号分割后加引号）
if "'" not in cols_clean:
cols_clean = ", ".join(f"'{col.strip()}'" for col in cols_raw.split(','))
try:
fieldnames = ast.literal_eval(f'[{cols_clean}]')
except (SyntaxError, ValueError) as e:
raise ValueError(f"Invalid column definition: {cols_raw}") from e
# 解析 VALUES 部分：合并为单个字符串并解析为 tuple 列表
values_str = " ".join(values_lines)
# 确保以括号包围（ast.literal_eval 要求合法字面量）
if not values_str.strip().startswith('('):
values_str = '(' + values_str.strip() + ')'
try:
# ast.literal_eval 支持嵌套元组，能正确处理带引号的字符串、数字、None 等
data_tuples = ast.literal_eval(values_str)
# 统一为 list of tuples（兼容单行或多个元组）
if isinstance(data_tuples, tuple) and len(data_tuples) > 0 and isinstance(data_tuples[0], tuple):
data = list(data_tuples)
else:
data = [data_tuples] if isinstance(data_tuples, tuple) else []
except (SyntaxError, ValueError) as e:
raise ValueError(f"Failed to parse VALUES: {values_str[:100]}...") from e
# 构建字典列表
return [dict(zip(fieldnames, row)) for row in data]
# 使用示例
sql_example = """INSERT INTO `examle` (`myid`, `title`, `value`, `usefull`, `picture`, `short_description`, `description`, `wikipedia`, `category`, `date`) VALUES
(1, 'Cat', '3,45', 'No', 'cat.jpg', 'A very useless domestic cat.', 'The cat (Felis catus), also known as the domestic cat or housecat to distinguish it from other felines and felids, is a small carnivorous mammal that is valued by humans for its companionship and its ability to hunt vermin and household pests. It has been associated with humans for at least 9,500 years and is currently the most popular pet in the world.', 'http://en.wikipedia.org/wiki/Cat', 'Sport,Food,Creature', '2009-10-27 16:37:49'),
(3, 'Basketball', '9,45', 'Yes', 'basketball.jpg', 'A Baskeball sport utility.', 'Basketball is a team sport in which two teams of 5 players try to score points against one another by placing a ball through a 10 foot (3.048 m) high hoop (the goal) under organized rules. Basketball is one of the most popular and widely viewed sports in the world.', 'http://en.wikipedia.org/wiki/Basketball', 'Sport', '2009-10-27 16:36:39');"""
result = parse_sql_inserts(sql_example)
for item in result[:2]: # 打印前两条
print(item)⚠️ 注意事项与局限性
-
安全性：
ast.literal_eval()仅解析基础字面量（str,int,float,tuple,list,dict,None,True,False），不执行任意代码，比eval()安全得多； -
SQL 格式要求：该方法假设
INSERT语句格式规范（列名用反引号/单引号包裹，值为标准 SQL 字面量）。不支持子查询、函数调用（如NOW()）、变量或 MySQL 特有语法（如DEFAULT）； -
编码与转义：确保
.sql文件为 UTF-8 编码；若含转义字符（如\'），需预先解码（可用codecs.decode(..., 'unicode_escape')）； -
大规模文件处理：对超大文件，建议逐块读取 + 正则定位
INSERT块，避免内存溢出； -
健壮性增强：生产环境推荐结合
sqlparse库进行词法分析，或使用专用 SQL 解析器（如moz-sql-parser）处理更复杂语法。
该方案平衡了简洁性与可靠性，是处理“手工导出型”SQL 数据迁移任务的实用首选。
