"""Config backup + factory-default-reseed detection + restore (sqlite board_config).""" import json import os import shutil import sqlite3 import time CONFIG_KEYS = ("device_config", "protocol_config") def _read(db_path, key): con = sqlite3.connect(db_path, timeout=5.0) try: row = con.execute("SELECT value FROM board_config WHERE key=?", (key,)).fetchone() return row[0] if row else None finally: con.close() def _write(db_path, key, value): con = sqlite3.connect(db_path, timeout=5.0) try: con.execute( "INSERT INTO board_config(key,value) VALUES(?,?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value), ) con.commit() finally: con.close() def _write_atomic(db_path, key_value_pairs): """Write multiple board_config keys in a single SQLite transaction. v1.5.4.4 STAB-1 fix: detect_and_restore가 device_config + protocol_config을 별도 connection / commit으로 write하던 결함을 차단. 단일 connection + 명시적 BEGIN IMMEDIATE로 두 키를 묶어 원자성을 보장 — 두 번째 write 실패 시 SQLite가 첫 번째 write도 rollback 처리. DBManager.update_config 패턴과 일관. Args: db_path: SQLite DB path. key_value_pairs: iterable of (key, value) tuples. 빈 입력은 no-op. Raises: sqlite3.Error: 어느 한 write라도 실패하면 rollback 후 그대로 전파. """ if not key_value_pairs: return con = sqlite3.connect(db_path, timeout=5.0, isolation_level=None) try: con.execute("BEGIN IMMEDIATE") try: for key, value in key_value_pairs: con.execute( "INSERT INTO board_config(key,value) VALUES(?,?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value), ) con.execute("COMMIT") except Exception: try: con.execute("ROLLBACK") except sqlite3.Error: pass raise finally: con.close() def _coerce(v): """Type-insensitive canonical form for comparing one config value. Note: list comparison is order-sensitive by design — a reordered list reads as a change, which fails safe toward no-restore. bool check must come before int because bool is a subclass of int.""" if isinstance(v, bool): return "true" if v else "false" if isinstance(v, int): return str(v) if isinstance(v, float): return str(int(v)) if v.is_integer() else repr(v) if isinstance(v, (dict, list)): return json.dumps(v, sort_keys=True, ensure_ascii=False) if v is None: return "" return str(v).strip() def _as_dict(text): if text is None: return None try: d = json.loads(text) except (ValueError, TypeError): return None return d if isinstance(d, dict) else None class ResetAssessment: def __init__(self, is_reset, reason, reverted, user_changed, considered): self.is_reset = is_reset self.reason = reason self.reverted = reverted # keys reverted to factory default self.user_changed = user_changed # keys holding a genuinely new value self.considered = considered # user-distinguishing keys judged def assess_reset(current_value, backup_value, default_value): """Field-level reseed detection. current/backup/default are JSON strings. A "reseed" = the user's distinguishing fields were wholesale reverted to factory defaults AND nothing looks like a genuine new edit (do-no-harm). Robust to dpworldapp re-serializing the default differently than the file. """ cur, bak, dft = _as_dict(current_value), _as_dict(backup_value), _as_dict(default_value) if cur is None or bak is None or dft is None: return ResetAssessment(False, "missing or invalid config json", [], [], []) considered, reverted, user_changed = [], [], [] for k in (set(bak) & set(dft)): bv, dv = _coerce(bak.get(k)), _coerce(dft.get(k)) if bv == dv: continue # not user-distinguishing (user value already equals default) considered.append(k) if k not in cur: reverted.append(k) # device dropped the key on reseed -> treat as reverted continue cv = _coerce(cur.get(k)) if cv == dv: reverted.append(k) # user value lost back to factory default elif cv != bv: user_changed.append(k) # a genuinely new value (not backup, not default) # else cv == bv -> user value preserved n = len(considered) is_reset = ( n >= 2 and len(user_changed) == 0 and len(reverted) >= max(2, (n * 4 + 4) // 5) # >= ceil(0.8 * n), min 2 ) if is_reset: reason = "reseed: %d/%d user fields reverted to factory default, 0 new edits" % (len(reverted), n) else: reason = "not a clean reseed (reverted=%d, user_changed=%d, considered=%d)" % ( len(reverted), len(user_changed), n) return ResetAssessment(is_reset, reason, reverted, user_changed, considered) def backup_config(db_path, extra_files, out_dir, stamp=None): stamp = stamp or time.strftime("%Y%m%d-%H%M%S") dest = os.path.join(out_dir, stamp) os.makedirs(dest, exist_ok=True) saved = [] for key in CONFIG_KEYS: val = _read(db_path, key) if val is not None: with open(os.path.join(dest, key + ".json"), "w", encoding="utf-8") as f: f.write(val) saved.append(key) for fp in extra_files or []: if os.path.isfile(fp): shutil.copy2(fp, dest) with open(os.path.join(dest, "manifest.json"), "w", encoding="utf-8") as f: json.dump({"stamp": stamp, "keys": saved, "db": db_path}, f) return dest def latest_backup(out_dir): if not os.path.isdir(out_dir): return None subs = [os.path.join(out_dir, d) for d in os.listdir(out_dir)] subs = [d for d in subs if os.path.isdir(d)] return max(subs, key=os.path.getmtime, default=None) class RestoreResult: def __init__(self, restored, reason, backup=None): self.restored = restored self.reason = reason self.backup = backup def detect_and_restore(db_path, default_file_path, backups_dir): current = _read(db_path, "device_config") backup = latest_backup(backups_dir) if backup is None: return RestoreResult(False, "no backup available") backup_file = os.path.join(backup, "device_config.json") if not os.path.isfile(backup_file): return RestoreResult(False, "backup has no device_config", backup) with open(backup_file, "r", encoding="utf-8") as f: backup_value = f.read() if not os.path.isfile(default_file_path): return RestoreResult(False, "default config file not found", backup) with open(default_file_path, "r", encoding="utf-8") as f: default_value = f.read() assessment = assess_reset(current, backup_value, default_value) if not assessment.is_reset: return RestoreResult(False, assessment.reason, backup) if _as_dict(backup_value) is None: return RestoreResult(False, "backup JSON invalid", backup) # Validate protocol_config BEFORE writing anything — prevent a half-write. proto = os.path.join(backup, "protocol_config.json") proto_value = None if os.path.isfile(proto): with open(proto, "r", encoding="utf-8") as f: proto_value = f.read() if _as_dict(proto_value) is None: return RestoreResult(False, "backup protocol_config invalid", backup) # Both validated — now write atomically. # v1.5.4.4 STAB-1 fix: 이전엔 두 _write 호출이 별도 connection / commit이라 # 두 번째 write 실패 시 device_config만 복원된 zombie state가 발생할 수 있었다. # 이제 단일 BEGIN IMMEDIATE transaction으로 묶어 atomicity 보장. pairs = [("device_config", backup_value)] if proto_value is not None: pairs.append(("protocol_config", proto_value)) _write_atomic(db_path, pairs) return RestoreResult(True, "restored from backup (%s)" % assessment.reason, backup)