""" Watch daemon - periodically syncs LAN Excel files and migrates changed tables to SQL Server. """ import argparse import logging import os import sys import time import yaml 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 def run_once(cfg, logger): """Execute one sync + migrate cycle.""" excel_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Excel') # 1. Sync files from LAN 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) if not tables_to_migrate: logger.info("无文件变更,跳过迁移") return logger.info(f"需要迁移的表: {', '.join(sorted(tables_to_migrate))}") # 3. Open DB connection and migrate each changed table 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'] if table_name not in tables_to_migrate: continue 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) # Rollback any uncommitted transaction for this table 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 def main(): parser = argparse.ArgumentParser(description='Excel to SQL Server Watch Daemon') parser.add_argument('--config', default=None, help='配置文件路径 (默认: config.yaml)') parser.add_argument('--once', action='store_true', help='执行一次后退出') parser.add_argument('--interval', type=int, default=None, 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) interval = args.interval or cfg.get('check_interval_minutes', 30) logger.info("=" * 50) logger.info("Watch daemon 启动") 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) except Exception as e: logger.error(f"运行周期异常: {e}", exc_info=True) if args.once: logger.info("--once 模式,退出") break logger.info(f"下次检查: {interval} 分钟后") time.sleep(interval * 60) if __name__ == '__main__': main()