macos-usb-stick-creator/backend/app.py

import json
import logging
import os
import threading
import time

from flask import Flask, Response, jsonify, request, send_from_directory

import catalog
import installer
from devices import list_usb_sticks

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
log = logging.getLogger("app")

FRONTEND_DIR = os.path.join(os.path.dirname(__file__), "..", "frontend")
app = Flask(__name__, static_folder=None)

_catalog_cache = {"items": [], "catalog_url": None, "loaded_at": None, "error": None, "loading": True}
_catalog_lock = threading.Lock()


def _load_catalog():
    with _catalog_lock:
        _catalog_cache["loading"] = True
    try:
        items, url = catalog.list_macos_installers()
        with _catalog_lock:
            _catalog_cache.update(
                items=items, catalog_url=url, loaded_at=time.time(), error=None, loading=False
            )
        log.info("Katalog geladen: %d macOS-Installer gefunden (%s)", len(items), url)
    except Exception as exc:  # noqa: BLE001
        log.exception("Katalog-Laden fehlgeschlagen")
        with _catalog_lock:
            _catalog_cache.update(error=str(exc), loading=False)


# Beim Container-Start einmal im Hintergrund laden, damit die GUI sofort
# ansprechbar ist und nicht auf den ersten Request wartet.
threading.Thread(target=_load_catalog, daemon=True).start()


@app.get("/")
def index():
    return send_from_directory(FRONTEND_DIR, "index.html")


@app.get("/<path:filename>")
def static_files(filename):
    return send_from_directory(FRONTEND_DIR, filename)


@app.get("/api/versions")
def api_versions():
    with _catalog_lock:
        return jsonify(dict(_catalog_cache))


@app.post("/api/versions/refresh")
def api_versions_refresh():
    threading.Thread(target=_load_catalog, daemon=True).start()
    return jsonify({"ok": True})


@app.get("/api/devices")
def api_devices():
    try:
        return jsonify({"items": list_usb_sticks()})
    except Exception as exc:  # noqa: BLE001
        log.exception("device listing failed")
        return jsonify({"items": [], "error": str(exc)}), 500


@app.get("/api/library")
def api_library():
    """Was liegt schon dauerhaft lokal (Offline-Bibliothek), fuer 'nur
    herunterladen' Badges + die Groessen-Anzeige in der GUI."""
    try:
        return jsonify(installer.list_library())
    except Exception as exc:  # noqa: BLE001
        log.exception("library listing failed")
        return jsonify({"items": [], "total_bytes": 0, "error": str(exc)}), 500


@app.post("/api/jobs")
def api_create_job():
    body = request.get_json(force=True)
    job_type = body.get("job_type", "stick")
    device_path = body.get("device_path")
    confirm_path = body.get("confirm_device_path")

    if job_type not in ("stick", "download", "multiboot"):
        return jsonify({"error": "Unbekannter job_type."}), 400

    with _catalog_lock:
        catalog_items = list(_catalog_cache["items"])

    if job_type == "multiboot":
        product_ids = body.get("product_ids") or []
        if not isinstance(product_ids, list) or len(product_ids) < 2:
            return jsonify({"error": "Multiboot braucht eine Liste mit mindestens 2 macOS-Versionen (product_ids)."}), 400
        products = []
        for pid in product_ids:
            match = next((p for p in catalog_items if p["id"] == pid), None)
            if not match:
                return jsonify({"error": f"Unbekannte macOS-Version '{pid}' -- Katalog neu laden."}), 400
            products.append(match)
        product = products
    else:
        product_id = body.get("product_id")
        product = next((p for p in catalog_items if p["id"] == product_id), None)
        if not product:
            return jsonify({"error": "Unbekannte macOS-Version -- Katalog neu laden."}), 400

    try:
        job = installer.create_job(product, device_path, confirm_path, job_type=job_type)
    except ValueError as exc:
        return jsonify({"error": str(exc)}), 400

    return jsonify({"job_id": job.id})


@app.get("/api/jobs/<job_id>")
def api_job_status(job_id):
    job = installer.JOBS.get(job_id)
    if not job:
        return jsonify({"error": "unbekannter Job"}), 404
    return jsonify(job.snapshot())


@app.get("/api/jobs/<job_id>/stream")
def api_job_stream(job_id):
    job = installer.JOBS.get(job_id)
    if not job:
        return jsonify({"error": "unbekannter Job"}), 404

    def gen():
        for event in job.stream():
            yield f"data: {json.dumps(event)}\n\n"

    return Response(gen(), mimetype="text/event-stream")


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, threaded=True)