概述

业务中有时会需要解析excel中的数据，按照要求处理后，写入到db中；
用python处理这个正好简便快捷

demo

没有依赖就 pip install pymysql一下

import pymysql
from pymysql.converters import escape_string
from openpyxl import load_workbook
from Snowflake import Snowflakedef load_excel_data(snowflake):# 连接到MySQL数据库mydb = pymysql.connect(host="xxx.xxx.xxx.xxx",port=3306,user="xxx",passwd="xxx",db="xxxx")# 打开Excel文件wb = load_workbook(filename=r'D:\xx\test.xlsx')sheet = wb.active# 获取表头header = [cell.value for cell in sheet[1]]column_header = []# 表头转换列名for excel_head_name in header:if '11' == excel_head_name:column_header.append("xx")elif '22' == excel_head_name:column_header.append("xx")elif '33' == excel_head_name:column_header.append("xx")elif '1122' == excel_head_name:column_header.append("xx")# 遍历每一行数据，并将其插入到数据库中cursor = mydb.cursor()count = 0defaultUser = "'xxx'"for row in sheet.iter_rows(min_row=2, values_only=True):cId = snowflake.next_id()date = row[0]# datetime 转 datedate = date.date()a2 = row[1]reason = row[2]detail = row[3]# \'%s\' 将含有特殊内容的字符串整个塞进去sql = f"INSERT INTO test_table (id, store_id, num, handler, create_by, update_by, date, a2, reason, detail) VALUES ({cId}, 3, 0, 43, {defaultUser}, {defaultUser}, \'%s\', \'%s\', \'%s\', \'%s\')" % (date, self_escape_string(a2), self_escape_string(reason), self_escape_string(detail))print(sql)# cursor.execute(sql, row)cursor.execute(sql)count += 1print(f"正在插入{count}条数据")# 提交更改并关闭数据库连接mydb.commit()cursor.close()mydb.close()# 将字符串中的特殊字符转义
# python中没有null只有None
def self_escape_string(data):if data is None:return ""return escape_string(data)if __name__ == '__main__':worker_id = 1data_center_id = 1snowflake = Snowflake(worker_id, data_center_id)load_excel_data(snowflake)

雪花id生成主键

import time
import randomclass Snowflake:def __init__(self, worker_id, data_center_id):### 机器标识IDself.worker_id = worker_id### 数据中心IDself.data_center_id = data_center_id### 计数序列号self.sequence = 0### 时间戳self.last_timestamp = -1def next_id(self):timestamp = int(time.time() * 1000)if timestamp < self.last_timestamp:raise Exception("Clock moved backwards. Refusing to generate id for %d milliseconds" % abs(timestamp - self.last_timestamp))if timestamp == self.last_timestamp:self.sequence = (self.sequence + 1) & 4095if self.sequence == 0:timestamp = self.wait_for_next_millis(self.last_timestamp)else:self.sequence = 0self.last_timestamp = timestampreturn ((timestamp - 1288834974657) << 22) | (self.data_center_id << 17) | (self.worker_id << 12) | self.sequencedef next_id(self):timestamp = int(time.time() * 1000)if timestamp < self.last_timestamp:raise Exception("Clock moved backwards. Refusing to generate id for %d milliseconds" % abs(timestamp - self.last_timestamp))if timestamp == self.last_timestamp:self.sequence = (self.sequence + 1) & 4095if self.sequence == 0:timestamp = self.wait_for_next_millis(self.last_timestamp)else:self.sequence = 0self.last_timestamp = timestampreturn ((timestamp - 1288834974657) << 22) | (self.data_center_id << 17) | (self.worker_id << 12) | self.sequencedef wait_for_next_millis(self, last_timestamp):timestamp = int(time.time() * 1000)while timestamp <= last_timestamp:timestamp = int(time.time() * 1000)return timestamp