Files
LEDMatrix/web_interface/blueprints/pages_v3.py
Chuck 4ef3f8cad5 feat(web): add config backup & restore UI (#310)
* feat(web): add config backup & restore UI

Adds a Backup & Restore tab to the v3 web UI that packages user config,
secrets, WiFi, user-uploaded fonts, plugin image uploads, and the installed
plugin list into a single ZIP for safe reinstall recovery. Restore extracts
the bundle, snapshots current state via the existing atomic config manager
(so rollback stays available), reapplies the selected sections, and
optionally reinstalls missing plugins from the store.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(backup): address PR review findings

- backup_manager: read plugin state from "states" key (not "plugins") to
  match the actual plugin_state.json format written by state_manager
- backup_manager: stream ZIP directly to a temp file instead of building
  it in an io.BytesIO buffer to avoid OOM on Raspberry Pi
- backup_manager: tighten plugin-uploads path validation in validate_backup
  and restore_backup to require "/uploads/" in the path, rejecting any
  non-uploads files smuggled under assets/plugins/
- api_v3: enforce 200 MB upload limit by streaming in chunks rather than
  relying on validate_file_upload (which only checks the filename)
- api_v3: replace bool() with _coerce_to_bool() for RestoreOptions fields
  so string "false" is not treated as truthy
- api_v3: capture and log _save_config_atomic return value instead of
  discarding it; log rather than silence font-cache and config-reload errors
- backup_restore.html: track inspectedFile so runRestore always applies to
  the file the user inspected, not a subsequently selected file; clear on
  input change or clearRestore()
- backup_restore.html: throw on non-success restore payload so errors are
  surfaced via the error notification path instead of yellow "warnings"
- test: update fixture to use correct "states" key structure; import
  SCHEMA_VERSION constant instead of hardcoding 1; rename unused err -> _err

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(backup): address second round of PR review findings

- api_v3: guard opts_dict with isinstance check after json.loads so a
  non-object JSON payload (null, array, etc.) returns a 400 instead of a
  500 AttributeError
- backup_manager: wrap tmp ZIP creation and os.replace in try/except so
  the .zip.tmp temp file is always removed on any failure
- backup_manager: replace hardcoded Path("/tmp/_zip_check") sentinel in
  validate_backup with a proper tempfile.TemporaryDirectory() so path
  traversal checks are portable and leave no artifacts
- backup_restore.html: detect partial-success responses (plugins_failed or
  errors non-empty) even when status is 'success' and render yellow/warning
  styling and notify instead of green

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(backup): add post-install steps for restored plugins; conditional restart hint

- api_v3: after a successful plugin reinstall during restore, run the same
  post-install sequence used by the normal /plugins/install flow:
  invalidate schema cache, discover_plugins()/load_plugin(), and
  set_plugin_installed() so restored plugins are immediately available
- backup_restore.html: only show the "restart the display service" hint
  when at least one item was restored or at least one plugin was installed

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(backup): address Codacy findings

- api_v3: replace 'fonts' in ' '.join(result.restored) substring check
  with any(r.startswith("fonts") for r in result.restored) to avoid
  fragile joined-string membership testing
- api_v3: replace deprecated datetime.utcnow() and utcfromtimestamp()
  with datetime.now(timezone.utc) and fromtimestamp(..., timezone.utc);
  add timezone to import
- test: remove unused import io (backup_manager no longer uses BytesIO)
- src/backup_manager.py hardcoded /tmp sentinel was already fixed in a
  prior commit (tempfile.TemporaryDirectory)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-27 17:05:27 -04:00

534 lines
24 KiB
Python

from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify
import json
import logging
from pathlib import Path
from src.web_interface.secret_helpers import mask_secret_fields
logger = logging.getLogger(__name__)
# Will be initialized when blueprint is registered
config_manager = None
plugin_manager = None
plugin_store_manager = None
pages_v3 = Blueprint('pages_v3', __name__)
@pages_v3.route('/')
def index():
"""Main v3 interface page"""
try:
if pages_v3.config_manager:
# Load configuration data
main_config = pages_v3.config_manager.load_config()
schedule_config = main_config.get('schedule', {})
# Get raw config files for JSON editor
main_config_data = pages_v3.config_manager.get_raw_file_content('main')
secrets_config_data = pages_v3.config_manager.get_raw_file_content('secrets')
main_config_json = json.dumps(main_config_data, indent=4)
secrets_config_json = json.dumps(secrets_config_data, indent=4)
else:
raise Exception("Config manager not initialized")
except Exception as e:
flash(f"Error loading configuration: {e}", "error")
schedule_config = {}
main_config_json = "{}"
secrets_config_json = "{}"
main_config_data = {}
secrets_config_data = {}
main_config_path = ""
secrets_config_path = ""
return render_template('v3/index.html',
schedule_config=schedule_config,
main_config_json=main_config_json,
secrets_config_json=secrets_config_json,
main_config_path=pages_v3.config_manager.get_config_path() if pages_v3.config_manager else "",
secrets_config_path=pages_v3.config_manager.get_secrets_path() if pages_v3.config_manager else "",
main_config=main_config_data,
secrets_config=secrets_config_data)
@pages_v3.route('/partials/<partial_name>')
def load_partial(partial_name):
"""Load HTMX partials dynamically"""
try:
# Map partial names to specific data loading
if partial_name == 'overview':
return _load_overview_partial()
elif partial_name == 'general':
return _load_general_partial()
elif partial_name == 'display':
return _load_display_partial()
elif partial_name == 'durations':
return _load_durations_partial()
elif partial_name == 'schedule':
return _load_schedule_partial()
elif partial_name == 'weather':
return _load_weather_partial()
elif partial_name == 'stocks':
return _load_stocks_partial()
elif partial_name == 'plugins':
return _load_plugins_partial()
elif partial_name == 'fonts':
return _load_fonts_partial()
elif partial_name == 'logs':
return _load_logs_partial()
elif partial_name == 'raw-json':
return _load_raw_json_partial()
elif partial_name == 'backup-restore':
return _load_backup_restore_partial()
elif partial_name == 'wifi':
return _load_wifi_partial()
elif partial_name == 'cache':
return _load_cache_partial()
elif partial_name == 'operation-history':
return _load_operation_history_partial()
else:
return f"Partial '{partial_name}' not found", 404
except Exception as e:
return f"Error loading partial '{partial_name}': {str(e)}", 500
@pages_v3.route('/partials/plugin-config/<plugin_id>')
def load_plugin_config_partial(plugin_id):
"""Load plugin configuration partial via HTMX - server-side rendered form"""
try:
return _load_plugin_config_partial(plugin_id)
except Exception as e:
return f'<div class="text-red-500 p-4">Error loading plugin config: {str(e)}</div>', 500
def _load_overview_partial():
"""Load overview partial with system stats"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
# This would be populated with real system stats via SSE
return render_template('v3/partials/overview.html',
main_config=main_config)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_general_partial():
"""Load general settings partial"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
return render_template('v3/partials/general.html',
main_config=main_config)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_display_partial():
"""Load display settings partial"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
return render_template('v3/partials/display.html',
main_config=main_config)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_durations_partial():
"""Load display durations partial"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
return render_template('v3/partials/durations.html',
main_config=main_config)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_schedule_partial():
"""Load schedule settings partial"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
schedule_config = main_config.get('schedule', {})
dim_schedule_config = main_config.get('dim_schedule', {})
# Get normal brightness for display in dim schedule UI
normal_brightness = main_config.get('display', {}).get('hardware', {}).get('brightness', 90)
return render_template('v3/partials/schedule.html',
schedule_config=schedule_config,
dim_schedule_config=dim_schedule_config,
normal_brightness=normal_brightness)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_weather_partial():
"""Load weather configuration partial"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
return render_template('v3/partials/weather.html',
main_config=main_config)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_stocks_partial():
"""Load stocks configuration partial"""
try:
if pages_v3.config_manager:
main_config = pages_v3.config_manager.load_config()
return render_template('v3/partials/stocks.html',
main_config=main_config)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_plugins_partial():
"""Load plugins management partial"""
try:
import json
from pathlib import Path
# Load plugin data from the plugin system
plugins_data = []
# Get installed plugins if managers are available
if pages_v3.plugin_manager and pages_v3.plugin_store_manager:
try:
# Get all installed plugin info
all_plugin_info = pages_v3.plugin_manager.get_all_plugin_info()
# Load config once before the loop (not per-plugin)
full_config = pages_v3.config_manager.load_config() if pages_v3.config_manager else {}
# Format for the web interface
for plugin_info in all_plugin_info:
plugin_id = plugin_info.get('id')
# Re-read manifest from disk to ensure we have the latest metadata
manifest_path = Path(pages_v3.plugin_manager.plugins_dir) / plugin_id / "manifest.json"
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
fresh_manifest = json.load(f)
# Update plugin_info with fresh manifest data
plugin_info.update(fresh_manifest)
except Exception as e:
# If we can't read the fresh manifest, use the cached one
print(f"Warning: Could not read fresh manifest for {plugin_id}: {e}")
# Get enabled status from config (source of truth)
# Read from config file first, fall back to plugin instance if config doesn't have the key
enabled = None
if pages_v3.config_manager:
plugin_config = full_config.get(plugin_id, {})
# Check if 'enabled' key exists in config (even if False)
if 'enabled' in plugin_config:
enabled = bool(plugin_config['enabled'])
# Fallback to plugin instance if config doesn't have enabled key
if enabled is None:
plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id)
if plugin_instance:
enabled = plugin_instance.enabled
else:
# Default to True if no config key and plugin not loaded (matches BasePlugin default)
enabled = True
# Get verified status from store registry (no GitHub API calls needed)
store_info = pages_v3.plugin_store_manager.get_registry_info(plugin_id)
verified = store_info.get('verified', False) if store_info else False
last_updated = plugin_info.get('last_updated')
last_commit = plugin_info.get('last_commit') or plugin_info.get('last_commit_sha')
branch = plugin_info.get('branch')
if store_info:
last_updated = last_updated or store_info.get('last_updated') or store_info.get('last_updated_iso')
last_commit = last_commit or store_info.get('last_commit') or store_info.get('last_commit_sha')
branch = branch or store_info.get('branch') or store_info.get('default_branch')
plugins_data.append({
'id': plugin_id,
'name': plugin_info.get('name', plugin_id),
'author': plugin_info.get('author', 'Unknown'),
'category': plugin_info.get('category', 'General'),
'description': plugin_info.get('description', 'No description available'),
'tags': plugin_info.get('tags', []),
'enabled': enabled,
'verified': verified,
'loaded': plugin_info.get('loaded', False),
'last_updated': last_updated,
'last_commit': last_commit,
'branch': branch
})
except Exception as e:
print(f"Error loading plugin data: {e}")
return render_template('v3/partials/plugins.html',
plugins=plugins_data)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_fonts_partial():
"""Load fonts management partial"""
try:
# This would load font data from the font system
fonts_data = {} # Placeholder for font data
return render_template('v3/partials/fonts.html',
fonts=fonts_data)
except Exception as e:
return f"Error: {str(e)}", 500
def _load_logs_partial():
"""Load logs viewer partial"""
try:
return render_template('v3/partials/logs.html')
except Exception as e:
return f"Error: {str(e)}", 500
def _load_raw_json_partial():
"""Load raw JSON editor partial"""
try:
if pages_v3.config_manager:
main_config_data = pages_v3.config_manager.get_raw_file_content('main')
secrets_config_data = pages_v3.config_manager.get_raw_file_content('secrets')
main_config_json = json.dumps(main_config_data, indent=4)
secrets_config_json = json.dumps(secrets_config_data, indent=4)
return render_template('v3/partials/raw_json.html',
main_config_json=main_config_json,
secrets_config_json=secrets_config_json,
main_config_path=pages_v3.config_manager.get_config_path(),
secrets_config_path=pages_v3.config_manager.get_secrets_path())
except Exception as e:
return f"Error: {str(e)}", 500
def _load_backup_restore_partial():
"""Load backup & restore partial."""
try:
return render_template('v3/partials/backup_restore.html')
except Exception as e:
return f"Error: {str(e)}", 500
@pages_v3.route('/setup')
def captive_setup():
"""Lightweight captive portal setup page — self-contained, no frameworks."""
return render_template('v3/captive_setup.html')
def _load_wifi_partial():
"""Load WiFi setup partial"""
try:
return render_template('v3/partials/wifi.html')
except Exception as e:
return f"Error: {str(e)}", 500
def _load_cache_partial():
"""Load cache management partial"""
try:
return render_template('v3/partials/cache.html')
except Exception as e:
return f"Error: {str(e)}", 500
def _load_operation_history_partial():
"""Load operation history partial"""
try:
return render_template('v3/partials/operation_history.html')
except Exception as e:
return f"Error: {str(e)}", 500
def _load_plugin_config_partial(plugin_id):
"""
Load plugin configuration partial - server-side rendered form.
This replaces the client-side generateConfigForm() JavaScript.
"""
try:
if not pages_v3.plugin_manager:
return '<div class="text-red-500 p-4">Plugin manager not available</div>', 500
# Handle starlark app config (starlark:<app_id>)
if plugin_id.startswith('starlark:'):
return _load_starlark_config_partial(plugin_id[len('starlark:'):])
# Try to get plugin info first
plugin_info = pages_v3.plugin_manager.get_plugin_info(plugin_id)
# If not found, re-discover plugins (handles plugins added after startup)
if not plugin_info:
pages_v3.plugin_manager.discover_plugins()
plugin_info = pages_v3.plugin_manager.get_plugin_info(plugin_id)
if not plugin_info:
return f'<div class="text-red-500 p-4">Plugin "{plugin_id}" not found</div>', 404
# Get plugin instance (may be None if not loaded)
plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id)
# Get plugin configuration from config file
config = {}
if pages_v3.config_manager:
full_config = pages_v3.config_manager.load_config()
config = full_config.get(plugin_id, {})
# Load uploaded images from metadata file if images field exists in schema
# This ensures uploaded images appear even if config hasn't been saved yet
schema_path_temp = Path(pages_v3.plugin_manager.plugins_dir) / plugin_id / "config_schema.json"
if schema_path_temp.exists():
try:
with open(schema_path_temp, 'r', encoding='utf-8') as f:
temp_schema = json.load(f)
# Check if schema has an images field with x-widget: file-upload
if (temp_schema.get('properties', {}).get('images', {}).get('x-widget') == 'file-upload' or
temp_schema.get('properties', {}).get('images', {}).get('x_widget') == 'file-upload'):
# Load metadata file
# Get PROJECT_ROOT relative to this file
project_root = Path(__file__).parent.parent.parent
metadata_file = project_root / 'assets' / 'plugins' / plugin_id / 'uploads' / '.metadata.json'
if metadata_file.exists():
try:
with open(metadata_file, 'r', encoding='utf-8') as mf:
metadata = json.load(mf)
# Convert metadata dict to list of image objects
images_from_metadata = list(metadata.values())
# Only use metadata images if config doesn't have images or config images is empty
if not config.get('images') or len(config.get('images', [])) == 0:
config['images'] = images_from_metadata
else:
# Merge: add metadata images that aren't already in config
config_image_ids = {img.get('id') for img in config.get('images', []) if img.get('id')}
new_images = [img for img in images_from_metadata if img.get('id') not in config_image_ids]
if new_images:
config['images'] = config.get('images', []) + new_images
except Exception as e:
print(f"Warning: Could not load metadata for {plugin_id}: {e}")
except Exception:
pass # Will load schema properly below
# Get plugin schema
schema = {}
schema_path = Path(pages_v3.plugin_manager.plugins_dir) / plugin_id / "config_schema.json"
if schema_path.exists():
try:
with open(schema_path, 'r', encoding='utf-8') as f:
schema = json.load(f)
except Exception as e:
print(f"Warning: Could not load schema for {plugin_id}: {e}")
# Get web UI actions from plugin manifest
web_ui_actions = []
manifest_path = Path(pages_v3.plugin_manager.plugins_dir) / plugin_id / "manifest.json"
if manifest_path.exists():
try:
with open(manifest_path, 'r', encoding='utf-8') as f:
manifest = json.load(f)
web_ui_actions = manifest.get('web_ui_actions', [])
except Exception as e:
print(f"Warning: Could not load manifest for {plugin_id}: {e}")
# Mask secret fields before rendering template (fail closed — never leak secrets)
schema_properties = schema.get('properties') if isinstance(schema, dict) else None
if not isinstance(schema_properties, dict):
return '<div class="text-red-500 p-4">Error loading plugin config securely: schema unavailable.</div>', 500
config = mask_secret_fields(config, schema_properties)
# Determine enabled status
enabled = config.get('enabled', True)
if plugin_instance:
enabled = plugin_instance.enabled
# Build plugin data for template
plugin_data = {
'id': plugin_id,
'name': plugin_info.get('name', plugin_id),
'author': plugin_info.get('author', 'Unknown'),
'version': plugin_info.get('version', ''),
'description': plugin_info.get('description', ''),
'category': plugin_info.get('category', 'General'),
'tags': plugin_info.get('tags', []),
'enabled': enabled,
'last_commit': plugin_info.get('last_commit') or plugin_info.get('last_commit_sha', ''),
'branch': plugin_info.get('branch', ''),
}
return render_template(
'v3/partials/plugin_config.html',
plugin=plugin_data,
config=config,
schema=schema,
web_ui_actions=web_ui_actions
)
except Exception as e:
import traceback
traceback.print_exc()
return f'<div class="text-red-500 p-4">Error loading plugin config: {str(e)}</div>', 500
def _load_starlark_config_partial(app_id):
"""Load configuration partial for a Starlark app."""
try:
starlark_plugin = pages_v3.plugin_manager.get_plugin('starlark-apps') if pages_v3.plugin_manager else None
if starlark_plugin and hasattr(starlark_plugin, 'apps'):
app = starlark_plugin.apps.get(app_id)
if not app:
return f'<div class="text-red-500 p-4">Starlark app not found: {app_id}</div>', 404
return render_template(
'v3/partials/starlark_config.html',
app_id=app_id,
app_name=app.manifest.get('name', app_id),
app_enabled=app.is_enabled(),
render_interval=app.get_render_interval(),
display_duration=app.get_display_duration(),
config=app.config,
schema=app.schema,
has_frames=app.frames is not None,
frame_count=len(app.frames) if app.frames else 0,
last_render_time=app.last_render_time,
)
# Standalone: read from manifest file
manifest_file = Path(__file__).resolve().parent.parent.parent / 'starlark-apps' / 'manifest.json'
if not manifest_file.exists():
return f'<div class="text-red-500 p-4">Starlark app not found: {app_id}</div>', 404
with open(manifest_file, 'r') as f:
manifest = json.load(f)
app_data = manifest.get('apps', {}).get(app_id)
if not app_data:
return f'<div class="text-red-500 p-4">Starlark app not found: {app_id}</div>', 404
# Load schema from schema.json if it exists
schema = None
schema_file = Path(__file__).resolve().parent.parent.parent / 'starlark-apps' / app_id / 'schema.json'
if schema_file.exists():
try:
with open(schema_file, 'r') as f:
schema = json.load(f)
except (OSError, json.JSONDecodeError) as e:
logger.warning(f"[Pages V3] Could not load schema for {app_id}: {e}", exc_info=True)
# Load config from config.json if it exists
config = {}
config_file = Path(__file__).resolve().parent.parent.parent / 'starlark-apps' / app_id / 'config.json'
if config_file.exists():
try:
with open(config_file, 'r') as f:
config = json.load(f)
except (OSError, json.JSONDecodeError) as e:
logger.warning(f"[Pages V3] Could not load config for {app_id}: {e}", exc_info=True)
return render_template(
'v3/partials/starlark_config.html',
app_id=app_id,
app_name=app_data.get('name', app_id),
app_enabled=app_data.get('enabled', True),
render_interval=app_data.get('render_interval', 300),
display_duration=app_data.get('display_duration', 15),
config=config,
schema=schema,
has_frames=False,
frame_count=0,
last_render_time=None,
)
except Exception as e:
logger.exception(f"[Pages V3] Error loading starlark config for {app_id}")
return f'<div class="text-red-500 p-4">Error loading starlark config: {str(e)}</div>', 500