终端证书文件批量导入数据库的Python源码

用于将终端证书文件批量导入进数据库
【注意】Python2和Python3&＃xff0c;在数的进制转换上&＃xff0c;不太一样

Python 2

content_str &＃43;&＃61; binascii.hexlify(b)


Python 3

content_str &＃43;&＃61; hex(b)[2:].zfill(2)


完整源码如下&＃xff1a;

# -*- coding: utf-8 -*-
import os
import logging
import binascii#######################################################
# 运行程序&＃xff0c;需要先配置证书文件路径和结果文件名
# 需要处理的RTU_ID需要预先配置在文件rtu_id_list.txt中
# 如果程序运行正常&＃xff0c;结果文件中全部都是整齐的SQL语句
# 如果程序运行正常&＃xff0c;结果文件中会出现错误信息
# 所有错误信息均已[Error]作为行首
######################################################## 这个是证书文件&＃xff08;文件夹所在父级路径&＃xff09;
cert_file_base &＃61; &＃39;E:/test/PycharmProjects/ImportCert&＃39;
# 这个用于保存结果的文件名
sql_file_name &＃61; &＃39;import_cert.sql&＃39;# 从文件rtu_id_list.txt中读取rtu_id列表
def get_rtu_list():rtu_list &＃61; []try:with open(&＃39;rtu_id_list.txt&＃39;, &＃39;r&＃39;) as rtu_file:rtu_ids &＃61; rtu_file.readlines()for i in range(0, len(rtu_ids)):rtu_list.append(rtu_ids[i].strip())except IOError as err:logging.error(&＃39;[Error] when read rtu_id_list.txt: %s&＃39;, err)finally:return rtu_list# 根据rtu_id列表&＃xff0c;找到对应的证书文件&＃xff0c;读取并入库
def get_cert_file(rtu_list):for rtu_id in rtu_list:cert_file_dir &＃61; cert_file_base &＃43; &＃39;/&＃39; &＃43; rtu_idtry:cert_file_list &＃61; os.listdir(cert_file_dir)except OSError as err:logging.error(&＃39;[Error] when read %s: %s&＃39;, cert_file_dir, err)continueif len(cert_file_list) !&＃61; 1:logging.error("[Error] There be %d files in path: %s", len(cert_file_list), cert_file_dir)continueelse:print_sql(rtu_id, cert_file_dir, cert_file_list[0])# 针对一个具体的rtu&＃xff0c;生成SQL
def print_sql(rtu_id, cert_file_dir, cert_file_name):cert_file_path &＃61; cert_file_dir &＃43; "/" &＃43; cert_file_name# logging.info(&＃39;Cert file is: &＃39;&＃43;cert_file_path)try:with open(cert_file_path, &＃39;rb&＃39;) as cert_file:content_bin &＃61; cert_file.read()except IOError as err:logging.error(&＃39;[Error] when read %s: %s&＃39;, cert_file_path, err)content_str &＃61; &＃39;&＃39;for b in content_bin:content_str &＃43;&＃61; binascii.hexlify(b) # for Python 2# content_str &＃43;&＃61; hex(b)[2:].zfill(2) # for Python 3content_str &＃43;&＃61; &＃39; &＃39;sql &＃61; "insert into fes_cert(rtu_id,cert_file_name,cert_content) " \"values(" &＃43; rtu_id &＃43; ",&＃39;" &＃43; cert_file_name &＃43; "&＃39;,&＃39;" &＃43; content_str.strip() &＃43; "&＃39;); "logging.info(sql)# 程序启动时&＃xff0c;将结果文件清空
def clean_thing(file_name):try:file &＃61; open(file_name, &＃39;w&＃39;).close()except IOError as err:logging.error(&＃39;[Error] when clean %s: %s&＃39;, file_name, err)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:# 清空结果文件clean_thing(sql_file_name)# 配置日志&＃xff08;SQL文件&＃xff09;logging.basicConfig(level&＃61;logging.DEBUG, filename&＃61;sql_file_name, filemode&＃61;&＃39;a&＃39;, format&＃61;&＃39;%(message)s&＃39;)# 运行业务逻辑get_cert_file(get_rtu_list())