dzxj_dtu/backend/modules/config_backup.py

"""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'