"""Module to define pure functions for processing data
Each function takes input data and returns output data, warnings and errors
This makes it easy to test and reason about the code
"""

import ast
import calendar
import json
import logging
import math
import re
import unicodedata
from dataclasses import dataclass
from datetime import datetime
from typing import Any

import dateutil.parser as dateutil
import pytz

from loculus_preprocessing.external_services import TaxonomyService

from .datatypes import (
    AnnotationSourceType,
    FunctionArgs,
    InputData,
    InputMetadata,
    ProcessedMetadataValue,
    ProcessingAnnotation,
    ProcessingResult,
    RawProcessingResult,
    _internal_error_message,
    processing_error,
    raw_internal_error,
)

logger = logging.getLogger(__name__)

options_cache: dict[str, dict[str, str]] = {}


def compute_options_cache(output_field: str, options_list: list[str]) -> dict[str, str]:
    """Create a dictionary mapping option to standardized option. Add dict to the options_cache."""
    options: dict[str, str] = {}
    for option in options_list:
        options[standardize_option(option)] = option
    options_cache[output_field] = options
    return options


def standardize_option(option):
    return " ".join(option.lower().split())


def valid_name() -> str:
    chars = (
        r"\u0041-\u005A"  # A-Z
        r"\u0061-\u007A"  # a-z
        r"\u00C0-\u00D6"  # À-Ö
        r"\u00D8-\u00F6"  # Ø-ö
        r"\u00F8-\u00FF"  # ø-ÿ
        r"\u0100-\u017F"  # Latin Extended-A
        r"\u0180-\u024F"  # Latin Extended-B
    )

    # Ordinal must be a separate "word":
    # - preceded only by start-of-string or whitespace
    # - ends at a word boundary
    ordinal = r"(?<!\S)\d+(?:st|nd|rd|th)\b"
    alpha_or_ord = rf"\s*(?:[{chars}']|{ordinal})"  # First character: letter or ordinal
    name_chars = rf"(?:[{chars}\s.\-']|{ordinal})*"

    return alpha_or_ord + name_chars + r"," + name_chars


def valid_authors(authors: str) -> bool:
    name = valid_name()
    pattern = rf"{name}(;{name})*;?"

    return re.fullmatch(pattern, authors) is not None


def get_invalid_author_names(authors: str) -> list[str]:
    pattern = re.compile(f"^{valid_name()}$")
    invalid = []
    for author in authors.split(";"):
        if author and pattern.fullmatch(author) is None:
            invalid.append(author.strip())
    return invalid


def warn_potentially_invalid_authors(authors: str) -> bool:
    authors_split = re.split(r"[,\s]+", authors)
    return bool(";" not in authors and len(authors_split) > 3)  # noqa: PLR2004


def reformat_authors_from_latin_to_ascii(authors: str) -> str:
    return unicodedata.normalize("NFKD", authors).encode("ascii", "ignore").decode("ascii")


def check_latin_characters(authors: str) -> tuple[list[str], list[str]]:
    warnings: list[str] = []
    errors: list[str] = []
    # Check if all characters in the authors string are Latin letters or spaces
    # (transformable to ASCII)
    for char in authors:
        # If character is already ASCII, skip
        if ord(char) < 128:
            continue
        if char.isalpha() and not (0x0000 <= ord(char) <= 0x024F):
            errors.append(
                f"Unsupported non-Latin character encountered: {char} (U+{ord(char):04X})."
            )
    return (errors, warnings)


def format_authors(authors: str) -> str:
    authors_list = [author for author in authors.split(";") if author]
    loculus_authors = []
    for author in authors_list:
        author_single_white_space = re.sub(r"\s\s+", " ", author)
        last_name, first_name = (
            author_single_white_space.split(",")[0].strip(),
            author.split(",")[1].strip(),
        )
        suffix = ""
        match_suffix = re.search(r"[ .](II|III|IV|V|VI)$", first_name, re.IGNORECASE)
        if match_suffix:
            suffix = match_suffix.group(1).upper()
            first_name = re.sub(r"[ .](II|III|IV|V|VI)$", "", first_name, flags=re.IGNORECASE)
        first_names = []
        for name in first_name.split():
            if len(name) == 1:
                first_names.append(f"{name.upper()}.")
            elif len(name) == 2 and name[1] == ".":
                first_names.append(f"{name.upper()}")
            elif len(name) <= 4 and name.isupper() and "." not in name and not last_name.isupper():
                initials = " ".join(f"{char}." for char in name)
                first_names.append(initials)
            elif re.fullmatch(r"(?:[A-Za-z]\.)+[A-Za-z]+\.?", name):
                letters = [ch.upper() for ch in re.findall(r"[A-Za-z]", name)]
                initials = " ".join(f"{ch}." for ch in letters)
                first_names.append(initials)
            else:
                first_names.append(name)
        formated_first_name = " ".join(first_names)
        if suffix:
            formated_first_name = f"{formated_first_name} {suffix}"
        loculus_authors.append(f"{last_name}, {formated_first_name}")
    return "; ".join(loculus_authors).strip()


def regex_error(
    function_name: str, function_arg: str, input_data: InputMetadata, args: FunctionArgs
) -> RawProcessingResult:
    return raw_internal_error(
        f"{function_name} did not receive a valid {function_arg}, with input {input_data} and args {args}."
    )


@dataclass
class DateRange:
    date_range_lower: datetime
    date_range_upper: datetime
    message: str | None = None

    @property
    def date_range_string(self) -> str:
        return derive_date_range_string(self.date_range_lower, self.date_range_upper)


def convert_to_date_range(date_str: str) -> DateRange | None:
    formats_to_messages = {
        "%Y-%m-%d": None,
        "%Y-%m": "Day is missing. Assuming date is some time in the month.",
        "%Y": "Month and day are missing. Assuming date is some time in the year.",
    }

    for fmt, msg in formats_to_messages.items():
        try:
            parsed_date = datetime.strptime(date_str, fmt).replace(tzinfo=pytz.utc)
        except ValueError:
            continue
        match fmt:
            case "%Y-%m-%d":
                return DateRange(
                    date_range_lower=parsed_date,
                    date_range_upper=parsed_date,
                )
            case "%Y-%m":
                return DateRange(
                    date_range_lower=parsed_date.replace(day=1),
                    date_range_upper=(
                        parsed_date.replace(
                            day=calendar.monthrange(parsed_date.year, parsed_date.month)[1]
                        )
                    ),
                    message=msg,
                )
            case "%Y":
                return DateRange(
                    date_range_lower=parsed_date.replace(month=1, day=1),
                    date_range_upper=parsed_date.replace(month=12, day=31),
                    message=msg,
                )
        break
    return None


