all-project/教职工数据迁移/simple_import.py

#!/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": "nsyOfyOyoFCr%Brj-k_5PqbW",
    "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()