01
注意：该读取方法只针对SQLyog导出的mysql脚本测试，其他数据库可能代码需要根据实际情况微调。
from io import StringIO
import pandas as pd
import re
def read_sql_script_all(sql_file_path, quotechar="'") -> (str, dict):
insert_check = re.compile(r"insert +into +`?(\w+?)`?\(", re.I | re.A)
with open(sql_file_path, encoding="utf-8") as f:
sql_txt = f.read()
end_pos = -1
df_dict = {}
while True:
match_obj = insert_check.search(sql_txt, end_pos+1)
if not match_obj:
break
table_name = match_obj.group(1)
start_pos = match_obj.span()[1]+1
end_pos = sql_txt.find(";", start_pos)
tmp = re.sub(r"\)( values |,)\(", "\n", sql_txt[start_pos:end_pos])
tmp = re.sub(r"[`()]", "", tmp)
df = pd.read_csv(StringIO(tmp), quotechar=quotechar)
dfs = df_dict.setdefault(table_name, [])
dfs.append(df)
for table_name, dfs in df_dict.items():
df_dict[table_name] = pd.concat(dfs)
return df_dict
-
sql_file_path：sql脚本的位置
-
quotechar：脚本中字符串是单引号还是双引号，默认使用单引号解析
-
一个字典，键是表名，值是该表对应的数据所组成的datafream对象
index_test：
df_dict = read_sql_script_all("D:/tmp/test.sql")
df = df_dict['index_test']
df.head(10)
def read_sql_script_by_tablename(sql_file_path, table_name, quotechar="'") -> (str, dict):
insert_check = re.compile(r"insert +into +`?(\w+?)`?\(", re.I | re.A)
with open(sql_file_path, encoding="utf-8") as f:
sql_txt = f.read()
end_pos = -1
dfs = []
while True:
match_obj = insert_check.search(sql_txt, end_pos+1)
if not match_obj:
break
start_pos = match_obj.span()[1]+1
end_pos = sql_txt.find(";", start_pos)
if table_name != match_obj.group(1):
continue
tmp = re.sub(r"\)( values |,)\(", "\n", sql_txt[start_pos:end_pos])
tmp = re.sub(r"[`()]", "", tmp)
df = pd.read_csv(StringIO(tmp), quotechar=quotechar)
dfs.append(df)
return pd.concat(dfs)
-
sql_file_path：sql脚本的位置 -
table_name：被读取的表名 -
quotechar：脚本中字符串是单引号还是双引号，默认使用单引号解析
-
该表所对应的datafream对象
df = read_sql_script_by_tablename("D:/tmp/test.sql", "index_test")
df.head()
02
from sqlalchemy import create_engine
import pandas as pd
import re
def load_sql2sqlite_conn(sqltxt_path):
create_rule = re.compile("create +table [^;]+;", re.I)
insert_rule = re.compile("insert +into [^;]+;", re.I)
with open(sqltxt_path, encoding="utf-8") as f:
sqltxt = f.read()
engine = create_engine('sqlite:///:memory:')
pos = -1
while True:
match_obj = create_rule.search(sqltxt, pos+1)
if match_obj:
pos = match_obj.span()[1]
sql = match_obj.group(0).replace("AUTO_INCREMENT", "")
sql = re.sub("\).+;", ");", sql)
engine.execute(sql)
match_obj = insert_rule.search(sqltxt, pos+1)
if match_obj:
pos = match_obj.span()[1]
sql = match_obj.group(0)
engine.execute(sql)
else:
break
tablenames = [t[0] for t in engine.execute(
"SELECT tbl_name FROM sqlite_master WHERE type='table';").fetchall()]
return tablenames, engine.connect()
-
sql_file_path：sql脚本的位置
-
两个元素的元组，第一个元素是表名列表，第二个元素是sqlite内存虚拟连接
tablenames, conn = load_sql2sqlite_conn("D:/tmp/test.sql")
tablename = tablenames[0]
print(tablename)
df = pd.read_sql(f"select * from {tablename};", conn)
df
python-docx无法处理 “doc格式” 文件，于是我这样做......
Excel中的xls、xlsx、xlsm混合文件，看我用Python如何统一处理......
