Python搞定爬取表映射文档翻译成执行语句（DB2）
作者企鹅：595696297 技术交流群：127591054 工作中用到所以开发了一个，。映射文档如下格式。格式按照自己需求参考。我这里有N列，可以参考。做必要改动即可。 下面是运行结果 其他SQL语句都是基本差不多的，看一下代码，略微改动一下，就可以完美移植各种数据库的映射语句。 简单说一下怎么使用下面是目录的结构，文档的格式在ExcelFile文件夹中。 1、首先要保证安装Python3.x版本。我这里用的是3.02、安装所需要包，用win+R进入CMD窗口输入：python -m pip install xlrd3、按照标准化文档的格式写映射文档，然后放到ExcelFile目录下，名字格式：数据标准化拆分-XXXXX4、按照自己需求配置，配置文件。同级目录：Config.txt import xlrd #import pandas as pd import collections import random import time from datetime import date,datetime import sys,os,io #Author JackChiang #遇到问题：联系QQ595696297哦！ #配置名称置为空等待读取 userspace = '' sheetname = '' if_add_flog = '' if_auto_load = '' if_auto_mkdir = '' #获取当前脚本所在路径 def cur_file_dir(): path = sys.path[0] #获取脚本路径 if os.path.isdir(path):#判断脚本是文件还是编译后的文件，如果是脚本返回脚本目录，如果是编译文件，返回编译文件路径 return path elif os.path.isfile(path): return os.path.dirname(path) #读取配置文件 def read_config(fileone): #申明修改全局变量 global userspace,sheetname,if_add_flog,if_auto_load,if_auto_mkdir f1 = open(fileone,'r+') while 1: lines = f1.readlines(10) if not lines: break for line in lines: line=line.strip('\n') file_key = line.split('=')[0] file_value = line.split('=')[1] if file_key == 'USERSPACE': userspace = file_value if file_key == 'SHEETNAME': sheetname = file_value if file_key == 'IF_ADD_FLOG': if_add_flog = file_value if file_key == 'IF_AUTO_LOAD': if_auto_load = file_value if file_key == 'IF_AUTO_MKDIR': if_auto_mkdir = file_value if userspace != '' and sheetname != '' and if_add_flog != '' and if_auto_load != '' and if_auto_mkdir != '': logMsg.write('配置文件读取完成！\n') return True else: logMsg.write('配置文件读取失败！\n') return False #搜索当前标准化文档的文件 def file_name(file_dir): L = [] #TRUE代表开启自动搜索全部文档，否则代表读取file文件目录 if if_auto_load != 'TRUE': f2 = open(file_dir+"\\file.txt",'r+') while 1: lines = f2.readlines(100) if not lines: break for line in lines: line=line.strip('\n') L.append(line) f2.close() print('从文件读取目录文件完成！') logMsg.write('从文件读取目录文件完成！\n') else: f1 = open(file_dir+"\\file.txt",'w') ExcelFile = file_dir+'\\ExcelFile' #清空文件内容 f1.truncate() for root, dirs, files in os.walk(ExcelFile): for file in files: filename = os.path.splitext(file)[0] filespl = os.path.splitext(file)[1] if '数据标准化拆分' in filename: if filespl == '.xls': path = '%s%s%s' % (root, '\\', file) L.append(path) f1.write(path) f1.write('\n') logMsg.write(file + '文件写入！！\n') print('写入目录文件完成！') logMsg.write('写入目录文件完成！\n') f1.close() return tuple(L) #读取Excel文件中的内容，找到没张表的位置 def read_excel_one(sheet): #开始一行一行遍历锁定ALL位置 flog = 0 flog_count = 0 L = [] j = 0 L1 = [] #dict = {} #使用有序字典 dict = collections.OrderedDict() for i in range(sheet.nrows): value = sheet.cell_value(i, 3) table_name = sheet.cell_value(i, 1) mode_value = sheet.cell_value(i, 0) if value == 'ALL': j = i if j != 1: L.append(j) L.append(i) L.append(table_name) L.append(mode_value) flog_count = flog_count + 1 #print('%s 找到啦！第%s次有ALL:%s'%(flog_count,i,table_name)) L.append(sheet.nrows-1) #print(L) #找到第一次出现ALL的位置代表表的开始，第二次出现的前一个位置代表结束。 #遍历列表 print(len(L)) for i in range(len(L)): #print(i+1,L[i]) if(i+1)%4==0: if i+1 != len(L): #print(L[i-3]) #print(L[i-2]) #print(L[i-1]) #print(L[i]) L1.append(L[i-1]) L1.append(L[i-3]) L1.append(L[i]-1) dict[L[i-2]] = L1 L1 = [] else: #处理最后一个情况 L1.append(L[i-1]) L1.append(L[i-3]) L1.append(L[i]) dict[L[i-2]] = L1 L1 = [] #print(dict) return dict #创建目录 def mkdir_one(fileone,cell_value): path_one = fileone + '\\' + cell_value # 判断路径是否存在 # 存在 True # 不存在 False isExists=os.path.exists(path_one) # 判断结果 if not isExists: # 如果不存在则创建目录 # 创建目录操作函数 os.makedirs(path_one) logMsg.write(path_one +' 创建成功\n') else: # 如果目录存在则不创建，并提示目录已存在 logMsg.write(path_one +' 目录已存在\n') return path_one #翻译语句 def read_table_one(dict,sheet,fileone): ddlFile = fileone + '\\DDL' defFile = fileone + '\\DDL\\DEFAULT_DDL' #cell_value = sheet.cell_value(2,0) #path_one = mkdir_one(ddlFile,cell_value) #判断是否需要创建文件夹，还是使用默认 path_one = defFile #遍历字典取值 for k,v in dict.items(): cell_value = v[0] #BANK : ['IMBS',1, 18] if if_auto_mkdir == 'TRUE': path_one = mkdir_one(ddlFile,cell_value) logMsg.write('%s 表的行范围为： %s\n'%(k,v)) f1 = open('%s\\%s_%s.ddl'%(path_one,cell_value,k),'w',encoding='utf-8') page = '--------------------------------------------------\n' f1.write(page) f1.write('-- Create Table '+ cell_value + '.'+ k+'\n') f1.write(page) f1.write('Create Table '+ cell_value + '.'+ k+'(\n') row_pri_name = '' row_table_decs = '' Str_com = [] for i in range(v[1]+1,v[2]+1): row_data = sheet.row_values(i) if row_data[5] == 'INTEGER': table_cloumn_type = row_data[5] elif row_data[5] == 'TIMESTAMP': table_cloumn_type = row_data[5] elif row_data[5] == 'DATE': table_cloumn_type = row_data[5] elif row_data[5] == 'BIGINT': table_cloumn_type = row_data[5] elif row_data[5] == 'DECIMAL': row_data7 = row_data[7] if row_data[6] == '': logMsg.write('在%s表中的%s字段长度没有写！！\n'%(k,row_data[3])) return; if row_data[7] == '': logMsg.write('在%s表中的%s字段精度没有写！！默认为0 \n'%(k,row_data[3])) row_data7 = 0 table_cloumn_type = '%s(%d,%d)'%(row_data[5],row_data[6],row_data7) else: table_cloumn_type = '%s(%d)'%(row_data[5],row_data[6]) if row_data[10] == 'Physical Primary Key': table_pri = ' NOT NULL ,' if row_pri_name == '': row_pri_name = row_data[3]+',' else: row_pri_name = row_pri_name+row_data[3]+',' else: table_pri = ' ,' #读到最后一行，去掉逗号 if i == v[2] and if_add_flog != 'TRUE': table_pri = table_pri[:-1]+')' #拼接字段 f1.write(' '+row_data[3]+' '+table_cloumn_type+table_pri) f1.write('\n') #注解放入列表 if row_data[4] != '': Str_com.append('Comment on Column %s.%s.%s is \'%s\';'%(cell_value,k,row_data[3],row_data[4])) row_table_decs = row_data[2] #print(row_data) if if_add_flog == 'TRUE': f1.write(' EFF_DT DATE NOT NULL ,\n') f1.write(' END_DT DATE ,\n') f1.write(' JOB_SEQ_ID INTEGER )\n') f1.write('in %s\n'%(userspace)) f1.write('Partitioning Key ('+row_pri_name+'EFF_DT) Using Hashing\n') f1.write('Compress Yes;\n') else: f1.write('in %s\n'%(userspace)) f1.write('Partitioning Key ('+row_pri_name[:-1]+') Using Hashing\n') f1.write('Compress Yes;\n') #创建注解 f1.write('Comment on Table '+cell_value + '.'+ k+' is \''+ row_table_decs + '\';\n') for i in Str_com: f1.write(i) f1.write('\n') f1.write('\n') if if_add_flog == 'TRUE': #创建索引 f1.write(page) f1.write('-- Create Index '+cell_value + '.'+ k+'_'+time.strftime("%Y%m%d")+'_1\n') f1.write(page) f1.write('Create Index '+cell_value + '.'+ k+'_'+time.strftime("%Y%m%d")+'_1\n') f1.write(' on '+cell_value + '.'+ k+'\n') f1.write(' (END_DT) Allow Reverse Scans;\n') f1.write('\n') f1.write(page) f1.write('-- Create Index '+cell_value + '.'+ k+'_'+time.strftime("%Y%m%d")+'_2\n') f1.write(page) f1.write('Create Index '+cell_value + '.'+ k+'_'+time.strftime("%Y%m%d")+'_2\n') f1.write(' on '+cell_value + '.'+ k) f1.write(' (JOB_SEQ_ID) Allow Reverse Scans;\n') f1.write('\n') #创建主键 f1.write(page) SQLKey = random.randint(100000000000000,999999999999999) f1.write('-- Create Primary Key SQL%d\n'%SQLKey) f1.write(page) f1.write('alter table '+cell_value + '.'+ k+'\n') f1.write(' add constraint SQL%d\n'%SQLKey) if if_add_flog == 'TRUE': f1.write('Primary Key ('+row_pri_name+' EFF_DT);\n') else: f1.write('Primary Key ('+row_pri_name[:-1]+');\n') f1.close() def execute_mode(L_name): #循环处理文档 for L in L_name: ExcelFile=xlrd.open_workbook(L) path_name = L.split('\\')[-1] SheelList = sheetname.split(',') #获取目标EXCEL文件sheet名 L_Sheetname = [] Sheetname_one = ExcelFile.sheet_names() for Sheet_NM in Sheetname_one: for Shee_t in SheelList: if Sheet_NM == Shee_t: L_Sheetname.append(Sheet_NM) else: pass #print(L_Sheetname) for Sheet_NM in L_Sheetname: sheet=ExcelFile.sheet_by_name(Sheet_NM) print('文件名：%s Sheet名字：%s Sheet行数：%s Sheet列数：%s'%(path_name,sheet.name,sheet.nrows,sheet.ncols)) logMsg.write('文件名：%s Sheet名字：%s Sheet列数：%s Sheet行数：%s\n'%(path_name,sheet.name,sheet.nrows,sheet.ncols)) #rows=sheet.row_values(2)#第三行内容 #cols=sheet.col_values(1)#第二列内容 #value = sheet.cell_value(1, 3) #print(value) dict = read_excel_one(sheet) read_table_one(dict,sheet,fileone) L_Sheetname = [] print('文件：%s 处理完成！！！'%path_name) logMsg.write('文件：%s 处理完成！！！\n'%path_name) print('===========================================\n') logMsg.write('===========================================\n') print('===========================================\n') logMsg.write('===========================================\n') #用pandas 实现（暂时放弃） #def read_excel_two(filepath): # df = pd.read_excel(filepath) # print(df.shape) # print(df.dtypes) # #print(df[df.字段代码 == 'ALL']) ################################################### ###################开始################## #记录日志 fileone = cur_file_dir() logMsg = open(fileone+"\\RunLog.txt",'w',encoding='utf-8') con_flog = read_config(fileone+"\\Config.txt") v1 = '%s、表空间已经设置为：%s:'%(1,userspace); v2 = '%s、sheet页名字已经设定为：%s:'%(2,sheetname) v3 = '%s、是否加ODS末尾字段状态：%s:'%(3,if_add_flog) v4 = '%s、是否自动读取ExcelFile文件夹文件状态为：%s:'%(4,if_auto_load) v5 = '%s、是否自动根据模式名创建文件状态为：%s:'%(5,if_auto_mkdir) print(v1) print(v2) print(v3) print(v4) print(v5) logMsg.write(v1+'\n') logMsg.write(v2+'\n') logMsg.write(v3+'\n') logMsg.write(v4+'\n') logMsg.write(v5+'\n') #可以从L取数，但这里灵活一点从文件取数 #如果需要自动识别所有，配置文件修改为TRUE，否则手动决定目录 L = file_name(fileone) #print(L[1]) if con_flog == True: execute_mode(L) print('程序执行完毕！！！') logMsg.write('=========程序执行完毕！！！============\n') logMsg.close() else: print('配置文件格式错误！！')
