FabledSteward/steward/ansible/routes.py

# steward/ansible/routes.py
from __future__ import annotations
from datetime import datetime, timedelta, timezone

from quart import (
    Blueprint, Response, current_app, redirect, render_template,
    request, session, url_for,
)
from sqlalchemy import select

from steward.ansible import executor, guards, sources as src_module
from steward.ansible.inventory_gen import run_title, scope_labels
from steward.auth.middleware import require_role
from steward.core.audit import audit_request as _audit
from steward.core.paging import keyset, next_page
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
from steward.models.ansible_schedule import AnsibleSchedule
from steward.models.users import User, UserRole

INTERVAL_PRESETS = [
    (300, "Every 5 minutes"),
    (3600, "Hourly"),
    (21600, "Every 6 hours"),
    (43200, "Every 12 hours"),
    (86400, "Daily"),
    (604800, "Weekly"),
]

ansible_bp = Blueprint("ansible", __name__, url_prefix="/ansible")


def _get_sources() -> list[dict]:
    ansible_cfg = current_app.config.get("ANSIBLE", {})
    return src_module.get_sources(ansible_cfg)


@ansible_bp.get("/")
@require_role(UserRole.viewer)
async def index():
    """Every run, newest first, a page at a time as the reader scrolls;
    ?before=<cursor> is the next page's rows alone (core.paging)."""
    before = request.args.get("before")
    async with current_app.db_sessionmaker() as db:
        stmt = keyset(select(AnsibleRun), AnsibleRun.started_at, AnsibleRun.id, before)
        runs, after = next_page((await db.execute(stmt)).scalars().all(),
                                lambda r: (r.started_at, r.id))
        labels = await scope_labels(db, [r.inventory_scope for r in runs])
    now = datetime.now(timezone.utc)
    ctx = {"runs": [run_view(r, labels, now) for r in runs], "after": after}
    if before:
        return await render_template("ansible/_runs_page.html", **ctx)
    return await render_template("ansible/index.html", **ctx)


def run_duration(run: AnsibleRun, now: datetime) -> float | None:
    """Seconds a run took, or has taken so far while it is still going."""
    if run.started_at is None:
        return None
    return ((run.finished_at or now) - run.started_at).total_seconds()


def run_view(run: AnsibleRun, labels: dict[str, str], now: datetime) -> dict:
    """A run as the pages name it: its title, scope label and duration."""
    label = labels.get(run.inventory_scope, run.inventory_scope)
    return {"run": run, "title": run_title(run.playbook_path, label), "scope_label": label,
            "duration_s": run_duration(run, now),
            "check": bool((run.params or {}).get("check"))}


@ansible_bp.get("/browse")
@require_role(UserRole.viewer)
async def browse():
    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup

    all_sources = _get_sources()
    source_data = []
    for source in all_sources:
        playbooks = []
        for path in src_module.discover_playbooks(source["path"]):
            contents = src_module.read_playbook(source["path"], path)
            meta = (src_module.discover_playbook_meta(contents) if contents is not None
                    else {"description": "", "category": ""})
            playbooks.append({
                "path": path,
                "category": meta["category"],
                "description": meta["description"],
            })
        inventories = src_module.discover_inventories(source["path"])
        source_data.append({
            "source": source,
            "playbooks": playbooks,
            "inventories": inventories,
            "editable": src_module.is_editable_source(source),
        })

    async with current_app.db_sessionmaker() as db:
        targets = (await db.execute(
            select(AnsibleTarget).order_by(AnsibleTarget.name)
        )).scalars().all()
        groups = (await db.execute(
            select(AnsibleGroup).order_by(AnsibleGroup.name)
        )).scalars().all()

    return await render_template(
        "ansible/browse.html",
        source_data=source_data,
        targets=targets,
        groups=groups,
    )


@ansible_bp.get("/browse/<source_name>/<path:playbook_path>")
@require_role(UserRole.viewer)
async def view_playbook(source_name: str, playbook_path: str):
    all_sources = _get_sources()
    source = next((s for s in all_sources if s["name"] == source_name), None)
    if source is None:
        return "Source not found", 404
    contents = src_module.read_playbook(source["path"], playbook_path)
    if contents is None:
        return "Playbook not found", 404
    return await render_template(
        "ansible/browse.html",
        source_data=[],
        view_source=source_name,
        view_path=playbook_path,
        view_contents=contents,
        view_editable=src_module.is_editable_source(source),
    )


