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

曾经终败给现在 2022-12-30 10:54 7阅读 0赞

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

###### Python 2 ######

content_str += binascii.hexlify(b)

###### Python 3 ######

content_str += hex(b)[2:].zfill(2)

完整源码如下：

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