Refactor logging system to use unified log_utils module and enhance logging messages across multiple scripts
This commit is contained in:
Misaka_Company
@@ -13,26 +13,13 @@ 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, LoggerManager)
|
||||
from config import DB_CONFIG, CACHE_DIR, EXCEL_CONFIGS, BATCH_SIZE, TABLE_SCHEMA
|
||||
|
||||
# ================= 抑制 openpyxl 的数据验证警告 =================
|
||||
warnings.filterwarnings('ignore', category=UserWarning, module='openpyxl')
|
||||
|
||||
# ================= 日志配置 =================
|
||||
# 配置控制台输出使用 UTF-8 编码，确保中文正确显示
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
console_handler,
|
||||
logging.FileHandler("sync_log.txt", encoding='utf-8')
|
||||
]
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class DataSynchronizer:
|
||||
def __init__(self, force_sync=False):
|
||||
self.force_sync = force_sync
|
||||
@@ -49,7 +36,7 @@ class DataSynchronizer:
|
||||
f"DATABASE={DB_CONFIG['database']};"
|
||||
f"UID={DB_CONFIG['username']};"
|
||||
f"PWD={DB_CONFIG['password']};"
|
||||
f"TrustServerCertificate={DB_CONFIG.get('TrustServerCertificate', 'no')};"
|
||||
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)
|
||||
@@ -67,7 +54,7 @@ class DataSynchronizer:
|
||||
if remote_mtime > local_mtime + 1:
|
||||
return True, f"源文件更新"
|
||||
except OSError as e:
|
||||
logger.error(f"无法访问源文件: {remote_path}, Error: {e}")
|
||||
log_error(f"无法访问源文件: {remote_path}, Error: {e}")
|
||||
return False, "源文件无法访问"
|
||||
|
||||
return False, "文件未变更"
|
||||
@@ -82,7 +69,7 @@ class DataSynchronizer:
|
||||
df = df.dropna(subset=['总排号'])
|
||||
df = df[df['总排号'].astype(str).str.strip() != '']
|
||||
else:
|
||||
logger.error("数据源中找不到映射后的[总排号]列，跳过此 sheet")
|
||||
log_error("数据源中找不到映射后的[总排号]列，跳过此 sheet")
|
||||
return None, None
|
||||
|
||||
# ★ 新增：去除重复的总排号（保留第一条）
|
||||
@@ -90,7 +77,7 @@ class DataSynchronizer:
|
||||
df['总排号'] = df['总排号'].astype(str).str.strip()
|
||||
duplicates = df[df.duplicated(subset=['总排号'], keep='first')]
|
||||
if not duplicates.empty:
|
||||
logger.warning(f"发现 {len(duplicates)} 条重复的总排号，已自动去重。重复的总排号: {duplicates['总排号'].tolist()[:10]}")
|
||||
log_warning(f"发现 {len(duplicates)} 条重复的总排号，已自动去重。重复的总排号: {duplicates['总排号'].tolist()[:10]}")
|
||||
df = df.drop_duplicates(subset=['总排号'], keep='first')
|
||||
|
||||
# 3. 补全列
|
||||
@@ -156,19 +143,19 @@ class DataSynchronizer:
|
||||
df_update = df[df['总排号'].isin(existing_id_set)].copy()
|
||||
df_insert = df[~df['总排号'].isin(existing_id_set)].copy()
|
||||
|
||||
logger.info(f"分析结果: 需插入 {len(df_insert)} 条, 需更新 {len(df_update)} 条")
|
||||
log_info(f"分析结果: 需插入 {len(df_insert)} 条, 需更新 {len(df_update)} 条")
|
||||
|
||||
# 1. 插入新数据
|
||||
if not df_insert.empty:
|
||||
logger.info("正在执行批量插入...")
|
||||
log_info("正在执行批量插入...")
|
||||
df_insert.to_sql('executionCardData', self.engine, schema='warehouseOutbound',
|
||||
if_exists='append', index=False, chunksize=BATCH_SIZE,
|
||||
dtype=dtype_dict)
|
||||
logger.info("批量插入完成。")
|
||||
log_info("批量插入完成。")
|
||||
|
||||
# 2. 更新现有数据 - 改用逐条或小批量 UPDATE
|
||||
if not df_update.empty:
|
||||
logger.info("正在执行批量更新...")
|
||||
log_info("正在执行批量更新...")
|
||||
|
||||
cols = [c for c in df.columns if c != '总排号']
|
||||
set_clause = ", ".join([f"[{c}] = :{c}" for c in cols])
|
||||
@@ -191,11 +178,11 @@ class DataSynchronizer:
|
||||
update_count += result.rowcount
|
||||
|
||||
if (i + batch_size) % 5000 == 0:
|
||||
logger.info(f"已更新 {i + batch_size}/{total_rows} 条记录...")
|
||||
log_info(f"已更新 {i + batch_size}/{total_rows} 条记录...")
|
||||
|
||||
# 修复 SQL Server executemany 返回负数 rowcount 的问题
|
||||
affected_rows = abs(update_count) if update_count < 0 else total_rows
|
||||
logger.info(f"批量更新完成，共影响 {affected_rows} 行。")
|
||||
log_info(f"批量更新完成，共影响 {affected_rows} 行。")
|
||||
|
||||
def process_excel_files(self):
|
||||
for cfg in EXCEL_CONFIGS:
|
||||
@@ -206,7 +193,7 @@ class DataSynchronizer:
|
||||
should_sync, reason = self._should_process_file(remote_path, local_path)
|
||||
|
||||
if should_sync:
|
||||
logger.info(f"开始处理文件: {filename} ({reason})")
|
||||
log_info(f"开始处理文件: {filename} ({reason})")
|
||||
try:
|
||||
# 复制文件到本地缓存（只复制一次）
|
||||
if os.path.exists(remote_path):
|
||||
@@ -214,7 +201,7 @@ class DataSynchronizer:
|
||||
|
||||
# 遍历该文件的所有指定 sheet
|
||||
for sheet_name in cfg['sheet_names']:
|
||||
logger.info(f" → 处理工作表: {sheet_name} (合同年份: {cfg['contract_year']})")
|
||||
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]
|
||||
@@ -224,21 +211,21 @@ class DataSynchronizer:
|
||||
|
||||
if cleaned_df is not None:
|
||||
self._sync_to_db(cleaned_df, dtype_mapping)
|
||||
logger.info(f" 工作表 {sheet_name} 同步成功。")
|
||||
log_info(f" 工作表 {sheet_name} 同步成功。")
|
||||
else:
|
||||
logger.warning(f" 工作表 {sheet_name} 清洗失败，跳过。")
|
||||
log_warning(f" 工作表 {sheet_name} 清洗失败，跳过。")
|
||||
except Exception as e:
|
||||
logger.error(f" 处理工作表 {sheet_name} 时发生错误: {str(e)}", exc_info=True)
|
||||
log_error(f" 处理工作表 {sheet_name} 时发生错误: {str(e)}", exc_info=True)
|
||||
|
||||
logger.info(f"文件 {filename} 所有工作表处理完成。")
|
||||
log_info(f"文件 {filename} 所有工作表处理完成。")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理文件 {filename} 时发生错误: {str(e)}", exc_info=True)
|
||||
log_error(f"处理文件 {filename} 时发生错误: {str(e)}", exc_info=True)
|
||||
else:
|
||||
logger.info(f"跳过文件: {filename} ({reason})")
|
||||
log_info(f"跳过文件: {filename} ({reason})")
|
||||
|
||||
def generate_contract_data(self):
|
||||
logger.info("开始生成/更新 contractData 表...")
|
||||
log_info("开始生成/更新 contractData 表...")
|
||||
|
||||
merge_sql = """
|
||||
WITH SourceData AS (
|
||||
@@ -299,21 +286,30 @@ class DataSynchronizer:
|
||||
try:
|
||||
with self.engine.begin() as conn:
|
||||
result = conn.execute(text(merge_sql))
|
||||
logger.info(f"ContractData 表同步完成 (SQL Server 内部处理)。rowcount: {result.rowcount}")
|
||||
log_info(f"ContractData 表同步完成 (SQL Server 内部处理)。rowcount: {result.rowcount}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"生成 ContractData 失败: {e}", exc_info=True)
|
||||
log_error(f"生成 ContractData 失败: {e}", exc_info=True)
|
||||
|
||||
def main():
|
||||
# 初始化日志管理器
|
||||
LoggerManager("etl_manager", log_prefix="sync")
|
||||
|
||||
parser = argparse.ArgumentParser(description="Excel数据同步至SQL Server")
|
||||
parser.add_argument('--force', action='store_true', help='强制同步所有文件')
|
||||
args = parser.parse_args()
|
||||
|
||||
syncer = DataSynchronizer(force_sync=args.force)
|
||||
logger.info("================= 任务开始 =================")
|
||||
|
||||
if args.force:
|
||||
log_start("Excel 同步任务 (强制模式)")
|
||||
else:
|
||||
log_start("Excel 同步任务 (增量模式)")
|
||||
|
||||
syncer.process_excel_files()
|
||||
syncer.generate_contract_data()
|
||||
logger.info("================= 任务结束 =================")
|
||||
|
||||
log_complete("Excel 同步任务已完成")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user