import csv
import io
from collections import Counter
from collections.abc import Callable, Mapping, Sequence
from decimal import Decimal
from typing import TYPE_CHECKING, Any, TypedDict, cast, overload
from typing import Optional as TOptional
from uuid import UUID

from flask import current_app
from flask_wtf import FlaskForm
from flask_wtf.file import FileAllowed, FileField, FileRequired, FileSize
from govuk_frontend_wtf.wtforms_widgets import (
    GovCharacterCount,
    GovCheckboxesInput,
    GovCheckboxInput,
    GovFileInput,
    GovRadioInput,
    GovSelect,
    GovSubmitInput,
    GovTextArea,
    GovTextInput,
)
from wtforms import Field, FieldList, FormField, HiddenField, IntegerField, SelectField, SelectMultipleField
from wtforms.fields.choices import RadioField
from wtforms.fields.simple import BooleanField, StringField, SubmitField, TextAreaField
from wtforms.validators import DataRequired, Email, Optional, Regexp, StopValidation, ValidationError

from app.common.auth.authorisation_helper import AuthorisationHelper
from app.common.collections.types import DecimalAnswer, IntegerAnswer
from app.common.data.interfaces.collections import (
    group_name_exists,
)
from app.common.data.interfaces.grants import grant_code_exists, grant_name_exists
from app.common.data.interfaces.user import get_user_by_email
from app.common.data.types import (
    NOT_NEEDED_TASKLIST_SECTION_STATUSES,
    CollectionType,
    ConditionsOperator,
    ExpressionType,
    FileUploadTypes,
    GroupDisplayOptions,
    ManagedExpressionsEnum,
    MaximumFileSize,
    MultilineTextInputRows,
    NumberInputWidths,
    NumberTypeEnum,
    QuestionDataType,
)
from app.common.expressions import ExpressionContext
from app.common.expressions.references import ExpressionReference
from app.common.expressions.registry import get_registered_data_types
from app.common.forms.fields import InterpolationStatementField, MHCLGAccessibleAutocomplete, MHCLGSelectWithSearch
from app.common.forms.helpers import get_referenceable_questions
from app.common.forms.validators import CommunitiesEmail, WordRange
from app.common.helpers.collections import SubmissionHelper
from app.common.helpers.feature_flags import FeatureFlags
from app.common.safe_ids import safe_column_id
from app.common.utils import comma_join_items, uppercase_first
from app.constants import DATA_SET_EXTERNAL_ID_COLUMN_HEADER, DATA_SET_IDENTIFIER_COLUMN_HEADERS
from app.deliver_grant_funding.data_sets import (
    BritishPoundsError,
    CellError,
    CSVDecodeError,
    DataTypeError,
    DecimalError,
    PrefixError,
    SuffixError,
    decode_csv_bytes,
    validate_csv_cell_against_column_mapping,
)
from app.deliver_grant_funding.session_models import DataSetColumnMapping

if TYPE_CHECKING:
    from app.common.data.models import (
        Collection,
        Component,
        DataSource,
        Form,
        Grant,
        GrantRecipient,
        Group,
        Organisation,
        Question,
    )
    from app.deliver_grant_funding.session_models import AddContextToComponentSessionModel


@overload
def strip_string_if_not_empty(value: None) -> None: ...
@overload
def strip_string_if_not_empty[T: str](value: T) -> T: ...
def strip_string_if_not_empty(value: str | None) -> str | None:
    if value is None:
        return None

    if not value:
        return value

    # Keep it working with str subclasses like ExpressionStatement; we should return the same type
    stripped = value.strip()
    if type(value) is not str:
        return type(value)(stripped)

    return stripped


@overload
def strip_newlines(value: None) -> None: ...
@overload
def strip_newlines[T: str](value: T) -> T: ...
def strip_newlines(value: str | None) -> str | None:
    if not value:
        return value

    replaced = value.replace("\n", "")

    # Keep it working with str subclasses like ExpressionStatement; we should return the same type
    if type(value) is not str:
        return type(value)(replaced)

    return replaced


@overload
def empty_string_to_none(value: None) -> None: ...
@overload
def empty_string_to_none[T: str](value: T) -> T | None: ...
def empty_string_to_none(value: str | None) -> str | None:
    return value if value else None


def _validate_no_blank_lines(form: FlaskForm, field: Field) -> None:
    choices = field.data.split("\n")
    if any(choice.strip() == "" for choice in choices):
        raise ValidationError("Remove blank lines from the list")


def _validate_no_duplicates(form: FlaskForm, field: Field) -> None:
    choices = [choice.strip() for choice in field.data.split("\n")]
    if len(choices) != len(set(choices)):
        raise ValidationError("Remove duplicate options from the list")


def _validate_max_list_length(max_length: int) -> Callable[[Any, Any], None]:
    def validator(form: FlaskForm, field: Field) -> None:
        if len(field.data.split("\n")) > max_length:
            raise ValidationError(f"You have entered too many options. The maximum is {max_length}")

    return validator


def _validate_textarea_size(form: FlaskForm, field: Field) -> None:
    rows = int(field.data)
    if rows not in MultilineTextInputRows:
        raise ValidationError("Select a text area size")


class GrantSetupForm(FlaskForm):
    SUBMIT_BUTTON_TEXT_SETUP = "Save and continue"
    SUBMIT_BUTTON_TEXT_CHANGE = "Update"
    submit = SubmitField(SUBMIT_BUTTON_TEXT_SETUP, widget=GovSubmitInput())

    def __init__(self, *args: Any, is_update: bool = False, **kwargs: Any):
        super().__init__(*args, **kwargs)
        if is_update:
            self.submit.label.text = self.SUBMIT_BUTTON_TEXT_CHANGE


class GrantGGISForm(FlaskForm):
    has_ggis = RadioField(
        "Do you have a GGIS number?",
        # These choices have no effect on the frontend, but are used for validation. Frontend choices are found in the
        # template, currently at app/deliver_grant_funding/templates/deliver_grant_funding/grant_setup/ggis_number.html.
        # Developers will need to keep these in sync manually.
        choices=[("yes", "Yes"), ("no", "No")],
        validators=[DataRequired("Please select an option")],
        widget=GovRadioInput(),
    )
    ggis_number = StringField(
        "Enter your GGIS reference number",
        description="For example, G2-SCH-2025-05-12346",
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )
    submit = SubmitField("Save and continue", widget=GovSubmitInput())

    def validate(self, extra_validators: Mapping[str, Sequence[Any]] | None = None) -> bool:
        if not super().validate(extra_validators):
            return False

        if self.has_ggis.data == "yes" and not self.ggis_number.data:
            self.ggis_number.errors = list(self.ggis_number.errors) + ["Enter your GGIS reference number"]
            return False

        return True


class GrantChangeGGISForm(FlaskForm):
    ggis_number = StringField(
        "What is the GGIS reference number?",
        description="For example, G2-SCH-2025-05-12346",
        filters=[strip_string_if_not_empty],
        validators=[DataRequired("Enter your GGIS reference number")],
        widget=GovTextInput(),
    )
    submit = SubmitField("Update", widget=GovSubmitInput())


