Development

docker/routlin-dash/app/config_utils.py

@@ -144,6 +144,28 @@ def get_dashboard_pending():
     return _read_dashboard_pending()
 
 
+def get_dashboard_done():
+    """Return list of (uuid, applied_ts) from .dashboard-done, newest first."""
+    items = []
+    try:
+        lines = open(DASHBOARD_DONE).read().splitlines()
+    except Exception:
+        return items
+    for line in lines:
+        if not line.strip():
+            continue
+        try:
+            parts = line.split(None, 1)
+            if len(parts) >= 2:
+                items.append((parts[0], int(parts[1])))
+            elif len(parts) == 1:
+                items.append((parts[0], None))
+        except Exception:
+            pass
+    items.reverse()
+    return items
+
+
 def flush_pending_to_queue():
     """Move all entries from .dashboard-pending to .dashboard-queue and clear pending."""
     items = _read_dashboard_pending()
@@ -325,6 +347,64 @@ def _find_snapshot_dependencies(path, key):
         return []
 
 
+def _items_match(item, ref):
+    """Return True if item and ref refer to the same entity by a common identifier field."""
+    if not isinstance(item, dict) or not isinstance(ref, dict):
+        return item == ref
+    for field in ('ip', 'name', 'mac_address', 'host', 'id', 'address'):
+        if field in ref and field in item:
+            return item[field] == ref[field]
+    return item == ref
+
+
+def revert_snapshot_to_core(entry_uuid):
+    """Apply the inverse of a snapshot to core.json and queue a new pending change.
+
+    Returns (flash_message, success_bool).
+    """
+    snap = load_snapshot_for_uuid(entry_uuid)
+    if not snap:
+        return f'Snapshot not found for {entry_uuid[:8]}.', False
+
+    path      = snap['path']
+    key       = snap['key']
+    before    = snap['before']   # original state to restore
+    after     = snap['after']    # applied state to undo
+    operation = snap['operation']
+
+    if operation == 'revert':
+        return 'This change is already a revert; cannot revert again.', False
+
+    core = load_core()
+
+    if key == 'global':
+        if before is None:
+            core.pop(path, None)
+        else:
+            core[path] = before
+    else:
+        items = core.setdefault(path, [])
+        if operation == 'add':
+            core[path] = [x for x in items if not _items_match(x, after)]
+        elif operation == 'delete':
+            if before:
+                core[path].append(before)
+        else:
+            if before and after:
+                for i, item in enumerate(items):
+                    if _items_match(item, after):
+                        items[i] = before
+                        break
+
+    msg = save_core_with_snapshot(
+        core, path=path, key=key, operation='revert',
+        before=after, after=before,
+        description=f"Reverted: {snap.get('description', '')}",
+        cmd=snap.get('cmd', 'core apply'),
+    )
+    return msg or 'Reverted.', True
+
+
 def load_snapshot_for_uuid(entry_uuid):
     """Return the snapshot dict for the given UUID, or None if not found."""
     try: