通用数据入库Python源码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

import pymysql as py
import pandas as pd
from sqlalchemy import create_engine, inspect
from urllib.parse import quote_plus
import time

word='table_name' #table_name修改为你自己的表名称

sheet_name='sheet1' #你的sheet页名称
path=r'your_path' #你的excel表本地路径
path=path+"\\"+word+".xlsx"

# 运行开始时间
start_time = time.time()
print('开始时间：' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start_time)))
print("-"*30)

# 读取内容
df=pd.read_excel(path)
rows_before=df.shape[0] #行数

# 运行结束时间
end_time = time.time()
print('结束时间：' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end_time)))

# 计算运行时长
duration = int(end_time - start_time)

# 将秒数转换为"几小时几分几秒"的格式
hours = duration // 3600
minutes = (duration % 3600) // 60
seconds = duration % 60

# 输出格式化后的运行时长
print(f"读取完成，处理用时：{hours}小时 {minutes}分 {seconds}秒")
print("")

excel_file1=path

df1=pd.read_excel(excel_file1,sheet_name=sheet_name)

password='password' #你的数据库密码

# 创建MySQL引擎
engine = create_engine(f'mysql+pymysql://你的用户名:{quote_plus(password)}@你的IP地址:端口号/数据库名称')

# 获取数据库表的列名
inspector = inspect(engine)
columns_to_insert = inspector.get_columns('orders')
columns_to_insert = [col['name'] for col in columns_to_insert]  # 提取列名

# 筛选 DataFrame 中存在的列（忽略多余的列）
columns_to_insert = [col for col in columns_to_insert if col in df1.columns]
if not columns_to_insert:
    print("数据库表中没有匹配的列，无法插入数据。")
else:
    df_to_insert = df1[columns_to_insert].copy()  # 仅选择需要插入的列

    # 插入数据到数据库
    try:
        df_to_insert.to_sql(name='orders', con=engine, if_exists='append', index=False)
        print(f"{word}插入成功: {df_to_insert.shape[0]}行, {df_to_insert.shape[1]}列")
    except Exception as e:
        print(f"数据插入失败，错误信息：{e}")

engine.dispose()

# 运行结束时间
end_time1 = time.time()
print('结束时间：' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end_time1)))

# 计算运行时长
duration1 = int(end_time1 - end_time)

# 将秒数转换为"几小时几分几秒"的格式
hours = duration1 // 3600
minutes = (duration1 % 3600) // 60
seconds = duration1 % 60
print("-"*30)
# 输出格式化后的运行时长
print(f"{word}数据插入成功，用时：{hours}小时 {minutes}分 {seconds}秒")