feat: backup system for portal DB and WLED configs
Portal DB: - Download current DB button - Manual backup creation - Auto-backup daily at 02:00 (keeps last 14) - Download/delete individual backups WLED Config: - Backup all devices at once - Backup individual device - Download config as JSON - Restore config to any device - Delete old backups New sidebar menu item 'Backup' with dedicated page. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -1,12 +1,16 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Busylight WLED Controller - Web Application."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import FastAPI, Request, Form
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
from fastapi import FastAPI, Request, Form, UploadFile, File
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse, FileResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
@@ -35,11 +39,16 @@ async def lifespan(application: FastAPI):
|
||||
except Exception as e:
|
||||
logger.error("Graph-Client konnte nicht initialisiert werden: %s", e)
|
||||
|
||||
# Backup-Verzeichnis erstellen
|
||||
os.makedirs(BACKUP_DIR, exist_ok=True)
|
||||
os.makedirs(WLED_BACKUP_DIR, exist_ok=True)
|
||||
|
||||
# Start background polling
|
||||
interval = int(database.get_setting("poll_interval", "15"))
|
||||
bg_scheduler.add_job(scheduler.poll_status, "interval", seconds=interval, id="poll")
|
||||
bg_scheduler.add_job(scheduler.check_devices, "interval", seconds=240, id="device_check")
|
||||
# Initial device check nach 5 Sekunden (nicht blockierend)
|
||||
bg_scheduler.add_job(auto_backup_db, "cron", hour=2, minute=0, id="auto_backup")
|
||||
# Initial device check (nicht blockierend)
|
||||
bg_scheduler.add_job(scheduler.check_devices, "date", id="device_check_init")
|
||||
bg_scheduler.start()
|
||||
logger.info("Polling gestartet (Intervall: %ds)", interval)
|
||||
@@ -49,6 +58,25 @@ async def lifespan(application: FastAPI):
|
||||
bg_scheduler.shutdown()
|
||||
|
||||
|
||||
BACKUP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backups")
|
||||
WLED_BACKUP_DIR = os.path.join(BACKUP_DIR, "wled")
|
||||
|
||||
|
||||
def auto_backup_db():
|
||||
"""Create daily DB backup. Keeps last 14 backups."""
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
dst = os.path.join(BACKUP_DIR, f"busylight_{ts}.db")
|
||||
shutil.copy2(database.DB_PATH, dst)
|
||||
# Keep only last 14
|
||||
backups = sorted(
|
||||
[f for f in os.listdir(BACKUP_DIR) if f.startswith("busylight_") and f.endswith(".db")],
|
||||
reverse=True,
|
||||
)
|
||||
for old in backups[14:]:
|
||||
os.remove(os.path.join(BACKUP_DIR, old))
|
||||
logger.info("Auto-Backup erstellt: %s", dst)
|
||||
|
||||
|
||||
app = FastAPI(title="Busylight Manager", lifespan=lifespan)
|
||||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||||
templates = Jinja2Templates(directory="templates")
|
||||
@@ -473,6 +501,151 @@ async def trigger_poll():
|
||||
return RedirectResponse("/", status_code=303)
|
||||
|
||||
|
||||
# ── Backup ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@app.get("/backup", response_class=HTMLResponse)
|
||||
async def backup_page(request: Request):
|
||||
# Portal backups
|
||||
db_backups = []
|
||||
if os.path.exists(BACKUP_DIR):
|
||||
for f in sorted(os.listdir(BACKUP_DIR), reverse=True):
|
||||
if f.startswith("busylight_") and f.endswith(".db"):
|
||||
path = os.path.join(BACKUP_DIR, f)
|
||||
size = os.path.getsize(path) / 1024
|
||||
db_backups.append({"name": f, "size": f"{size:.0f} KB"})
|
||||
|
||||
# WLED config backups
|
||||
wled_backups = []
|
||||
if os.path.exists(WLED_BACKUP_DIR):
|
||||
for f in sorted(os.listdir(WLED_BACKUP_DIR), reverse=True):
|
||||
if f.endswith(".json"):
|
||||
path = os.path.join(WLED_BACKUP_DIR, f)
|
||||
size = os.path.getsize(path) / 1024
|
||||
wled_backups.append({"name": f, "size": f"{size:.1f} KB"})
|
||||
|
||||
conn = database.get_db()
|
||||
devices = conn.execute("SELECT id, hostname, name FROM wled_devices ORDER BY name, hostname").fetchall()
|
||||
conn.close()
|
||||
|
||||
return templates.TemplateResponse(*_ctx(request, "backup",
|
||||
db_backups=db_backups, wled_backups=wled_backups, devices=devices))
|
||||
|
||||
|
||||
@app.get("/api/backup/db/download")
|
||||
async def download_db():
|
||||
"""Download current database."""
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
return FileResponse(
|
||||
database.DB_PATH,
|
||||
media_type="application/octet-stream",
|
||||
filename=f"busylight_{ts}.db",
|
||||
)
|
||||
|
||||
|
||||
@app.post("/api/backup/db/create")
|
||||
async def create_db_backup():
|
||||
"""Create a manual DB backup."""
|
||||
auto_backup_db()
|
||||
return RedirectResponse("/backup?db_created=1", status_code=303)
|
||||
|
||||
|
||||
@app.get("/api/backup/db/{filename}")
|
||||
async def download_db_backup(filename: str):
|
||||
"""Download a specific DB backup."""
|
||||
path = os.path.join(BACKUP_DIR, filename)
|
||||
if not os.path.exists(path) or ".." in filename:
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
return FileResponse(path, media_type="application/octet-stream", filename=filename)
|
||||
|
||||
|
||||
@app.post("/api/backup/db/{filename}/delete")
|
||||
async def delete_db_backup(filename: str):
|
||||
path = os.path.join(BACKUP_DIR, filename)
|
||||
if os.path.exists(path) and ".." not in filename:
|
||||
os.remove(path)
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
|
||||
|
||||
@app.post("/api/backup/wled/all")
|
||||
async def backup_all_wled():
|
||||
"""Backup config of all WLED devices."""
|
||||
from wled_client import get_wled_config
|
||||
conn = database.get_db()
|
||||
devices = conn.execute("SELECT id, hostname, name FROM wled_devices").fetchall()
|
||||
conn.close()
|
||||
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
count = 0
|
||||
for dev in devices:
|
||||
config = get_wled_config(dev["hostname"])
|
||||
if config:
|
||||
safe_name = dev["hostname"].replace(".", "_")
|
||||
path = os.path.join(WLED_BACKUP_DIR, f"{safe_name}_{ts}.json")
|
||||
with open(path, "w") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
count += 1
|
||||
|
||||
return RedirectResponse(f"/backup?wled_backed_up={count}", status_code=303)
|
||||
|
||||
|
||||
@app.post("/api/backup/wled/{device_id}")
|
||||
async def backup_single_wled(device_id: int):
|
||||
"""Backup config of a single WLED device."""
|
||||
from wled_client import get_wled_config
|
||||
conn = database.get_db()
|
||||
dev = conn.execute("SELECT hostname FROM wled_devices WHERE id=?", (device_id,)).fetchone()
|
||||
conn.close()
|
||||
if not dev:
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
|
||||
config = get_wled_config(dev["hostname"])
|
||||
if config:
|
||||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
safe_name = dev["hostname"].replace(".", "_")
|
||||
path = os.path.join(WLED_BACKUP_DIR, f"{safe_name}_{ts}.json")
|
||||
with open(path, "w") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
|
||||
return RedirectResponse("/backup?wled_backed_up=1", status_code=303)
|
||||
|
||||
|
||||
@app.get("/api/backup/wled/download/{filename}")
|
||||
async def download_wled_backup(filename: str):
|
||||
path = os.path.join(WLED_BACKUP_DIR, filename)
|
||||
if not os.path.exists(path) or ".." in filename:
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
return FileResponse(path, media_type="application/json", filename=filename)
|
||||
|
||||
|
||||
@app.post("/api/backup/wled/delete/{filename}")
|
||||
async def delete_wled_backup(filename: str):
|
||||
path = os.path.join(WLED_BACKUP_DIR, filename)
|
||||
if os.path.exists(path) and ".." not in filename:
|
||||
os.remove(path)
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
|
||||
|
||||
@app.post("/api/backup/wled/restore/{filename}")
|
||||
async def restore_wled_backup(filename: str, device_id: int = Form(...)):
|
||||
"""Restore a WLED config backup to a device."""
|
||||
from wled_client import restore_wled_config
|
||||
path = os.path.join(WLED_BACKUP_DIR, filename)
|
||||
if not os.path.exists(path) or ".." in filename:
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
|
||||
conn = database.get_db()
|
||||
dev = conn.execute("SELECT hostname FROM wled_devices WHERE id=?", (device_id,)).fetchone()
|
||||
conn.close()
|
||||
if not dev:
|
||||
return RedirectResponse("/backup", status_code=303)
|
||||
|
||||
with open(path) as f:
|
||||
config = json.load(f)
|
||||
|
||||
success, msg = restore_wled_config(dev["hostname"], config)
|
||||
return RedirectResponse(f"/backup?restored={'1' if success else '0'}", status_code=303)
|
||||
|
||||
|
||||
# ── CLI Auth Command ───────────────────────────────────────────────────────
|
||||
|
||||
def cli_auth():
|
||||
|
||||
Reference in New Issue
Block a user