def _caller_role() -> UserRole:
    return UserRole(session.get("user_role", "viewer"))


def _run_vars(params: dict | None, secret_vars: dict | None = None) -> dict:
    return {**((params or {}).get("extra_vars_map") or {}), **(secret_vars or {})}


def check_run_form(form, source: dict | None, playbook_path: str,
                   params: dict | None, secret_vars: dict | None) -> str | None:
    """The server-side gates a web run must pass (the form's own checks are
    only a convenience): who may set these variables, and the playbook's
    steward:confirm flag, which a real (non-check) run must have ticked."""
    err = guards.check_vars_for_role(_run_vars(params, secret_vars), _caller_role())
    if err:
        return err
    content = src_module.read_playbook(source["path"], playbook_path) if source else None
    meta = src_module.discover_playbook_meta(content or "")
    if meta["confirm"] and not (params or {}).get("check") and "confirmed" not in form:
        return (f"{playbook_path} is flagged destructive by its author (steward:confirm). "
                "Tick the confirmation to run it, or run it in check mode.")
    return None


def _parse_run_params(form) -> tuple[dict | None, dict | None, str | None]:
    """Build executor params + the unpersisted secret-vars dict from form fields.

    Returns (params, secret_vars, error). Discovered variable fields are named
    ``var__<name>``; a sibling hidden ``secret__<name>`` marks the sensitive
    ones, which go into secret_vars (never persisted) instead of params. A
    free-form ``extra_vars`` textarea (key=value per line) is also folded in.
    All run-time values flow through a JSON extra-vars file downstream, so
    values may contain spaces/quotes safely.
    """
    extra_vars_map: dict[str, str] = {}
    secret_vars: dict[str, str] = {}

    for key in set(form.keys()):
        if not key.startswith("var__"):
            continue
        name = key[len("var__"):]
        val = (form.get(key) or "").strip()
        if not val:
            continue  # untouched → fall through to inventory/play default
        if f"secret__{name}" in form:
            secret_vars[name] = val
        else:
            extra_vars_map[name] = val

    for line in (form.get("extra_vars", "") or "").splitlines():
        line = line.strip()
        if not line:
            continue
        if "=" not in line:
            return None, None, f"Invalid extra var (expected key=value): {line!r}"
        k, v = line.split("=", 1)
        extra_vars_map[k.strip()] = v

    params: dict = {}
    if extra_vars_map:
        params["extra_vars_map"] = extra_vars_map
    limit = (form.get("limit", "") or "").strip()
    if limit:
        params["limit"] = limit
    tags = (form.get("tags", "") or "").strip()
    if tags:
        params["tags"] = tags
    if "check" in form:
        params["check"] = True
    return (params or None), (secret_vars or None), None


@ansible_bp.get("/playbook-options")
@require_role(UserRole.operator)
async def playbook_options():
    """HTMX fragment: <option>s of playbooks in a source, to populate a playbook
    dropdown when a source is chosen. `selected` pre-selects one (edit forms)."""
    source_name = (request.args.get("source_name", "") or "").strip()
    selected = (request.args.get("selected", "") or "").strip()
    source = next((s for s in _get_sources() if s["name"] == source_name), None)
    playbooks = src_module.discover_playbooks(source["path"]) if source else []
    return await render_template(
        "ansible/_playbook_options.html", playbooks=playbooks, selected=selected)


@ansible_bp.get("/playbook-vars")
@require_role(UserRole.operator)
async def playbook_vars():
    """HTMX fragment: fill-in fields for a playbook's declared variables
    (vars/vars_prompt), loaded when a playbook is chosen from a dropdown."""
    source_name = (request.args.get("source_name", "") or "").strip()
    playbook_path = (request.args.get("playbook_path", "") or "").strip()
    # Schedule form: schedules can't prompt, so secret vars render disabled.
    schedule = (request.args.get("schedule", "") or "").strip() == "1"
    source = next((s for s in _get_sources() if s["name"] == source_name), None)
    variables: list = []
    meta: dict = {"description": "", "category": "", "confirm": False}
    if source and playbook_path:
        contents = src_module.read_playbook(source["path"], playbook_path)
        if contents is not None:
            variables = src_module.discover_playbook_variables(contents)
            meta = src_module.discover_playbook_meta(contents)
    return await render_template(
        "ansible/_playbook_vars.html", variables=variables,
        description=meta["description"], category=meta["category"],
        confirm=meta["confirm"], schedule=schedule)