def derive_date_range_string(lower: datetime, upper: datetime) -> str:
    """
    Convert a date range into the most compact string representation possible.

    - Single day:           2026-05-28
    - Full month:           2026-05
    - Full year:            2026
    - Multiple full years:  2024/2026 (ISO range)
    - Multiple full months: 2025-01/2026-03 (ISO range)
    - Arbitrary range:      2026-05-10/2026-05-28 (ISO range)
    """
    if lower == upper:
        return lower.strftime("%Y-%m-%d")
    if (
        lower.day == 1
        and upper.day == calendar.monthrange(upper.year, upper.month)[1]
        and lower.month == upper.month
        and lower.year == upper.year
    ):
        return lower.strftime("%Y-%m")
    if (
        lower.month == 1
        and lower.day == 1
        and upper.month == 12  # noqa: PLR2004
        and upper.day == 31  # noqa: PLR2004
        and lower.year == upper.year
    ):
        return lower.strftime("%Y")
    if (
        lower.month == 1
        and lower.day == 1
        and upper.month == 12  # noqa: PLR2004
        and upper.day == 31  # noqa: PLR2004
    ):
        return f"{lower.strftime('%Y')}/{upper.strftime('%Y')}"
    if lower.day == 1 and upper.day == calendar.monthrange(upper.year, upper.month)[1]:
        return f"{lower.strftime('%Y-%m')}/{upper.strftime('%Y-%m')}"
    return f"{lower.strftime('%Y-%m-%d')}/{upper.strftime('%Y-%m-%d')}"


