diff --git a/.gitignore b/.gitignore index e3e787b..692f895 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,7 @@ Desktop.ini # Logs *.log +watch.log # Environment variables / secrets .env diff --git a/README.md b/README.md index d46bfb9..0dc77e4 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,10 @@ Excel/ PG/ # 压力表 Excel 文件 TM/ # 温度计 Excel 文件 -migrate.py # 迁移脚本 -config.yaml # 配置文件（含数据库连接、数据源映射） +migrate.py # 迁移脚本（一次性全量 / TRUNCATE 模式） +sync.py # LAN 文件同步模块 +watch.py # 定时守护进程（自动同步 + 增量迁移） +config.yaml # 配置文件（含数据库连接、LAN 路径、数据源映射） requirements.txt ``` @@ -30,6 +32,8 @@ python -m venv .venv ## 使用方式 +### 一次性迁移（migrate.py） + ```bash # 激活虚拟环境 .venv/Scripts/activate @@ -37,9 +41,12 @@ python -m venv .venv # 预览模式（不写入数据库） python migrate.py --dry-run -# 全表迁移 +# 全表迁移（默认 DROP + CREATE） python migrate.py +# 全表迁移（TRUNCATE 模式，保留表结构） +python migrate.py --truncate + # 单表迁移 python migrate.py --table PG_2022 @@ -50,11 +57,44 @@ python migrate.py --table PG_2022 TM_2025 python migrate.py --config /path/to/config.yaml ``` +### 定时同步守护进程（watch.py） + +自动从局域网 UNC 路径同步 Excel 文件到本地，仅当文件发生变化时触发对应年份表的迁移。 + +```bash +# 启动守护进程（默认每 30 分钟检查一次） +python watch.py + +# 单次执行后退出（用于测试或手动触发） +python watch.py --once + +# 自定义同步间隔（1 分钟，用于调试） +python watch.py --interval 1 + +# 指定配置文件 +python watch.py --config /path/to/config.yaml +``` + +**运行逻辑：** +1. 首次运行：同步所有 LAN 文件并强制迁移全部表 +2. 后续周期：仅同步有变化的文件，只迁移受影响的表 +3. 使用 TRUNCATE 模式（保留表结构，清空数据后重新插入） +4. 单表迁移失败不影响其他表（错误隔离） +5. 日志同时输出到控制台和 `watch.log` 文件 + +## 迁移模式对比 + +| 模式 | 命令 | 行为 | +|------|------|------| +| DROP + CREATE | `python migrate.py` | 每次删除并重建表，适用于表结构可能变化的场景 | +| TRUNCATE | `python migrate.py --truncate` | 保留表结构仅清空数据，性能更好，`watch.py` 自动使用此模式 | + +TRUNCATE 模式在检测到列名变化时会自动回退到 DROP + CREATE。 + ## SQL Server 表结构 - Schema: `executionCard` - 每张表结构：`id INT IDENTITY(1,1)` + 主键 `总排号 NVARCHAR(50)` + 各年份数据列 -- 迁移前已存在的同名表会被 **DROP 后重建** | 表名 | 数据来源 | 行数 | |------|---------|------| @@ -67,10 +107,14 @@ python migrate.py --config /path/to/config.yaml - `tables` — 目标表与 Excel 文件/工作表的映射关系，sources 中后出现的覆盖先出现的（同总排号时） - `type_rules` — 列名到 SQL 类型的推断规则 - `column_mapping` — 列名重映射（如 `ID` → `CRM订单明细ID`） +- `check_interval_minutes` — 定时检查间隔（分钟），默认 30 +- `lan_sources` — 本地相对路径到局域网 UNC 路径的映射，用于自动同步 ## 注意事项 -- 脚本会 **DROP 并重建** 目标表，每次迁移都是全量写入 +- 默认模式（无 `--truncate`）会 **DROP 并重建** 目标表，每次迁移都是全量写入 +- TRUNCATE 模式保留表结构，仅在列名变化时才重建 - 多数据源合并时按主键去重，后读覆盖先读 - 非法的日期/数值会被自动转为 NULL -- `Excel/` 目录已加入 `.gitignore`，不纳入版本控制 +- `Excel/` 目录和 `watch.log` 已加入 `.gitignore`，不纳入版本控制 +- 文件同步使用原子写入（先写 `.tmp` 再 `os.replace`），防止复制中断导致文件损坏 diff --git a/migrate.py b/migrate.py index 636c421..ffd24bb 100644 --- a/migrate.py +++ b/migrate.py @@ -216,6 +216,62 @@ def infer_sql_type(col_name, type_rules, primary_key): return type_rules.get('default_type', 'NVARCHAR(500)') +def table_exists(cursor, schema, table_name): + """检查表是否已存在""" + full_name = f'[{schema}].[{table_name}]' + cursor.execute(f"SELECT OBJECT_ID('{full_name}', 'U')") + row = cursor.fetchone() + return row[0] is not None + + +def get_existing_columns(cursor, schema, table_name): + """查询现有表的列名集合（排除 id 自增列）""" + full_name = f'[{schema}].[{table_name}]' + cursor.execute(f""" + SELECT COLUMN_NAME + FROM INFORMATION_SCHEMA.COLUMNS + WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? + ORDER BY ORDINAL_POSITION + """, (schema, table_name)) + cols = {row[0] for row in cursor.fetchall()} + cols.discard('id') + return cols + + +def ensure_table(cursor, table_name, headers, cfg, actual_pk, use_truncate=False): + """确保目标表就绪。 + - use_truncate=False（默认）→ 调用 create_table()，DROP + CREATE + - use_truncate=True → + 1. 表不存在 → CREATE TABLE + 2. 表存在且列名集合一致 → TRUNCATE TABLE + 3. 表存在但列名有变化 → DROP + CREATE（兜底） + """ + if not use_truncate: + return create_table(cursor, table_name, headers, cfg, actual_pk) + + schema = cfg['schema'] + full_name = f'[{schema}].[{table_name}]' + + if not table_exists(cursor, schema, table_name): + # 表不存在，走正常 CREATE + return create_table(cursor, table_name, headers, cfg, actual_pk) + + # 比较列名集合 + existing_cols = get_existing_columns(cursor, schema, table_name) + new_cols = set(headers) + # 主键列也应包含在比较中 + new_cols.add(actual_pk) + + if existing_cols == new_cols: + # 列名一致，TRUNCATE + cursor.execute(f"TRUNCATE TABLE {full_name}") + cursor.commit() + return f"TRUNCATE TABLE {full_name}" + else: + # 列名有变化，兜底 DROP + CREATE + return create_table(cursor, table_name, headers, cfg, actual_pk) + + def create_table(cursor, table_name, headers, cfg, actual_pk): """动态创建表。如已存在则先 DROP。""" schema = cfg['schema'] @@ -366,7 +422,7 @@ def batch_insert(cursor, table_name, headers, data_dict, cfg, actual_pk): # 主流程 # ============================================================ -def migrate_table(cursor, table_config, cfg, dry_run=False): +def migrate_table(cursor, table_config, cfg, dry_run=False, use_truncate=False): """迁移单个表""" table_name = table_config['table'] sources = table_config['sources'] @@ -400,8 +456,9 @@ def migrate_table(cursor, table_config, cfg, dry_run=False): return len(merged) # 2. 建表 - create_sql = create_table(cursor, table_name, headers, cfg, actual_pk) - print(f" 建表: [{cfg['schema']}].[{table_name}] ({len(headers)} 列)") + 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)} 列)") # 3. 插入数据 total = batch_insert(cursor, table_name, headers, merged, cfg, actual_pk) @@ -414,6 +471,8 @@ def main(): parser = argparse.ArgumentParser(description='Excel to SQL Server Migration') parser.add_argument('--config', default=None, help='配置文件路径 (默认: config.yaml)') parser.add_argument('--dry-run', action='store_true', help='仅预览，不写入数据库') + parser.add_argument('--truncate', action='store_true', + help='使用 TRUNCATE 模式（表存在时截断而非重建）') parser.add_argument('--table', nargs='+', default=None, help='只迁移指定的表，例如 --table PG_2022 TM_2025') args = parser.parse_args() @@ -455,7 +514,8 @@ def main(): total_rows = 0 for i, table_cfg in enumerate(tables, 1): print(f"\n[{i}/{total_tables}]", end='') - rows = migrate_table(cursor, table_cfg, cfg, dry_run=args.dry_run) + rows = migrate_table(cursor, table_cfg, cfg, dry_run=args.dry_run, + use_truncate=args.truncate) total_rows += rows if not args.dry_run: diff --git a/sync.py b/sync.py new file mode 100644 index 0000000..f28e1a1 --- /dev/null +++ b/sync.py @@ -0,0 +1,89 @@ +""" +LAN file sync module. +Detects file changes, copies from LAN UNC paths to local, and determines which tables need migration. +""" + +import os +import shutil + + +def sync_file(local_path, lan_path): + """Sync a single file from LAN to local. + + Returns: + (status, error_msg) + status: 'copied' | 'updated' | 'skipped' | 'error' + """ + try: + if not os.path.exists(lan_path): + return 'error', f'LAN path not accessible: {lan_path}' + + lan_mtime = os.path.getmtime(lan_path) + local_existed = os.path.exists(local_path) + + if local_existed: + local_mtime = os.path.getmtime(local_path) + if local_mtime == lan_mtime: + return 'skipped', None + + # Ensure parent directory exists + os.makedirs(os.path.dirname(local_path), exist_ok=True) + + # Atomic copy: write to .tmp first, then os.replace + tmp_path = local_path + '.tmp' + try: + shutil.copy2(lan_path, tmp_path) + os.replace(tmp_path, local_path) + except Exception: + # Clean up temp file on failure + if os.path.exists(tmp_path): + os.remove(tmp_path) + raise + + if not local_existed: + return 'copied', None + else: + return 'updated', None + + except Exception as e: + return 'error', str(e) + + +def sync_all_files(cfg, excel_dir): + """Sync all LAN source files to local. + + Returns: + {rel_path: (status, error_msg)} + """ + lan_sources = cfg.get('lan_sources', {}) + results = {} + + for rel_path, lan_path in lan_sources.items(): + local_path = os.path.join(excel_dir, rel_path) + status, error_msg = sync_file(local_path, lan_path) + results[rel_path] = (status, error_msg) + + return results + + +def determine_tables_to_migrate(cfg, sync_results): + """Determine which tables need migration based on sync results. + + A table needs migration if ANY of its source files were copied or updated. + + Returns: + set of table names that need migration + """ + tables_to_migrate = set() + + for table_cfg in cfg.get('tables', []): + table_name = table_cfg['table'] + for src in table_cfg['sources']: + file_rel = src['file'] + if file_rel in sync_results: + status, _ = sync_results[file_rel] + if status in ('copied', 'updated'): + tables_to_migrate.add(table_name) + break + + return tables_to_migrate diff --git a/watch.py b/watch.py new file mode 100644 index 0000000..9a5c476 --- /dev/null +++ b/watch.py @@ -0,0 +1,186 @@ +""" +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()