Enhance Excel synchronization by adding Uptime Kuma heartbeat functionality, new Excel sync configuration, and additional file mappings
This commit is contained in:
Misaka_Company
394
excel_sync_to_sql.py
Normal file
394
excel_sync_to_sql.py
Normal file
@@ -0,0 +1,394 @@
|
||||
import os
|
||||
import sys
|
||||
import shutil
|
||||
import logging
|
||||
import argparse
|
||||
import datetime
|
||||
import urllib.parse
|
||||
import warnings
|
||||
import time
|
||||
import requests
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.engine import URL
|
||||
from sqlalchemy.types import NVARCHAR, Integer, Date
|
||||
|
||||
# 导入配置
|
||||
from log_utils import (log_error, log_warning, log_info, log_processing, log_file, log_sync,
|
||||
log_start, log_complete, log_stop, LoggerManager)
|
||||
from config import (DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA,
|
||||
EXCEL_SYNC_INTERVAL, EXCEL_SYNC_UPTIME_KUMA_CONFIG)
|
||||
|
||||
# ================= 抑制 openpyxl 的数据验证警告 =================
|
||||
warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl')
|
||||
|
||||
class DataSynchronizer:
|
||||
def __init__(self, force_sync=False):
|
||||
self.force_sync = force_sync
|
||||
self.engine = self._get_db_connection()
|
||||
self.cache_dir = CACHE_DIR
|
||||
|
||||
if not os.path.exists(self.cache_dir):
|
||||
os.makedirs(self.cache_dir)
|
||||
|
||||
def _get_db_connection(self):
|
||||
connection_string = (
|
||||
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={DB_CONFIG.get('TrustServerCertificate', 'yes')};"
|
||||
)
|
||||
connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})
|
||||
return create_engine(connection_url, fast_executemany=True)
|
||||
|
||||
def _should_process_file(self, remote_path, local_path):
|
||||
if self.force_sync:
|
||||
return True, "强制同步"
|
||||
|
||||
if not os.path.exists(local_path):
|
||||
return True, "缓存不存在"
|
||||
|
||||
try:
|
||||
remote_mtime = os.path.getmtime(remote_path)
|
||||
local_mtime = os.path.getmtime(local_path)
|
||||
if remote_mtime > local_mtime + 1:
|
||||
return True, f"源文件更新"
|
||||
except OSError as e:
|
||||
log_error(f"无法访问源文件: {remote_path}, Error: {e}")
|
||||
return False, "源文件无法访问"
|
||||
|
||||
return False, "文件未变更"
|
||||
|
||||
def _clean_dataframe(self, df, contract_year):
|
||||
"""主表数据清洗与验证"""
|
||||
# 1. 设置合同年份
|
||||
df['合同年份'] = contract_year
|
||||
|
||||
# 2. 移除总排号为空的行
|
||||
if '总排号' in df.columns:
|
||||
df = df.dropna(subset=['总排号'])
|
||||
df = df[df['总排号'].astype(str).str.strip() != '']
|
||||
else:
|
||||
log_error("数据源中找不到映射后的[总排号]列，跳过此 sheet")
|
||||
return None, None
|
||||
|
||||
# ★ 新增：去除重复的总排号（保留第一条）
|
||||
if '总排号' in df.columns:
|
||||
df['总排号'] = df['总排号'].astype(str).str.strip()
|
||||
duplicates = df[df.duplicated(subset=['总排号'], keep='first')]
|
||||
if not duplicates.empty:
|
||||
log_warning(f"发现 {len(duplicates)} 条重复的总排号，已自动去重。重复的总排号: {duplicates['总排号'].tolist()[:10]}")
|
||||
df = df.drop_duplicates(subset=['总排号'], keep='first')
|
||||
|
||||
# 3. 补全列
|
||||
for col in TABLE_SCHEMA.keys():
|
||||
if col not in df.columns:
|
||||
df[col] = None
|
||||
|
||||
# 用于存储每一列的 SQL 类型
|
||||
dtype_dict = {}
|
||||
|
||||
# 4. 字段清洗
|
||||
for col, rules in TABLE_SCHEMA.items():
|
||||
if col not in df.columns:
|
||||
continue
|
||||
|
||||
if rules['type'] == 'int':
|
||||
# ★ 修改：先转换为数值，然后四舍五入到整数
|
||||
df[col] = pd.to_numeric(df[col], errors='coerce')
|
||||
# 将浮点数四舍五入为整数（处理如 123.5 这样的值）
|
||||
df[col] = df[col].round(0)
|
||||
# 转换为可空整数类型
|
||||
df[col] = df[col].astype('Int64')
|
||||
# 将 NaN 替换为 None
|
||||
df[col] = df[col].replace({pd.NA: None})
|
||||
dtype_dict[col] = Integer()
|
||||
|
||||
elif rules['type'] == 'date':
|
||||
df[col] = pd.to_datetime(df[col], errors='coerce')
|
||||
df[col] = df[col].apply(lambda x: x.date() if pd.notnull(x) else None)
|
||||
dtype_dict[col] = Date()
|
||||
|
||||
elif rules['type'] == 'str':
|
||||
# 先转换为字符串
|
||||
df[col] = df[col].fillna('').astype(str)
|
||||
# 替换各种空值表示
|
||||
df[col] = df[col].replace({'nan': '', 'None': '', '<NA>': ''})
|
||||
|
||||
# 强制截断
|
||||
max_len = rules.get('max_len', 255)
|
||||
df[col] = df[col].str.slice(0, max_len)
|
||||
# 将空字符串转为 None
|
||||
df[col] = df[col].replace('', None)
|
||||
|
||||
dtype_dict[col] = NVARCHAR(max_len)
|
||||
|
||||
final_cols = list(TABLE_SCHEMA.keys())
|
||||
|
||||
return df[final_cols], dtype_dict
|
||||
|
||||
def _sync_to_db(self, df, dtype_dict):
|
||||
"""同步主表数据 - 使用更稳健的方法"""
|
||||
if df is None or df.empty:
|
||||
return
|
||||
|
||||
target_table = "[warehouseOutbound].[executionCardData]"
|
||||
|
||||
with self.engine.connect() as conn:
|
||||
existing_ids = pd.read_sql(f"SELECT [总排号] FROM {target_table}", conn)
|
||||
|
||||
existing_id_set = set(existing_ids['总排号'].astype(str))
|
||||
df['总排号'] = df['总排号'].astype(str).str.strip()
|
||||
|
||||
df_update = df[df['总排号'].isin(existing_id_set)].copy()
|
||||
df_insert = df[~df['总排号'].isin(existing_id_set)].copy()
|
||||
|
||||
log_info(f"分析结果: 需插入 {len(df_insert)} 条, 需更新 {len(df_update)} 条")
|
||||
|
||||
# 1. 插入新数据
|
||||
if not df_insert.empty:
|
||||
log_info("正在执行批量插入...")
|
||||
df_insert.to_sql('executionCardData', self.engine, schema='warehouseOutbound',
|
||||
if_exists='append', index=False, chunksize=BATCH_SIZE,
|
||||
dtype=dtype_dict)
|
||||
log_info("批量插入完成。")
|
||||
|
||||
# 2. 更新现有数据 - 改用逐条或小批量 UPDATE
|
||||
if not df_update.empty:
|
||||
log_info("正在执行批量更新...")
|
||||
|
||||
cols = [c for c in df.columns if c != '总排号']
|
||||
set_clause = ", ".join([f"[{c}] = :{c}" for c in cols])
|
||||
update_sql = f"""
|
||||
UPDATE [warehouseOutbound].[executionCardData]
|
||||
SET {set_clause}
|
||||
WHERE [总排号] = :总排号
|
||||
"""
|
||||
|
||||
with self.engine.begin() as conn:
|
||||
batch_size = 1000
|
||||
total_rows = len(df_update)
|
||||
update_count = 0
|
||||
|
||||
for i in range(0, total_rows, batch_size):
|
||||
batch = df_update.iloc[i:i+batch_size]
|
||||
records = batch.to_dict('records')
|
||||
|
||||
result = conn.execute(text(update_sql), records)
|
||||
update_count += result.rowcount
|
||||
|
||||
if (i + batch_size) % 5000 == 0:
|
||||
log_info(f"已更新 {i + batch_size}/{total_rows} 条记录...")
|
||||
|
||||
# 修复 SQL Server executemany 返回负数 rowcount 的问题
|
||||
affected_rows = abs(update_count) if update_count < 0 else total_rows
|
||||
log_info(f"批量更新完成，共影响 {affected_rows} 行。")
|
||||
|
||||
def process_excel_files(self):
|
||||
for cfg in EXCEL_CONFIGS:
|
||||
remote_path = cfg['file_path']
|
||||
filename = os.path.basename(remote_path)
|
||||
local_path = os.path.join(self.cache_dir, filename)
|
||||
|
||||
should_sync, reason = self._should_process_file(remote_path, local_path)
|
||||
|
||||
if should_sync:
|
||||
log_info(f"开始处理文件: {filename} ({reason})")
|
||||
try:
|
||||
# 复制文件到本地缓存（只复制一次）
|
||||
if os.path.exists(remote_path):
|
||||
shutil.copy2(remote_path, local_path)
|
||||
|
||||
# 遍历该文件的所有指定 sheet
|
||||
for sheet_name in cfg['sheet_names']:
|
||||
log_info(f" → 处理工作表: {sheet_name} (合同年份: {cfg['contract_year']})")
|
||||
try:
|
||||
df = pd.read_excel(local_path, sheet_name=sheet_name, header=0, engine='openpyxl')
|
||||
df.columns = [str(c).strip() for c in df.columns]
|
||||
df.rename(columns=cfg['field_mapping'], inplace=True)
|
||||
|
||||
cleaned_df, dtype_mapping = self._clean_dataframe(df, cfg['contract_year'])
|
||||
|
||||
if cleaned_df is not None:
|
||||
self._sync_to_db(cleaned_df, dtype_mapping)
|
||||
log_info(f" 工作表 {sheet_name} 同步成功。")
|
||||
else:
|
||||
log_warning(f" 工作表 {sheet_name} 清洗失败，跳过。")
|
||||
except Exception as e:
|
||||
log_error(f" 处理工作表 {sheet_name} 时发生错误: {str(e)}", exc_info=True)
|
||||
|
||||
log_info(f"文件 {filename} 所有工作表处理完成。")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"处理文件 {filename} 时发生错误: {str(e)}", exc_info=True)
|
||||
else:
|
||||
log_info(f"跳过文件: {filename} ({reason})")
|
||||
|
||||
def generate_contract_data(self):
|
||||
log_info("开始生成/更新 contractData 表...")
|
||||
|
||||
merge_sql = """
|
||||
WITH SourceData AS (
|
||||
SELECT
|
||||
CAST(ISNULL([合同年份], '') AS NVARCHAR(10)) AS [合同年份],
|
||||
CAST(ISNULL([车间号], '') AS NVARCHAR(20)) AS [车间号],
|
||||
CAST(ISNULL([工令号], '') AS NVARCHAR(200)) AS [工令号],
|
||||
CAST([订单号] AS NVARCHAR(150)) AS [订单号],
|
||||
CAST([客户名称] AS NVARCHAR(200)) AS [客户名称],
|
||||
CAST([产品名称] AS NVARCHAR(200)) AS [产品型号],
|
||||
CAST([量程] AS NVARCHAR(150)) AS [量程],
|
||||
TRY_CAST([数量] AS INT) AS [数量],
|
||||
CAST(NULL AS INT) AS [单价],
|
||||
TRY_CAST([序号] AS INT) AS [ID],
|
||||
CAST([位号] AS NVARCHAR(500)) AS [位号],
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY [合同年份], [车间号], [工令号]
|
||||
ORDER BY [总排号] DESC
|
||||
) as rn
|
||||
FROM [warehouseOutbound].[executionCardData]
|
||||
WHERE
|
||||
[车间号] IS NOT NULL AND [车间号] <> ''
|
||||
AND [工令号] IS NOT NULL AND [工令号] <> ''
|
||||
)
|
||||
|
||||
MERGE INTO [warehouseOutbound].[contractData] AS Target
|
||||
USING (SELECT * FROM SourceData WHERE rn = 1) AS Source
|
||||
ON (
|
||||
Target.[合同年份] = Source.[合同年份]
|
||||
AND Target.[车间号] = Source.[车间号]
|
||||
AND Target.[工令号] = Source.[工令号]
|
||||
)
|
||||
|
||||
WHEN MATCHED THEN
|
||||
UPDATE SET
|
||||
Target.[订单号] = Source.[订单号],
|
||||
Target.[客户名称] = Source.[客户名称],
|
||||
Target.[产品型号] = Source.[产品型号],
|
||||
Target.[量程] = Source.[量程],
|
||||
Target.[数量] = Source.[数量],
|
||||
Target.[ID] = Source.[ID],
|
||||
Target.[位号] = Source.[位号]
|
||||
|
||||
WHEN NOT MATCHED BY TARGET THEN
|
||||
INSERT (
|
||||
[合同年份], [车间号], [工令号],
|
||||
[订单号], [客户名称], [产品型号],
|
||||
[量程], [数量], [单价], [ID], [位号]
|
||||
)
|
||||
VALUES (
|
||||
Source.[合同年份], Source.[车间号], Source.[工令号],
|
||||
Source.[订单号], Source.[客户名称], Source.[产品型号],
|
||||
Source.[量程], Source.[数量], Source.[单价], Source.[ID], Source.[位号]
|
||||
)
|
||||
;
|
||||
"""
|
||||
|
||||
try:
|
||||
with self.engine.begin() as conn:
|
||||
result = conn.execute(text(merge_sql))
|
||||
log_info(f"ContractData 表同步完成 (SQL Server 内部处理)。rowcount: {result.rowcount}")
|
||||
|
||||
except Exception as e:
|
||||
log_error(f"生成 ContractData 失败: {e}", exc_info=True)
|
||||
|
||||
# ================= Uptime Kuma 心跳 =================
|
||||
|
||||
_last_heartbeat_time = 0
|
||||
|
||||
def send_heartbeat():
|
||||
"""发送心跳信号到 Uptime Kuma"""
|
||||
global _last_heartbeat_time
|
||||
|
||||
if not EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False):
|
||||
return
|
||||
|
||||
try:
|
||||
url = EXCEL_SYNC_UPTIME_KUMA_CONFIG['push_url']
|
||||
params = {
|
||||
'status': 'up',
|
||||
'msg': 'OK',
|
||||
'ping': ''
|
||||
}
|
||||
response = requests.get(url, params=params, timeout=5)
|
||||
response.raise_for_status()
|
||||
_last_heartbeat_time = time.time()
|
||||
except Exception as e:
|
||||
log_warning(f"心跳发送失败: {e}")
|
||||
|
||||
def main():
|
||||
# 初始化日志管理器
|
||||
LoggerManager("excel_sync", log_prefix="excel_sync")
|
||||
|
||||
# 解析参数
|
||||
parser = argparse.ArgumentParser(description="Excel数据同步至SQL Server")
|
||||
parser.add_argument('--force', action='store_true', help='强制同步所有文件')
|
||||
parser.add_argument('--once', action='store_true', help='只运行一次后退出')
|
||||
args = parser.parse_args()
|
||||
|
||||
syncer = DataSynchronizer(force_sync=args.force)
|
||||
|
||||
# 启动信息
|
||||
mode = "强制模式" if args.force else "增量模式"
|
||||
if args.once:
|
||||
log_start(f"Excel 同步任务 ({mode}, 单次运行)")
|
||||
syncer.process_excel_files()
|
||||
syncer.generate_contract_data()
|
||||
log_complete("Excel 同步任务已完成")
|
||||
return
|
||||
|
||||
# 周期性运行模式
|
||||
log_start(f"Excel 同步服务已启动 ({mode})")
|
||||
log_info(f"同步周期: {EXCEL_SYNC_INTERVAL} 秒 ({EXCEL_SYNC_INTERVAL//60} 分钟)")
|
||||
if EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False):
|
||||
log_info(f"心跳间隔: {EXCEL_SYNC_UPTIME_KUMA_CONFIG['heartbeat_interval']} 秒")
|
||||
log_info("=" * 70)
|
||||
|
||||
# 启动时发送第一次心跳
|
||||
send_heartbeat()
|
||||
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
# 执行同步任务
|
||||
log_info(f"开始执行周期性同步检查...")
|
||||
syncer.process_excel_files()
|
||||
syncer.generate_contract_data()
|
||||
log_info(f"周期性同步检查完成")
|
||||
|
||||
# 下次同步时间
|
||||
next_sync_time = time.time() + EXCEL_SYNC_INTERVAL
|
||||
log_info(f"下次同步将在 {EXCEL_SYNC_INTERVAL//60} 分钟后进行")
|
||||
|
||||
# 等待下次同步，期间持续发送心跳
|
||||
while time.time() < next_sync_time:
|
||||
# 检查是否需要发送心跳
|
||||
time_since_last_heartbeat = time.time() - _last_heartbeat_time
|
||||
if time_since_last_heartbeat >= EXCEL_SYNC_UPTIME_KUMA_CONFIG['heartbeat_interval']:
|
||||
send_heartbeat()
|
||||
|
||||
# 短暂休眠
|
||||
time.sleep(1)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
log_info("=" * 70)
|
||||
log_stop("收到停止信号，服务正在关闭...")
|
||||
break
|
||||
except Exception as e:
|
||||
log_error(f"同步任务异常: {e}", exc_info=True)
|
||||
log_info(f"将在 {EXCEL_SYNC_INTERVAL//60} 分钟后重试...")
|
||||
time.sleep(EXCEL_SYNC_INTERVAL)
|
||||
finally:
|
||||
# 停止时发送心跳停止信号
|
||||
if EXCEL_SYNC_UPTIME_KUMA_CONFIG.get('enabled', False):
|
||||
try:
|
||||
url = EXCEL_SYNC_UPTIME_KUMA_CONFIG['push_url']
|
||||
params = {'status': 'down', 'msg': 'Service stopped'}
|
||||
requests.get(url, params=params, timeout=5)
|
||||
except:
|
||||
pass
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user