class GrantNameForm(GrantSetupForm):
    name = StringField(
        "Enter the grant name",
        description="Use the full and official name of the grant - no abbreviations or acronyms",
        validators=[
            DataRequired("Enter the grant name"),
        ],
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )

    def __init__(self, *args: Any, existing_grant_id: UUID | None = None, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.existing_grant_id = existing_grant_id

    def validate_name(self, field: StringField) -> None:
        if field.data and grant_name_exists(field.data, exclude_grant_id=self.existing_grant_id):
            raise ValidationError("Grant name already in use")


class GrantCodeForm(GrantSetupForm):
    code = StringField(
        "Enter the grant code",
        validators=[
            DataRequired("Enter a unique grant code"),
            Regexp(
                r"^[A-Z0-9-]+$", message="The grant code should only contain uppercase letters, numbers, and dashes"
            ),
        ],
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )

    def __init__(self, *args: Any, existing_grant_id: UUID | None = None, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.existing_grant_id = existing_grant_id

    def validate_code(self, field: StringField) -> None:
        if field.data and (grant := grant_code_exists(field.data, exclude_grant_id=self.existing_grant_id)):
            raise ValidationError(f"Grant code already in use by {grant.name}")


class GrantDescriptionForm(GrantSetupForm):
    DESCRIPTION_MAX_WORDS = 200

    description = TextAreaField(
        "Enter the main purpose of this grant",
        validators=[
            DataRequired("Enter the main purpose of this grant"),
            WordRange(max_words=DESCRIPTION_MAX_WORDS, field_display_name="description"),
        ],
        filters=[strip_string_if_not_empty],
        widget=GovCharacterCount(),
    )


class GrantContactForm(GrantSetupForm):
    primary_contact_name = StringField(
        "Full name",
        validators=[DataRequired("Enter the full name")],
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )
    primary_contact_email = StringField(
        "Email address",
        description="Use the shared email address for the grant team",
        validators=[
            DataRequired("Enter the email address"),
            Email(message="Enter an email address in the correct format, like name@example.com"),
        ],
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )


class QuestionTypeForm(FlaskForm):
    question_data_type = RadioField(
        "What type of question do you need?",
        choices=[(qdt.name, qdt.value) for qdt in QuestionDataType],
        validators=[DataRequired("Select a question type")],
        widget=GovRadioInput(),
    )
    parent = HiddenField(
        "Parent",
        description="The parent this question will belong to. If not set the question belongs to the form directly",
    )
    submit = SubmitField(widget=GovSubmitInput())


class GroupForm(FlaskForm):
    name = StringField(
        "Question group name",
        validators=[DataRequired("Enter the question group name")],
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())

    def __init__(self, *args: Any, check_name_exists: bool = False, group_form_id: UUID | None = None, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.check_name_exists = check_name_exists
        self.group_form_id = group_form_id

    def validate_name(self, field: StringField) -> None:
        if field.data and self.check_name_exists:
            if not self.group_form_id:
                raise ValueError("group_form_id must be provided if check_name_exists is True")
            if group_name_exists(field.data, self.group_form_id):
                raise ValidationError("A question group with this name already exists")


class GroupDisplayOptionsForm(FlaskForm):
    show_questions_on_the_same_page = RadioField(
        "How do you want this question group to be displayed?",
        choices=[
            (GroupDisplayOptions.ONE_QUESTION_PER_PAGE, "One question per page"),
            (GroupDisplayOptions.ALL_QUESTIONS_ON_SAME_PAGE, "All questions on the same page"),
        ],
        default=GroupDisplayOptions.ONE_QUESTION_PER_PAGE,
        validators=[DataRequired("Select how you want this question group to be displayed")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())


class GroupAddAnotherOptionsForm(FlaskForm):
    question_group_is_add_another = RadioField(
        "Should people be able to answer all questions in this question group more than once?",
        choices=[
            ("yes", "Yes"),
            ("no", "No - questions can only be answered once"),
        ],
        default="no",
        validators=[
            DataRequired(
                "Select whether people should be able to answer all questions in this question group more than once"
            )
        ],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())


class GroupAddAnotherSummaryForm(FlaskForm):
    questions_to_show_in_add_another_summary = SelectMultipleField(
        "Which question answers should be included when showing a summary of each add another answer?",
        default=[],
        widget=GovCheckboxesInput(),
        choices=[],
        validators=[
            DataRequired(
                "Select which question answers should be included when showing a summary of each add another answer"
            )
        ],
        render_kw={"params": {"fieldset": {"legend": {"classes": "govuk-visually-hidden"}}}},
    )

    def __init__(self, *args: Any, group: Group, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.questions_to_show_in_add_another_summary.choices = [
            (str(question.id), question.name) for question in group.cached_questions
        ]

        if not self.is_submitted():
            self.questions_to_show_in_add_another_summary.data = [
                str(question.id) for question in group.questions_in_add_another_summary
            ]

    submit = SubmitField(widget=GovSubmitInput())


class ConditionsOperatorForm(FlaskForm):
    conditions_operator = RadioField(
        "When should this component be shown?",
        choices=[
            (ConditionsOperator.ALL, "When all of these conditions match"),
            (ConditionsOperator.ANY, "When any of these conditions match"),
        ],
        default=ConditionsOperator.ALL,
        validators=[DataRequired("Select when this should be shown")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())


class QuestionForm(FlaskForm):
    text = InterpolationStatementField(
        "Question text",
        description="The text grant recipients will see on their report",
        validators=[DataRequired("Enter the question text")],
        filters=[strip_string_if_not_empty, strip_newlines],
        widget=GovTextArea(),
    )
    hint = InterpolationStatementField(
        "Question hint (optional)",
        filters=[strip_string_if_not_empty],
        widget=GovTextArea(),
        description=(
            "A single sentence to help someone answer the question, for example, ‘Must be between 6 and 8 digits long’"
        ),
        render_kw={"params": {"rows": 2}},
    )
    name = StringField(
        "Question name",
        validators=[DataRequired("Enter the question name")],
        description=(
            "A short name for this question that will be used for reference in monitoring reports (use lower-case text)"
        ),
        filters=[strip_string_if_not_empty, strip_newlines],
        widget=GovTextInput(),
    )
    add_context = StringField(widget=GovSubmitInput())

    # Note: the next fields all read from properties on the `Question` model because the names match. This
    # implicit connection needs to be maintained.
    data_source_items = StringField(
        "List of options",
        validators=[Optional()],
        description="Enter each option on a new line",
        filters=[strip_string_if_not_empty, lambda val: val.replace("\r", "") if val else val],
        widget=GovTextArea(),
    )
    separate_option_if_no_items_match = BooleanField(
        "Include an ‘other’ option",
        validators=[Optional()],
        widget=GovCheckboxInput(),
    )
    none_of_the_above_item_text = StringField(
        "‘Other’ option text",
        default="Other",
        validators=[Optional()],
        widget=GovTextInput(),
    )

    # Multiline textarea field presentation options
    rows = SelectField(
        "Text area size",
        widget=GovSelect(),
        validators=[Optional()],
        choices=[(opt.value, f"{opt.name.title()} ({opt.value} rows)") for opt in MultilineTextInputRows],
        default=MultilineTextInputRows.MEDIUM.value,
    )
    word_limit = IntegerField(
        "Word limit (optional)",
        widget=GovTextInput(),
        validators=[Optional()],
    )

    # Number options
    number_type = RadioField(
        "Type of number",
        choices=[(number_type.value, number_type.value) for number_type in NumberTypeEnum],
        widget=GovRadioInput(),
        validators=[Optional()],
    )
    max_decimal_places = IntegerField(
        "Maximum number of decimal places",
        widget=GovTextInput(),
        validators=[Optional()],
    )
    # File upload options
    file_types_supported = SelectMultipleField(
        "Accepted file types",
        choices=[(file_type.value, file_type.value) for file_type in FileUploadTypes],
        widget=GovCheckboxesInput(),
        validators=[Optional()],
        default=[file_type.value for file_type in FileUploadTypes],
    )
    maximum_file_size = RadioField(
        "Maximum file size",
        choices=[(size.value, f"{size.value} ({size.human_readable})") for size in MaximumFileSize],
        widget=GovRadioInput(),
        validators=[Optional()],
        default=MaximumFileSize.SMALL.value,
    )

    # Number field presentation options
    prefix = StringField(
        "Prefix (optional)",
        widget=GovTextInput(),
        validators=[Optional()],
        filters=[strip_string_if_not_empty, empty_string_to_none],
    )
    suffix = StringField(
        "Suffix (optional)",
        widget=GovTextInput(),
        validators=[Optional()],
        filters=[strip_string_if_not_empty, empty_string_to_none],
    )
    width = SelectField(
        "Input width",
        description="Reduce the size of the input if you know the answer will be smaller",
        widget=GovSelect(),
        validators=[Optional()],
        choices=[(opt.value, f"{opt.name.title()}") for opt in NumberInputWidths],
        default=NumberInputWidths.BILLIONS.value,
    )

    # Date field presentation options
    approximate_date = BooleanField(
        "Ask for an approximate date (month and year only)",
        validators=[Optional()],
        widget=GovCheckboxInput(),
    )

    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        *args: Any,
        question_type: QuestionDataType,
        obj: TOptional[Question | AddContextToComponentSessionModel] = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, obj=obj, **kwargs)

        self._question_type = question_type
        self._original_separate_option_if_no_items_match = self.separate_option_if_no_items_match.data

        match question_type:
            case QuestionDataType.RADIOS | QuestionDataType.CHECKBOXES:
                max_length = (
                    current_app.config["MAX_DATA_SOURCE_ITEMS_RADIOS"]
                    if question_type == QuestionDataType.RADIOS
                    else current_app.config["MAX_DATA_SOURCE_ITEMS_CHECKBOXES"]
                )
                self.data_source_items.validators = [
                    DataRequired("Enter the options for your list"),
                    _validate_no_blank_lines,
                    _validate_no_duplicates,
                    _validate_max_list_length(max_length=max_length),
                ]

                if self.separate_option_if_no_items_match.raw_data:
                    self.none_of_the_above_item_text.validators = [
                        DataRequired("Enter the text to show for the fallback option")
                    ]

                if question_type == QuestionDataType.CHECKBOXES:
                    self.data_source_items.description = (
                        "Enter each option on a new line - you can add a maximum of "
                        f"{current_app.config['MAX_DATA_SOURCE_ITEMS_CHECKBOXES']} options"
                    )

            case QuestionDataType.TEXT_MULTI_LINE:
                self.rows.validators = [_validate_textarea_size]

    @property
    def normalised_data_source_items(self) -> list[str] | None:
        """For radios questions, we might want to display a final item beneath an 'or' divider, to signify that
        the choice is semantically unrelated to all of the other answers. The most common usecase for this is something
        like a "Other" answer.

        This answer is stored in the data source like a normal item. We store it as the last item and then record on
        the question that the last item in the data source should be presented distinctly.

        This form is essentially just responsible for appending the "Other" item to the data source items
        explicitly set by the form builder.
        """
        if self._question_type not in [QuestionDataType.RADIOS, QuestionDataType.CHECKBOXES]:
            return None

        data_source_items: list[str] = []
        if self.data_source_items.data is not None:
            data_source_items.extend(item.strip() for item in self.data_source_items.data.split("\n") if item.strip())

            if self.separate_option_if_no_items_match.data is True:
                data_source_items.append(cast(str, self.none_of_the_above_item_text.data))

        return data_source_items

    def validate_prefix(self, field: Field) -> None:
        if self.prefix.data and self.suffix.data:
            raise ValidationError("Remove the suffix if you need a prefix")

    def validate_suffix(self, field: Field) -> None:
        if self.prefix.data and self.suffix.data:
            raise ValidationError("Remove the prefix if you need a suffix")

    def is_submitted_to_add_context(self) -> bool:
        return bool(self.is_submitted() and self.add_context.data and not self.submit.data)

    def get_component_form_data(self) -> dict[str, Any]:
        return {key: data for key, data in self.data.items() if key not in {"csrf_token", "submit"}}

    def validate(self, extra_validators=None):
        if self.is_submitted_to_add_context():
            return True

        # Only need to validate number fields if the question type is NUMBER
        if self._question_type == QuestionDataType.NUMBER:
            self.number_type.validators = [DataRequired("Select the type of number")]

        if self.number_type.data == NumberTypeEnum.DECIMAL.value:
            self.max_decimal_places.validators = [DataRequired("Enter the maximum number of decimal places")]

        if self._question_type == QuestionDataType.FILE_UPLOAD:
            self.file_types_supported.validators = [DataRequired("Select at least one file type")]
            self.maximum_file_size.validators = [DataRequired("Select a maximum file size")]

        return super().validate(extra_validators=extra_validators)


class AddContextSelectSourceForm(FlaskForm):
    data_source = RadioField(
        "Select a data source",
        choices=[],
        widget=GovRadioInput(),
    )

    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        *args: Any,
        form: Form,
        current_component: TOptional[Component],
        parent_component: TOptional[Group] = None,
        include_this_component: bool = False,
        **kwargs: Any,
    ):
        super().__init__(*args, **kwargs)
        self.form = form
        self.current_component = current_component
        self.parent_component = parent_component
        self.include_this_component = include_this_component

        if FeatureFlags.NEW_CONTEXT_SOURCES.is_enabled:
            # A soft feature flag that will (when implemented) allow platform admins to test new context sources
            # before releasing to a wider audience eg form builders.
            self.data_source.choices = [(choice.name, choice.value) for choice in ExpressionContext.ContextSources]
        else:
            self.data_source.choices = [
                (ExpressionContext.ContextSources.SECTION.name, ExpressionContext.ContextSources.SECTION.value),
                (
                    ExpressionContext.ContextSources.PREVIOUS_SECTION.name,
                    ExpressionContext.ContextSources.PREVIOUS_SECTION.value,
                ),
                (
                    ExpressionContext.ContextSources.DATASET.name,
                    ExpressionContext.ContextSources.DATASET.value,
                ),
            ]

        if include_this_component and current_component and current_component.is_question:
            self.data_source.choices.insert(
                0,
                (
                    "THIS_QUESTION",
                    "This question",
                ),
            )

    def validate_data_source(self, field: Field) -> None:
        choice = None

        if field.data == "THIS_QUESTION" and not self.include_this_component:
            raise ValidationError("You cannot select this question")

        try:
            choice = ExpressionContext.ContextSources[field.data]
        except KeyError:
            return

        if choice == ExpressionContext.ContextSources.SECTION:
            if not get_referenceable_questions(
                form=self.form,
                current_component=self.current_component,
                parent_component=self.parent_component,
                include_this_component=self.include_this_component,
            ):
                raise ValidationError("There are no available questions before this one in the section")


class SelectDataSourceSectionForm(FlaskForm):
    section = RadioField(
        "Select a previous section",
        choices=[],
        validators=[DataRequired("Select a previous section")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        current_form: Form,
        *args: Any,
        **kwargs: Any,
    ):
        super().__init__(*args, **kwargs)

        # TODO: when using this for conditions and validation, we also need to filter the 'available' questions
        # based on the usable data types.
        self.section.choices = [(f.id, f.title) for f in current_form.earlier_forms]


class SelectDataSourceQuestionForm(FlaskForm):
    question = SelectField(
        "Select which question's answer to use",
        choices=[],
        coerce=ExpressionReference,
        validators=[DataRequired("Select the question")],
        widget=MHCLGAccessibleAutocomplete(),
    )

    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        form: Form,
        interpolate: Callable[[str], str],
        current_component: TOptional[Component],
        subject_reference: TOptional[ExpressionReference],
        *args: Any,
        expression_type: TOptional[ExpressionType],
        managed_expression_name: TOptional[ManagedExpressionsEnum],
        parent_component: TOptional[Group] = None,
        include_this_component: bool = False,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.include_this_component = include_this_component

        # NOTE: I think this logic might sit better inside `get_referenceable_questions` or a separate helper for a
        # future refactor, but not no time to pull that thread currently - sorry future us.
        limit_to_data_types: set[QuestionDataType] = get_registered_data_types()
        if expression_type is not None:
            if managed_expression_name is None:
                limit_to_data_types = {QuestionDataType.NUMBER}
            else:
                if subject_reference:
                    limit_to_data_types = {subject_reference.data_type}
                elif current_component and current_component.data_type:
                    limit_to_data_types = {current_component.data_type}

        limit_to_component = (
            subject_reference.question
            if subject_reference and subject_reference.question
            else current_component
            if current_component and current_component.form == form
            else None
        )

        referenceable_questions = get_referenceable_questions(
            form,
            limit_to_component,
            parent_component if parent_component and parent_component.form == form else None,
            limit_to_data_type=limit_to_data_types,
            include_this_component=self.include_this_component,
        )

        if referenceable_questions:
            self.question.choices = [("", "")] + [
                (ExpressionReference.from_question(question), interpolate(question.text))
                for question in referenceable_questions
            ]


class SelectDataSourceDataSetForm(FlaskForm):
    data_set = RadioField(
        "Select uploaded data set",
        choices=[],
        validators=[DataRequired("Select a data set")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        collection: Collection,
        *args: Any,
        number_columns_only: bool = False,
        **kwargs: Any,
    ):
        super().__init__(*args, **kwargs)
        self.number_columns_only = number_columns_only

        data_sources = collection.data_sources
        if number_columns_only:
            data_sources = [
                d
                for d in data_sources
                if d.schema and any(col.data_type == QuestionDataType.NUMBER for col in d.schema.root.values())
            ]

        self.data_set.choices = [(d.id, cast(str, d.name)) for d in data_sources]


class SelectDataSourceDataSetColumnForm(FlaskForm):
    column = RadioField(
        choices=[],
        validators=[DataRequired("Select a column")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        data_set: DataSource,
        *args: Any,
        number_columns_only: bool = False,
        **kwargs: Any,
    ):
        super().__init__(*args, **kwargs)
        self.number_columns_only = number_columns_only

        assert data_set.schema is not None

        self.column.label.text = f"Select column in {data_set.name} data set"

        columns = data_set.schema.ordered_items()
        if number_columns_only:
            columns = [
                (safe_column_id, column_schema)
                for safe_column_id, column_schema in columns
                if column_schema.data_type == QuestionDataType.NUMBER
            ]

        self.column.choices = [
            (safe_column_id, uppercase_first(column_schema.original_column_name) or "")
            for safe_column_id, column_schema in columns
        ]


class GrantAddUserForm(FlaskForm):
    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.grant = kwargs["grant"]

    user_email = StringField(
        description="This needs to be the user’s personal 'communities.gov.uk' "
        "email address, not a shared email address.",
        validators=[
            DataRequired("Enter an email address"),
            CommunitiesEmail(),
        ],
        filters=[strip_string_if_not_empty],
        widget=GovTextInput(),
    )
    submit = SubmitField("Continue", widget=GovSubmitInput())

    def validate(self, extra_validators: Any = None) -> bool:
        if not super().validate(extra_validators):
            return False

        if self.user_email.data:
            user_to_add = get_user_by_email(self.user_email.data)
            if not user_to_add:
                return True

            if AuthorisationHelper.is_deliver_grant_admin(grant_id=self.grant.id, user=user_to_add):
                self.user_email.errors = list(self.user_email.errors) + [
                    f'This user already is an admin of "{self.grant.name}" so you cannot add them'
                ]
                return False
            if AuthorisationHelper.is_deliver_grant_member(grant_id=self.grant.id, user=user_to_add):
                self.user_email.errors = list(self.user_email.errors) + [
                    f'This user already is a member of "{self.grant.name}" so you cannot add them'
                ]
                return False

        return True


class SetUpCollectionForm(FlaskForm):
    name = StringField(
        widget=GovTextInput(),
        validators=[DataRequired()],
    )

    submit = SubmitField(widget=GovSubmitInput())

    def __init__(self, *args: Any, collection_type: CollectionType, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.name.label.text = f"{uppercase_first(collection_type.constants.singular)} name"
        self.name.validators = [DataRequired(f"Enter a name for the {collection_type.constants.singular}")]
        self.submit.label.text = f"Create {collection_type.constants.singular}"


class CollectionCreationMethodForm(FlaskForm):
    method = RadioField(widget=GovRadioInput())
    submit = SubmitField("Continue", widget=GovSubmitInput())

    def __init__(self, *args: Any, collection_type: CollectionType, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        singular = collection_type.constants.singular
        self.method.label.text = f"Do you want to create a new {singular} or copy an existing {singular}?"
        self.method.choices = [
            ("create", f"Create a new {singular}"),
            ("copy", f"Copy an existing {singular}"),
        ]
        self.method.validators = [DataRequired(f"Select how you want to create the {singular}")]


class SelectCollectionToCopyForm(FlaskForm):
    collection = SelectField(widget=MHCLGSelectWithSearch())
    submit = SubmitField("Continue", widget=GovSubmitInput())

    def __init__(
        self,
        *args: Any,
        collection_type: CollectionType,
        collections: Sequence["Collection"],
        grant: Grant,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        singular = collection_type.constants.singular
        self.collection.label.text = f"Which {singular} do you want to copy?"
        self.collection.description = f"Search by {singular} or grant name"

        sorted_collections = sorted(
            collections, key=lambda c: (0 if c.grant == grant else 1, c.grant.name, -c.created_at_utc.timestamp())
        )
        self.collection.choices = [("", "")] + [
            (str(collection.id), collection.name, {"data-hint": collection.grant.name})
            for collection in sorted_collections
        ]
        self.collection.validators = [DataRequired(f"Select the {singular} to copy")]


class AddSectionForm(FlaskForm):
    title = StringField(
        "Section name",
        widget=GovTextInput(),
        validators=[DataRequired("Enter a name for the section")],
    )
    submit = SubmitField("Add section", widget=GovSubmitInput())


class AddGuidanceForm(FlaskForm):
    guidance_heading = StringField(
        "Give your page a heading",
        description=(
            "When you add guidance your question text will no longer be the main page heading, "
            "so you need to use a different one. "
            "Use a heading that’s a statement rather than a question - for example, ‘Interview needs’."
        ),
        widget=GovTextInput(),
        filters=[strip_string_if_not_empty],
    )
    guidance_body = InterpolationStatementField(
        "Add guidance text",
        description="Use Markdown if you need to format your guidance content. Formatting help can be found below.",
        widget=GovTextArea(),
        filters=[strip_string_if_not_empty],
    )
    add_context = StringField(widget=GovSubmitInput())

    preview = SubmitField("Save and preview guidance", widget=GovSubmitInput())
    submit = SubmitField("Save guidance", widget=GovSubmitInput())

    def __init__(self, *args: Any, heading_required: bool | None = True, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.heading_required = heading_required

    def validate(self, extra_validators: Mapping[str, Sequence[Any]] | None = None) -> bool:
        result: bool = super().validate(extra_validators=extra_validators)

        if not result:
            return result

        if (
            self.heading_required
            and (self.guidance_heading.data or self.guidance_body.data)
            and not (self.guidance_heading.data and self.guidance_body.data)
        ):
            self.form_errors.append("Provide both a page heading and guidance text, or neither")
            return False

        return result

    def is_submitted_to_add_context(self) -> bool:
        return bool(self.is_submitted() and self.add_context.data and not (self.submit.data or self.preview.data))

    def get_component_form_data(self) -> dict[str, Any]:
        return {key: data for key, data in self.data.items() if key not in {"csrf_token", "submit"}}


class PreviewGuidanceForm(FlaskForm):
    guidance = StringField()


class TestGrantRecipientJourneyForm(FlaskForm):
    organisation = SelectField(
        "Select a test organisation:",
        choices=[],
        validators=[DataRequired("Select a test organisation")],
        # TODO: replace with select-with-search; deprecated accessible autocomplete
        widget=MHCLGAccessibleAutocomplete(),
    )

    def __init__(self, *args: Any, users_test_grant_recipients: list[GrantRecipient], **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.organisation.choices = [("", "")] + [
            (str(grant_recipient.id), grant_recipient.organisation.name)
            for grant_recipient in users_test_grant_recipients
        ]
        if len(users_test_grant_recipients) == 1:
            self.organisation.default = str(users_test_grant_recipients[0].id)

    submit = SubmitField("Start test submission journey", widget=GovSubmitInput())


class PublicSignUpSettingsForm(FlaskForm):
    allow_public_sign_up = RadioField(
        "Should this collection allow public self sign up?",
        choices=[(True, "Yes"), (False, "No")],
        validators=[DataRequired("Select whether the collection should allow public sign up")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())


class MultipleSubmissionsForm(FlaskForm):
    allow_multiple_submissions = RadioField(
        "Are organisations allowed to have multiple submissions?",
        choices=[(True, "Yes"), (False, "No")],
        validators=[DataRequired("Select whether organisations are allowed to have multiple submissions")],
        widget=GovRadioInput(),
    )
    submit = SubmitField("Save and continue", widget=GovSubmitInput())


class MultipleSubmissionsNamingForm(FlaskForm):
    MANAGED = "managed-by-service"
    USER_NAMED = "named-by-user"

    submissions_naming = RadioField(
        "How will the submissions be named?",
        choices=[
            (MANAGED, "Grant policy team will provide a list of names"),
            (USER_NAMED, "Users will create and name each submission"),
        ],
        validators=[DataRequired("Select how the submissions will be named")],
        widget=GovRadioInput(),
    )
    submit = SubmitField("Save and continue", widget=GovSubmitInput())


class MultipleSubmissionsSettingsForm(FlaskForm):
    submission_name_question = SelectField(
        "Which question should be used for the submission names?",
        choices=[],
        widget=MHCLGSelectWithSearch(),
        validators=[DataRequired("Select a question to use as the submission name")],
    )
    guidance_body = StringField(
        "Add guidance for completing multiple submissions (optional)",
        widget=GovTextArea(),
        filters=[strip_string_if_not_empty],
    )
    preview = SubmitField("Save and preview guidance", widget=GovSubmitInput())
    submit = SubmitField("Save setting", widget=GovSubmitInput())

    def __init__(self, questions: list[Question], *args: Any, interpolate: Callable[[str], str], **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)

        self.submission_name_question.choices = [("", "")] + [(str(q.id), interpolate(q.text)) for q in questions]


class CertificationSettingsForm(FlaskForm):
    requires_certification = RadioField(
        "Do submissions require certification?",
        choices=[(True, "Yes"), (False, "No, allow submitting without certification")],
        validators=[DataRequired("Select whether certification is required")],
        widget=GovRadioInput(),
    )
    submit = SubmitField("Save certification setting", widget=GovSubmitInput())


class ReopeningSettingsForm(FlaskForm):
    allow_submission_reopening = RadioField(
        "Can submissions be reopened?",
        choices=[(True, "Yes"), (False, "No")],
        validators=[DataRequired("Select whether submissions can be reopened")],
        widget=GovRadioInput(),
    )
    submit = SubmitField("Save reopening setting", widget=GovSubmitInput())


class CollectionSettingsSelectSectionForm(FlaskForm):
    section = RadioField(
        "Select a section",
        choices=[],
        validators=[DataRequired("Select a section")],
        widget=GovRadioInput(),
    )
    submit = SubmitField(widget=GovSubmitInput())

    def __init__(self, *args: Any, collection_forms: list[Form], **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.section.choices = [(str(f.id), f.title) for f in collection_forms]


class CollectionSettingsSelectQuestionForm(FlaskForm):
    question = SelectField(
        "Select which question's answer to use as the submission name",
        choices=[],
        validators=[DataRequired("Select the question")],
        widget=MHCLGAccessibleAutocomplete(),
    )
    submit = SubmitField(widget=GovSubmitInput())

    def __init__(
        self,
        *args: Any,
        form: Form,
        interpolate: Callable[[str], str],
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.question.choices = [("", "")] + [
            (str(question.id), interpolate(question.text)) for question in form.cached_questions
        ]


class UploadDataSetForm(FlaskForm):
    name = StringField(
        "Data set name",
        widget=GovTextInput(),
        filters=[strip_string_if_not_empty],
        validators=[DataRequired("Enter the name for this data set")],
    )

    file = FileField(
        "Upload a file",
        widget=GovFileInput(),
        validators=[
            FileRequired("Select a file"),
            FileAllowed(["csv"], "The file must be a CSV"),
            FileSize(max_size=10485760, message="The file must be smaller than 10MB"),
        ],
    )

    submit = SubmitField("Continue and map columns", widget=GovSubmitInput())

    def __init__(
        self,
        *args: Any,
        existing_data_source_names: list[str | None],
        collection: Collection,
        existing_datasource: DataSource | None = None,
        locked_orgs: list[Organisation] | None = None,
        all_organisations: list[Organisation] | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self._existing_data_source_names = existing_data_source_names or []
        self.existing_datasource = existing_datasource
        self.data_errors = []
        self.locked_orgs = locked_orgs or []
        self.all_organisations = all_organisations or []
        self.collection = collection
        self.removed_column_errors = {}
        self.changed_column_errors = {}

    def validate_name(self, field: StringField) -> None:
        if field.data and field.data in self._existing_data_source_names:
            raise ValidationError("A data set with this name already exists for this monitoring report")

    @staticmethod
    def _validate_minimum_columns(fieldnames) -> None:
        if not fieldnames or not any(fieldname.strip() for fieldname in fieldnames):
            raise ValidationError("The CSV file must have at least one column")

        if any(not fieldname.strip() for fieldname in fieldnames):
            raise ValidationError("The CSV file must have a name for each column")

        missing = [col for col in DATA_SET_IDENTIFIER_COLUMN_HEADERS if col not in fieldnames]
        if missing:
            raise ValidationError(f"The CSV file must contain the columns: {', '.join(missing)}")

        data_columns = [col for col in fieldnames if col.strip() and col not in DATA_SET_IDENTIFIER_COLUMN_HEADERS]
        if not data_columns:
            raise ValidationError("The CSV file must contain at least one column of data")

    @staticmethod
    def _validate_duplicate_column_names_in_csv(fieldnames) -> None:
        if len(fieldnames) != len(set(fieldnames)):
            counter = Counter(fieldnames)
            duplicate_fieldnames = [fieldname for fieldname, count in counter.items() if count > 1]
            raise ValidationError(f"The CSV file contains duplicate column names: {', '.join(duplicate_fieldnames)}")

        fieldnames_to_safe_ids = {fieldname: safe_column_id(fieldname) for fieldname in fieldnames}
        counter = Counter(fieldnames_to_safe_ids.values())
        duplicate_safe_ids = {safe_id for safe_id, count in counter.items() if count > 1}
        if duplicate_safe_ids:
            duplicate_originals = [
                name for name, safe_id in fieldnames_to_safe_ids.items() if safe_id in duplicate_safe_ids
            ]
            raise ValidationError(f"The CSV file contains duplicate column names: {', '.join(duplicate_originals)}")

    @staticmethod
    def _validate_columns_in_each_row(rows) -> None:
        if any(None in row or None in row.values() for row in rows):
            raise ValidationError("The CSV file contains rows which are longer or shorter than the number of columns")

    @staticmethod
    def _validate_max_rows(rows) -> None:
        row_count = len(rows)
        if row_count > 10000:
            raise ValidationError("The file must contain no more than 10,000 rows")

    @staticmethod
    def _validate_min_rows(rows) -> None:
        row_count = len(rows)
        if row_count < 1:
            raise ValidationError("The CSV file must contain at least one row of data")

    def _validate_data_for_existing_submissions(  # noqa: C901
        self, existing_datasource: DataSource, rows: list[dict[str, str]]
    ) -> None:
        removed_column_errors = {}
        changed_column_errors = {}
        for org in self.locked_orgs:
            existing_org_item = existing_datasource.get_filtered_organisation_item(org.external_id)
            if not existing_org_item:
                # Nothing to validate if there was no data for this org in the first place
                continue
            new_org_item_row = next(
                (row for row in rows if row[DATA_SET_EXTERNAL_ID_COLUMN_HEADER] == org.external_id), None
            )
            if not new_org_item_row:
                raise ValidationError(f"The file does not contain a row for grant recipient {org.name}")

            for column_id, column_def in existing_datasource.schema.root.items():  # ty:ignore[unresolved-attribute]
                orig_col_name = column_def.original_column_name
                existing_value_answer = (
                    existing_org_item.data[column_id] if column_id in existing_org_item.data else None
                )
                if not existing_value_answer:
                    continue
                if orig_col_name not in new_org_item_row:
                    orgs_with_errors_in_this_column = removed_column_errors.get(orig_col_name, [])
                    orgs_with_errors_in_this_column.append(org.name)
                    removed_column_errors[orig_col_name] = orgs_with_errors_in_this_column
                    continue

                org_has_error = False
                new_value = new_org_item_row[orig_col_name]
                if not new_value:
                    org_has_error = True

                # This allows matches for values with prefix or suffix, eg 100 == £100
                if not org_has_error:
                    if isinstance(existing_value_answer, (IntegerAnswer, DecimalAnswer)):
                        existing_value_answer = cast(IntegerAnswer | DecimalAnswer, existing_value_answer)
                        stripped_value = (
                            new_value.strip()
                            .removeprefix(existing_value_answer.prefix if existing_value_answer.prefix else "")
                            .removesuffix(existing_value_answer.suffix if existing_value_answer.suffix else "")
                            .replace(",", "")
                        )
                        typed_new_value = existing_value_answer.model_copy()
                        if isinstance(existing_value_answer, IntegerAnswer):
                            cast(IntegerAnswer, typed_new_value).value = int(stripped_value)
                        else:
                            cast(DecimalAnswer, typed_new_value).value = Decimal(stripped_value)

                        org_has_error = existing_value_answer != typed_new_value
                    else:
                        if existing_value_answer.get_value_for_text_export() != new_value.strip():
                            org_has_error = True
                if org_has_error:
                    orgs_with_errors_in_this_column = changed_column_errors.get(orig_col_name, [])
                    orgs_with_errors_in_this_column.append(org.name)
                    changed_column_errors[orig_col_name] = orgs_with_errors_in_this_column

        self.removed_column_errors = removed_column_errors
        self.changed_column_errors = changed_column_errors
        if removed_column_errors or changed_column_errors:
            raise StopValidation("There is a problem")

    def _validate_no_dropped_organisation_items(self, rows: list[dict[str, str]]) -> None:
        if self.collection.is_editable_for_current_status:
            return

        # Only require orgs that are both a) already part of this data set and b) still a current grant recipient.
        # This means a recipient removed from the grant entirely is fine to drop, and a recipient who's never had data
        # in this data set (eg they were added to the grant for a later, still-in-draft report or are a last-minute
        # addition) isn't expected to appear here either as there'd be nothing to drop.
        assert self.existing_datasource is not None
        existing_item_external_ids = {item.external_id for item in self.existing_datasource.organisation_items}
        current_recipient_external_ids = {org.external_id for org in self.all_organisations}
        required_external_ids = existing_item_external_ids.intersection(current_recipient_external_ids)

        new_external_ids = {row.get(DATA_SET_EXTERNAL_ID_COLUMN_HEADER, "").strip() for row in rows}
        missing_external_ids = required_external_ids - new_external_ids
        if not missing_external_ids:
            return

        name_by_external_id = {org.external_id: org.name for org in self.all_organisations}
        missing_names = sorted(
            name_by_external_id.get(external_id, external_id) for external_id in missing_external_ids
        )

        gr = "grant recipient" if len(missing_names) == 1 else "grant recipients"
        quoted_names = [f"‘{name}’" for name in missing_names]
        raise ValidationError(
            f"Missing {gr} {comma_join_items(quoted_names)} in the selected file. "
            f"Make sure the data set includes the {gr} so they can complete the open "
            f"{self.collection.type.constants.singular}"
        )

    def _validate_existing_references(self, existing_datasource: DataSource, fieldnames: list) -> list[str]:
        fieldnames_to_safe_ids = {fieldname: safe_column_id(fieldname) for fieldname in fieldnames}

        errors = []

        for dependent_reference in existing_datasource.depended_on_by_columns:
            if dependent_reference.depends_on_column_name not in fieldnames_to_safe_ids.values():
                column = existing_datasource.schema.root[dependent_reference.depends_on_column_name]  # ty:ignore[unresolved-attribute, invalid-argument-type]
                col_name = column.original_column_name
                q_text = dependent_reference.component.name
                usage_text = "used in "
                if dependent_reference.expression:
                    match dependent_reference.expression.type_:
                        case ExpressionType.VALIDATION:
                            usage_text = "referenced in validation for "
                        case ExpressionType.CONDITION:
                            usage_text += "a condition for "
                        case _:
                            raise ValueError(
                                (
                                    "Reference checking not implemented for expression of type "
                                    + f"{dependent_reference.expression.type_}"
                                )
                            )
                errors.append(
                    (
                        f"Column '{col_name}' is missing from the selected file but is being "
                        + usage_text
                        + f"'{q_text}' Add the column to the file or remove the form reference"
                    )
                )
        return errors

    def _validate_data_for_existing_columns(
        self, datasource: DataSource, fieldnames: list[str], rows: list[dict[str, str]]
    ) -> list[str]:
        safe_ids_to_column_names = {safe_column_id(fieldname): fieldname for fieldname in fieldnames}
        data_errors = []
        for col_safe_id, column_schema in datasource.schema.root.items():  # ty:ignore[unresolved-attribute]
            if col_safe_id in safe_ids_to_column_names.keys():
                column_mapping = DataSetColumnMapping.build_from_data_source_schema_column(column_schema)
                for row in rows:
                    cell_value = row[safe_ids_to_column_names[col_safe_id]]
                    if cell_value and len(cell_value.strip()) > 0:
                        data_errors.extend(
                            validate_csv_cell_against_column_mapping(
                                safe_ids_to_column_names[col_safe_id],
                                cell_value,
                                column_mapping,
                            )
                        )

        errors_to_show = set()
        for bp_error in [e for e in data_errors if (isinstance(e, BritishPoundsError))]:
            errors_to_show.add(
                (
                    f"One or more numbers in column '{bp_error.column}' are not formatted as British pounds to 2 "
                    + "decimal places with the '£' prefix. For example, £100.00"
                )
            )
        for prefix_error in [e for e in data_errors if (isinstance(e, PrefixError))]:
            errors_to_show.add(
                f"One or more numbers in column '{prefix_error.column}' do not match the prefix "
                + f"('{prefix_error.prefix}')"
            )
        for suffix_error in [e for e in data_errors if (isinstance(e, SuffixError))]:
            errors_to_show.add(
                f"One or more numbers in column '{suffix_error.column}' do not match the suffix "
                + f"('{suffix_error.suffix}')"
            )
        for type_error in [
            e for e in data_errors if isinstance(e, DataTypeError) and e.expected_type == NumberTypeEnum.INTEGER
        ]:
            errors_to_show.add(f"One or more numbers in column '{type_error.column}' are not whole numbers")
        for type_error in [
            e for e in data_errors if isinstance(e, DataTypeError) and e.expected_type == NumberTypeEnum.DECIMAL
        ]:
            errors_to_show.add(f"One or more numbers in column '{type_error.column}' are not decimal numbers")
        for decimal_error in [e for e in data_errors if (isinstance(e, DecimalError))]:
            errors_to_show.add(
                f"One or more numbers in column '{decimal_error.column}' has more than "
                + f"{decimal_error.max_decimal_places} decimal places"
            )

        return list(errors_to_show)

    def validate_file(self, field: Field) -> None:
        if not field.data or not hasattr(field.data, "stream"):
            return

        field.data.stream.seek(0)
        try:
            try:
                content = decode_csv_bytes(field.data.stream.read())
            except CSVDecodeError as e:
                raise ValidationError(str(e)) from e
            reader = csv.DictReader(io.StringIO(content))
            fieldnames = [fieldname.strip() for fieldname in reader.fieldnames or []]
            reader.fieldnames = fieldnames
            rows = list(reader)

            UploadDataSetForm._validate_minimum_columns(fieldnames)
            UploadDataSetForm._validate_duplicate_column_names_in_csv(fieldnames)
            UploadDataSetForm._validate_columns_in_each_row(rows)
            UploadDataSetForm._validate_max_rows(rows)
            UploadDataSetForm._validate_min_rows(rows)

            if self.existing_datasource:
                errors = self._validate_data_for_existing_columns(self.existing_datasource, fieldnames, rows)
                errors.extend(self._validate_existing_references(self.existing_datasource, fieldnames))
                if errors:
                    self.data_errors = errors
                    raise ValidationError(errors[0])
                self._validate_no_dropped_organisation_items(rows)
                self._validate_data_for_existing_submissions(self.existing_datasource, rows)
        finally:
            field.data.stream.seek(0)


class ColumnDataTypeMappingForm(FlaskForm):
    # Our template sets the CSRF token for the main MapDataSetColumnsForm, which includes any number of these mini forms
    # for the individual column select fields. We need to set csrf=False for the mini forms so the POST requests don't
    # expect a CSRF for every single one, just the main form.
    class Meta:
        csrf = False

    column_name = HiddenField()
    column_type = SelectField(
        "Select data type",
        choices=[
            ("", "Select data type"),
            ("BRITISH_POUNDS", "British pounds"),
            ("DECIMAL", "Decimal number"),
            ("INTEGER", "Whole number"),
            ("TEXT", "Text"),
        ],
        validators=[],
        widget=GovSelect(),
    )

    def validate_column_type(self, field: Field) -> None:
        if not field.data or field.data == "":
            raise ValidationError(f"Select a data type for {self.column_name.data}")


class MapDataSetColumnsForm(FlaskForm):
    columns = FieldList(FormField(ColumnDataTypeMappingForm))
    submit = SubmitField("Continue", widget=GovSubmitInput())

    def __init__(self, *args: Any, data_columns: list[str], **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.data_columns = data_columns

        # Column name is a hidden field so would be lost on POST but we need them to give nice error messages so need to
        # make sure they persist
        for idx, col in enumerate(data_columns):
            if idx < len(self.columns.entries):
                self.columns.entries[idx].form.column_name.data = col
            else:
                entry = self.columns.append_entry()
                entry.form.column_name.data = col

    def get_column_mappings(self) -> list[DataSetColumnMapping]:
        mappings = []
        for idx, column in enumerate(self.data_columns):
            selected_value = self.columns.entries[idx].form.column_type.data
            match selected_value:
                case "BRITISH_POUNDS":
                    mapping = DataSetColumnMapping(
                        column_name=column,
                        column_type=selected_value,
                    )
                case "TEXT" | "INTEGER" | "DECIMAL" | _:
                    mapping = DataSetColumnMapping(
                        column_name=column,
                        column_type=selected_value,
                    )
            mappings.append(mapping)
        return mappings

    def has_british_pounds_columns(self) -> bool:
        return any(entry.form.column_type.data == "BRITISH_POUNDS" for entry in self.columns.entries)

    def get_numerical_columns(self) -> list[str]:
        return [
            column
            for idx, column in enumerate(self.data_columns)
            if self.columns.entries[idx].form.column_type.data in ["DECIMAL", "INTEGER"]
        ]

    def build_british_pounds_form_errors(
        self,
        column_errors: dict[str, list[CellError]],
    ) -> list[dict[str, list[str]]]:
        columns_error_list: list[dict[str, list[str]]] = []
        for entry in self.columns.entries:
            subform = entry.form
            column_name = subform.column_name.data
            col_errs = column_errors.get(column_name, []) if column_name else []
            subform_errors: dict[str, list[str]] = {}

            for error in col_errs:
                if isinstance(error, BritishPoundsError):
                    message = (
                        f"One or more numbers in column '{column_name}' are not formatted as British pounds to 2 "
                        f"decimal places with the '£' prefix. For example, £100.00"
                    )
                    subform.column_type.errors = list(subform.column_type.errors) + [message]
                    subform_errors.setdefault("column_type", []).append(message)

            columns_error_list.append(subform_errors)

        return columns_error_list


class NumberColumnOptionsForm(FlaskForm):
    class Meta:
        csrf = False

    column_name = HiddenField()
    number_type = HiddenField()
    prefix = StringField(
        "Prefix (optional)",
        widget=GovTextInput(),
        validators=[Optional()],
        filters=[strip_string_if_not_empty, empty_string_to_none],
    )
    suffix = StringField(
        "Suffix (optional)",
        widget=GovTextInput(),
        validators=[Optional()],
        filters=[strip_string_if_not_empty, empty_string_to_none],
    )
    max_decimal_places = IntegerField(
        "Decimal places",
        widget=GovTextInput(),
        validators=[],
    )

    def validate_prefix(self, field: Field) -> None:
        if self.prefix.data and self.suffix.data:
            raise ValidationError(f"Remove the suffix if you need a prefix for {self.column_name.data}")

    def validate_suffix(self, field: Field) -> None:
        if self.prefix.data and self.suffix.data:
            raise ValidationError(f"Remove the prefix if you need a suffix for {self.column_name.data}")

    def update_validators(self) -> None:
        if self.number_type.data == NumberTypeEnum.DECIMAL:
            self.max_decimal_places.validators = [
                DataRequired(f"Enter the maximum number of decimal places for {self.column_name.data}")
            ]
        else:
            self.max_decimal_places.validators = [Optional()]

    def validate(self, extra_validators=None):
        self.update_validators()
        return super().validate(extra_validators=extra_validators)


class NumberColumnFormattingOptions(TypedDict):
    prefix: str | None
    suffix: str | None
    max_decimal_places: int | None


class MapNumberColumnsForm(FlaskForm):
    columns = FieldList(FormField(NumberColumnOptionsForm))
    submit = SubmitField("Continue", widget=GovSubmitInput())

    def __init__(self, *args: Any, numerical_columns: list[DataSetColumnMapping], **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.numerical_columns = numerical_columns
        for idx, col in enumerate(numerical_columns):
            if idx < len(self.columns.entries):
                self.columns.entries[idx].form.column_name.data = col.column_name
                self.columns.entries[idx].form.number_type.data = col.number_type
            else:
                entry = self.columns.append_entry()
                entry.form.column_name.data = col.column_name
                entry.form.number_type.data = col.number_type

    def get_number_column_formatting_options_mappings(self) -> dict[str, NumberColumnFormattingOptions]:
        settings = {}
        for idx, col in enumerate(self.numerical_columns):
            entry = self.columns.entries[idx].form
            settings[col.column_name] = NumberColumnFormattingOptions(
                prefix=entry.prefix.data,
                suffix=entry.suffix.data,
                max_decimal_places=int(entry.max_decimal_places.data) if entry.max_decimal_places.data else None,
            )
        return settings

    def build_number_column_form_errors(
        self,
        column_errors: dict[str, list[CellError]],
    ) -> list[dict[str, list[str]]]:
        columns_error_list: list[dict[str, list[str]]] = []
        for entry in self.columns.entries:
            subform = entry.form
            column_name = subform.column_name.data
            col_errs = column_errors.get(column_name, []) if column_name else []
            subform_errors: dict[str, list[str]] = {}

            for error in col_errs:
                match error:
                    case PrefixError():
                        message = (
                            f"One or more numbers in '{column_name}' do not match the prefix '{subform.prefix.data}'"
                        )
                        subform.prefix.errors = list(subform.prefix.errors) + [message]
                        subform_errors.setdefault("prefix", []).append(message)

                    case SuffixError():
                        message = (
                            f"One or more numbers in '{column_name}' do not match the suffix '{subform.suffix.data}'"
                        )
                        subform.suffix.errors = list(subform.suffix.errors) + [message]
                        subform_errors.setdefault("suffix", []).append(message)

                    case DecimalError():
                        message = (
                            f"One or more numbers in '{column_name}' have more than "
                            f"{subform.max_decimal_places.data} decimal places"
                        )
                        subform.max_decimal_places.errors = list(subform.max_decimal_places.errors) + [message]
                        subform_errors.setdefault("max_decimal_places", []).append(message)

                    case DataTypeError():
                        number_type_label = subform.number_type.data.lower() if subform.number_type.data else "number"
                        message = f"One or more values in '{column_name}' are not a valid {number_type_label}"
                        subform_errors.setdefault("data_type", []).append(message)

                    case _:
                        current_app.logger.error("Invalid data upload form error", error)
                        raise RuntimeError()

            columns_error_list.append(subform_errors)

        return columns_error_list


class SelectConditionCalculationForm(FlaskForm):
    need_calculation = RadioField(
        "Do you need a calculation for the condition?",
        choices=[
            ("yes", "Yes"),
            ("no", "No"),
        ],
        widget=GovRadioInput(),
        validators=[DataRequired("Please select an option")],
    )
    submit = SubmitField("Continue", widget=GovSubmitInput())


class ReopenSubmissionForm(FlaskForm):
    REASON_MAX_WORDS = 200
    reopened_reason = TextAreaField(
        "Why are you reopening this submission?",
        validators=[
            DataRequired("Enter the reason for reopening this submission"),
            WordRange(max_words=REASON_MAX_WORDS, field_display_name="reason for reopening the submission"),
        ],
        widget=GovCharacterCount(),
    )
    submit = SubmitField("Reopen submission", widget=GovSubmitInput())


class RequestOrAllowChangesSubmissionForm(FlaskForm):
    request_changes = RadioField(
        choices=[("yes", "Yes"), ("no", "No, just allow changes")],
        validators=[DataRequired("Please select an option")],
        widget=GovRadioInput(),
    )
    submit = SubmitField("Continue", widget=GovSubmitInput())


class RequestChangesSubmissionForm(FlaskForm):
    REASON_MAX_WORDS = 200
    section_ids = SelectMultipleField(
        "Which sections need to be changed? (optional)",
        choices=[],
        coerce=UUID,
        widget=GovCheckboxesInput(),
        validators=[Optional()],
    )
    changes_requested_reason = TextAreaField(
        validators=[
            DataRequired("Enter the changes needed"),
            WordRange(max_words=REASON_MAX_WORDS, field_display_name="reason for requesting changes to the submission"),
        ],
        widget=GovCharacterCount(),
    )
    submit = SubmitField("Confirm and request changes", widget=GovSubmitInput())

    def __init__(self, *args: Any, submission_helper: SubmissionHelper, **kwargs: Any):
        super().__init__(*args, **kwargs)

        self.section_ids.choices = [
            (str(form.id), form.title)
            for form in submission_helper.collection.forms
            if submission_helper.get_tasklist_status_for_form(form) not in NOT_NEEDED_TASKLIST_SECTION_STATUSES
        ]
        self.changes_requested_reason.label.text = (
            f"What changes are needed to this {submission_helper.long_collection_name}?"
        )


class ApproveOrRejectSubmissionForm(FlaskForm):
    REASON_MAX_WORDS = 200
    is_approved = RadioField(
        choices=[("yes", "Mark as approved"), ("no", "Mark as rejected")],
        validators=[DataRequired("Select an option to mark this submission as approved or rejected")],
        widget=GovRadioInput(),
    )
    approved_reason = TextAreaField(
        "Additional comments (optional)",
        widget=GovCharacterCount(),
    )
    rejected_reason = TextAreaField(
        "Why are you marking this submission as rejected?",
        widget=GovCharacterCount(),
    )
    submit = SubmitField("Confirm and submit decision", widget=GovSubmitInput())

    def validate(self, extra_validators=None):
        if self.is_approved.data == "no":
            self.rejected_reason.validators = [
                DataRequired("Enter the reason for rejecting this submission"),
                WordRange(
                    max_words=self.REASON_MAX_WORDS, field_display_name="The reason for rejecting this submission"
                ),
            ]
        elif self.is_approved.data == "yes":
            self.approved_reason.validators = [
                WordRange(
                    max_words=self.REASON_MAX_WORDS, field_display_name="The reason for approving this submission"
                ),
            ]

        return super().validate(extra_validators=extra_validators)
