AllForCompany/Excel-to-SQL_Server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: unified logging module and remove redundant first_run logic

Browse Source 
- Add logger.py with TimedRotatingFileHandler (daily rotation, 30-day retention)
- Replace all print() in migrate.py with logging calls
- Add logging to sync.py for sync status tracking
- Refactor watch.py to use logger.py, remove duplicate first_run branch
- All logs now go to logs/app.log consistently
- Add logging config section to config.yaml, add logs/ to .gitignore

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
2026-06-01 13:56:23 +08:00
parent 7edbd987f0
commit d224b98893
5 changed files with 124 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -21,6 +21,7 @@ Desktop.ini
.DS_Store
# Logs
logs/
*.log
watch.log

65
logger.py Normal file
View File

@@ -0,0 +1,65 @@
"""
Unified logging module.
Provides a pre-configured logger with console + rotating file handlers.
"""
import logging
import os
from logging.handlers import TimedRotatingFileHandler
def setup_logger(name='app', log_dir=None, level='INFO', max_keep_days=30):
"""Create and configure a logger.
Args:
name: Logger name (used as basename for log file).
log_dir: Directory for log files. Defaults to <project_root>/logs.
level: Logging level string (DEBUG, INFO, WARNING, ERROR).
max_keep_days: Number of days to retain rotated log files.
Returns:
Configured logging.Logger instance.
"""
if log_dir is None:
log_dir = 'logs'
# Resolve relative paths against project root
project_root = os.path.dirname(os.path.abspath(__file__))
if not os.path.isabs(log_dir):
log_dir = os.path.join(project_root, log_dir)
os.makedirs(log_dir, exist_ok=True)
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
# Avoid adding duplicate handlers on repeated calls
if logger.handlers:
return logger
fmt = logging.Formatter(
'%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# Console handler
ch = logging.StreamHandler()
ch.setLevel(getattr(logging, level.upper(), logging.INFO))
ch.setFormatter(fmt)
logger.addHandler(ch)
# File handler — rotate daily, keep max_keep_days backups
log_file = os.path.join(log_dir, f'{name}.log')
fh = TimedRotatingFileHandler(
log_file, when='midnight', interval=1,
backupCount=max_keep_days, encoding='utf-8'
)
fh.setLevel(getattr(logging, level.upper(), logging.INFO))
fh.setFormatter(fmt)
fh.suffix = '%Y-%m-%d'
logger.addHandler(fh)
return logger
def get_logger(name='app'):
"""Get an existing logger by name (does not reconfigure)."""
return logging.getLogger(name)

55
migrate.py
View File

@@ -5,6 +5,7 @@ Configuration is loaded from config.yaml.
"""
import argparse
import logging
import os
import sys
from collections import OrderedDict
@@ -14,6 +15,8 @@ import openpyxl
import pyodbc
import yaml
logger = logging.getLogger('app')
# ============================================================
# 配置加载
@@ -429,7 +432,7 @@ def migrate_table(cursor, table_config, cfg, dry_run=False, use_truncate=False):
primary_key = cfg['primary_key']
excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel')
print(f"\n 迁移 {table_name} ...")
logger.info(f"迁移 {table_name} ...")
# 1. 读取并合并数据
headers, actual_pk, merged = merge_sources(sources, excel_dir, primary_key, cfg)
@@ -438,31 +441,31 @@ def migrate_table(cursor, table_config, cfg, dry_run=False, use_truncate=False):
for src in sources:
fpath = os.path.join(excel_dir, src['file'])
_, _, data = read_sheet(fpath, src['sheet'], primary_key, cfg)
print(f" 读取: {src['file']} > {src['sheet']} → {len(data):,} 行")
logger.info(f" 读取: {src['file']} > {src['sheet']} → {len(data):,} 行")
print(f" 合并去重后: {len(merged):,} 行, {len(headers)} 列 (主键列: {actual_pk})")
logger.info(f" 合并去重后: {len(merged):,} 行, {len(headers)} 列 (主键列: {actual_pk})")
if dry_run:
print(f" [DRY-RUN] 建表 SQL:")
logger.info(f" [DRY-RUN] 建表 SQL:")
type_rules = cfg['type_rules']
col_defs = ['[id] INT IDENTITY(1,1)']
for h in headers:
sql_type = infer_sql_type(h, type_rules, actual_pk)
col_defs.append(f' [{h}] {sql_type}')
col_defs.append(f' CONSTRAINT [PK_{table_name}] PRIMARY KEY ([{actual_pk}])')
print(f" CREATE TABLE [{cfg['schema']}].[{table_name}] (")
print(',\n'.join(col_defs))
print(" );")
logger.info(f" CREATE TABLE [{cfg['schema']}].[{table_name}] (")
logger.info(',\n'.join(col_defs))
logger.info(" );")
return len(merged)
# 2. 建表
create_sql = ensure_table(cursor, table_name, headers, cfg, actual_pk, use_truncate=use_truncate)
action = "TRUNCATE" if use_truncate and "TRUNCATE" in (create_sql or "") else "建表"
print(f" {action}: [{cfg['schema']}].[{table_name}] ({len(headers)} 列)")
logger.info(f" {action}: [{cfg['schema']}].[{table_name}] ({len(headers)} 列)")
# 3. 插入数据
total = batch_insert(cursor, table_name, headers, merged, cfg, actual_pk)
print(f" 插入: {total:,} 行 done")
logger.info(f" 插入: {total:,} 行 done")
return total
@@ -480,15 +483,25 @@ def main():
# 加载配置
cfg = load_config(args.config)
# 初始化日志
from logger import setup_logger
log_cfg = cfg.get('logging', {})
setup_logger(
'app',
log_dir=log_cfg.get('log_dir'),
level=log_cfg.get('level', 'INFO'),
max_keep_days=log_cfg.get('max_keep_days', 30),
)
# 按参数筛选表
if args.table:
specified = set(args.table)
tables = [t for t in cfg['tables'] if t['table'] in specified]
missing = specified - {t['table'] for t in tables}
if missing:
print(f" 错误: 未找到表 {', '.join(missing)}")
logger.error(f"未找到表 {', '.join(missing)}")
available = ', '.join(t['table'] for t in cfg['tables'])
print(f" 可用表: {available}")
logger.info(f"可用表: {available}")
sys.exit(1)
else:
tables = cfg['tables']
@@ -496,11 +509,11 @@ def main():
total_tables = len(tables)
primary_key = cfg['primary_key']
print(f"{'=' * 60}")
print(f" Excel → SQL Server 迁移工具")
print(f" 目标: [{cfg['schema']}], {total_tables} 张表")
print(f" 模式: {'DRY-RUN' if args.dry_run else 'LIVE'}")
print(f"{'=' * 60}")
logger.info("=" * 50)
logger.info(f"Excel → SQL Server 迁移工具")
logger.info(f" 目标: [{cfg['schema']}], {total_tables} 张表")
logger.info(f" 模式: {'DRY-RUN' if args.dry_run else 'LIVE'}")
logger.info("=" * 50)
conn = None
cursor = None
@@ -509,11 +522,11 @@ def main():
conn = get_connection(cfg)
cursor = conn.cursor()
create_schema(cursor, cfg['schema'])
print(f"\n Schema [{cfg['schema']}] 已就绪")
logger.info(f"Schema [{cfg['schema']}] 已就绪")
total_rows = 0
for i, table_cfg in enumerate(tables, 1):
print(f"\n[{i}/{total_tables}]", end='')
logger.info(f"[{i}/{total_tables}]")
rows = migrate_table(cursor, table_cfg, cfg, dry_run=args.dry_run,
use_truncate=args.truncate)
total_rows += rows
@@ -522,9 +535,9 @@ def main():
cursor.close()
conn.close()
print(f"\n{'=' * 60}")
print(f" 完成! 共迁移 {total_tables} 张表, {total_rows:,} 行")
print(f"{'=' * 60}")
logger.info("=" * 50)
logger.info(f"完成! 共迁移 {total_tables} 张表, {total_rows:,} 行")
logger.info("=" * 50)
if __name__ == '__main__':

8
sync.py
View File

@@ -3,9 +3,12 @@ LAN file sync module.
Detects file changes, copies from LAN UNC paths to local, and determines which tables need migration.
"""
import logging
import os
import shutil
logger = logging.getLogger('app')
def sync_file(local_path, lan_path):
"""Sync a single file from LAN to local.
@@ -16,6 +19,7 @@ def sync_file(local_path, lan_path):
"""
try:
if not os.path.exists(lan_path):
logger.error(f"LAN 路径不可达: {lan_path}")
return 'error', f'LAN path not accessible: {lan_path}'
lan_mtime = os.path.getmtime(lan_path)
@@ -24,6 +28,7 @@ def sync_file(local_path, lan_path):
if local_existed:
local_mtime = os.path.getmtime(local_path)
if local_mtime == lan_mtime:
logger.debug(f"跳过 (mtime一致): {os.path.basename(local_path)}")
return 'skipped', None
# Ensure parent directory exists
@@ -41,11 +46,14 @@ def sync_file(local_path, lan_path):
raise
if not local_existed:
logger.info(f"copied: {os.path.basename(local_path)}")
return 'copied', None
else:
logger.info(f"updated: {os.path.basename(local_path)}")
return 'updated', None
except Exception as e:
logger.error(f"同步失败: {local_path} - {e}")
return 'error', str(e)

108
watch.py
View File

@@ -5,41 +5,16 @@ Watch daemon - periodically syncs LAN Excel files and migrates changed tables to
import argparse
import logging
import os
import sys
import time
import yaml
from logger import setup_logger
from migrate import load_config, get_connection, create_schema, migrate_table
from sync import sync_all_files, determine_tables_to_migrate
def setup_logging(log_file):
"""Configure logging to both console and file."""
logger = logging.getLogger()
logger.setLevel(logging.INFO)
fmt = logging.Formatter(
'%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# Console handler
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(logging.INFO)
ch.setFormatter(fmt)
logger.addHandler(ch)
# File handler
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setLevel(logging.INFO)
fh.setFormatter(fmt)
logger.addHandler(fh)
return logger
logger = logging.getLogger('app')
def run_once(cfg, logger):
def run_once(cfg):
"""Execute one sync + migrate cycle."""
excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel')
@@ -47,14 +22,6 @@ def run_once(cfg, logger):
logger.info("开始同步 LAN 文件...")
sync_results = sync_all_files(cfg, excel_dir)
for rel_path, (status, error_msg) in sync_results.items():
if status == 'error':
logger.error(f" 同步失败: {rel_path} - {error_msg}")
elif status in ('copied', 'updated'):
logger.info(f" {status}: {rel_path}")
else:
logger.debug(f" 跳过: {rel_path}")
# 2. Determine which tables need migration
tables_to_migrate = determine_tables_to_migrate(cfg, sync_results)
@@ -78,9 +45,9 @@ def run_once(cfg, logger):
try:
rows = migrate_table(cursor, table_cfg, cfg, use_truncate=True)
logger.info(f" 迁移完成: {table_name} ({rows:,} 行)")
logger.info(f"迁移完成: {table_name} ({rows:,} 行)")
except Exception as e:
logger.error(f" 迁移失败: {table_name} - {e}", exc_info=True)
logger.error(f"迁移失败: {table_name} - {e}", exc_info=True)
# Rollback any uncommitted transaction for this table
try:
conn.rollback()
@@ -106,12 +73,18 @@ def main():
help='覆盖配置文件中的 check_interval_minutes')
args = parser.parse_args()
# Setup logging
log_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'watch.log')
logger = setup_logging(log_file)
# Load config
cfg = load_config(args.config)
# Setup logging
log_cfg = cfg.get('logging', {})
setup_logger(
'app',
log_dir=log_cfg.get('log_dir'),
level=log_cfg.get('level', 'INFO'),
max_keep_days=log_cfg.get('max_keep_days', 30),
)
interval = args.interval or cfg.get('check_interval_minutes', 30)
logger.info("=" * 50)
@@ -119,58 +92,9 @@ def main():
logger.info(f" 同步间隔: {interval} 分钟")
logger.info("=" * 50)
first_run = True
while True:
try:
if first_run:
logger.info("首次运行，强制同步并迁移所有表...")
# On first run, force all tables to migrate by marking all files as 'updated'
excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel')
sync_results = sync_all_files(cfg, excel_dir)
# Log sync results
for rel_path, (status, error_msg) in sync_results.items():
if status == 'error':
logger.error(f" 同步失败: {rel_path} - {error_msg}")
else:
logger.info(f" {status}: {rel_path}")
# Force all tables to migrate on first run
all_tables = {t['table'] for t in cfg['tables']}
conn = None
try:
conn = get_connection(cfg)
cursor = conn.cursor()
create_schema(cursor, cfg['schema'])
for table_cfg in cfg['tables']:
table_name = table_cfg['table']
try:
rows = migrate_table(cursor, table_cfg, cfg, use_truncate=True)
logger.info(f" 迁移完成: {table_name} ({rows:,} 行)")
except Exception as e:
logger.error(f" 迁移失败: {table_name} - {e}", exc_info=True)
try:
conn.rollback()
except Exception:
pass
cursor.close()
except Exception as e:
logger.error(f"数据库连接失败: {e}", exc_info=True)
finally:
if conn:
try:
conn.close()
except Exception:
pass
first_run = False
else:
run_once(cfg, logger)
run_once(cfg)
except Exception as e:
logger.error(f"运行周期异常: {e}", exc_info=True)