sfera/services/api-server/src/api_server/html5_setup_controller.py

from __future__ import annotations

import asyncio
from collections.abc import AsyncIterator, Callable, Iterable
from typing import Any

from fastapi import HTTPException

from api_server.html5_forms import form_value
from api_server.html5_setup import (
    render_html5_import_check,
    render_html5_import_job,
    render_html5_project_setup,
    render_html5_settings_panel,
    render_html5_setup_summary,
)
from api_server.html5_sse import html5_sse_comment, html5_sse_if_changed


def html5_setup_page(
    *,
    project_id: str,
    projects: Iterable[object],
    setup: object,
) -> str:
    return render_html5_project_setup(project_id=project_id, projects=projects, setup=setup)


def html5_setup_summary(*, project_id: str, setup: object) -> str:
    return render_html5_setup_summary(project_id, setup)


async def html5_setup_event_stream(
    *,
    project_id: str,
    setup_response: Callable[[str], object],
    latest_import_job: Callable[[str], object | None],
    once: bool = False,
) -> AsyncIterator[str]:
    last_fragments: dict[str, str] = {}
    while True:
        yield html5_sse_comment(f"setup {project_id} heartbeat")
        try:
            setup = setup_response(project_id)
        except HTTPException as error:
            setup_error = f'<div class="setup-summary" data-html5-setup-summary><p class="muted padded">{error.detail}</p></div>'
            for event_text in html5_sse_if_changed(last_fragments, "setup-summary", setup_error):
                yield event_text
            if once:
                break
            await asyncio.sleep(2)
            continue
        for event_text in html5_sse_if_changed(last_fragments, "setup-summary", render_html5_setup_summary(project_id, setup)):
            yield event_text
        for event_text in html5_sse_if_changed(
            last_fragments,
            "setup-import-job",
            render_html5_import_job(project_id, latest_import_job(project_id)),
        ):
            yield event_text
        if once:
            break
        await asyncio.sleep(2)


async def html5_setup_source(
    *,
    project_id: str,
    form: dict[str, list[str]],
    import_source_kind: Callable[[str], Any],
    setup_response: Callable[[str], object],
    save_settings: Callable[[str, object], Any],
) -> str:
    source = import_source_kind(form_value(form, "source") or "XML_DUMP")
    current = setup_response(project_id)
    settings = current.settings.model_copy(update={"structure_source": source})
    setup = await save_settings(project_id, settings)
    return render_html5_setup_summary(project_id, setup)


async def html5_setup_settings(
    *,
    project_id: str,
    form: dict[str, list[str]],
    setup_response: Callable[[str], object],
    save_settings: Callable[[str, object], Any],
) -> str:
    current = setup_response(project_id)
    settings = current.settings.model_copy(
        update={
            "name": form_value(form, "name") or current.settings.name,
            "platform_version": form_value(form, "platform_version"),
            "compatibility_mode": form_value(form, "compatibility_mode"),
        }
    )
    setup = await save_settings(project_id, settings)
    return render_html5_settings_panel(project_id, setup, saved=True)


def html5_setup_check(
    *,
    project_id: str,
    form: dict[str, list[str]],
    import_source_kind: Callable[[str], Any],
    import_request: Callable[..., object],
    current_import_source: Callable[[str], object],
    import_check: Callable[[str, object, object], object],
) -> str:
    source = import_source_kind(form_value(form, "source") or current_import_source(project_id).value)
    check = import_check(project_id, source, import_request(source=source))
    return render_html5_import_check(project_id, check)


def html5_setup_import(
    *,
    project_id: str,
    form: dict[str, list[str]],
    import_source_kind: Callable[[str], Any],
    import_request: Callable[..., object],
    current_import_source: Callable[[str], object],
    execute_import: Callable[[str, object], object],
    setup_response: Callable[[str], object],
) -> str:
    source = import_source_kind(form_value(form, "source") or current_import_source(project_id).value)
    structure_only = form_value(form, "structure_only") in {"1", "true", "on", "yes"}
    execute_import(project_id, import_request(source=source, structure_only=structure_only))
    return render_html5_setup_summary(project_id, setup_response(project_id))


async def html5_setup_import_job(
    *,
    project_id: str,
    form: dict[str, list[str]],
    import_source_kind: Callable[[str], Any],
    import_request: Callable[..., object],
    current_import_source: Callable[[str], object],
    start_import_job: Callable[[str, object, object], Any],
) -> str:
    source = import_source_kind(form_value(form, "source") or current_import_source(project_id).value)
    job = await start_import_job(project_id, source, import_request(source=source))
    return render_html5_import_job(project_id, job)


def html5_setup_job(
    *,
    project_id: str,
    job_id: str,
    jobs_by_id: dict[str, object],
) -> str:
    job = jobs_by_id.get(job_id)
    if job is None or (getattr(job, "payload", {}) or {}).get("project_id") != project_id:
        raise HTTPException(status_code=404, detail=f"Unknown import job: {job_id}")
    return render_html5_import_job(project_id, job)


async def html5_setup_reindex(
    *,
    project_id: str,
    start_reindex_job: Callable[[str], Any],
) -> str:
    job = await start_reindex_job(project_id)
    return render_html5_import_job(project_id, job)