pxe-server/webapp/views/dell_updates.py

"""Updates > Dell: Win11 driver packs and WinPE BIOS files from Dell's catalogs.

Check and download run as background jobs (systemd-run) that write progress
JSON the page polls; preview, apply, discard and rollback are quick renames
and small file rewrites, so they run inline. Logic lives in
services/dell_updates.py; this module is the route surface.
"""

from flask import Blueprint, flash, jsonify, redirect, render_template, request, url_for

from services import dell_updates, updates_common
from services.audit import audit
from views import save_speed_cap

bp = Blueprint("dell_updates", __name__, url_prefix="/updates/dell")


def _back():
    return redirect(url_for("dell_updates.page"))


def _render(preview=None, preview_keys=()):
    prog = dell_updates.progress()
    items = sorted((prog["download"].get("items") or {}).values(),
                   key=lambda i: (i.get("status") == "applied", i.get("kind", ""), i.get("name", "")))
    return render_template(
        "updates_dell.html",
        scan=dell_updates.last_scan(),
        prog=prog,
        staged=items,
        history=dell_updates.history()[:15],
        settings=updates_common.settings(),
        preview=preview,
        preview_keys=list(preview_keys),
    )


@bp.route("/")
def page():
    return _render()


@bp.route("/status.json")
def status():
    prog = dell_updates.progress()
    items = {k: {"status": v.get("status"), "percent": v.get("percent"), "error": v.get("error", "")}
             for k, v in (prog["download"].get("items") or {}).items()}
    return jsonify({
        "check": {"state": prog["check"].get("state", ""), "step": prog["check"].get("step", ""),
                  "error": prog["check"].get("error", "")},
        "download": {"state": prog["download"].get("state", ""), "items": items},
        "check_running": prog["check_running"],
        "download_running": prog["download_running"],
    })


@bp.route("/check", methods=["POST"])
def check():
    try:
        dell_updates.start("check")
    except Exception as ex:
        flash(f"Check not started: {ex}", "danger")
        return _back()
    audit("DELL_CHECK", "catalog check started")
    flash("Checking Dell's catalogs; this page updates when the scan is ready (a minute or two).", "info")
    return _back()


@bp.route("/download", methods=["POST"])
def download():
    drivers = set(request.form.getlist("drv"))
    picks = {}
    for row_id in request.form.getlist("bios"):
        try:
            picks[row_id] = int(request.form.get(f"pick_{row_id}", "0"))
        except ValueError:
            picks[row_id] = 0
    try:
        count = dell_updates.select(drivers, picks)
        dell_updates.start("download")
    except Exception as ex:
        flash(f"Download not started: {ex}", "danger")
        return _back()
    audit("DELL_DOWNLOAD_START", f"{count} item(s): {len(drivers)} driver row(s), {len(picks)} BIOS line(s)")
    cap = updates_common.settings().get("max_mbps") or 0
    flash(f"Downloading {count} item(s) at {'%g MB/s' % cap if cap else 'full speed'}. "
          "Driver packs are hash checked and repacked here; nothing on the share changes until you apply.", "info")
    return _back()


@bp.route("/cancel", methods=["POST"])
def cancel():
    job = request.form.get("job", "download")
    if job == "check":
        updates_common.stop_job(dell_updates.UNITS["check"])
        audit("DELL_CHECK_CANCEL", "")
        flash("Check stopped. The previous scan, if any, is still shown.", "warning")
    else:
        dell_updates.cancel_download()
        audit("DELL_DOWNLOAD_CANCEL", "")
        flash("Download stopped. Partial files are kept; downloading the same items again resumes them.", "warning")
    return _back()


@bp.route("/preview", methods=["POST"])
def preview():
    keys = request.form.getlist("key")
    if not keys:
        flash("Select at least one staged item to apply.", "warning")
        return _back()
    return _render(preview=dell_updates.preview(keys), preview_keys=keys)


@bp.route("/apply", methods=["POST"])
def apply():
    keys = request.form.getlist("key")
    if not keys:
        flash("Nothing selected to apply.", "warning")
        return _back()
    if updates_common.job_running(dell_updates.UNITS["download"]):
        flash("A download is still running; apply when it has finished.", "danger")
        return _back()
    try:
        report = dell_updates.apply(keys)
    except Exception as ex:
        flash(f"Apply failed: {ex}", "danger")
        return _back()
    parts = []
    if report["drivers"]:
        parts.append(f"{len(report['drivers'])} driver pack(s) replaced in the maps")
    if report["bios"]:
        parts.append(f"{len(report['bios'])} models.txt line(s) updated")
    if parts:
        flash("Applied: " + ", ".join(parts) + ". Old zips and BIOS files were kept; the maps and "
              "models.txt were backed up first (Rollback in the history below).", "success")
    for msg in report["skipped"]:
        flash(f"Skipped {msg}.", "warning")
    for msg in report["errors"]:
        flash(f"Not applied: {msg}", "danger")
    if not parts and not report["skipped"] and not report["errors"]:
        flash("Nothing was applied (no selected item was ready).", "warning")
    return _back()


@bp.route("/discard", methods=["POST"])
def discard():
    keys = request.form.getlist("key")
    if updates_common.job_running(dell_updates.UNITS["download"]):
        flash("A download is running; cancel it first.", "danger")
        return _back()
    gone = dell_updates.discard(keys)
    if gone:
        audit("DELL_DISCARD", ", ".join(gone))
        flash(f"Removed {len(gone)} staged item(s) from the staging area.", "success")
    else:
        flash("Nothing removed.", "warning")
    return _back()


@bp.route("/rollback", methods=["POST"])
def rollback():
    apply_id = request.form.get("apply_id", "")
    try:
        restored, refused = dell_updates.rollback(apply_id)
    except Exception as ex:
        flash(f"Rollback failed: {ex}", "danger")
        return _back()
    if restored:
        flash(f"Restored {len(restored)} file(s) from before apply {apply_id}. The new zips/BIOS files stay "
              "on the share, unused.", "success")
    for msg in refused:
        flash(f"Not restored: {msg}.", "danger")
    return _back()


@bp.route("/settings", methods=["POST"])
def save_settings():
    save_speed_cap()  # shared with the other Updates pages (views/__init__.py)
    return _back()