class ProcessingFunctions:
    @classmethod
    def call_function(
        cls,
        function_name: str,
        args: FunctionArgs,
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
    ) -> ProcessingResult:
        if not hasattr(cls, function_name):
            msg = (
                f"CRITICAL: No processing function matches: {function_name}."
                "This is a configuration error."
            )
            raise ValueError(msg)
        func = getattr(cls, function_name)
        try:
            result = func(input_data, output_field, input_fields=input_fields, args=args)
        except Exception as e:
            result = raw_internal_error(
                f"{function_name} raised an unexpected exception for output field '{output_field}': {e}. "
            )
        if isinstance(result, RawProcessingResult):
            return ProcessingResult(
                datum=result.datum,
                warnings=[
                    ProcessingAnnotation.from_fields(
                        input_fields,
                        [output_field],
                        AnnotationSourceType.METADATA,
                        message=msg,
                    )
                    for msg in result.warnings
                ],
                errors=[
                    ProcessingAnnotation.from_fields(
                        input_fields,
                        [output_field],
                        AnnotationSourceType.METADATA,
                        message=msg,
                    )
                    for msg in result.errors
                ],
            )
        if not isinstance(result, ProcessingResult):
            return ProcessingResult(
                datum=None,
                warnings=[],
                errors=[
                    ProcessingAnnotation.from_fields(
                        input_fields,
                        [output_field],
                        AnnotationSourceType.METADATA,
                        message=_internal_error_message(
                            f"{function_name} returned an unexpected type '{type(result).__name__}. "
                        ),
                    )
                ],
            )
        return result

    @staticmethod
    def check_date(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,  # args is essential - even if Pylance says it's not used
    ) -> RawProcessingResult:
        """Check that date is complete YYYY-MM-DD
        If not according to format return error
        If in future, return warning
        Expects input_data to be an ordered dictionary with a single key "date"
        """
        date = input_data["date"]

        if not date:
            return RawProcessingResult()

        warnings: list[str] = []
        try:
            parsed_date = datetime.strptime(date, "%Y-%m-%d").astimezone(pytz.utc)
            if parsed_date > datetime.now(tz=pytz.utc):
                warnings.append("Date is in the future.")
            return RawProcessingResult(datum=date, warnings=warnings)
        except ValueError as e:
            return processing_error(
                f"Date is {date} which is not in the required format YYYY-MM-DD. Parsing error: {e}"
            )

    @staticmethod
    def parse_date_into_range(  # noqa: C901, PLR0912, PLR0915
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,  # args is essential - even if Pylance says it's not used
    ) -> RawProcessingResult:
        """
        Parse date string (`input.date`) with input formats:
        - YYYY
        - YYYY-MM
        - YYYY-MM-DD
        Or date string as a range formatted as:
        - ISO format: date/date (where date is in one of the above formats)
        - lucene format: [date TO date] (where date is in one of the above formats)
        Uses (`input.releaseDate`) as an upper bound when parsing into a range.
        Return value determined FunctionArgs:
        fieldType: "dateRangeString" | "dateRangeLower" | "dateRangeUpper"
        Default fieldType is "dateRangeString" - in the case of a range the date is returned
        as a single month/year or an ISO date range format by function `derive_date_range_string`
        """
        if not args:
            args = {"fieldType": "dateRangeString"}

        input_date_str = input_data["date"]
        release_date_str = input_data.get("releaseDate", "") or ""
        try:
            release_date = dateutil.parse(release_date_str).replace(tzinfo=pytz.utc)
        except Exception:
            release_date = None

        try:
            submitted_at = datetime.fromtimestamp(float(str(args["submittedAt"])), tz=pytz.utc)
        except Exception:
            return raw_internal_error(
                f"parse_into_ranges did not receive a valid submittedAt date, with input {input_data} and args {args}."
            )

        max_upper_limit = min(submitted_at, release_date) if release_date else submitted_at

        if not input_date_str:
            return RawProcessingResult(
                datum=max_upper_limit.strftime("%Y-%m-%d")
                if args["fieldType"] == "dateRangeUpper"
                else None,
            )

        warnings: list[str] = []
        errors: list[str] = []

        datum = convert_to_date_range(input_date_str)

        if datum is None:
            # Try lucene format ("[YYYY-MM-DD TO YYYY-MM-DD]")
            # Try ISO range ("YYYY-MM-DD/YYYY-MM-DD")
            range_patterns = [
                r"^\s*\[([0-9-]+)\s*TO\s*([0-9-]+)\s*\]\s*$",  # Lucene
                r"^\s*([0-9-]+)\s*/\s*([0-9-]+)\s*$",  # ISO-ish
            ]

            for pattern in range_patterns:
                match = re.fullmatch(pattern, input_date_str)
                if not match:
                    continue

                lower_date = convert_to_date_range(match.group(1))
                upper_date = convert_to_date_range(match.group(2))

                if lower_date is None or upper_date is None:
                    return processing_error(
                        f"Metadata field {output_field}: "
                        f"Detected date range but could not parse date: {input_date_str}."
                    )
                msg = None
                if lower_date.message or upper_date.message:
                    msg = "Detected date range."
                    msg += (
                        f" For lower date: {match.group(1)} - {lower_date.message}"
                        if lower_date.message
                        else ""
                    )
                    msg += (
                        f" For upper date: {match.group(2)} - {upper_date.message}"
                        if upper_date.message
                        else ""
                    )

                datum = DateRange(
                    date_range_lower=lower_date.date_range_lower,
                    date_range_upper=upper_date.date_range_upper,
                    message=msg,
                )

        if datum and datum.message:
            warnings.append(f"Metadata field {output_field}:'{input_date_str}' - " + datum.message)

        if datum is None:
            return processing_error(
                f"Metadata field {output_field}: Date {input_date_str} could not be parsed."
            )

        logger.debug(f"parsed_date: {datum}")

        if datum.date_range_lower > datum.date_range_upper:
            logger.debug(
                f"Lower range of date: {datum.date_range_lower} > upper: {datum.date_range_upper}"
            )
            errors.append(
                f"Metadata field {output_field}:'{input_date_str}' is an invalid date "
                f"range. Lower bound: {datum.date_range_lower} is after upper "
                f"bound: {datum.date_range_upper}."
            )

        if datum.date_range_upper > max_upper_limit:
            logger.debug(
                "Tightening upper limit due to release date or current date. "
                f"Original upper limit: {datum.date_range_upper},"
                f"new upper limit: {max_upper_limit}"
            )
            datum.date_range_upper = max_upper_limit

        if datum.date_range_lower and datum.date_range_lower > datetime.now(tz=pytz.utc):
            logger.debug(
                f"Lower range of date: {datum.date_range_lower} > {datetime.now(tz=pytz.utc)}"
            )
            errors.append(f"Metadata field {output_field}:'{input_date_str}' is in the future.")

        if release_date and datum.date_range_lower and (datum.date_range_lower > release_date):
            logger.debug(
                f"Lower range of date: {datum.date_range_lower} > release_date: {release_date}"
            )
            errors.append(
                f"Metadata field {output_field}:'{input_date_str}' is after release date."
            )

        match args["fieldType"]:
            case "dateRangeString":
                return_value = datum.date_range_string
            case "dateRangeLower":
                if datum.date_range_lower is None:
                    return_value = None
                else:
                    return_value = datum.date_range_lower.strftime("%Y-%m-%d")
                warnings = []
                errors = []
            case "dateRangeUpper":
                return_value = datum.date_range_upper.strftime("%Y-%m-%d")
                warnings = []
                errors = []
            case _:
                msg = f"Config error: Unknown fieldType: {args['fieldType']}"
                raise ValueError(msg)

        return RawProcessingResult(datum=return_value, warnings=warnings, errors=errors)

    @staticmethod
    def parse_and_assert_past_date(  # noqa: C901
        input_data: InputMetadata,
        output_field,
        input_fields: list[str],
        args: FunctionArgs,  # args is essential - even if Pylance says it's not used
    ) -> RawProcessingResult:
        """Parse date string. If it's incomplete, add 01-01, if no year, return null and error
        input_data:
            date: str, date string to parse
            release_date: str, optional release date to compare against if None use today
        """
        date_str = input_data["date"]

        if not date_str:
            return RawProcessingResult()
        release_date_str = input_data.get("release_date", "") or ""
        try:
            release_date = dateutil.parse(release_date_str)
        except Exception:
            release_date = None

        formats_to_messages = {
            "%Y-%m-%d": None,
            "%Y-%m": "Day is missing. Assuming the 1st.",
            "%Y": "Month and day are missing. Assuming January 1st.",
        }

        warnings: list[str] = []
        errors: list[str] = []

        for format, message in formats_to_messages.items():
            try:
                parsed_date = datetime.strptime(date_str, format).replace(tzinfo=pytz.utc)
                match format:
                    case "%Y-%m-%d":
                        datum = parsed_date.strftime("%Y-%m-%d")
                    case "%Y-%m":
                        datum = f"{parsed_date.strftime('%Y-%m')}-01"
                    case "%Y":
                        datum = f"{parsed_date.strftime('%Y')}-01-01"

                logger.debug(f"parsed_date: {parsed_date}")

                if message:
                    warnings.append(f"Metadata field {output_field}:'{date_str}' - " + message)

                if parsed_date > datetime.now(tz=pytz.utc):
                    logger.debug(f"parsed_date: {parsed_date} > {datetime.now(tz=pytz.utc)}")
                    errors.append(f"Metadata field {output_field}:'{date_str}' is in the future.")

                if release_date and parsed_date > release_date:
                    logger.debug(f"parsed_date: {parsed_date} > release_date: {release_date}")
                    errors.append(
                        f"Metadata field {output_field}:'{date_str}'is after release date."
                    )

                return RawProcessingResult(datum=datum, warnings=warnings, errors=errors)
            except ValueError:
                continue

        # If all parsing attempts fail, it's an unrecognized format
        return processing_error(f"Metadata field {output_field}: Date format is not recognized.")

    @staticmethod
    def parse_timestamp(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,  # args is essential - even if Pylance says it's not used
    ) -> RawProcessingResult:
        """Parse a timestamp string, e.g. 2022-11-01T00:00:00Z and return a YYYY-MM-DD string"""
        timestamp = input_data["timestamp"]

        if not timestamp:
            return RawProcessingResult()

        try:
            parsed_timestamp = dateutil.parse(timestamp)
            return RawProcessingResult(datum=parsed_timestamp.strftime("%Y-%m-%d"))
        except ValueError as e:
            return processing_error(
                f"Timestamp is {timestamp} which is not in parseable YYYY-MM-DD. Parsing error: {e}"
            )

    @staticmethod
    def concatenate(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        """Concatenates input fields using the "/" separator in the order
        specified by the order argument. Optionally, a 'fallback_value' argument can be provided.
        This should be a string to use in place of metadata that is not available. If fallback_value
        is not provided, the empty string will be used in place of missing metadata.
        """
        warnings: list[str] = []
        errors: list[str] = []

        if not isinstance(args["ACCESSION_VERSION"], str):
            return raw_internal_error(
                f"concatenate did not receive a valid ACCESSION_VERSION (got: {args['ACCESSION_VERSION']!r})."
            )

        accession_version: str = args["ACCESSION_VERSION"]
        order = args["order"]
        field_types = args["type"]
        fallback_value = (
            str(args["fallback_value"]).strip() if args.get("fallback_value") is not None else ""
        )

        if not isinstance(order, list):
            return raw_internal_error(
                f"concatenate expected 'order' to be a list (ACCESSION_VERSION: {accession_version})."
            )

        n_inputs = len(input_data.keys())
        # exclude ACCESSION_VERSION as it's provided by _call_preprocessing_function() and should not be an input_metadata field
        n_expected = len(
            [i for i in order if i != "ACCESSION_VERSION" and not i.startswith("ARG:")]
        )
        if n_inputs < n_expected:
            return raw_internal_error(
                f"concatenate expected {n_expected} input fields but got {n_inputs} (ACCESSION_VERSION: {accession_version})."
            )
        if not isinstance(field_types, list):
            return raw_internal_error(
                f"concatenate expected 'type' to be a list (ACCESSION_VERSION: {accession_version})."
            )

        formatted_input_data: list[str] = []
        try:
            for i in range(len(order)):
                if field_types[i] == "date":
                    processed = ProcessingFunctions.parse_and_assert_past_date(
                        {"date": input_data[order[i]]}, output_field, input_fields, args
                    )
                    formatted_input_data.append(
                        fallback_value
                        if null_per_backend(processed.datum)
                        else str(processed.datum)
                    )
                elif field_types[i] == "dateRangeString":
                    # Assumes date ranges are in ISO format
                    raw = input_data[order[i]]
                    if null_per_backend(raw):
                        formatted_input_data.append(fallback_value)
                        continue
                    raw_value = str(raw).strip()
                    if raw_value.count("/") > 1:
                        date_string = None
                        errors.append(
                            _internal_error_message(
                                f"dateRangeString field '{order[i]}' has an unexpected format: '{raw_value}' (ACCESSION_VERSION: {accession_version})."
                            )
                        )
                    else:
                        date_string = raw_value.replace("/", " TO ")
                    formatted_input_data.append(
                        fallback_value if null_per_backend(date_string) else str(date_string)
                    )
                elif field_types[i] == "timestamp":
                    processed = ProcessingFunctions.parse_timestamp(
                        {"timestamp": input_data[order[i]]}, output_field, input_fields, args
                    )
                    formatted_input_data.append(
                        fallback_value
                        if null_per_backend(processed.datum)
                        else str(processed.datum)
                    )
                elif field_types[i] == "ACCESSION_VERSION":
                    formatted_input_data.append(accession_version)
                elif field_types[i].startswith("ARG:"):
                    if field_types[i][4:] not in args:
                        return raw_internal_error(
                            f"concatenate is missing argument '{field_types[i][4:]}' in args (ACCESSION_VERSION: {accession_version})."
                        )
                    formatted_input_data.append(str(args[field_types[i][4:]]))
                elif order[i] in input_data:
                    formatted_input_data.append(
                        fallback_value
                        if null_per_backend(input_data[order[i]])
                        else str(input_data[order[i]]).strip()
                    )
                else:
                    return raw_internal_error(
                        f"concatenate could not find field '{order[i]}' (type: {field_types[i]}) in input_data (ACCESSION_VERSION: {accession_version})."
                    )

            result = "/".join(formatted_input_data)
            # To avoid downstream issues do not let the result start or end in a "/"
            # Also replace white space with '_'
            result = result.strip("/").replace(" ", "_")

            return RawProcessingResult(datum=result, warnings=warnings, errors=errors)
        except ValueError as e:
            return RawProcessingResult(
                warnings=warnings,
                errors=[
                    *errors,
                    _internal_error_message(
                        f"Concatenation failed for '{output_field}' with error: {e} (ACCESSION_VERSION: {accession_version})."
                    ),
                ],
            )

    @staticmethod
    def check_authors(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        authors = input_data["authors"]

        author_format_description = (
            "Please ensure that "
            "authors are separated by semi-colons. Each author's name should be in the format "
            "'last name, first name;'. Last name(s) is mandatory, a comma is mandatory to "
            "separate first names/initials from last name. Only ASCII alphabetical characters A-Z "
            "are allowed. For example: 'Smith, Anna; Perez, Tom J.; Xu, X.L.;' "
            "or 'Xu,;' if the first name is unknown."
        )

        if not authors:
            return RawProcessingResult()
        errors, warnings = check_latin_characters(authors)
        if errors or warnings:
            return RawProcessingResult(warnings=warnings, errors=errors)

        if valid_authors(authors):
            formatted_authors = format_authors(authors)
            if warn_potentially_invalid_authors(authors):
                warnings.append(
                    "The authors list might not be using the Loculus format. "
                    + author_format_description
                )
            if " and " in authors:
                warnings.append(
                    "Authors list contains 'and'. This may indicate a misformatted "
                    "authors list. Authors should always be separated by semi-colons only e.g. "
                    "`Smith, Anna; Perez, Tom J.; Xu, X.L.`."
                )
            return RawProcessingResult(datum=formatted_authors, warnings=warnings)
        invalid_names = get_invalid_author_names(authors)
        if invalid_names:
            names_to_show = "; ".join(f"'{name}'" for name in invalid_names[:3])
            if len(invalid_names) > 3:  # noqa: PLR2004
                names_to_show += f" ... and {len(invalid_names) - 3} others"
            error_message = f"Invalid name(s): {names_to_show}. " + author_format_description
        else:
            error_message = (
                "The authors list is not in a recognized format. " + author_format_description
            )
        return RawProcessingResult(errors=[error_message], warnings=warnings)

    @staticmethod
    def extract_regex(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        """
        Extracts a substring from the `regex_field` using the provided regex `pattern`
        with a `capture_group`, if `uppercase` is set to true the extracted value is capitalized.
        e.g. ^(?P<segment>[^-]+)-(?P<subtype>[^-]+)$ where segment or subtype could be used
        as a capture_group to extract their respective value from the regex_field.
        """
        regex_field = input_data["regex_field"]

        errors: list[str] = []

        pattern = args.get("pattern")
        capture_group = args.get("capture_group")
        uppercase = args.get("uppercase", False)

        if not regex_field:
            return RawProcessingResult()
        if not isinstance(pattern, str):
            return regex_error("extract_regex", "pattern", input_data, args)
        if not isinstance(capture_group, str):
            return regex_error("extract_regex", "capture_group", input_data, args)
        match = re.match(pattern, regex_field.strip())
        if match:
            try:
                result = match.group(capture_group)
                if result is not None and uppercase:
                    result = result.upper()
                return RawProcessingResult(datum=result)
            except IndexError:
                return raw_internal_error(
                    f"Pattern '{pattern}' does not contain capture group '{capture_group}'."
                )
        else:
            errors.append(
                f"The value '{regex_field}' does not match the expected regex pattern: '{pattern}'."
            )
        return RawProcessingResult(errors=errors)

    @staticmethod
    def check_regex(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        """
        Validates that the field regex_field matches the regex expression.
        If not return error
        """
        regex_field = input_data["regex_field"]

        pattern = args["pattern"]

        if not regex_field:
            return RawProcessingResult()
        if not isinstance(pattern, str):
            return regex_error("check_regex", "pattern", input_data, args)
        regex_field = regex_field.strip()
        if re.match(pattern, regex_field):
            return RawProcessingResult(datum=regex_field)
        return processing_error(
            f"The value '{regex_field}' does not match the expected regex pattern: '{pattern}'."
        )

    @staticmethod
    def identity(  # noqa: C901, PLR0912
        input_data: InputMetadata, output_field: str, input_fields: list[str], args: FunctionArgs
    ) -> RawProcessingResult:
        """Identity function, takes input_data["input"] and returns it as output"""
        if "input" not in input_data:
            return processing_error(f"No data found for output field: {output_field}")
        input_datum = input_data["input"]
        if not input_datum:
            return RawProcessingResult()

        errors: list[str] = []
        output_datum: ProcessedMetadataValue
        if args and "type" in args:
            match args["type"]:
                case "int":
                    try:
                        output_datum = int(input_datum)
                    except ValueError:
                        output_datum = None
                        errors.append(f"Invalid int value: {input_datum} for field {output_field}.")
                case "float":
                    try:
                        output_datum = float(input_datum)
                        if math.isnan(output_datum):
                            output_datum = None
                        elif math.isinf(output_datum):
                            raise ValueError
                    except ValueError:
                        output_datum = None
                        errors.append(
                            f"Invalid float value: {input_datum} for field {output_field}."
                        )
                case "boolean":
                    if input_datum.lower() == "true":
                        output_datum = True
                    elif input_datum.lower() == "false":
                        output_datum = False
                    else:
                        output_datum = None
                        errors.append(
                            f"Invalid boolean value: {input_datum} for field {output_field}."
                        )
                case _:
                    if isinstance(input_datum, str):
                        output_datum = input_datum.strip()
                    else:
                        output_datum = input_datum
        else:
            output_datum = input_datum
        return RawProcessingResult(datum=output_datum, errors=errors)

    @staticmethod
    def process_options(
        input_data: InputMetadata, output_field: str, input_fields: list[str], args: FunctionArgs
    ) -> RawProcessingResult:
        """Checks that option is in options"""
        if "options" not in args or not isinstance(args["options"], list):
            return processing_error(
                "Website configuration error: no options list specified for field "
                f"{output_field}, please contact an administrator."
            )
        input_datum = input_data["input"]
        if not input_datum:
            return RawProcessingResult()

        output_datum: ProcessedMetadataValue
        standardized_input_datum = standardize_option(input_datum)
        if output_field in options_cache:
            options = options_cache[output_field]
        else:
            options = compute_options_cache(output_field, args["options"])
        error_msg = (
            f"Metadata field {output_field}:'{input_datum}' - not in list of accepted options."
        )
        if standardized_input_datum in options:
            output_datum = options[standardized_input_datum]
        # Allow ingested data to include fields not in options
        elif args["is_insdc_ingest_group"]:
            return RawProcessingResult(datum=input_datum, warnings=[error_msg])
        else:
            return processing_error(error_msg)
        return RawProcessingResult(datum=output_datum)

    @staticmethod
    def is_above_threshold(
        input_data: InputMetadata, output_field: str, input_fields: list[str], args: FunctionArgs
    ) -> RawProcessingResult:
        """Flag if input value is above a threshold specified in args"""
        if "threshold" not in args:
            return processing_error(
                f"Field {output_field} is missing threshold argument."
                " Please report this error to the administrator."
            )
        input_datum = input_data["input"]
        if input_datum is None or (isinstance(input_datum, str) and not input_datum.strip()):
            return RawProcessingResult()
        try:
            threshold = float(args["threshold"])  # type: ignore
            input = float(input_datum)
        except (ValueError, TypeError):
            msg = f"Field {output_field} has non-numeric threshold value."
            logger.error(msg)
            return processing_error(msg)
        return RawProcessingResult(datum=(input > threshold))

    @staticmethod
    def is_variant(
        input_data: InputMetadata, output_field: str, input_fields: list[str], args: FunctionArgs
    ) -> RawProcessingResult:
        """Flag if number of mutations is above mutation rate (specified in args) times length"""
        if "mu" not in args:
            return processing_error(
                f"Field {output_field} is missing mu argument."
                " Please report this error to the administrator."
            )
        length_datum = input_data.get("length")
        num_mutations_datum = input_data.get("numMutations")
        if length_datum is None or num_mutations_datum is None:
            return RawProcessingResult()
        try:
            mu = float(args["mu"])  # type: ignore
            length = float(length_datum)
            threshold = mu * length
            is_above_threshold_result = ProcessingFunctions.is_above_threshold(
                input_data={"input": num_mutations_datum},
                output_field=output_field,
                input_fields=input_fields,
                args={"threshold": threshold},
            )
        except (ValueError, TypeError):
            return processing_error(
                f"Field {output_field} has non-numeric length or numMutations value."
            )
        return RawProcessingResult(
            datum=is_above_threshold_result.datum,
            warnings=is_above_threshold_result.warnings,
            errors=is_above_threshold_result.errors,
        )

    @staticmethod
    def assign_custom_lineage(  # noqa: C901
        input_data: InputMetadata, output_field: str, input_fields: list[str], args: FunctionArgs
    ) -> RawProcessingResult:
        """
        Assign flu lineage based on seg4 and seg6.
        Add reassortant flag if different lineages are detected for internal segments,
        add and variant flag if any segment is a variant.
        It expects the following input_data fields to be present:
            - subtype_seg4: the subtype assigned to seg4
            - subtype_seg6: the subtype assigned to seg6
            - reference_seg1,...,reference_seg8: the reference sequence assigned to each segment
            - variant_seg1,...,variant_seg8: boolean flag indicating whether a segment is a variant
        It expects the following args to be present:
            - pattern: regex pattern to extract lineage from reference
                e.g. ^(?P<segment>[^-]+)-(?P<lineage>[^-]+)$
            - uppercase: boolean flag indicating whether to uppercase the extracted lineage
            - capture_group: the name of the capture group in the regex pattern to extract
                e.g. "lineage"
        """
        logger.debug(
            f"Starting custom lineage assignment with input_data: {input_data} and args: {args}"
        )
        if not input_data:
            return RawProcessingResult()
        ha_subtype = input_data.get("subtype_seg4")
        na_subtype = input_data.get("subtype_seg6")
        references: dict[str, str | None] = {}
        extracted_lineages: dict[str, str | None] = {}
        variant: dict[str, bool | None] = {}
        for i in range(1, 9):
            segment = f"seg{i}"
            reference_field = f"reference_seg{i}"
            variant_field = f"variant_seg{i}"
            if reference_field in input_data:
                references[segment] = input_data.get(reference_field)
                variant[segment] = (
                    bool(input_data.get(variant_field)) if variant_field in input_data else None
                )
        try:
            for i in range(1, 9):
                segment = f"seg{i}"
                extracted_lineages[segment] = ProcessingFunctions.call_function(  # type: ignore
                    "extract_regex",
                    {
                        "pattern": args["pattern"],
                        "uppercase": args["uppercase"],
                        "capture_group": args["capture_group"],
                    },
                    {"regex_field": references.get(segment, "")},
                    "output_field",
                    ["segment_name"],
                ).datum
            logger.debug(f"Extracted lineages: {extracted_lineages} from references: {references}")
            if not ha_subtype or not na_subtype:
                return RawProcessingResult()
            lineage = f"{ha_subtype}{na_subtype}"
            if (
                extracted_lineages.get("seg4") == "H1N1PDM"
                and extracted_lineages.get("seg6") == "H1N1PDM"
            ):
                lineage = "H1N1pdm"
            logger.debug(
                f"Determined preliminary lineage {lineage} based on segments seg4 and seg6"
            )
            if lineage in {"H1N1", "H3N2", "H2N2", "H1N1pdm"}:
                logger.debug(
                    f"Lineage {lineage} is a human lineage, checking for reassortment and variants"
                )
                # only assign human lineages
                if len({v for v in extracted_lineages.values() if v is not None}) > 1:
                    lineage += " reassortant"
                if any(v for v in variant.values() if v):
                    lineage += " (variant)"
                return RawProcessingResult(datum=lineage)
        except (ValueError, TypeError):
            return processing_error(
                f"Internal error processing custom lineage for field {output_field}."
            )
        return RawProcessingResult()

    @staticmethod
    def build_display_name(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        """Builds a displayName from input_fields. The identifier field in the displayName is based
        on specimenCollectorSampleId or - if it is not set - submissionId (direct submissions only).

        This method wraps ProcessingFunctions.concatenate(). Thus, it has the same required input
        args, as well as adding some additional checks and requirements:
            - submissionId and specimenCollectorSampleId must be in the input_data
            - IDENTIFIER keyword must be in args['order'] and args['type']
            - the IDENTIFIER is resolved by trying specimenCollectorSampleId first, then
              submissionId; if neither yields a usable value it is replaced with ACCESSION_VERSION
            - if fallback_value is not in args, { 'fallback_value': 'unknown' } is added to the args
              before passing them on to concatenate()
            - for sequences ingested from INSDC, we do not try to parse the IDENTIFIER field using
              regex. We will use the Isolate Name as IDENTIFIER field if it contains no slashes or
              spaces (otherwise we fall back to ACCESSION_VERSION)
            - if regex_pattern is provided, human_readable_pattern must also be provided. It is a
              submitter-friendly rendering of the pattern to show in error messages
              (e.g. '<any>/<any>/<identifier>/<date>')
        """
        collector_id = input_data.get("specimenCollectorSampleId", None)
        submission_id = input_data.get("submissionId", None)
        warnings: list[str] = []

        order = args.get("order")
        field_types = args.get("type")
        if (
            not isinstance(order, list)
            or not isinstance(field_types, list)
            or len(order) != len(field_types)
            or "IDENTIFIER" not in order
        ):
            return raw_internal_error(
                "'order' and 'type' must be lists of equal length and 'order' must contain IDENTIFIER."
            )

        regex_pattern = args.get("regex_pattern")
        regex_pattern = str(regex_pattern) if regex_pattern is not None else None
        if regex_pattern is not None and "identifier" not in re.compile(regex_pattern).groupindex:
            return raw_internal_error(
                "If provided, 'regex_pattern' must contain a named capture group called 'identifier'."
            )

        human_readable_pattern = args.get("human_readable_pattern")
        human_readable_pattern = (
            str(human_readable_pattern) if human_readable_pattern is not None else None
        )
        if regex_pattern is not None and human_readable_pattern is None:
            return raw_internal_error(
                "If 'regex_pattern' is provided, 'human_readable_pattern' must also be provided."
            )

        concatenate_order = order.copy()
        concatenate_field_types = field_types.copy()

        insdc_ingested = bool(args["is_insdc_ingest_group"])

        # Try to parse the specimenCollectorSampleId first
        identifier = parse_identifier_string(collector_id, insdc_ingested, regex_pattern)
        if identifier is None and not insdc_ingested:
            # For direct submissions only: try to parse the submissionId
            # Don't do this for ingested since there the submissionId is just the
            # (concatenation of) nuccore accession(s) of the sequence(s)
            identifier = parse_identifier_string(submission_id, insdc_ingested, regex_pattern)

        def replace_identifier(values, replacement):
            return [replacement if v == "IDENTIFIER" else v for v in values]

        if identifier is not None:
            # We were able to parse an IDENTIFIER, treat it as a string
            concatenate_field_types = replace_identifier(field_types, "string")
            input_data["IDENTIFIER"] = identifier
        else:
            # Unable to parse specimenCollectorSampleId and submissionID, use ACCESSION_VERSION
            if not insdc_ingested and regex_pattern is not None:
                warnings.append(
                    f"specimenCollectorSampleId and submissionId could not be parsed, using "
                    f"ACCESSION_VERSION in displayName instead. To include your own identifier, "
                    f"remove whitespace and '/' characters or use the format '{human_readable_pattern}' and we will parse the `identifier` from the submission."
                )
            concatenate_order = replace_identifier(order, "ACCESSION_VERSION")
            concatenate_field_types = replace_identifier(field_types, "ACCESSION_VERSION")

        new_args = args.copy()
        new_args.update(
            {
                "order": concatenate_order,
                "type": concatenate_field_types,
                "fallback_value": args.get("fallback_value", "unknown"),
                "ACCESSION_VERSION": args["ACCESSION_VERSION"],
            }
        )

        concat_result = ProcessingFunctions.concatenate(
            input_data,
            output_field,
            input_fields,
            new_args,
        )

        return RawProcessingResult(
            datum=concat_result.datum,
            warnings=warnings + concat_result.warnings,
            errors=concat_result.errors,
        )

    @staticmethod
    def resolve_host_taxon_id(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        """Validates that the host exists
        in NCBI's taxonomy. Checks either the hostTaxonId or the
        hostNameScientific, depending on the input.

        If validation succeeds, returns the tax_id of the associated taxon.

        It is possible that multiple taxa have the same scientific name. In these cases,
        return the tax_id of the most generic taxon (i.e., the one that's closest to
        the root of the taxonomy)
        """
        unvalidated_host = input_data.get("host")
        if not unvalidated_host:
            return RawProcessingResult()

        taxonomy_service: TaxonomyService = args["taxonomy_service"]  # type: ignore
        return taxonomy_service.get_tax_id(
            unvalidated_host, not bool(args["is_insdc_ingest_group"])
        )

    @staticmethod
    def scientific_name_from_id(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        tax_id: str | None = input_data.get("hostTaxonId")
        if not tax_id:
            return RawProcessingResult()

        taxonomy_service: TaxonomyService = args["taxonomy_service"]  # type: ignore
        return taxonomy_service.get_scientific_name(tax_id, not bool(args["is_insdc_ingest_group"]))

    @staticmethod
    def common_name_from_id(
        input_data: InputMetadata,
        output_field: str,
        input_fields: list[str],
        args: FunctionArgs,
    ) -> RawProcessingResult:
        tax_id: str | None = input_data.get("hostTaxonId")
        if not tax_id:
            return RawProcessingResult()

        taxonomy_service: TaxonomyService = args["taxonomy_service"]  # type: ignore
        return taxonomy_service.get_common_name(tax_id)


def single_metadata_annotation(
    source_name: str,
    message: str,
) -> list[ProcessingAnnotation]:
    return [
        ProcessingAnnotation.from_single(
            source_name,
            AnnotationSourceType.METADATA,
            message=message,
        )
    ]


def process_frameshifts(input: str | None) -> InputData:
    """Converts frameshift string to InputData for processing"""
    try:
        return InputData(datum=format_frameshift(input))
    except Exception as e:
        return InputData(
            datum=None,
            errors=single_metadata_annotation(
                "frameshifts",
                _internal_error_message(f"Frameshift formatting failed: {e}. "),
            ),
        )


def format_frameshift(input: str | None) -> str | None:
    """
    In nextclade frameshifts have the json format:
    [{
          "cdsName": "GPC",
          "nucRel": {
            "begin": 5,
            "end": 20
          },
          "nucAbs": [
            {
              "begin": 97,
              "end": 112
            }
          ],
          "codon": {
            "begin": 2,
            "end": 7
          },
          "gapsLeading": {
            "begin": 1,
            "end": 2
          },
          "gapsTrailing": {
            "begin": 7,
            "end": 8
          }
        },...
    ]

    This function:
    * converts this json object to a comma separated list of frameshift elements:
    cdsName:codon.begin-codon.end(nucAbs.begin-nucAbs.end;...)
    * Additionally, if there is only one element in this range only output the first element
    e.g. cdsName:codon.begin(nucAbs.begin)
    * Converts frameshift positions from index-0 to index-1 (this aligns with other metrics)
    * Makes the range [] have an inclusive start and inclusive end
    (the default in nextclade is exclusive end)
    """
    if input is None:
        return None
    if input == "[]":
        return ""

    def range_string(_start: str | int, _end: str | int) -> str:
        """Converts 0-indexed exclusive range to 1-indexed inclusive range string"""
        start = int(_start) + 1
        end = int(_end)
        if end > start:
            return f"{start}-{end}"
        return str(start)

    frame_shifts = json.loads(
        input.replace("'", '"')
    )  # Required for json.loads to recognize input as json string and convert to dict
    frame_shift_strings = []
    for frame_shift in frame_shifts:
        nuc_range_list = [range_string(nuc["begin"], nuc["end"]) for nuc in frame_shift["nucAbs"]]
        codon_range = range_string(frame_shift["codon"]["begin"], frame_shift["codon"]["end"])
        frame_shift_strings.append(
            frame_shift["cdsName"] + f":{codon_range}(nt:" + ";".join(nuc_range_list) + ")"
        )
    return ",".join(frame_shift_strings)


def process_stop_codons(input: str | None) -> InputData:
    """Converts stop codon string to InputData for processing"""
    try:
        return InputData(datum=format_stop_codon(input))
    except Exception as e:
        return InputData(
            datum=None,
            errors=single_metadata_annotation(
                "stopCodons",
                _internal_error_message(f"Stop codon formatting failed: {e}. "),
            ),
        )


def format_stop_codon(result: str | None) -> str | None:
    """
    In nextclade stop codons have the json format:
    [   {
            cdsName: String,
            codon: usize,
        },...
    ]

    This function:
    * converts this to a comma-separated list of strings: cdsName:codon
    * Converts stop codon positions from index-0 to index-1 (this aligns with other metrics)
    """
    if result is None:
        return None
    if result == "[]":
        return ""
    result = result.replace("'", '"')
    stop_codons = json.loads(result)
    stop_codon_strings = []
    for stop_codon in stop_codons:
        stop_codon_string = f"{stop_codon['cdsName']}:{stop_codon['codon'] + 1}"
        stop_codon_strings.append(stop_codon_string)
    return ",".join(stop_codon_strings)


def _format_aa_substitution(mutation: dict) -> str:
    return f"{mutation['cdsName']}:{mutation['refAa']}{int(mutation['pos']) + 1}{mutation['qryAa']}"


def process_mutations_from_clade_founder(input: str | None, args: FunctionArgs | None) -> InputData:
    """Processes substitutions from clade founder InputData for processing"""
    if input is None:
        return InputData(datum=None)
    try:
        data = ast.literal_eval(input)
        mutations = []
        for value in data.values():
            if not value.get("privateSubstitutions"):
                continue
            for mutation in value["privateSubstitutions"]:
                substitution = _format_aa_substitution(mutation)
                mutations.append(substitution)
        if mutations:
            return InputData(datum=" ".join(mutations))
    except Exception as e:
        return InputData(
            datum=None,
            errors=single_metadata_annotation(
                "mutationsFromCladeFounder",
                _internal_error_message(f"Clade founder mutation processing failed: {e}. "),
            ),
        )
    return InputData(datum=None)


def process_labeled_mutations(input: str | None, args: FunctionArgs | None) -> InputData:
    """Processes labeledMutations from privateMutations list for processing"""
    if input is None:
        return InputData(datum=None)
    try:
        data = ast.literal_eval(input)
        mutations = []
        for value in data.values():
            if not value.get("labeledSubstitutions"):
                continue
            for labeled_mutation in value["labeledSubstitutions"]:
                if not labeled_mutation.get("substitution"):
                    continue
                mutation = labeled_mutation["substitution"]
                substitution = _format_aa_substitution(mutation)
                mutations.append(substitution)
        if mutations:
            return InputData(datum=" ".join(mutations))
    except Exception as e:
        msg = _internal_error_message(f"Labeled mutation processing failed: {e}. ")
        return InputData(
            datum=None,
            errors=single_metadata_annotation(
                "labeledMutations",
                msg,
            ),
        )
    return InputData(datum=None)


def process_phenotype_values(input: str | None, args: FunctionArgs | None) -> InputData:
    """Processes phenotype values string to InputData for processing"""
    if input is None:
        return InputData(datum=None)
    name = args.get("name", "") if args else ""
    try:
        data = ast.literal_eval(input)
        for entry in data:
            if entry.get("name") == name:
                value = entry.get("value")
                return InputData(datum=str(value) if value is not None else None)
    except Exception as e:
        msg = _internal_error_message(f"Phenotype value processing failed: {e}. ")
        return InputData(
            datum=None,
            errors=single_metadata_annotation(
                "phenotypeValues",
                msg,
            ),
        )
    return InputData(datum=None)


def parse_identifier_string(
    input: ProcessedMetadataValue, insdc_ingested: bool, regex_pattern: str | None = None
) -> str | None:
    """Return an IDENTIFIER string to use in the displayName or None if `input` cannot be used
    as an identifier.
    """
    if not isinstance(input, str) or not input.strip():
        return None
    has_forbidden_char = any(c.isspace() for c in input) or "/" in input

    if insdc_ingested:
        # For INSDC ingested sequences: use the value as-is unless it contains whitespace or '/'
        # Don't attempt to parse these as the format on INSDC isolate names is very inconsistent
        return None if has_forbidden_char else input

    if not has_forbidden_char:
        # Direct submission without forbidden_char: use the value as-is, no regex parsing
        return input

    # Direct submission containing forbidden_char: attempt regex extraction of identifier field
    if regex_pattern is None:
        return None
    extract_result = ProcessingFunctions.extract_regex(
        input_data={"regex_field": input},
        output_field="IDENTIFIER",
        input_fields=[],
        args={"pattern": regex_pattern, "capture_group": "identifier"},
    )
    return None if extract_result.datum is None else str(extract_result.datum)


def trim_ns(sequence: str) -> str:
    """
    Trims 'N' and 'n' characters from the start and end of a nucleotide sequence.

    Args:
        sequence (str): The nucleotide sequence to process.

    Returns:
        str: The trimmed sequence.
    """
    return sequence.strip("Nn")


def null_per_backend(x: Any) -> bool:
    match x:
        case None:
            return True
        case "":
            return True
        case _:
            return False
