kirche-einfach/backend/components/kirche/form_inputs.py

"""Structured form-input component for kirche-einfach flows.

A flow that starts with :class:`CustomInputForm` declares the fields of its
web form. The API reads the exported node template
(``template`` minus ``code``) and derives ``{json_schema, ui_schema}`` from
the ``lfx.io`` input definitions below — one JSON-Schema property per field,
keyed by ``name``. Field metadata maps as follows:

* ``display_name`` → ``title``, ``info`` → ``description``,
  ``value`` → ``default``, ``options`` → ``enum``
* ``required`` → required array, ``advanced`` → ``ui:collapsed`` hint
* ``_input_type`` → JSON type + widget (see ``app/flows/input_schema.py``)

``collect()`` is fully dynamic: it emits every input except the plumbing
ones (``form_values``, ``code``), so fields can be renamed, added, or
removed in the Langflow code editor without touching this method.
``name`` is the stable JSON key (freely renamable, must stay unique and
non-empty); downstream components read the keys as-is.

``collect()`` returns ``Data`` (the LFX JSON type), not ``Message``: the
values travel as a real JSON object. A terminal ``ChatOutput`` renders it
as a fenced ```json block; components consuming it should read ``.data``
(or ``get_text()``), not ``.text``.

``FileInput`` is a stub for now: it carries a string reference only, real
upload comes later.
"""

from __future__ import annotations

import json
from typing import ClassVar

from lfx.custom import Component
from lfx.field_typing.range_spec import RangeSpec
from lfx.io import (
    BoolInput,
    DropdownInput,
    FileInput,
    FloatInput,
    IntInput,
    MessageTextInput,
    MultilineInput,
    MultiselectInput,
    Output,
    SecretStrInput,
    SliderInput,
    StrInput,
)
from lfx.schema.data import Data


def _text(value) -> str:
    if value is None:
        return ""
    if hasattr(value, "text"):
        return value.text or ""
    return str(value)


def _json_safe(value):
    """Plain JSON value for one field: unwrap LFX wrappers, keep scalars."""
    if value is None or isinstance(value, (bool, int, float, str)):
        return value
    if isinstance(value, list):
        return [_json_safe(item) for item in value]
    if isinstance(value, dict):
        return {str(key): _json_safe(item) for key, item in value.items()}
    text = getattr(value, "text", None)
    if isinstance(text, str):
        return text
    data = getattr(value, "data", None)
    if isinstance(data, dict):
        return _json_safe(data)
    return str(value)


#: Inputs that are plumbing, not form fields (never part of the output).
PLUMBING_INPUTS = frozenset({"form_values", "code"})


class CustomInputForm(Component):
    display_name = "CustomInput Form"
    description = "Collects structured form values into a single JSON object."
    icon = "ClipboardList"

    inputs: ClassVar[list] = [
        # Plumbing, not a form field: the API posts submitted web-form
        # values as the run envelope (same {"payload": {...}} shape the
        # other kirche components use); they arrive here through a ChatInput
        # entry node and override the defaults below. Hidden in the GUI and
        # skipped by the form parser (show=False).
        MessageTextInput(
            name="form_values",
            display_name="Form values (JSON envelope)",
            info="Submitted values as JSON; merged over the defaults",
            value="",
            show=False,
        ),
        StrInput(
            name="full_name",
            display_name="Full name",
            info="Your first and last name",
            value="",
            required=True,
        ),
        MultilineInput(
            name="notes",
            display_name="Notes",
            info="Anything else we should know",
            value="",
        ),
        SecretStrInput(
            name="password",
            display_name="Password",
            info="Optional access phrase",
            value="",
        ),
        IntInput(
            name="age",
            display_name="Age",
            info="Your age in years",
            value=18,
        ),
        FloatInput(
            name="amount",
            display_name="Amount",
            info="Amount in euros",
            value=0.0,
        ),
        BoolInput(
            name="newsletter",
            display_name="Newsletter",
            info="Send me the parish newsletter",
            value=False,
        ),
        DropdownInput(
            name="department",
            display_name="Department",
            info="Which team is this for",
            options=["Music", "Youth", "Seniors", "Admin"],
            value="Music",
        ),
        MultiselectInput(
            name="topics",
            display_name="Topics",
            info="What are you interested in",
            options=["Worship", "Outreach", "Courses"],
            value=[],
        ),
        SliderInput(
            name="satisfaction",
            display_name="Satisfaction",
            info="How satisfied are you (1-5)",
            value=3,
            range_spec=RangeSpec(min=1, max=5, step=1),
        ),
        FileInput(
            name="attachment",
            display_name="Attachment",
            info="String reference only for now (no upload yet)",
            file_types=["pdf", "png", "jpg"],
        ),
    ]
    outputs: ClassVar[list] = [
        Output(display_name="Form Data", name="form_data", method="collect"),
    ]

    def _field_names(self) -> list:
        """Names of all form fields (every input except the plumbing ones)."""
        inputs = getattr(self, "_inputs", None)
        if isinstance(inputs, dict) and inputs:
            return [name for name in inputs if name not in PLUMBING_INPUTS]
        seen = []
        declared = getattr(self, "inputs", None) or getattr(type(self), "inputs", None) or []
        for inp in declared:
            name = getattr(inp, "name", None)
            if name and name not in PLUMBING_INPUTS and name not in seen:
                seen.append(name)
        return seen

    def _field_value(self, name: str):
        """Runtime value of one input.

        Reads ``_inputs`` directly: a field renamed to e.g. ``name`` would
        otherwise resolve to the component attribute instead of its value.
        """
        inputs = getattr(self, "_inputs", None)
        if isinstance(inputs, dict) and name in inputs:
            return getattr(inputs[name], "value", None)
        try:
            return getattr(self, name, None)
        except (AttributeError, ValueError):
            return None

    def collect(self) -> Data:
        data = {}
        for name in self._field_names():
            data[name] = _json_safe(self._field_value(name))
        # Submitted values (via form_values) win over the field defaults.
        # Accepts the raw dict or the run envelope {"payload": {...}}.
        # Unknown keys are kept: the frontend may already know renamed
        # fields this code revision does not declare.
        try:
            override = json.loads(_text(self.form_values) or "{}")
        except (json.JSONDecodeError, ValueError, AttributeError):
            override = {}
        if isinstance(override, dict):
            if isinstance(override.get("payload"), dict):
                override = override["payload"]
            for key, val in override.items():
                if isinstance(key, str) and key:
                    data[key] = _json_safe(val)
        return Data(data=data)