100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
简单的教职工数据导入脚本
|
||
"""
|
||
|
||
import os
|
||
import pandas as pd
|
||
from sqlalchemy import create_engine, text
|
||
|
||
# 数据库配置
|
||
DB_CONFIG = {
|
||
"host": "47.99.164.66",
|
||
"port": 3306,
|
||
"database": "hzdyjs_xg",
|
||
"user": "root",
|
||
"password": "hn02le.34lkdLKD",
|
||
"charset": "utf8mb4"
|
||
}
|
||
|
||
def main():
|
||
# 构建数据库连接
|
||
user = DB_CONFIG.get("user")
|
||
password = DB_CONFIG.get("password")
|
||
host = DB_CONFIG.get("host", "localhost")
|
||
port = DB_CONFIG.get("port", 3306)
|
||
database = DB_CONFIG.get("database")
|
||
charset = DB_CONFIG.get("charset", "utf8mb4")
|
||
uri = f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset={charset}"
|
||
engine = create_engine(uri, pool_pre_ping=True)
|
||
|
||
# Excel文件路径
|
||
excel_file = "人员清单列表1@正式员工.xlsx"
|
||
table_name = "temp_staff_data"
|
||
|
||
if not os.path.exists(excel_file):
|
||
print(f"找不到文件: {excel_file}")
|
||
return
|
||
|
||
print(f"读取Excel文件: {excel_file}")
|
||
|
||
# 读取Excel文件
|
||
df = pd.read_excel(excel_file, engine='openpyxl')
|
||
print(f"读取到 {len(df)} 行数据")
|
||
print(f"列名: {list(df.columns)}")
|
||
|
||
# 清理列名
|
||
df.columns = [col.replace('（', '_').replace('）', '_').replace('、', '_').replace(' ', '_') for col in df.columns]
|
||
print(f"清理后的列名: {list(df.columns)}")
|
||
|
||
# 创建临时表
|
||
print(f"创建临时表: {table_name}")
|
||
|
||
# 先删除已存在的表
|
||
with engine.begin() as conn:
|
||
conn.execute(text(f"DROP TABLE IF EXISTS `{table_name}`"))
|
||
|
||
# 根据实际列名创建表结构
|
||
create_sql = f"CREATE TABLE `{table_name}` (\n"
|
||
create_sql += " `id` BIGINT(20) NOT NULL AUTO_INCREMENT,\n"
|
||
|
||
for col in df.columns:
|
||
# 清理列名，移除特殊字符
|
||
clean_col = col.replace('（', '_').replace('）', '_').replace('、', '_').replace(' ', '_')
|
||
create_sql += f" `{clean_col}` TEXT DEFAULT NULL,\n"
|
||
|
||
create_sql += " `import_time` DATETIME DEFAULT CURRENT_TIMESTAMP,\n"
|
||
create_sql += " PRIMARY KEY (`id`)\n"
|
||
create_sql += ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;"
|
||
|
||
print("创建表SQL:")
|
||
print(create_sql)
|
||
|
||
with engine.begin() as conn:
|
||
conn.execute(text(create_sql))
|
||
|
||
# 导入数据
|
||
print("导入数据...")
|
||
df.to_sql(name=table_name, con=engine, index=False, if_exists="append", chunksize=1000, method="multi")
|
||
|
||
# 验证结果
|
||
with engine.begin() as conn:
|
||
result = conn.execute(text(f"SELECT COUNT(*) as count FROM {table_name}"))
|
||
count = result.fetchone()[0]
|
||
print(f"成功导入 {count} 条记录")
|
||
|
||
# 显示前5行
|
||
result = conn.execute(text(f"SELECT * FROM {table_name} LIMIT 5"))
|
||
columns = result.keys()
|
||
print(f"\n前5行数据预览:")
|
||
print("-" * 80)
|
||
print(f"{' | '.join(columns)}")
|
||
print("-" * 80)
|
||
for row in result:
|
||
print(f"{' | '.join(str(cell) if cell is not None else '' for cell in row)}")
|
||
|
||
print("\n导入完成！")
|
||
|
||
if __name__ == "__main__":
|
||
main() |