sql转csv脚本

progppy

功能：通过识别sql中的insert等语句，提取出信息并导出output.csv
更多介绍见代码注释

[Python] 纯文本查看 复制代码

?

001 002 003 004 005 006 007 008 009 010 011 012 013 014 015 016 017 018 019 020 021 022 023 024 025 026 027 028 029 030 031 032 033 034 035 036 037 038 039 040 041 042 043 044 045 046 047 048 049 050 051 052 053 054 055 056 057 058 059 060 061 062 063 064 065 066 067 068 069 070 071 072 073 074 075 076 077 078 079 080 081 082 083 084 085 086 087 088 089 090 091 092 093 094 095 096 097 098 099 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168

''' 把data.sql放到同级目录下，运行程序会输出output.csv   这段代码的功能是从一个 .sql 文件中解析数据，并将其导出到一个 .csv 文件中。以下是对代码的详细解释： 1. 文件读取与解析   import sqlparse from ast import literal_eval   with open('data.sql', encoding='utf-8') as f:     sql = f.read()   parsed = sqlparse.parse(sql) rows = [] sqlparse 用于解析 SQL 文件：通过 sqlparse.parse 方法将 SQL 文件的内容解析为一个列表，每个元素是一个 SQL 语句。 rows 用于存储解析后的数据：最终会将符合要求的数据行存入 rows 列表中。 2. 筛选 INSERT 语句   for stmt in parsed:     if stmt.get_type() == 'INSERT': stmt.get_type()：获取 SQL 语句的类型。这里只处理类型为 INSERT 的语句。 3. 提取 VALUES 数据 Python复制 split_result = str(stmt).split('VALUES') if len(split_result) < 2:     continue   values_part = re.search(r'VALUES\s*\((.*)\);?$', str(stmt), re.DOTALL | re.IGNORECASE) if not values_part:     continue split('VALUES')：将 SQL 语句按 VALUES 关键字分割，提取出 VALUES 后面的部分。 re.search：使用正则表达式提取 VALUES 的内容： VALUES\s*\((.*)\);?$：匹配 VALUES 后面的括号内的内容，并忽略行末的分号。 re.DOTALL：让点 (.) 匹配换行符。 re.IGNORECASE：不区分大小写。 4. 清理 Values 数据 Python复制 clean_values = (     values_part.group(1)     .replace('\\n', '[NEWLINE]')  # 将换行符替换成标记     .replace(r"\'", "'")          # 处理被转义的单引号     .replace('NULL', 'None')      # 将 NULL 转换为 None     .replace('"[', '[')           # 处理数组类型字段     .replace(']"', ']')           # 处理数组类型字段     .replace('[NEWLINE]', '\\n')  # 将标记换回换行符 ) values_part.group(1)：提取正则匹配的第一组内容（即括号内的数据部分）。 replace 方法：清理和转换数据，以使其符合 Python 的语法要求。 5. 构造 Python 元组 Python复制 clean_values = f"({clean_values})"  # 包裹为合法的 Python 元组 将清理后的字符串包裹成合法的 Python 元组形式（如 ('value1', 'value2')）。 6. 安全解析数据 Python复制 try:     parsed_data = literal_eval(clean_values)     if isinstance(parsed_data, tuple):         rows.append(parsed_data)     elif isinstance(parsed_data, list):         rows.extend(parsed_data) except (SyntaxError, ValueError) as e:     print(f"[解析失败] 错误类型: {type(e).__name__}")     print(f"问题数据片段: {clean_values[:100]}...")     continue literal_eval：安全地将字符串转换为 Python 数据结构。 异常处理：如果解析失败（如语法错误或值错误），会捕获异常并打印错误信息。 7. 导出数据到 CSV 文件 Python复制 import csv   with open('SecondScriptOutput.csv', 'w', newline='', encoding='utf-8') as f:     writer = csv.writer(f,         quoting=csv.QUOTE_ALL,  # 强制所有字段加引号         escapechar='\\'         # 明确转义符     )     writer.writerow(['ID', 'Category', 'Remark'])       clean_rows = []     for row in rows:         clean_row = [str(field).replace('\n', '\\n') if isinstance(field, str) else field                      for field in row]         clean_rows.append(clean_row)       writer.writerows(clean_rows) csv.writer：定义一个 CSV 写入器。 quoting=csv.QUOTE_ALL：强制将所有字段用双引号包裹。 escapechar='\\'：明确转义符为 \。 数据清理：将换行符 \n 替换为 \\n，避免 CSV 文件格式混乱。 8. 代码要点说明 正则表达式：re.search(r'VALUES\s*\((.*)\);?$', str(stmt), re.DOTALL | re.IGNORECASE) 用于提取 VALUES 部分的内容。 异常处理：try-except 用于处理解析失败的情况，避免程序崩溃。 换行符处理：在清理数据时，将换行符 \n 替换为 [NEWLINE]，并在导出时换回 \n，以防止 CSV 文件格式混乱。 9. 运行结果 程序会输出以下内容： 复制 成功加载 X 条数据 数据已导出到 SecondScriptOutput.csv 10. 注意事项 依赖库：需要安装 sqlparse 库。如果未安装，可以运行 pip install sqlparse。 编码问题：确保 data.sql 文件的编码为 UTF-8，否则可能读取异常。 SQL 语法兼容性：目前仅处理 INSERT 语句，对于复杂 SQL 语句（如包含子查询等）可能无法解析。 性能问题：对于大规模数据文件，建议优化字符串处理和正则表达式部分。 希望这段讲解能帮助你更好地理解代码！   '''   import sqlparse from ast import literal_eval   with open('data.sql', encoding='utf-8') as f:     sql = f.read()   parsed = sqlparse.parse(sql) rows = []   for stmt in parsed:     if stmt.get_type() == 'INSERT':         # 防御性提取 VALUES         import re         # 新增正则表达式提取VALUES内容         values_part = re.search(r'VALUES\s*\((.*)\);?$', str(stmt), re.DOTALL | re.IGNORECASE)         if not values_part: continue           clean_values = (             values_part.group(1)             .replace('\\n', '[NEWLINE]')             # 使用正则处理带引号的字段             .replace(r"\'", "'")  # 处理被转义的单引号             .replace('NULL', 'None')             # 构造Python元组             .replace('"[', '[').replace(']"', ']')  # 处理数组类型字段             .replace('[NEWLINE]', '\\n')         )           # 包裹为合法Python元组         clean_values = f"({clean_values})"           # 安全解析         try:             parsed_data = literal_eval(clean_values)             if isinstance(parsed_data, tuple):                 rows.append(parsed_data)             elif isinstance(parsed_data, list):                 rows.extend(parsed_data)         except (SyntaxError, ValueError) as e:             print(f"[解析失败] 错误类型: {type(e).__name__}")             print(f"问题数据片段: {clean_values[:100]}...")             continue   print(f"成功加载 {len(rows)} 条数据") # 在代码最后添加（print语句之前）   import csv with open('output.csv', 'w', newline='', encoding='utf-8') as f:     writer = csv.writer(f,         quoting=csv.QUOTE_ALL,  # 强制所有字段加引号         escapechar='\\'         # 明确转义符     )     writer.writerow(['ID', 'Category', 'Remark'])     clean_rows = []     for row in rows:         clean_row = [str(field).replace('\n', '\\n') if isinstance(field, str) else field                      for field in row]         clean_rows.append(clean_row)       writer.writerows(clean_rows)   print("数据已导出到 output.csv")

kover

没看到连接数据库在哪里设置
而且为什么是导出insert的，不应该是select吗，从数据库查数据导出

zxd12138

牛掰，有帮助

Jutean

好支持顶

JakinLeung

非常好代码，使我sql转换

jixuxu

牛掰，谢大佬

longtitude

膜拜大佬

zhou2957

正好学习了，感谢楼主

maduse

学习sql中

mujita666

厉害了，居然可以这样

Sdersweden

牛掰，感谢大佬分享。