@ansible_bp.get("/run-form/<source_name>/<path:playbook_path>")
@require_role(UserRole.operator)
async def run_form(source_name: str, playbook_path: str):
    """HTMX fragment: the run form for one playbook, with a field per declared
    variable (vars/vars_prompt) discovered from the playbook itself."""
    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup

    source = next((s for s in _get_sources() if s["name"] == source_name), None)
    if source is None:
        return "Source not found", 404
    contents = src_module.read_playbook(source["path"], playbook_path)
    if contents is None:
        return "Playbook not found", 404
    variables = src_module.discover_playbook_variables(contents)
    meta = src_module.discover_playbook_meta(contents)
    inventories = src_module.discover_inventories(source["path"])

    async with current_app.db_sessionmaker() as db:
        targets = (await db.execute(
            select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
        groups = (await db.execute(
            select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()

    return await render_template(
        "ansible/_run_form.html",
        source_name=source_name, playbook_path=playbook_path,
        variables=variables, description=meta["description"], category=meta["category"],
        confirm=meta["confirm"], targets=targets, groups=groups,
        inventories=inventories,
    )


@ansible_bp.post("/runs")
@require_role(UserRole.operator)
async def create_run():
    from steward.ansible.runner import trigger_run

    form = await request.form
    playbook_path = form.get("playbook_path", "").strip()
    source_name = form.get("source_name", "").strip()
    inventory_scope = form.get("inventory_scope", "steward:all").strip()

    if not playbook_path:
        return "playbook_path is required", 400

    params_or_none, secret_vars, err = _parse_run_params(form)
    if err:
        return err, 400
    source = next((s for s in _get_sources() if s["name"] == source_name), None)
    err = check_run_form(form, source, playbook_path, params_or_none, secret_vars)
    if err:
        return err, 400

    run, source, err = await trigger_run(
        current_app._get_current_object(),  # type: ignore[attr-defined]
        source_name=source_name,
        playbook_path=playbook_path,
        inventory_scope=inventory_scope,
        params=params_or_none,
        secret_vars=secret_vars,
        triggered_by=session["user_id"],
    )
    if err == "Source not found":
        return err, 404
    if err:
        return err, 400
    await _audit("ansible.run.started", "ansible_run", run.id, {
        "source": source_name, "playbook": playbook_path, "scope": inventory_scope,
        "check": bool((params_or_none or {}).get("check")),
        "vars": sorted(_run_vars(params_or_none, secret_vars))})

    async with current_app.db_sessionmaker() as db:
        labels = await scope_labels(db, [run.inventory_scope])
    return await render_template(
        "ansible/run_started.html",
        run=run,
        title=run_title(run.playbook_path, labels.get(run.inventory_scope, run.inventory_scope)),
    )


@ansible_bp.get("/runs/<run_id>/stream")
@require_role(UserRole.viewer)
async def stream_run(run_id: str):
    async with current_app.db_sessionmaker() as db:
        result = await db.execute(select(AnsibleRun).where(AnsibleRun.id == run_id))
        run = result.scalar_one_or_none()
    if run is None:
        return "Not found", 404

    async def generate():
        if run.status not in (AnsibleRunStatus.running, AnsibleRunStatus.queued):
            yield f"event: done\ndata: {run.status.value}\n\n"
            return

        q = executor.register_listener(run_id)
        try:
            while True:
                msg = await q.get()
                if isinstance(msg, executor._Done):
                    yield f"event: done\ndata: {msg.status}\n\n"
                    break
                # Escape newlines in SSE data field
                safe = msg.replace("\n", " ")
                yield f"event: output\ndata: {safe}\n\n"
        finally:
            executor.deregister_listener(run_id, q)

    return Response(
        generate(),
        content_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@ansible_bp.get("/runs/<run_id>")
@require_role(UserRole.viewer)
async def run_detail(run_id: str):
    async with current_app.db_sessionmaker() as db:
        result = await db.execute(select(AnsibleRun).where(AnsibleRun.id == run_id))
        run = result.scalar_one_or_none()
        if run is None:
            return "Not found", 404
        # NULL triggered_by = an automated (system) run; otherwise resolve the user.
        triggered_label = "system"
        if run.triggered_by:
            res = await db.execute(
                select(User.username).where(User.id == run.triggered_by))
            triggered_label = res.scalar_one_or_none() or "(deleted user)"
        labels = await scope_labels(db, [run.inventory_scope])
    return await render_template(
        "ansible/run_detail.html", run=run, triggered_label=triggered_label,
        row=run_view(run, labels, datetime.now(timezone.utc)))


@ansible_bp.post("/runs/<run_id>/cancel")
@require_role(UserRole.operator)
async def cancel_run(run_id: str):
    async with current_app.db_sessionmaker() as db:
        run = (await db.execute(
            select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one_or_none()
    if run is None:
        return "Not found", 404
    if run.status in (AnsibleRunStatus.running, AnsibleRunStatus.queued):
        executor.cancel_run(run_id)
        await _audit("ansible.run.cancelled", "ansible_run", run_id,
                     {"playbook": run.playbook_path})
    return redirect(url_for("ansible.run_detail", run_id=run_id))


@ansible_bp.get("/runs/<run_id>/log")
@require_role(UserRole.viewer)
async def run_log(run_id: str):
    """Download the full run log (persistent artifact; falls back to DB output)."""
    import os
    from steward.ansible.executor import artifact_path

    async with current_app.db_sessionmaker() as db:
        run = (await db.execute(
            select(AnsibleRun).where(AnsibleRun.id == run_id))).scalar_one_or_none()
    if run is None:
        return "Not found", 404

    path = artifact_path(run_id)
    if os.path.exists(path):
        with open(path, encoding="utf-8", errors="replace") as f:
            body = f.read()
    else:
        body = run.output or ""
    return Response(
        body, mimetype="text/plain",
        headers={"Content-Disposition": f'attachment; filename="ansible-run-{run_id[:8]}.log"'},
    )


# ── Scheduled recurring runs ──────────────────────────────────────────────────

@ansible_bp.get("/schedules")
@require_role(UserRole.viewer)
async def schedules():
    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup

    editing_id = request.args.get("edit")
    async with current_app.db_sessionmaker() as db:
        scheds = (await db.execute(
            select(AnsibleSchedule).order_by(AnsibleSchedule.name))).scalars().all()
        groups = (await db.execute(
            select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()
        targets = (await db.execute(
            select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
        run_ids = [s.last_run_id for s in scheds if s.last_run_id]
        runs = {
            r.id: r for r in (await db.execute(
                select(AnsibleRun).where(AnsibleRun.id.in_(run_ids)))).scalars().all()
        } if run_ids else {}
        labels = await scope_labels(db, [s.inventory_scope for s in scheds])

    source_data = [
        {"name": s["name"], "playbooks": src_module.discover_playbooks(s["path"])}
        for s in _get_sources()
    ]

    rows = []
    editing = None
    for s in scheds:
        rows.append({
            "s": s,
            "last_run": runs.get(s.last_run_id) if s.last_run_id else None,
            "scope_label": labels.get(s.inventory_scope, s.inventory_scope),
            "next_run": (s.last_run_at + timedelta(seconds=s.interval_seconds)) if s.last_run_at else None,
        })
        if editing_id and s.id == editing_id:
            editing = s
    rows.sort(key=_schedule_order)

    # Default the source dropdown to the first source (or the edited schedule's
    # source), and pre-render that source's playbooks so the playbook dropdown
    # starts on a real item rather than a placeholder.
    sel_source = editing.source_name if editing else (source_data[0]["name"] if source_data else "")
    sel_playbooks = next((sd["playbooks"] for sd in source_data if sd["name"] == sel_source), [])

    # Edit mode: pre-render the selected playbook's declared variables so their
    # saved values show (fresh loads fetch these via HTMX from /playbook-vars).
    edit_variables: list = []
    edit_meta: dict = {"description": "", "category": "", "confirm": False}
    if editing:
        src = next((s for s in _get_sources() if s["name"] == editing.source_name), None)
        if src:
            contents = src_module.read_playbook(src["path"], editing.playbook_path)
            if contents is not None:
                edit_variables = src_module.discover_playbook_variables(contents)
                edit_meta = src_module.discover_playbook_meta(contents)

    return await render_template(
        "ansible/schedules.html",
        rows=rows, groups=groups, targets=targets,
        source_data=source_data, sel_source=sel_source, sel_playbooks=sel_playbooks,
        editing=editing, interval_presets=INTERVAL_PRESETS,
        edit_variables=edit_variables, edit_description=edit_meta["description"],
        edit_category=edit_meta["category"], edit_confirm=edit_meta["confirm"],
    )


def _schedule_order(row: dict) -> tuple:
    """Schedules list order: failing first (its last run failed or was cut
    short, or its last firing could not start one), then running, then the
    rest, paused last; by name within each."""
    s, run = row["s"], row["last_run"]
    status = run.status.value if run else None
    if not s.enabled:
        rank = 3
    elif status in ("failed", "interrupted") or (s.last_error and run is None):
        rank = 0
    elif status in ("queued", "running"):
        rank = 1
    else:
        rank = 2
    return rank, s.name.lower()


def _schedule_form_fields(form) -> tuple[dict | None, str | None]:
    """Validate + extract schedule fields from a form. Returns (fields, error)."""
    name = (form.get("name", "") or "").strip()
    source_name = (form.get("source_name", "") or "").strip()
    playbook_path = (form.get("playbook_path", "") or "").strip()
    inventory_scope = (form.get("inventory_scope", "steward:all") or "steward:all").strip()
    try:
        interval = int(form.get("interval_seconds", "0"))
    except (TypeError, ValueError):
        interval = 0
    if not name or not source_name or not playbook_path or interval <= 0:
        return None, "name, source, playbook, and a positive interval are required"
    # Scheduled runs can't prompt, so secret vars are intentionally dropped —
    # automation should rely on global creds / inventory vars, not run-time secrets.
    params, _secret_vars, err = _parse_run_params(form)
    if err:
        return None, err
    return {
        "name": name, "source_name": source_name, "playbook_path": playbook_path,
        "inventory_scope": inventory_scope, "interval_seconds": interval, "params": params,
    }, None


def _schedule_detail(fields: dict) -> dict:
    """What a schedule audit row records: what runs, where, how often — and
    which variables, by name only."""
    return {"name": fields["name"], "source": fields["source_name"],
            "playbook": fields["playbook_path"], "scope": fields["inventory_scope"],
            "interval_seconds": fields["interval_seconds"],
            "vars": sorted(_run_vars(fields["params"]))}


@ansible_bp.post("/schedules")
@require_role(UserRole.operator)
async def create_schedule():
    form = await request.form
    fields, err = _schedule_form_fields(form)
    if err:
        return err, 400
    err = guards.check_vars_for_role(_run_vars(fields["params"]), _caller_role())
    if err:
        return err, 400
    sched = AnsibleSchedule(enabled=True, **fields)
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            db.add(sched)
    await _audit("ansible.schedule.created", "ansible_schedule", sched.id, _schedule_detail(fields))
    return redirect(url_for("ansible.schedules"))


@ansible_bp.post("/schedules/<schedule_id>")
@require_role(UserRole.operator)
async def update_schedule(schedule_id: str):
    form = await request.form
    fields, err = _schedule_form_fields(form)
    if err:
        return err, 400
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            sched = await db.get(AnsibleSchedule, schedule_id)
            if sched is None:
                return "Not found", 404
            err = guards.check_vars_for_role(
                guards.changed_vars(_run_vars(fields["params"]), _run_vars(sched.params)),
                _caller_role())
            if err:
                return err, 400
            for k, v in fields.items():
                setattr(sched, k, v)
    await _audit("ansible.schedule.updated", "ansible_schedule", schedule_id, _schedule_detail(fields))
    return redirect(url_for("ansible.schedules"))


@ansible_bp.post("/schedules/<schedule_id>/toggle")
@require_role(UserRole.operator)
async def toggle_schedule(schedule_id: str):
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            sched = await db.get(AnsibleSchedule, schedule_id)
            if sched is None:
                return "Not found", 404
            sched.enabled = not sched.enabled
            enabled, name = sched.enabled, sched.name
    await _audit("ansible.schedule.enabled" if enabled else "ansible.schedule.disabled",
                 "ansible_schedule", schedule_id, {"name": name})
    return redirect(url_for("ansible.schedules"))


@ansible_bp.post("/schedules/<schedule_id>/delete")
@require_role(UserRole.operator)
async def delete_schedule(schedule_id: str):
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            sched = await db.get(AnsibleSchedule, schedule_id)
            if sched is None:
                return redirect(url_for("ansible.schedules"))
            name = sched.name
            await db.delete(sched)
    await _audit("ansible.schedule.deleted", "ansible_schedule", schedule_id, {"name": name})
    return redirect(url_for("ansible.schedules"))


@ansible_bp.post("/schedules/<schedule_id>/run-now")
@require_role(UserRole.operator)
async def run_schedule_now(schedule_id: str):
    from steward.ansible.runner import trigger_run

    async with current_app.db_sessionmaker() as db:
        sched = await db.get(AnsibleSchedule, schedule_id)
        if sched is None:
            return "Not found", 404
        source_name, playbook_path = sched.source_name, sched.playbook_path
        inventory_scope, params = sched.inventory_scope, sched.params

    run, _source, err = await trigger_run(
        current_app._get_current_object(),  # type: ignore[attr-defined]
        source_name=source_name, playbook_path=playbook_path,
        inventory_scope=inventory_scope, params=params,
        triggered_by=session["user_id"],
    )
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            fresh = await db.get(AnsibleSchedule, schedule_id)
            if fresh:
                fresh.last_run_at = datetime.now(timezone.utc)
                fresh.last_run_id = run.id if run else None
                fresh.last_error = err
    await _audit("ansible.schedule.run_now", "ansible_schedule", schedule_id,
                 {"playbook": playbook_path, "run_id": run.id if run else None, "error": err})
    if run:
        return redirect(url_for("ansible.run_detail", run_id=run.id))
    return redirect(url_for("ansible.schedules"))


# ── In-app playbook authoring/editor (admin only) ─────────────────────────────

def _editable_sources() -> list[dict]:
    return [s for s in _get_sources() if src_module.is_editable_source(s)]


def _editable_source(name: str) -> dict | None:
    return next((s for s in _editable_sources() if s["name"] == name), None)


@ansible_bp.get("/playbooks/new")
@require_role(UserRole.admin)
async def new_playbook():
    sources = _editable_sources()
    return await render_template(
        "ansible/playbook_editor.html",
        editable_sources=sources, editing=False,
        source_name=(sources[0]["name"] if sources else ""),
        playbook_path="", content=_NEW_PLAYBOOK_TEMPLATE, error=None,
    )


@ansible_bp.get("/playbooks/edit/<source_name>/<path:playbook_path>")
@require_role(UserRole.admin)
async def edit_playbook(source_name: str, playbook_path: str):
    source = _editable_source(source_name)
    if source is None:
        return "Source is not editable", 404
    content = src_module.read_playbook(source["path"], playbook_path)
    if content is None:
        return "Playbook not found", 404
    return await render_template(
        "ansible/playbook_editor.html",
        editable_sources=_editable_sources(), editing=True,
        source_name=source_name, playbook_path=playbook_path,
        content=content, error=None,
    )


@ansible_bp.post("/playbooks/save")
@require_role(UserRole.admin)
async def save_playbook():
    form = await request.form
    source_name = (form.get("source_name", "") or "").strip()
    playbook_path = (form.get("playbook_path", "") or "").strip()
    content = form.get("content", "") or ""
    editing = (form.get("editing", "") == "1")

    source = _editable_source(source_name)
    err = None
    if source is None:
        err = "Source is not editable"
    elif not playbook_path:
        err = "Filename is required"
    else:
        ok, verr = src_module.validate_playbook_yaml(content)
        if not ok:
            err = verr
    if err is None:
        ok, werr = src_module.write_playbook(source["path"], playbook_path, content)
        if not ok:
            err = werr

    if err is not None:
        return await render_template(
            "ansible/playbook_editor.html",
            editable_sources=_editable_sources(), editing=editing,
            source_name=source_name, playbook_path=playbook_path,
            content=content, error=err,
        ), 400

    await _audit("ansible.playbook.updated" if editing else "ansible.playbook.created",
                 "ansible_playbook", f"{source_name}:{playbook_path}",
                 {"source": source_name, "playbook": playbook_path, "bytes": len(content)})
    return redirect(url_for(
        "ansible.view_playbook", source_name=source_name, playbook_path=playbook_path))


@ansible_bp.post("/playbooks/delete")
@require_role(UserRole.admin)
async def delete_playbook():
    form = await request.form
    source = _editable_source((form.get("source_name", "") or "").strip())
    if source is None:
        return "Source is not editable", 400
    playbook_path = (form.get("playbook_path", "") or "").strip()
    ok, err = src_module.delete_playbook(source["path"], playbook_path)
    if not ok:
        return err, 400
    await _audit("ansible.playbook.deleted", "ansible_playbook", f"{source['name']}:{playbook_path}",
                 {"source": source["name"], "playbook": playbook_path})
    return redirect(url_for("ansible.browse"))


_NEW_PLAYBOOK_TEMPLATE = """\
---
- name: My playbook
  hosts: all
  gather_facts: false
  tasks:
    - name: Ping
      ansible.builtin.ping:
"""