AllForCompany/BLD_sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add ntfy notification system and enhance logging for error handling

Browse Source
This commit is contained in:
Misaka_Company
2026-01-08 13:30:38 +08:00
parent 1e91071769
commit c50573fd78
5 changed files with 211 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -4,3 +4,4 @@ build
dist
log
*.spec
temp

12
config.py
View File

@@ -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 # 批量处理大小

214
migration.py
View File

@@ -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()

65
ntfy_utils.py Normal file
View File

@@ -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"]
)

6
run_incremental_sync.py
View File

@@ -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):
"""文件操作消息"""