"""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. When the run envelope names the asked fields (``fields`` list, sent by the API from the version's form schema), defaults are emitted only for asked fields — the template may declare fields unknown to this code and unasked code defaults stay out of the output. Without the list (raw dict, Langflow playground), all declared defaults apply. ``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 # noqa: F401 -- used by inlined _form_core at runtime import re # noqa: F401 -- used by inlined _form_core at runtime 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 from kirche._form_core import ( PLUMBING_INPUTS, _field_conditions, _json_safe, collect_form_data, parse_envelope, ) # <-- _form_core inlined by scripts/make_flow_fixtures.py:kirche_node() # (LFX executes embedded source without the kirche package) --> 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 _is_visible(self, name: str, effective: dict) -> bool: """Whether a field's branch conditions hold. Single pass over effective values (submitted first, code defaults after): no recursion, so trigger cycles cannot hang — a stale or cyclic marker simply matches or not. Unknown triggers never match. """ inputs = getattr(self, "_inputs", None) info = "" structured = None if isinstance(inputs, dict) and name in inputs: info = getattr(inputs[name], "info", "") or "" structured = getattr(inputs[name], "visible_if", None) for trigger, expected in _field_conditions(info, structured): if trigger == name or trigger not in effective: # Self-reference or unknown trigger: fail open (visible), # mirroring the backend parser. continue actual = effective[trigger] if isinstance(actual, list): if expected not in actual: return False elif actual != expected: return False return True def collect(self) -> Data: payload, asked = parse_envelope(self.form_values) # Effective trigger values: submitted first, code defaults after. declared = self._field_names() field_values = {name: _json_safe(self._field_value(name)) for name in declared} effective, scope = collect_form_data(declared, field_values, payload, asked) # Defaults only for asked AND visible fields (hidden branch fields # stay out, like the frontend never submits them); without an asked # list, all visible declared fields (Langflow playground). data = {name: effective[name] for name in scope if self._is_visible(name, effective)} # Submitted values win over the field defaults. Unknown keys are # kept: the frontend may already know renamed fields this code # revision does not declare. for key, val in payload.items(): if isinstance(key, str) and key: data[key] = _json_safe(val) return Data(data=data)