"""DTU 运行数据备份/快照工具。 功能: - 原子写 JSON 配置(write_json_atomic) - 写前自动快照(snapshot_before_write): 内容变化才快照, 每类保留最近 N 份, 节流去重 - 快照列表/读取/删除(list_snapshots/read_snapshot/delete_snapshot/find_latest_snapshot) - 整机备份包构建与校验(build_bundle/validate_bundle) 整机备份包(bundle)结构: { "format": "dzxj_dtu_backup", "format_version": 1, "created_at": "2026-10-13T14:00:00+0800", "device": {"dtu_id": ..., "customer_id": ..., ...}, "effective_uid_prefix": "A1B2", # 导入时固化, 保证换机后 jumper_uid 前缀不变 "files": { "dtu_config": {...}, "serial_config": {...}, "mqtt_config": {...}, "devices": {...}, "system_env": "..." } } """ import json import logging import os import re import time logger = logging.getLogger('serial_mqtt_gateway') BUNDLE_FORMAT = 'dzxj_dtu_backup' BUNDLE_FORMAT_VERSION = 1 MAX_BUNDLE_BYTES = 2 * 1024 * 1024 DEFAULT_KEEP = 20 DEFAULT_MIN_INTERVAL_S = 30 # backend/backups(模块位于 backend/modules/ 下) BACKUP_DIR = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'backups' ) _SNAPSHOT_RE = re.compile(r'^(.+)_(\d{8}_\d{6})(?:_(\d+))?\.json$') _UID_RE = re.compile(r'^[0-9a-fA-F]{24}$') _PREFIX_RE = re.compile(r'^[0-9a-fA-F]{4}$') BUNDLE_FILES = ('dtu_config', 'serial_config', 'mqtt_config', 'devices', 'system_env') def ensure_backup_dir(backup_dir=BACKUP_DIR): """确保备份目录存在""" try: os.makedirs(backup_dir, exist_ok=True) except OSError as e: logger.error(f"创建备份目录失败: {e}") return backup_dir def write_json_atomic(path, data): """原子写 JSON 文件(tmp + os.replace), 避免写一半损坏""" tmp_path = f"{path}.tmp" with open(tmp_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) os.replace(tmp_path, path) def read_json_file(path): """读取 JSON 文件, 失败抛出异常""" with open(path, 'r', encoding='utf-8') as f: return json.load(f) def _snapshot_name(raw_name): """校验快照文件名, 防路径穿越; 非法返回 None""" if not raw_name or os.path.basename(raw_name) != raw_name: return None return raw_name if _SNAPSHOT_RE.match(raw_name) else None def snapshot_stem(name): """从快照文件名提取原始配置文件 stem, 非法返回 None""" safe = _snapshot_name(name) if not safe: return None return _SNAPSHOT_RE.match(safe).group(1) def list_snapshots(backup_dir=BACKUP_DIR, stem=None): """列出快照, 按修改时间倒序(最新在前) 返回: [{'name','stem','size','mtime'}, ...] """ result = [] if not os.path.isdir(backup_dir): return result for name in os.listdir(backup_dir): m = _SNAPSHOT_RE.match(name) if not m: continue if stem and m.group(1) != stem: continue path = os.path.join(backup_dir, name) try: st = os.stat(path) except OSError: continue result.append({ 'name': name, 'stem': m.group(1), 'size': st.st_size, 'mtime': st.st_mtime, }) result.sort(key=lambda x: x['mtime'], reverse=True) return result def find_latest_snapshot(backup_dir=BACKUP_DIR, stem=None): """返回最新一份快照信息, 无则 None""" snaps = list_snapshots(backup_dir, stem=stem) return snaps[0] if snaps else None def read_snapshot(name, backup_dir=BACKUP_DIR): """读取快照内容(JSON), 返回 (data, error)""" safe = _snapshot_name(name) if not safe: return None, '非法快照文件名' path = os.path.join(backup_dir, safe) if not os.path.exists(path): return None, '快照不存在' try: with open(path, 'r', encoding='utf-8') as f: return json.load(f), None except Exception as e: return None, f'快照解析失败: {e}' def delete_snapshot(name, backup_dir=BACKUP_DIR): """删除快照, 返回 (ok, message)""" safe = _snapshot_name(name) if not safe: return False, '非法快照文件名' path = os.path.join(backup_dir, safe) if not os.path.exists(path): return False, '快照不存在' try: os.remove(path) return True, '已删除' except OSError as e: return False, f'删除失败: {e}' def _rotate_snapshots(backup_dir, stem, keep): """每类快照只保留最近 keep 份""" snaps = list_snapshots(backup_dir, stem=stem) # 倒序 for item in snaps[keep:]: try: os.remove(os.path.join(backup_dir, item['name'])) logger.info(f"清理旧配置快照: {item['name']}") except OSError: pass def snapshot_before_write(path, backup_dir=BACKUP_DIR, keep=DEFAULT_KEEP, min_interval_s=DEFAULT_MIN_INTERVAL_S, now=None): """写文件前对现有内容做快照。 规则: - 目标文件不存在或为空 -> 不快照 - 内容与最新快照相同 -> 跳过(去重) - 距最新快照不足 min_interval_s -> 跳过(节流) 返回快照文件名, 未创建返回 None """ now = time.time() if now is None else now try: if not os.path.exists(path) or os.path.getsize(path) == 0: return None with open(path, 'rb') as f: current = f.read() stem = os.path.splitext(os.path.basename(path))[0] latest = find_latest_snapshot(backup_dir, stem=stem) if latest: try: with open(os.path.join(backup_dir, latest['name']), 'rb') as f: if f.read() == current: return None except OSError: pass if now - latest['mtime'] < min_interval_s: return None ensure_backup_dir(backup_dir) base = os.path.join( backup_dir, f"{stem}_{time.strftime('%Y%m%d_%H%M%S', time.localtime(now))}.json" ) target = base counter = 1 while os.path.exists(target): target = base[:-5] + f"_{counter}.json" counter += 1 with open(target, 'wb') as f: f.write(current) try: os.utime(target, (now, now)) # 统一 mtime 与逻辑时间, 保证节流/轮转排序一致 except OSError: pass _rotate_snapshots(backup_dir, stem, keep) logger.info(f"配置快照已保存: {os.path.basename(target)}") return os.path.basename(target) except Exception as e: logger.error(f"配置快照失败({path}): {e}") return None def latest_snapshot_content(path, backup_dir=BACKUP_DIR): """取某配置文件最新快照的内容(用于损坏恢复), 返回 (data, name) 或 (None, None)""" stem = os.path.splitext(os.path.basename(path))[0] latest = find_latest_snapshot(backup_dir, stem=stem) if not latest: return None, None data, err = read_snapshot(latest['name'], backup_dir=backup_dir) if err: logger.warning(f"读取快照 {latest['name']} 失败: {err}") return None, latest['name'] return data, latest['name'] def build_bundle(files, device_meta=None, uid_prefix=None, now=None): """构建整机备份包""" now = time.time() if now is None else now prefix = (uid_prefix or '').strip().upper() return { 'format': BUNDLE_FORMAT, 'format_version': BUNDLE_FORMAT_VERSION, 'created_at': time.strftime('%Y-%m-%dT%H:%M:%S', time.localtime(now)), 'device': device_meta or {}, 'effective_uid_prefix': prefix if _PREFIX_RE.match(prefix) else None, 'files': files, } def _validate_address_range(value): if not isinstance(value, dict): return 'address_range 必须是对象' start = value.get('start') end = value.get('end') if not isinstance(start, int) or not isinstance(end, int): return 'address_range.start/end 必须是整数' if not (1 <= start <= 247) or not (1 <= end <= 247) or start > end: return f'address_range 非法: {start}-{end}(需 1-247 且 start<=end)' return None def validate_bundle(obj): """校验整机备份包, 返回 (ok, message)""" if not isinstance(obj, dict): return False, '备份内容不是 JSON 对象' if obj.get('format') != BUNDLE_FORMAT: return False, '不是本系统的备份文件(format 不匹配)' fv = obj.get('format_version') if not isinstance(fv, int) or not (1 <= fv <= BUNDLE_FORMAT_VERSION): return False, f'备份格式版本不支持: {fv}' files = obj.get('files') if not isinstance(files, dict) or not files: return False, '缺少 files 字段或为空' unknown = set(files) - set(BUNDLE_FILES) if unknown: return False, f'包含未知文件项: {sorted(unknown)}' if not any(v not in (None, {} , '') for v in files.values()): return False, '备份包内没有任何有效文件' for key, val in files.items(): if val is None: continue if key == 'system_env': if not isinstance(val, str): return False, 'system_env 必须是文本内容' continue if not isinstance(val, dict): return False, f'{key} 必须是 JSON 对象' dtu = files.get('dtu_config') if isinstance(dtu, dict): if 'address_range' in dtu and dtu['address_range'] is not None: err = _validate_address_range(dtu['address_range']) if err: return False, err if 'uid_prefix' in dtu and dtu['uid_prefix'] is not None: if not _PREFIX_RE.match(str(dtu['uid_prefix'])): return False, f"uid_prefix 非法: {dtu['uid_prefix']}" pm = dtu.get('port_mappings') if pm is not None and not isinstance(pm, dict): return False, 'port_mappings 必须是对象' dev = files.get('devices') if isinstance(dev, dict): for uid, addr in dev.items(): if not _UID_RE.match(str(uid)): return False, f'devices 中存在非法线架 UID: {uid}' if not isinstance(addr, int) or not (1 <= addr <= 247): return False, f'devices 中地址非法: {uid} -> {addr}' sc = files.get('serial_config') if isinstance(sc, dict) and 'port' in sc and not isinstance(sc['port'], str): return False, 'serial_config.port 必须是字符串' mc = files.get('mqtt_config') if isinstance(mc, dict) and 'broker' in mc and not isinstance(mc['broker'], str): return False, 'mqtt_config.broker 必须是字符串' try: size = len(json.dumps(obj, ensure_ascii=False).encode('utf-8')) except (TypeError, ValueError): return False, '备份内容无法序列化' if size > MAX_BUNDLE_BYTES: return False, f'备份内容过大({size} 字节, 上限 {MAX_BUNDLE_BYTES})' return True, 'ok'