From c50573fd78ed5d2d02a5c58b9baaae3e055dd96d Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Thu, 8 Jan 2026 13:30:38 +0800 Subject: [PATCH] Add ntfy notification system and enhance logging for error handling --- .gitignore | 1 + config.py | 12 ++- migration.py | 214 ++++++++++++++++++++++++---------------- ntfy_utils.py | 65 ++++++++++++ run_incremental_sync.py | 6 ++ 5 files changed, 211 insertions(+), 87 deletions(-) create mode 100644 ntfy_utils.py diff --git a/.gitignore b/.gitignore index 2a8af98..7a970b5 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ build dist log *.spec +temp \ No newline at end of file diff --git a/config.py b/config.py index acccace..9d61206 100644 --- a/config.py +++ b/config.py @@ -323,7 +323,17 @@ SYNC_MAPPING = { } } } - +# ================= ntfy 配置 ================= +NTFY_CONFIG = { + 'enabled': True, # 是否启用通知 + 'server_url': 'https://ntfy.server10086.icu', # 如果是自建服务器，请修改为自己的 URL + 'topic': 'bld', # 你的订阅主题 + 'token': 'tk_eop5fs66acxtwxf6vlkiojhdvkgb0', # <--- 在这里填入你的 Access Token + 'priority': { + 'error': 'high', # 错误消息优先级 + 'critical': 'urgent' # 严重错误优先级 + } +} # ================= 运行参数 ================= POLL_INTERVAL = 5 # 轮询间隔(秒) BATCH_SIZE = 10000 # 批量处理大小 \ No newline at end of file diff --git a/migration.py b/migration.py index 220f06e..a136536 100644 --- a/migration.py +++ b/migration.py @@ -1,55 +1,54 @@ import pandas as pd import os +import shutil import urllib -from sqlalchemy import create_engine +from sqlalchemy import create_engine, text +import ntfy_utils # 确保该文件在同一目录下 # ========================================== -# 1. 全局配置 (Global Configuration) +# 1. 脚本配置 (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" # 确保已安装此驱动 + "server": "192.168.110.114", + "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' +TARGET_DB_SCHEMA = "warehouseOutbound" +TARGET_TABLE_NAME = "customerProductType" +SQL_SOURCE_FILE_COL = "SourceFile" # 你在SQL中新增的字段名 -# Excel 列名映射到 SQL 字段名的逻辑键 (用于后续代码逻辑引用) -# 这里的 value 必须与 SQL 数据库中的实际字段名完全一致 -SQL_COL_YEAR = "合同年份" # 数据库中存年份的字段名 -SQL_COL_WORKSHOP = "车间号" # 数据库中存车间号的字段名 -SQL_COL_ORDER = "工令号" # 数据库中存工令号的字段名 -SQL_COL_MODEL = "客户型号" # 数据库中存客户型号的字段名 +# 字段映射常量 +SQL_COL_YEAR = "合同年份" +SQL_COL_WORKSHOP = "车间号" +SQL_COL_ORDER = "工令号" +SQL_COL_MODEL = "客户型号" -# ========================================== -# 2. 迁移任务清单 (Migration Tasks) -# ========================================== -# 可以在这里添加任意数量的文件配置 +# 运行参数 +FORCE_UPDATE = False # 如果设为 True，则无视时间对比，强制更新所有文件 +TEMP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "temp") + +# 迁移任务清单 MIGRATION_TASKS = [ - # --- 任务 1 --- { - "file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2022.xlsm", # Excel文件路径 - "year": 2022, # 该文件对应的合同年份 - "sheet_names": ["Sheet1"], # 指定要迁移的工作表名称列表 - # 映射表: Excel列名 -> SQL字段名 + "file_path": r"\\192.168.110.113\生产执行卡\往年生产执行卡\生产执行卡2022.xlsm", + "year": 2022, + "sheet_names": ["Sheet1"], "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"], # 只迁移 "汇总" 表 + "sheet_names": ["Sheet1"], "mapping": { "车间号": SQL_COL_WORKSHOP, "工令号": SQL_COL_ORDER, @@ -59,7 +58,7 @@ MIGRATION_TASKS = [ ] # ========================================== -# 3. 核心逻辑 +# 2. 核心辅助函数 # ========================================== def get_db_engine(): @@ -71,94 +70,137 @@ def get_db_engine(): f"PWD={DB_CONFIG['password']};" f"TrustServerCertificate=yes;" ) - # 使用 fast_executemany 提高写入速度 + # fast_executemany 极大提高写入速度 return create_engine(f"mssql+pyodbc:///?odbc_connect={params}", fast_executemany=True) +def get_file_mtime(path): + """获取文件最后修改时间戳""" + try: + return os.path.getmtime(path) + except OSError: + return 0 + +def delete_old_data(engine, filename): + """根据 SourceFile 字段精确删除旧数据""" + full_table = f"[{TARGET_DB_SCHEMA}].[{TARGET_TABLE_NAME}]" + sql = text(f"DELETE FROM {full_table} WHERE [{SQL_SOURCE_FILE_COL}] = :fname") + with engine.begin() as conn: + conn.execute(sql, {"fname": filename}) + +# ========================================== +# 3. 迁移主逻辑 +# ========================================== + def run_migration(): + # 初始化环境 + if not os.path.exists(TEMP_DIR): + os.makedirs(TEMP_DIR) + engine = get_db_engine() - print(f"连接数据库... [{TARGET_DB_SCHEMA}].[{TARGET_TABLE_NAME}]") + sync_count = 0 + error_count = 0 + + print(f"🚀 开始增量同步任务 (强制更新={FORCE_UPDATE})") for task in MIGRATION_TASKS: - file_path = task['file_path'] - year_val = task['year'] - # mapping 的键(Key)是Excel列名，值(Value)是SQL列名 - mapping = task['mapping'] + remote_path = task['file_path'] + filename = os.path.basename(remote_path) + local_path = os.path.join(TEMP_DIR, filename) - if not os.path.exists(file_path): - print(f"文件不存在: {file_path}") + # 1. 检查源文件 + if not os.path.exists(remote_path): + msg = f"远程文件未找到: {remote_path}" + print(f"❌ {msg}") + ntfy_utils.send_error(msg) continue - print(f"\n-------- 处理文件: {os.path.basename(file_path)} --------") - + # 2. 增量判定 + remote_mtime = get_file_mtime(remote_path) + local_mtime = get_file_mtime(local_path) + + if not FORCE_UPDATE and os.path.exists(local_path) and remote_mtime <= local_mtime: + print(f"⏭️ 跳过: {filename} (文件未变更)") + continue + + print(f"🔄 正在处理: {filename} ...") + try: - # 读取 Excel - xls_dict = pd.read_excel(file_path, sheet_name=task['sheet_names']) + # 3. 复制文件到本地 temp + shutil.copy2(remote_path, local_path) + + # 4. 读取 Excel + xls_dict = pd.read_excel(local_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} + xls_dict = {task['sheet_names'][0]: xls_dict} + + # 准备存放该文件所有 Sheet 的合并数据 + df_all_sheets = [] 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()) + source_cols = list(task['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}") + missing = [c for c in source_cols if c not in df.columns] + if missing: + print(f" ⚠️ Sheet[{sheet_name}] 缺失列: {missing}") continue - # 3. 提取数据并重命名 - # 先提取 -> 只有这几列 + # 提取并重命名 df_subset = df[source_cols].copy() - # 后重命名 -> 变成数据库的列名 - df_subset.rename(columns=mapping, inplace=True) + df_subset.rename(columns=task['mapping'], inplace=True) - # 4. 注入年份字段 - df_subset[SQL_COL_YEAR] = year_val - - # 此时 df_subset 的列名应该完全等于：SQL字段列表 + # 注入年份和来源文件名 + df_subset[SQL_COL_YEAR] = task['year'] + df_subset[SQL_SOURCE_FILE_COL] = filename # 存入文件名，用于下次精准删除 - # 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 + if not df_subset.empty: + df_all_sheets.append(df_subset) - 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(" -> [成功] 写入完成") + # 5. 写入数据库 + if df_all_sheets: + final_df = pd.concat(df_all_sheets, ignore_index=True) + + # 执行删除并插入 (事务) + with engine.begin() as conn: + # A. 删除旧记录 + delete_sql = text(f"DELETE FROM [{TARGET_DB_SCHEMA}].[{TARGET_TABLE_NAME}] WHERE [{SQL_SOURCE_FILE_COL}] = :fname") + conn.execute(delete_sql, {"fname": filename}) - except Exception as e: - print(f" -> [写入错误] {e}") - # 如果报错，打印一下列名帮助排查 - print(f" 当前DataFrame列名: {df_subset.columns.tolist()}") + # B. 插入新记录 + final_df.to_sql( + name=TARGET_TABLE_NAME, + schema=TARGET_DB_SCHEMA, + con=conn, + if_exists='append', + index=False, + chunksize=1000 + ) + + print(f" ✅ 成功同步: {len(final_df)} 行记录") + sync_count += 1 + else: + print(f" ⚠️ 警告: 文件内容为空或格式不符") except Exception as e: - print(f" -> [文件处理异常] {e}") + error_msg = f"文件 [{filename}] 处理失败: {str(e)}" + print(f"❌ {error_msg}") + ntfy_utils.send_error(error_msg) + error_count += 1 + + # 结束汇总 + summary = f"同步完成: 成功 {sync_count} 个文件, 失败 {error_count} 个文件。" + print(f"\n🏁 {summary}") + if sync_count > 0: + # 只有在有实际更新时才发送成功通知 + ntfy_utils.send_ntfy(summary, title="📊 数据迁移报告", tags=["package"]) if __name__ == "__main__": run_migration() \ No newline at end of file diff --git a/ntfy_utils.py b/ntfy_utils.py new file mode 100644 index 0000000..71a5e06 --- /dev/null +++ b/ntfy_utils.py @@ -0,0 +1,65 @@ +# ntfy_utils.py +import requests +import config + +def send_ntfy(message, title="数据库同步消息", priority="default", tags=None): + """ + 向加密的 ntfy 服务器发送消息 + """ + conf = config.NTFY_CONFIG + if not conf.get('enabled', False): + return + + # 确保 URL 正确（末尾不要多余斜杠） + server_url = conf['server_url'].rstrip('/') + url = f"{server_url}/{conf['topic']}" + + # 构造请求头 + headers = { + "Title": title.encode('utf-8'), + "Priority": priority, + "Tags": ",".join(tags) if tags else "" + } + + # --- 核心：配置秘钥认证 --- + token = conf.get('token') + if token: + # ntfy 使用 Bearer Token 模式 + headers["Authorization"] = f"Bearer {token}" + + try: + # 发送请求 + response = requests.post( + url, + data=message.encode('utf-8'), + headers=headers, + timeout=10 + ) + + # 针对认证失败的处理 + if response.status_code == 401: + print("ntfy 认证失败：Token 无效") + elif response.status_code == 403: + print("ntfy 权限不足：该 Token 无权发布消息") + + response.raise_for_status() + except Exception as e: + print(f"发送 ntfy 通知失败: {e}") + +def send_error(msg): + """便捷方法：发送错误通知""" + send_ntfy( + message=str(msg), + title="❌ 同步任务错误", + priority=config.NTFY_CONFIG['priority']['error'], + tags=["warning", "database"] + ) + +def send_critical(msg): + """便捷方法：发送严重崩溃通知""" + send_ntfy( + message=str(msg), + title="🔥 同步服务崩溃", + priority=config.NTFY_CONFIG['priority']['critical'], + tags=["skull", "critical"] + ) \ No newline at end of file diff --git a/run_incremental_sync.py b/run_incremental_sync.py index 46beef7..cbcaa51 100644 --- a/run_incremental_sync.py +++ b/run_incremental_sync.py @@ -6,6 +6,7 @@ import config import db_utils import logging from logging.handlers import TimedRotatingFileHandler +import ntfy_utils # ================= 日志系统配置 ================= def setup_logger(): @@ -43,10 +44,12 @@ def log_success(message): def log_error(message): """错误消息 - 红色感觉""" logger.error(f"❌ [错误] {message}") + ntfy_utils.send_error(f"❌ [错误] {message}") def log_warning(message): """警告消息 - 黄色感觉""" logger.warning(f"⚠️ [警告] {message}") + ntfy_utils.send_error(f"⚠️ [警告] {message}") def log_info(message): """信息消息""" @@ -63,14 +66,17 @@ def log_skip(message): def log_critical(message): """严重错误""" logger.critical(f"🔥 [严重] {message}") + ntfy_utils.send_critical(f"🔥 [严重] {message}") def log_start(message): """启动消息""" logger.info(f"🚀 [启动] {message}") + ntfy_utils.send_ntfy(f"🚀 [启动] {message}") def log_stop(message): """停止消息""" logger.info(f"🛑 [停止] {message}") + ntfy_utils.send_ntfy(f"🛑 [停止] {message}") def log_file(message): """文件操作消息"""