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, 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]) -> None: 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)) @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", [])] 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() json_response = False if not name: return Response("name is required", status=400) with session_scope() as db: 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) blueprint_id = blueprint.id if json_response: return jsonify({"id": blueprint_id}), 201 return redirect(f"/blueprints/{blueprint_id}") @bp.post("/blueprints/") @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) 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, ordered_overlay_ids_from_form()) return redirect(f"/blueprints/{blueprint_id}") @bp.delete("/blueprints/") @require_login def delete_blueprint(blueprint_id: int) -> Response: user = current_user() assert user is not None 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) 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 Response(status=204)