|
|
@@ -0,0 +1,315 @@
|
|
|
+"""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'
|