left4me/l4d2web/routes/blueprint_routes.py

import json

from flask import Blueprint, Response, jsonify, redirect, request
from sqlalchemy import delete, func, select

from l4d2web.auth import current_user, require_login
from l4d2web.db import session_scope
from l4d2web.models import Blueprint as BlueprintModel
from l4d2web.models import BlueprintOverlay, Overlay, Server


bp = Blueprint("blueprint", __name__)


def split_textarea_lines(raw: str) -> list[str]:
    return [line.strip() for line in raw.splitlines() if line.strip()]


def ordered_overlay_ids_from_form() -> list[int]:
    ordered = []
    for fallback_position, value in enumerate(request.form.getlist("overlay_ids")):
        if not value:
            continue
        overlay_id = int(value)
        raw_position = request.form.get(f"overlay_position_{overlay_id}", "").strip()
        try:
            position = int(raw_position)
        except ValueError:
            position = fallback_position + 1
        ordered.append((position, fallback_position, overlay_id))
    return [overlay_id for _, _, overlay_id in sorted(ordered)]


def replace_blueprint_overlays(
    db,
    blueprint_id: int,
    overlay_ids: list[int],
    expose_ids: set[int] | None = None,
) -> None:
    expose_ids = expose_ids or set()
    db.execute(delete(BlueprintOverlay).where(BlueprintOverlay.blueprint_id == blueprint_id))
    for position, overlay_id in enumerate(overlay_ids):
        db.add(
            BlueprintOverlay(
                blueprint_id=blueprint_id,
                overlay_id=overlay_id,
                position=position,
                expose_server_cfg=overlay_id in expose_ids,
            )
        )


def expose_overlay_ids_from_form() -> set[int]:
    out: set[int] = set()
    for value in request.form.getlist("expose_server_cfg_ids"):
        if not value:
            continue
        try:
            out.add(int(value))
        except ValueError:
            continue
    return out


def overlay_ids_authorized(db, overlay_ids: list[int], user_id: int) -> bool:
    unique_ids = set(overlay_ids)
    if not unique_ids:
        return True
    allowed_count = db.scalar(
        select(func.count(Overlay.id)).where(
            Overlay.id.in_(unique_ids),
            Overlay.user_id.is_(None) | (Overlay.user_id == user_id),
        )
    )
    return allowed_count == len(unique_ids)


@bp.post("/blueprints")
@require_login
def create_blueprint() -> Response:
    user = current_user()
    assert user is not None

    if request.is_json:
        payload = request.get_json(silent=True) or {}
        name = str(payload.get("name", "")).strip()
        arguments = [str(item) for item in payload.get("arguments", [])]
        config = [str(item) for item in payload.get("config", [])]
        overlay_ids = [int(item) for item in payload.get("overlay_ids", [])]
        expose_ids = {int(item) for item in payload.get("expose_server_cfg_ids", [])}
        json_response = True
    else:
        name = request.form.get("name", "").strip()
        arguments = split_textarea_lines(request.form.get("arguments", ""))
        config = split_textarea_lines(request.form.get("config", ""))
        overlay_ids = ordered_overlay_ids_from_form()
        expose_ids = expose_overlay_ids_from_form()
        json_response = False

    if not name:
        return Response("name is required", status=400)

    with session_scope() as db:
        if not overlay_ids_authorized(db, overlay_ids, user.id):
            return Response("overlay not authorized", status=403)
        blueprint = BlueprintModel(user_id=user.id, name=name, arguments=json.dumps(arguments), config=json.dumps(config))
        db.add(blueprint)
        db.flush()
        replace_blueprint_overlays(db, blueprint.id, overlay_ids, expose_ids)
        blueprint_id = blueprint.id

    if json_response:
        return jsonify({"id": blueprint_id}), 201
    return redirect(f"/blueprints/{blueprint_id}")


@bp.post("/blueprints/<int:blueprint_id>")
@require_login
def update_blueprint_form(blueprint_id: int) -> Response:
    user = current_user()
    assert user is not None
    name = request.form.get("name", "").strip()
    if not name:
        return Response("name is required", status=400)

    with session_scope() as db:
        blueprint = db.scalar(
            select(BlueprintModel).where(BlueprintModel.id == blueprint_id, BlueprintModel.user_id == user.id)
        )
        if blueprint is None:
            return Response(status=404)
        overlay_ids = ordered_overlay_ids_from_form()
        expose_ids = expose_overlay_ids_from_form()
        if not overlay_ids_authorized(db, overlay_ids, user.id):
            return Response("overlay not authorized", status=403)

        blueprint.name = name
        blueprint.arguments = json.dumps(split_textarea_lines(request.form.get("arguments", "")))
        blueprint.config = json.dumps(split_textarea_lines(request.form.get("config", "")))
        replace_blueprint_overlays(db, blueprint.id, overlay_ids, expose_ids)

    return redirect(f"/blueprints/{blueprint_id}")


def _delete_blueprint(db, user_id: int, blueprint_id: int) -> Response | None:
    blueprint = db.scalar(
        select(BlueprintModel).where(
            BlueprintModel.id == blueprint_id,
            BlueprintModel.user_id == user_id,
        )
    )
    if blueprint is None:
        return Response(status=404)

    linked_count = db.scalar(
        select(func.count(Server.id)).where(Server.blueprint_id == blueprint.id)
    ) or 0
    if linked_count > 0:
        return Response("blueprint is in use", status=409)

    db.execute(delete(BlueprintOverlay).where(BlueprintOverlay.blueprint_id == blueprint.id))
    db.delete(blueprint)
    return None


@bp.delete("/blueprints/<int:blueprint_id>")
@require_login
def delete_blueprint(blueprint_id: int) -> Response:
    user = current_user()
    assert user is not None

    with session_scope() as db:
        error = _delete_blueprint(db, user.id, blueprint_id)
        if error is not None:
            return error

    return Response(status=204)


@bp.post("/blueprints/<int:blueprint_id>/delete")
@require_login
def delete_blueprint_form(blueprint_id: int) -> Response:
    user = current_user()
    assert user is not None

    with session_scope() as db:
        error = _delete_blueprint(db, user.id, blueprint_id)
        if error is not None:
            return error

    return redirect("/blueprints")