From c6ffe7f15333e164ca053ce33de1219071399865 Mon Sep 17 00:00:00 2001 From: jessikitty Date: Thu, 24 Sep 2026 14:31:28 +1000 Subject: [PATCH] Add admin backup routes (create/status/download/verify/delete/restore) --- app/routes/backup.py | 164 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 164 insertions(+) create mode 100644 app/routes/backup.py diff --git a/app/routes/backup.py b/app/routes/backup.py new file mode 100644 index 0000000..d1b171a --- /dev/null +++ b/app/routes/backup.py @@ -0,0 +1,164 @@ +""" +Admin routes for backup, verify and restore. + +Restores are deliberately awkward: they need an explicit typed confirmation +and, for a wipe, a second flag. The CLI (`flask backup restore`) is the +better path on a fresh Docker instance where no admin user exists yet. + +Created by: Jess Rogerson (yelling commands at Claude.AI) +""" + +import os + +from flask import (Blueprint, render_template, redirect, url_for, flash, + request, jsonify, send_file, current_app, abort) +from flask_login import login_required, current_user + +from app.routes.admin import admin_required +from app.services import backup_service as bsvc + +backup_bp = Blueprint('backup', __name__, url_prefix='/admin/backup') + + +@backup_bp.route('/') +@login_required +@admin_required +def index(): + """Backup management page.""" + backups = bsvc.list_backups() + total_bytes = sum(b['size_bytes'] for b in backups) + + uploads_root = bsvc.upload_folder() + upload_files = list(bsvc.walk_uploads(uploads_root)) + uploads_bytes = sum(os.path.getsize(p) for p, _ in upload_files) + + expected = bsvc.referenced_files() + on_disk = {rel for _, rel in upload_files} + missing = sorted(expected - on_disk) + orphaned = sorted(on_disk - expected) + + return render_template( + 'admin/backup.html', + backups=backups, + backup_folder=bsvc.backup_folder(), + backup_total_human=bsvc.human_size(total_bytes), + uploads_root=uploads_root, + upload_file_count=len(upload_files), + uploads_human=bsvc.human_size(uploads_bytes), + missing=missing, + orphaned=orphaned, + dialect=bsvc.database_dialect(), + sqlite_path=bsvc.sqlite_db_path(), + keep=current_app.config.get('BACKUP_KEEP'), + ) + + +@backup_bp.route('/create', methods=['POST']) +@login_required +@admin_required +def create(): + """Start a backup in the background and hand back a job id.""" + label = (request.form.get('label') or '').strip() or None + job_id = bsvc.start_backup_job( + current_app._get_current_object(), + created_by=current_user.username, + label=label, + keep=current_app.config.get('BACKUP_KEEP'), + ) + return jsonify({'job_id': job_id}) + + +@backup_bp.route('/status/') +@login_required +@admin_required +def status(job_id): + job = bsvc.get_job(job_id) + if not job: + return jsonify({'error': 'Unknown job'}), 404 + return jsonify(job) + + +@backup_bp.route('/download/') +@login_required +@admin_required +def download(filename): + folder = bsvc.backup_folder() + safe = os.path.basename(filename) + path = os.path.join(folder, safe) + if not os.path.isfile(path): + abort(404) + return send_file(path, as_attachment=True, download_name=safe) + + +@backup_bp.route('/verify/') +@login_required +@admin_required +def verify(filename): + folder = bsvc.backup_folder() + safe = os.path.basename(filename) + path = os.path.join(folder, safe) + if not os.path.isfile(path): + return jsonify({'error': 'Not found'}), 404 + + report = bsvc.verify_backup(path) + return jsonify({ + 'ok': report['ok'], + 'checked': report['checked'], + 'failed': report['failed'][:20], + 'missing': report['missing'][:20], + 'errors': report['errors'], + }) + + +@backup_bp.route('/delete/', methods=['POST']) +@login_required +@admin_required +def delete(filename): + try: + removed = bsvc.delete_backup(filename) + flash(f'Deleted backup {removed}', 'success') + except FileNotFoundError: + flash('That backup no longer exists.', 'warning') + except ValueError as exc: + flash(str(exc), 'danger') + return redirect(url_for('backup.index')) + + +@backup_bp.route('/restore/', methods=['POST']) +@login_required +@admin_required +def restore(filename): + """Restore an archive that already sits in the backup folder.""" + folder = bsvc.backup_folder() + safe = os.path.basename(filename) + path = os.path.join(folder, safe) + if not os.path.isfile(path): + flash('That backup no longer exists.', 'warning') + return redirect(url_for('backup.index')) + + if request.form.get('confirm', '').strip().upper() != 'RESTORE': + flash('Restore cancelled: confirmation text did not match.', 'warning') + return redirect(url_for('backup.index')) + + mode = request.form.get('mode', 'merge') + force = request.form.get('force') == 'on' + overwrite = request.form.get('overwrite_files') == 'on' + + try: + summary = bsvc.restore_backup(path, mode=mode, force=force, + overwrite_files=overwrite) + except Exception as exc: # noqa: BLE001 - reported to the admin + current_app.logger.exception('Restore failed') + flash(f'Restore failed: {exc}', 'danger') + return redirect(url_for('backup.index')) + + inserted = ', '.join(f"{v} {k}" for k, v in summary['inserted'].items() if v) + flash( + f"Restore complete ({mode} mode). Rows added: {inserted or 'none'}. " + f"Files written: {summary['files_written']} " + f"({summary['files_bytes_human']}), skipped: {summary['files_skipped']}.", + 'success' + ) + for warning in summary['warnings']: + flash(warning, 'info') + return redirect(url_for('backup.index'))