import time import os import sys import pyodbc import config import db_utils import logging from logging.handlers import TimedRotatingFileHandler from collections import defaultdict # ================= 日志系统配置 (新增) ================= def setup_logger(): """ 配置双向日志系统：控制台输出 + 按天轮转的文件日志 """ # 1. 确保日志目录存在 log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'log') if not os.path.exists(log_dir): os.makedirs(log_dir) # 2. 创建 Logger logger = logging.getLogger("SyncService") logger.setLevel(logging.INFO) logger.handlers = [] # 清除旧句柄，防止重复打印 # 3. 文件处理器 (按天轮转) # filename: 正在写入的日志名 # when='midnight': 每天午夜轮转 # interval=1: 间隔1天 # backupCount=30: 保留最近30个文件 log_file_path = os.path.join(log_dir, 'sync_service.log') file_handler = TimedRotatingFileHandler( log_file_path, when='midnight', interval=1, backupCount=30, encoding='utf-8' ) file_handler.suffix = "%Y-%m-%d" # 轮转后的文件名后缀 # 文件日志格式：[时间] [级别] 消息 file_fmt = logging.Formatter('%(asctime)s - [%(levelname)s] - %(message)s') file_handler.setFormatter(file_fmt) # 4. 控制台处理器 (屏幕输出) console_handler = logging.StreamHandler(sys.stdout) # 控制台日志格式：[时间] 消息 (保持简洁) console_fmt = logging.Formatter('%(asctime)s - %(message)s', datefmt='%H:%M:%S') console_handler.setFormatter(console_fmt) # 5. 添加处理器 logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # 初始化全局 Logger logger = setup_logger() # ================= 辅助函数 ================= def clean_access_path(raw_addr): """清洗 TableAddress 字段""" if not raw_addr: return "" s = str(raw_addr).strip() if s.upper().startswith(";DATABASE="): return s[10:] if s.upper().startswith("LOCAL:"): return s[6:] return s def get_norm_path(path): """获取标准化路径""" if not path: return "" return os.path.normpath(path).lower() # ================= 主逻辑 ================= def process_logs(): sql_conn = db_utils.get_sql_conn() sql_cursor = sql_conn.cursor() sql_cursor.fast_executemany = True cols = config.LOG_TABLE_CONFIG # --- 0. 预处理配置映射 --- normalized_mapping = {} for cfg_path, cfg_tables in config.SYNC_MAPPING.items(): norm_key = get_norm_path(cfg_path) normalized_mapping[norm_key] = { 'original_path': cfg_path, 'tables': cfg_tables } try: log_full_name = db_utils.fmt_table(cols['schema'], cols['table_name']) # --- 1. 读取日志 --- query_log = f""" SELECT TOP 1000 {cols['col_log_id']}, {cols['col_table_name']}, {cols['col_record_id']}, {cols['col_address']} FROM {log_full_name} WHERE {cols['col_synced']} = 0 AND TableType = 'LINKED_ACCESS' ORDER BY {cols['col_log_id']} ASC """ sql_cursor.execute(query_log) logs = sql_cursor.fetchall() if not logs: return False logger.info(f"📥 收到 {len(logs)} 条链接表变更日志...") # --- 2. 任务分组 --- tasks = defaultdict(lambda: defaultdict(lambda: {'record_ids': set(), 'log_ids': []})) for row in logs: log_id, acc_table, record_id, raw_address = row clean_path = clean_access_path(raw_address) norm_lookup_key = get_norm_path(clean_path) # 校验配置 if norm_lookup_key not in normalized_mapping: # 记录一条警告日志，但不刷屏 logger.warning(f"忽略未配置的文件路径: {clean_path}") continue mapping_node = normalized_mapping[norm_lookup_key] if acc_table not in mapping_node['tables']: logger.warning(f"忽略未配置的表: {acc_table} (在文件 {os.path.basename(clean_path)} 中)") continue tasks[clean_path][acc_table]['record_ids'].add(record_id) tasks[clean_path][acc_table]['log_ids'].append(log_id) # --- 3. 执行同步循环 --- for file_path, tables_data in tasks.items(): if not os.path.exists(file_path): logger.error(f"无法访问文件: {file_path}") continue try: acc_conn = db_utils.get_access_conn(file_path) acc_cursor = acc_conn.cursor() except Exception as conn_err: logger.error(f"连接 Access 失败 [{file_path}]: {conn_err}") continue norm_key = get_norm_path(file_path) config_node = normalized_mapping[norm_key] for acc_table, data in tables_data.items(): record_ids = list(data['record_ids']) log_ids = data['log_ids'] target_conf = config_node['tables'][acc_table] target_schema = target_conf['target_schema'] target_table = target_conf['target_table'] pk_col = target_conf['pk_col'] target_full_name = db_utils.fmt_table(target_schema, target_table) try: ids_placeholders = ','.join(['?'] * len(record_ids)) # A. Access 读取 acc_sql = f"SELECT * FROM [{acc_table}] WHERE [{pk_col}] IN ({ids_placeholders})" acc_cursor.execute(acc_sql, record_ids) new_rows = acc_cursor.fetchall() acc_cols = [col[0] for col in acc_cursor.description] # B. SQL Server 写入 del_sql = f"DELETE FROM {target_full_name} WHERE [{pk_col}] IN ({ids_placeholders})" sql_cursor.execute(del_sql, record_ids) if new_rows: insert_sql = db_utils.generate_insert_sql(target_schema, target_table, acc_cols) try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} ON") except: pass sql_cursor.executemany(insert_sql, new_rows) try: sql_cursor.execute(f"SET IDENTITY_INSERT {target_full_name} OFF") except: pass # C. 更新日志状态 log_placeholders = ','.join(['?'] * len(log_ids)) update_log_sql = f""" UPDATE {log_full_name} SET {cols['col_synced']} = 1 WHERE {cols['col_log_id']} IN ({log_placeholders}) """ sql_cursor.execute(update_log_sql, log_ids) sql_conn.commit() logger.info(f"✅ 同步成功: {target_table} 更新 {len(record_ids)} 条 (来源: {os.path.basename(file_path)})") except Exception as tbl_err: logger.error(f"处理表 {acc_table} 异常: {tbl_err}") sql_conn.rollback() acc_conn.close() return True except Exception as e: logger.critical(f"全局严重异常: {e}") return False finally: try: sql_conn.close() except: pass if __name__ == "__main__": logger.info("=== 增量同步服务已启动 (日志模式: log/sync_service.log) ===") while True: try: has_work = process_logs() time.sleep(0.5 if has_work else config.POLL_INTERVAL) except KeyboardInterrupt: logger.info("收到停止指令，服务正在关闭...") break except Exception as e: logger.critical(f"主循环崩溃: {e}") time.sleep(5) # 防止死循环刷屏