164 lines
6.5 KiB
Python
164 lines
6.5 KiB
Python
import pandas as pd
|
||
import os
|
||
import urllib
|
||
from sqlalchemy import create_engine
|
||
|
||
# ==========================================
|
||
# 1. 全局配置 (Global Configuration)
|
||
# ==========================================
|
||
|
||
# 数据库连接信息
|
||
DB_CONFIG = {
|
||
"server": "192.168.110.114", # 你的服务器地址，例如: 192.168.1.100
|
||
"database": "CompanyDB", # 你的数据库名
|
||
"username": "peng", # 用户名
|
||
"password": "Cqbld123456.", # 密码
|
||
"driver": "ODBC Driver 18 for SQL Server" # 确保已安装此驱动
|
||
}
|
||
|
||
# 目标表配置
|
||
TARGET_TABLE_NAME = "customerProductType" # SQL Server 表名
|
||
TARGET_DB_SCHEMA = "warehouseOutbound" # [关键] 这里指定架构，例如 'dbo' 或 'production'
|
||
|
||
# Excel 列名映射到 SQL 字段名的逻辑键 (用于后续代码逻辑引用)
|
||
# 这里的 value 必须与 SQL 数据库中的实际字段名完全一致
|
||
SQL_COL_YEAR = "合同年份" # 数据库中存年份的字段名
|
||
SQL_COL_WORKSHOP = "车间号" # 数据库中存车间号的字段名
|
||
SQL_COL_ORDER = "工令号" # 数据库中存工令号的字段名
|
||
SQL_COL_MODEL = "客户型号" # 数据库中存客户型号的字段名
|
||
|
||
# ==========================================
|
||
# 2. 迁移任务清单 (Migration Tasks)
|
||
# ==========================================
|
||
# 可以在这里添加任意数量的文件配置
|
||
MIGRATION_TASKS = [
|
||
# --- 任务 1 ---
|
||
{
|
||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2022.xlsm", # Excel文件路径
|
||
"year": 2022, # 该文件对应的合同年份
|
||
"sheet_names": ["Sheet1"], # 指定要迁移的工作表名称列表
|
||
# 映射表: Excel列名 -> SQL字段名
|
||
"mapping": {
|
||
"车间号": SQL_COL_WORKSHOP,
|
||
"工令号": SQL_COL_ORDER,
|
||
"产品型号": SQL_COL_MODEL
|
||
# 可以添加其他非关键字段...
|
||
}
|
||
},
|
||
# --- 任务 2 ---
|
||
{
|
||
"file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2023（1-5月）.xlsm",
|
||
"year": 2023,
|
||
"sheet_names": ["Sheet1"], # 只迁移 "汇总" 表
|
||
"mapping": {
|
||
"车间号": SQL_COL_WORKSHOP,
|
||
"工令号": SQL_COL_ORDER,
|
||
"产品型号": SQL_COL_MODEL
|
||
}
|
||
}
|
||
]
|
||
|
||
# ==========================================
|
||
# 3. 核心逻辑
|
||
# ==========================================
|
||
|
||
def get_db_engine():
|
||
params = urllib.parse.quote_plus(
|
||
f"DRIVER={{{DB_CONFIG['driver']}}};"
|
||
f"SERVER={DB_CONFIG['server']};"
|
||
f"DATABASE={DB_CONFIG['database']};"
|
||
f"UID={DB_CONFIG['username']};"
|
||
f"PWD={DB_CONFIG['password']};"
|
||
f"TrustServerCertificate=yes;"
|
||
)
|
||
# 使用 fast_executemany 提高写入速度
|
||
return create_engine(f"mssql+pyodbc:///?odbc_connect={params}", fast_executemany=True)
|
||
|
||
def run_migration():
|
||
engine = get_db_engine()
|
||
print(f"连接数据库... [{TARGET_DB_SCHEMA}].[{TARGET_TABLE_NAME}]")
|
||
|
||
for task in MIGRATION_TASKS:
|
||
file_path = task['file_path']
|
||
year_val = task['year']
|
||
# mapping 的键(Key)是Excel列名，值(Value)是SQL列名
|
||
mapping = task['mapping']
|
||
|
||
if not os.path.exists(file_path):
|
||
print(f"文件不存在: {file_path}")
|
||
continue
|
||
|
||
print(f"\n-------- 处理文件: {os.path.basename(file_path)} --------")
|
||
|
||
try:
|
||
# 读取 Excel
|
||
xls_dict = pd.read_excel(file_path, sheet_name=task['sheet_names'])
|
||
if not isinstance(xls_dict, dict):
|
||
first_sheet = task['sheet_names'][0] if task['sheet_names'] else "Sheet1"
|
||
xls_dict = {first_sheet: xls_dict}
|
||
|
||
for sheet_name, df in xls_dict.items():
|
||
if df.empty: continue
|
||
|
||
# 1. 清洗表头：去除列名前后的空格 (防止 "车间 " 匹配不上 "车间")
|
||
df.columns = df.columns.astype(str).str.strip()
|
||
|
||
# 2. 【关键步骤】只筛选指定的源字段
|
||
# 我们只提取 mapping 字典中 key 定义的列
|
||
source_cols = list(mapping.keys())
|
||
|
||
# 检查 Excel 里是否缺列
|
||
missing_source = [c for c in source_cols if c not in df.columns]
|
||
if missing_source:
|
||
print(f" [跳过] 工作表 {sheet_name} 缺少源列: {missing_source}")
|
||
continue
|
||
|
||
# 3. 提取数据并重命名
|
||
# 先提取 -> 只有这几列
|
||
df_subset = df[source_cols].copy()
|
||
# 后重命名 -> 变成数据库的列名
|
||
df_subset.rename(columns=mapping, inplace=True)
|
||
|
||
# 4. 注入年份字段
|
||
df_subset[SQL_COL_YEAR] = year_val
|
||
|
||
# 此时 df_subset 的列名应该完全等于：SQL字段列表
|
||
|
||
# 5. 数据清洗
|
||
# 确保关键字段非空
|
||
subset_keys = [SQL_COL_YEAR, SQL_COL_WORKSHOP, SQL_COL_ORDER]
|
||
df_subset.dropna(subset=subset_keys, inplace=True)
|
||
|
||
# 确保唯一性
|
||
df_subset.drop_duplicates(subset=subset_keys, keep='first', inplace=True)
|
||
|
||
if df_subset.empty:
|
||
print(f" -> 工作表 {sheet_name} 清洗后无数据")
|
||
continue
|
||
|
||
print(f" -> 工作表 {sheet_name}: 准备写入 {len(df_subset)} 行...")
|
||
|
||
# 6. 写入数据库
|
||
try:
|
||
# 使用 engine.connect() 显式连接
|
||
with engine.connect() as conn:
|
||
df_subset.to_sql(
|
||
name=TARGET_TABLE_NAME,
|
||
schema=TARGET_DB_SCHEMA,
|
||
con=conn,
|
||
if_exists='append', # 追加模式
|
||
index=False,
|
||
chunksize=1000
|
||
)
|
||
print(" -> [成功] 写入完成")
|
||
|
||
except Exception as e:
|
||
print(f" -> [写入错误] {e}")
|
||
# 如果报错，打印一下列名帮助排查
|
||
print(f" 当前DataFrame列名: {df_subset.columns.tolist()}")
|
||
|
||
except Exception as e:
|
||
print(f" -> [文件处理异常] {e}")
|
||
|
||
if __name__ == "__main__":
|
||
run_migration() |