"""Base classes and core functionality for pydantic-settings sources."""

from __future__ import annotations as _annotations

import json
from abc import ABC, abstractmethod
from collections.abc import Callable, Hashable, Mapping, Sequence
from dataclasses import asdict, is_dataclass
from dataclasses import fields as dataclass_fields
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast, get_args, get_type_hints

from pydantic import AliasChoices, AliasPath, TypeAdapter
from pydantic._internal._typing_extra import (  # type: ignore[attr-defined]
    get_origin,
)
from pydantic._internal._utils import deep_update, is_model_class
from pydantic.dataclasses import is_pydantic_dataclass
from pydantic.fields import FieldInfo
from typing_inspection.introspection import is_union_origin

from ..exceptions import SettingsError
from .types import (
    ConfigFileSourceType,
    EnvNoneType,
    EnvPrefixTarget,
    ForceDecode,
    NoDecode,
    PydanticModel,
    _CliSubCommand,
)
from .utils import (
    InitState,
    _annotation_is_complex,
    _get_alias_names,
    _get_field_annotation_and_metadata,
    _get_field_metadata,
    _get_model_config,
    _get_model_fields,
    _strip_annotated,
    _union_is_complex,
    _validate_by_alias_enabled,
    _validate_by_name_enabled,
    _warn_if_field_info_incomplete,
)

if TYPE_CHECKING:
    from pydantic_settings.main import BaseSettings

    from .types import Traversable


def get_subcommand(
    model: PydanticModel,
    is_required: bool = True,
    cli_exit_on_error: bool | None = None,
    _suppress_errors: list[SettingsError | SystemExit] | None = None,
) -> PydanticModel | None:
    """
    Get the subcommand from a model.

    Args:
        model: The model to get the subcommand from.
        is_required: Determines whether a model must have subcommand set and raises error if not
            found. Defaults to `True`.
        cli_exit_on_error: Determines whether this function exits with error if no subcommand is found.
            Defaults to model_config `cli_exit_on_error` value if set. Otherwise, defaults to `True`.

    Returns:
        The subcommand model if found, otherwise `None`.

    Raises:
        SystemExit: When no subcommand is found and is_required=`True` and cli_exit_on_error=`True`
            (the default).
        SettingsError: When no subcommand is found and is_required=`True` and
            cli_exit_on_error=`False`.
    """

    model_cls = type(model)
    if cli_exit_on_error is None and is_model_class(model_cls):
        model_default = model_cls.model_config.get('cli_exit_on_error')
        if isinstance(model_default, bool):
            cli_exit_on_error = model_default
    if cli_exit_on_error is None:
        cli_exit_on_error = True

    subcommands: list[str] = []
    for field_name, field_info in _get_model_fields(model_cls).items():
        if _CliSubCommand in field_info.metadata:
            if getattr(model, field_name) is not None:
                return getattr(model, field_name)
            subcommands.append(field_name)
        # TODO: we could warn if no sub command was found and one of `field_info`s
        # isn't complete.

    if is_required:
        error_message = (
            f'Error: CLI subcommand is required {{{", ".join(subcommands)}}}'
            if subcommands
            else 'Error: CLI subcommand is required but no subcommands were found.'
        )
        err = SystemExit(error_message) if cli_exit_on_error else SettingsError(error_message)
        if _suppress_errors is None:
            raise err
        _suppress_errors.append(err)

    return None


class PydanticBaseSettingsSource(ABC):
    """
    Abstract base class for settings sources, every settings source classes should inherit from it.
    """

    def __init__(self, settings_cls: type[BaseSettings], _init_state: InitState | None = None):
        self._init_state: InitState = {} if _init_state is None else _init_state
        self.settings_cls = settings_cls
        self.config = settings_cls.model_config
        self._current_state: dict[str, Any] = {}
        self._settings_sources_data: dict[str, dict[str, Any]] = {}

    def _set_current_state(self, state: dict[str, Any]) -> None:
        """
        Record the state of settings from the previous settings sources. This should
        be called right before __call__.
        """
        self._current_state = state

    def _set_settings_sources_data(self, states: dict[str, dict[str, Any]]) -> None:
        """
        Record the state of settings from all previous settings sources. This should
        be called right before __call__.
        """
        self._settings_sources_data = states

    @property
    def current_state(self) -> dict[str, Any]:
        """
        The current state of the settings, populated by the previous settings sources.
        """
        return self._current_state

    @property
    def settings_sources_data(self) -> dict[str, dict[str, Any]]:
        """
        The state of all previous settings sources.
        """
        return self._settings_sources_data

    @abstractmethod
    def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
        """
        Gets the value, the key for model creation, and a flag to determine whether value is complex.

        This is an abstract method that should be overridden in every settings source classes.

        Args:
            field: The field.
            field_name: The field name.

        Returns:
            A tuple that contains the value, key and a flag to determine whether value is complex.
        """

    def field_is_complex(self, field: FieldInfo) -> bool:
        """
        Checks whether a field is complex, in which case it will attempt to be parsed as JSON.

        Args:
            field: The field.

        Returns:
            Whether the field is complex.
        """
        return _annotation_is_complex(field.annotation, field.metadata, self._init_state)

    def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool) -> Any:
        """
        Prepares the value of a field.

        Args:
            field_name: The field name.
            field: The field.
            value: The value of the field that has to be prepared.
            value_is_complex: A flag to determine whether value is complex.

        Returns:
            The prepared value.
        """
        if value is not None and (self.field_is_complex(field) or value_is_complex):
            return self.decode_complex_value(field_name, field, value)
        return value

    def decode_complex_value(self, field_name: str, field: FieldInfo, value: Any) -> Any:
        """
        Decode the value for a complex field

        Args:
            field_name: The field name.
            field: The field.
            value: The value of the field that has to be prepared.

        Returns:
            The decoded value for further preparation
        """
        metadata = _get_field_metadata(field) if field else []
        if field and (
            NoDecode in metadata or (self.config.get('enable_decoding') is False and ForceDecode not in metadata)
        ):
            return value

        return json.loads(value)

    @abstractmethod
    def __call__(self) -> dict[str, Any]:
        pass


class ConfigFileSourceMixin(ABC):
    @staticmethod
    def _existing_files(files: ConfigFileSourceType | None) -> list[Path | Traversable]:
        """Return the given files that exist, with `~` expanded in native paths."""
        if files is None:
            return []
        if not isinstance(files, Sequence) or isinstance(files, str):
            files = [files]
        existing: list[Path | Traversable] = []
        for file in files:
            file_path: Path | Traversable = Path(file) if isinstance(file, str) else file
            if isinstance(file_path, Path):
                file_path = file_path.expanduser()
            if file_path.is_file():
                existing.append(file_path)
        return existing

    def _read_files(self, files: ConfigFileSourceType | None, deep_merge: bool = False) -> dict[str, Any]:
        vars: dict[str, Any] = {}
        for file_path in self._existing_files(files):
            updating_vars = self._read_file(file_path)
            if deep_merge:
                vars = deep_update(vars, updating_vars)
            else:
                vars.update(updating_vars)
        return vars

    @abstractmethod
    def _read_file(self, path: Path | Traversable) -> dict[str, Any]:
        pass


def _unwrap_optional_annotation(annotation: Any) -> Any:
    """Return the inner type of an ``Optional[T]`` annotation, otherwise the annotation unchanged."""
    if is_union_origin(get_origin(annotation)):
        args = get_args(annotation)
        if len(args) == 2 and type(None) in args:
            for arg in args:  # pragma: no branch
                if arg is not type(None):
                    return arg
    return annotation


def _is_model_or_dataclass(annotation: Any) -> bool:
    """Whether the annotation is a pydantic model or a pydantic dataclass, i.e. a type with nested fields."""
    return is_model_class(annotation) or is_pydantic_dataclass(annotation)


def _get_discriminator(field_info: FieldInfo) -> str | Callable[[Any], Hashable] | None:
    """Get the discriminator of a field using a discriminated union (via Annotated or Field(discriminator=...))."""
    from pydantic import Discriminator

    discriminator = field_info.discriminator
    if discriminator is None:
        discriminator = next((m for m in field_info.metadata if isinstance(m, Discriminator)), None)
    return discriminator.discriminator if isinstance(discriminator, Discriminator) else discriminator


def _get_member_field(model_cls: type[Any], name: str) -> FieldInfo | None:
    """Get a field of a discriminated union member: a model, a pydantic dataclass or a stdlib dataclass.

    Returns ``None`` if a stdlib dataclass field's annotation can't be resolved, e.g. when it names a class that
    is only in a local scope, so a `Field()` in its `Annotated` metadata is then not found.
    """
    if _is_model_or_dataclass(model_cls):
        return _get_model_fields(model_cls).get(name)
    # Pydantic doesn't attach `__pydantic_fields__` to a stdlib dataclass, so find the `FieldInfo` itself:
    # `Field()` is kept as the dataclass field's default, or in its `Annotated` metadata.
    dc_field = next((f for f in dataclass_fields(model_cls) if f.name == name), None)
    if dc_field is None:
        return None
    if isinstance(dc_field.default, FieldInfo):
        return dc_field.default
    # Resolve this field's annotation alone, as another field may name a class that is only in a local scope.
    # `typing` can't evaluate a single annotation, so it is put on a class created in the module of the class
    # declaring the field, which may be a base in another module.
    try:
        # On Python 3.14+ reading `__annotations__` evaluates them, so it can raise `NameError` too
        owner = next((c for c in model_cls.__mro__ if name in getattr(c, '__annotations__', {})), model_cls)
        annotations_holder = type(
            owner.__name__, (), {'__module__': owner.__module__, '__annotations__': {name: dc_field.type}}
        )
        hint = get_type_hints(annotations_holder, localns=dict(vars(owner)), include_extras=True)[name]
    except NameError:
        return None
    return next((m for m in get_args(hint)[1:] if isinstance(m, FieldInfo)), None)


def _get_discriminator_keys(model_cls: type[Any], discriminator: str) -> tuple[str, ...]:
    """Get the keys that can hold the discriminator value of a discriminated union member: its name and alias."""
    alias = getattr(_get_member_field(model_cls, discriminator), 'validation_alias', None)
    return (discriminator, alias) if isinstance(alias, str) else (discriminator,)


def _get_nested_discriminators(annotation: Any) -> list[str] | None:
    """Get the discriminators nested inside a discriminated union's annotation.

    A member can be guarded by more than one discriminator, e.g. in
    ``Annotated[Annotated[BlackCat | WhiteCat, Discriminator('color')] | Dog, Discriminator('pet_type')]``
    reaching ``BlackCat`` requires both ``pet_type='cat'`` and ``color='black'``. Each one has to keep the
    default's value for its default to still apply.

    A nested discriminator can be given as ``Discriminator(...)`` or ``Field(discriminator=...)``.

    Returns ``None`` if a nested discriminator can't be resolved to a field name (i.e. it is callable), as
    the member it selects can't be looked up in the incoming value.
    """
    from pydantic import Discriminator

    discriminators: list[str] = []
    for arg in get_args(annotation):
        if arg is type(None):
            continue
        for metadata in get_args(arg)[1:]:
            # A nested discriminator is set either with `Discriminator(...)` or `Field(discriminator=...)`
            if isinstance(metadata, FieldInfo):
                discriminator = _get_discriminator(metadata)
            elif isinstance(metadata, Discriminator):
                discriminator = metadata.discriminator
            else:
                continue
            if discriminator is None:
                continue
            if not isinstance(discriminator, str):
                return None
            discriminators.append(discriminator)
        nested = _get_nested_discriminators(arg)
        if nested is None:
            return None
        discriminators.extend(nested)
    return discriminators


class DefaultSettingsSource(PydanticBaseSettingsSource):
    """
    Source class for loading default object values.

    Args:
        settings_cls: The Settings class.
        nested_model_default_partial_update: Whether to allow partial updates on nested model default object fields.
            Defaults to `False`.
    """

    def __init__(
        self,
        settings_cls: type[BaseSettings],
        nested_model_default_partial_update: bool | None = None,
        _init_state: InitState | None = None,
    ):
        super().__init__(settings_cls, _init_state)
        self.defaults: dict[str, Any] = {}
        # Discriminated union fields: per discriminator guarding the default's member, the keys that can hold
        # its value, and the default's value
        self._discriminator_tags: dict[str, list[tuple[tuple[str, ...], Any]]] = {}
        self.nested_model_default_partial_update = (
            nested_model_default_partial_update
            if nested_model_default_partial_update is not None
            else self.config.get('nested_model_default_partial_update', False)
        )
        if self.nested_model_default_partial_update:
            validate_by_alias = _validate_by_alias_enabled(self.config)
            for field_name, field_info in settings_cls.model_fields.items():
                _warn_if_field_info_incomplete(field_info, field_name, self._init_state)
                discriminator = _get_discriminator(field_info)
                if callable(discriminator):
                    # The member selected by a callable discriminator can't be looked up in the incoming value
                    continue
                nested = _get_nested_discriminators(field_info.annotation) if discriminator is not None else []
                if nested is None:
                    # Same for a nested callable discriminator
                    continue
                alias_names, *_ = _get_alias_names(field_name, field_info, validate_by_alias=validate_by_alias)
                preferred_alias = alias_names[0]
                if is_dataclass(type(field_info.default)):
                    self.defaults[preferred_alias] = asdict(field_info.default)
                elif is_model_class(type(field_info.default)):
                    self.defaults[preferred_alias] = field_info.default.model_dump()
                if discriminator is not None and preferred_alias in self.defaults:
                    default_cls = type(field_info.default)
                    self._discriminator_tags[preferred_alias] = [
                        (_get_discriminator_keys(default_cls, name), getattr(field_info.default, name, None))
                        for name in (discriminator, *nested)
                    ]

    def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
        # Nothing to do here. Only implement the return statement to make mypy happy
        return None, '', False

    def __call__(self) -> dict[str, Any]:
        defaults = self.defaults.copy()
        for alias, discriminators in self._discriminator_tags.items():
            value = self.current_state.get(alias)
            if not isinstance(value, dict):
                continue
            # Pydantic reads each tag from the discriminator field's name or its alias, so a value under any of
            # those keys selects a member. A key that is absent leaves the default's member in place.
            # The default's fields don't apply when any discriminator selects another member.
            if any(value[key] != default_tag for keys, default_tag in discriminators for key in keys if key in value):
                del defaults[alias]
        return defaults

    def __repr__(self) -> str:
        return (
            f'{self.__class__.__name__}(nested_model_default_partial_update={self.nested_model_default_partial_update})'
        )


class InitSettingsSource(PydanticBaseSettingsSource):
    """
    Source class for loading values provided during settings class initialization.
    """

    def __init__(
        self,
        settings_cls: type[BaseSettings],
        init_kwargs: dict[str, Any],
        nested_model_default_partial_update: bool | None = None,
        _init_state: InitState | None = None,
    ):
        super().__init__(settings_cls, _init_state)
        case_sensitive = self.config.get('case_sensitive', False)
        include_name = _validate_by_name_enabled(self.config)
        validate_by_alias = _validate_by_alias_enabled(self.config)

        def normalize(name: str) -> str:
            # When case_sensitive is False, matching is done on lower-cased names.
            return name if case_sensitive else name.lower()

        self.init_kwargs = {}
        init_kwarg_names = set(init_kwargs.keys())
        # Map each normalized name to all provided keys sharing it, so field aliases can
        # be matched without losing the original key used to look up the value. A 1-to-many
        # mapping is required because case_sensitive=False can collapse distinct keys
        # (e.g. ``TeSt`` and ``TEST``) onto the same normalized name.
        normalized_provided: dict[str, list[str]] = {}
        for key in init_kwargs:
            normalized_provided.setdefault(normalize(key), []).append(key)
        for field_name, field_info in settings_cls.model_fields.items():
            _warn_if_field_info_incomplete(field_info, field_name, self._init_state)
            # Canonical (case-preserving) alias names determine the output key, so the value
            # still matches pydantic's validation aliases, which are always case-sensitive.
            canonical_alias_names, *_ = _get_alias_names(field_name, field_info, validate_by_alias=validate_by_alias)
            # Match alias names respect case_sensitive when matching against provided keys.
            match_alias_names, *_ = _get_alias_names(
                field_name, field_info, case_sensitive=case_sensitive, validate_by_alias=validate_by_alias
            )
            # When populate_by_name is True, allow using the field name as an input key,
            # but normalize to the preferred alias to keep keys consistent across sources.
            matchable_names = list(match_alias_names)
            if include_name:
                name = normalize(field_name)
                if name not in matchable_names:
                    matchable_names.append(name)
            # All provided keys matching this field, in matchable_names preference order
            # (aliases first, then the field name). Only keys not yet consumed by an
            # earlier field are considered.
            matched_keys = [
                original
                for name in matchable_names
                if name in normalized_provided
                for original in normalized_provided[name]
                if original in init_kwarg_names
            ]
            if matched_keys:
                preferred_alias = canonical_alias_names[0] if canonical_alias_names else field_name
                # Prefer the first match in preference order; drop the remaining matches
                # so they are not re-added as extras.
                init_kwarg_names.difference_update(matched_keys)
                self.init_kwargs[preferred_alias] = init_kwargs[matched_keys[0]]
        # Include any remaining init kwargs (e.g., extras) unchanged
        # Note: If populate_by_name is True and the provided key is the field name, but
        # no alias exists, we keep it as-is so it can be processed as extra if allowed.
        self.init_kwargs.update({key: val for key, val in init_kwargs.items() if key in init_kwarg_names})

        self.nested_model_default_partial_update = (
            nested_model_default_partial_update
            if nested_model_default_partial_update is not None
            else self.config.get('nested_model_default_partial_update', False)
        )

    def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
        # Nothing to do here. Only implement the return statement to make mypy happy
        return None, '', False

    def __call__(self) -> dict[str, Any]:
        return (
            TypeAdapter(dict[str, Any]).dump_python(self.init_kwargs)
            if self.nested_model_default_partial_update
            else self.init_kwargs
        )

    def __repr__(self) -> str:
        return f'{self.__class__.__name__}(init_kwargs={self.init_kwargs!r})'


class PydanticBaseEnvSettingsSource(PydanticBaseSettingsSource):
    def __init__(
        self,
        settings_cls: type[BaseSettings],
        case_sensitive: bool | None = None,
        env_prefix: str | None = None,
        env_prefix_target: EnvPrefixTarget | None = None,
        env_ignore_empty: bool | None = None,
        env_parse_none_str: str | None = None,
        env_parse_enums: bool | None = None,
        _init_state: InitState | None = None,
    ) -> None:
        super().__init__(settings_cls, _init_state)
        self.case_sensitive = case_sensitive if case_sensitive is not None else self.config.get('case_sensitive', False)
        self.env_prefix = env_prefix if env_prefix is not None else self.config.get('env_prefix', '')
        self.env_prefix_target = (
            env_prefix_target if env_prefix_target is not None else self.config.get('env_prefix_target', 'variable')
        )
        self.env_ignore_empty = (
            env_ignore_empty if env_ignore_empty is not None else self.config.get('env_ignore_empty', False)
        )
        self.env_parse_none_str = (
            env_parse_none_str if env_parse_none_str is not None else self.config.get('env_parse_none_str')
        )
        self.env_parse_enums = env_parse_enums if env_parse_enums is not None else self.config.get('env_parse_enums')

    def _apply_case_sensitive(self, value: str) -> str:
        return value.lower() if not self.case_sensitive else value

    def _extract_field_info(
        self, field: FieldInfo, field_name: str, model_config: Mapping[str, Any] | None = None
    ) -> list[tuple[str, str, bool]]:
        """
        Extracts field info. This info is used to get the value of field from environment variables.

        It returns a list of tuples, each tuple contains:
            * field_key: The key of field that has to be used in model creation.
            * env_name: The environment variable name of the field.
            * value_is_complex: A flag to determine whether the value from environment variable
              is complex and has to be parsed.

        Args:
            field (FieldInfo): The field.
            field_name (str): The field name.
            model_config (Mapping[str, Any] | None): The config of the model owning the field, which decides
                whether its alias and name are accepted. Defaults to the settings config, pass it for a field
                of a nested model.

        Returns:
            list[tuple[str, str, bool]]: List of tuples, each tuple contains field_key, env_name, and value_is_complex.
        """
        _warn_if_field_info_incomplete(field, field_name, self._init_state)
        field_info: list[tuple[str, str, bool]] = []
        v_alias: str | list[str | int] | list[list[str | int]] | None
        if model_config is None:
            model_config = self.config
        if not _validate_by_alias_enabled(model_config):
            # pydantic ignores the aliases, so only the field name can populate the field.
            v_alias = None
        elif isinstance(field.validation_alias, (AliasChoices, AliasPath)):
            v_alias = field.validation_alias.convert_to_aliases()
        else:
            v_alias = field.validation_alias

        if v_alias:
            env_prefix = self.env_prefix if self.env_prefix_target in ('alias', 'all') else ''
            if isinstance(v_alias, list):  # AliasChoices, AliasPath
                if isinstance(v_alias[0], list):  # AliasChoices: one entry per choice
                    for alias in v_alias:
                        first_arg = cast(str, alias[0])  # first item of an AliasChoices must be a str
                        field_info.append(
                            (
                                first_arg,
                                self._apply_case_sensitive(env_prefix + first_arg),
                                len(alias) > 1,
                            )
                        )
                else:  # bare AliasPath: a single flat path, e.g. ['a', 0, 'b']
                    # Only the head names the env var to read; the remaining segments are
                    # navigated in-memory once the value has been JSON-decoded, so they must
                    # never be looked up as env vars of their own (previously each string
                    # segment produced its own candidate). Complexity must reflect the number
                    # of segments in the whole path, not the length of the head's name.
                    head = cast(str, v_alias[0])  # first item of an AliasPath must be a str
                    field_info.append((head, self._apply_case_sensitive(env_prefix + head), len(v_alias) > 1))
            else:  # string validation alias
                field_info.append((v_alias, self._apply_case_sensitive(env_prefix + v_alias), False))

        if not v_alias or _validate_by_name_enabled(model_config):
            annotation, metadata = _get_field_annotation_and_metadata(field)
            env_prefix = self.env_prefix if self.env_prefix_target in ('variable', 'all') else ''
            if is_union_origin(get_origin(annotation)) and _union_is_complex(annotation, metadata, self._init_state):
                field_info.append((field_name, self._apply_case_sensitive(env_prefix + field_name), True))
            else:
                field_info.append((field_name, self._apply_case_sensitive(env_prefix + field_name), False))

        return field_info

    def _replace_field_names_case_insensitively(self, field: FieldInfo, field_values: dict[str, Any]) -> dict[str, Any]:
        """
        Replace field names in values dict by looking in models fields insensitively.

        By having the following models:

            ```py
            class SubSubSub(BaseModel):
                VaL3: str

            class SubSub(BaseModel):
                Val2: str
                SUB_sub_SuB: SubSubSub

            class Sub(BaseModel):
                VAL1: str
                SUB_sub: SubSub

            class Settings(BaseSettings):
                nested: Sub

                model_config = SettingsConfigDict(env_nested_delimiter='__')
            ```

        Then:
            _replace_field_names_case_insensitively(
                field,
                {"val1": "v1", "sub_SUB": {"VAL2": "v2", "sub_SUB_sUb": {"vAl3": "v3"}}}
            )
            Returns {'VAL1': 'v1', 'SUB_sub': {'Val2': 'v2', 'SUB_sub_SuB': {'VaL3': 'v3'}}}
        """
        values: dict[str, Any] = {}

        for name, value in field_values.items():
            sub_model_field: FieldInfo | None = None

            annotation = _strip_annotated(_unwrap_optional_annotation(field.annotation))

            if not _is_model_or_dataclass(annotation):
                values[name] = value
                continue
            model_fields = _get_model_fields(annotation)
            validate_by_alias = _validate_by_alias_enabled(_get_model_config(annotation))

            # Find field in sub model by looking in fields case insensitively
            field_key: str | None = None
            for sub_model_field_name, sub_model_field in model_fields.items():
                _warn_if_field_info_incomplete(sub_model_field, sub_model_field_name, self._init_state)
                aliases, _ = _get_alias_names(
                    sub_model_field_name, sub_model_field, validate_by_alias=validate_by_alias
                )
                _search = (alias for alias in aliases if alias.lower() == name.lower())
                if field_key := next(_search, None):
                    break

            if not field_key:
                values[name] = value
                continue

            if (
                sub_model_field is not None
                and _is_model_or_dataclass(_strip_annotated(_unwrap_optional_annotation(sub_model_field.annotation)))
                and isinstance(value, dict)
            ):
                values[field_key] = self._replace_field_names_case_insensitively(sub_model_field, value)
            else:
                values[field_key] = value

        return values

    def _replace_env_none_type_values(self, field_value: dict[str, Any]) -> dict[str, Any]:
        """
        Recursively parse values that are of "None" type(EnvNoneType) to `None` type(None).
        """
        values: dict[str, Any] = {}

        for key, value in field_value.items():
            if not isinstance(value, EnvNoneType):
                values[key] = value if not isinstance(value, dict) else self._replace_env_none_type_values(value)
            else:
                values[key] = None

        return values

    def _get_resolved_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
        """
        Gets the value, the preferred alias key for model creation, and a flag to determine whether value
        is complex.

        Note:
            In V3, this method should either be made public, or, this method should be removed and the
            abstract method get_field_value should be updated to include a "use_preferred_alias" flag.

        Args:
            field: The field.
            field_name: The field name.

        Returns:
            A tuple that contains the value, preferred key and a flag to determine whether value is complex.
        """
        field_value, field_key, value_is_complex = self.get_field_value(field, field_name)
        if not (value_is_complex or (_validate_by_name_enabled(self.config) and (field_key == field_name))):
            field_infos = self._extract_field_info(field, field_name)
            preferred_key, _, preferred_is_complex = field_infos[0]
            # Only normalize to preferred_key when it's a simple string alias.
            # When the preferred key comes from an AliasPath (complex entry), skip normalization
            # to avoid using the AliasPath's first element as the key (see #766).
            if not preferred_is_complex:
                return field_value, preferred_key, value_is_complex
        return field_value, field_key, value_is_complex

    def __call__(self) -> dict[str, Any]:
        data: dict[str, Any] = {}

        for field_name, field in self.settings_cls.model_fields.items():
            _warn_if_field_info_incomplete(field, field_name, self._init_state)
            try:
                field_value, field_key, value_is_complex = self._get_resolved_field_value(field, field_name)
            except Exception as e:
                raise SettingsError(
                    f'error getting value for field "{field_name}" from source "{self.__class__.__name__}"'
                ) from e

            try:
                field_value = self.prepare_field_value(field_name, field, field_value, value_is_complex)
            except ValueError as e:
                raise SettingsError(
                    f'error parsing value for field "{field_name}" from source "{self.__class__.__name__}"'
                ) from e

            if field_value is not None:
                if self.env_parse_none_str is not None:
                    if isinstance(field_value, dict):
                        field_value = self._replace_env_none_type_values(field_value)
                    elif isinstance(field_value, EnvNoneType):
                        field_value = None
                if (
                    not self.case_sensitive
                    # and _lenient_issubclass(field.annotation, BaseModel)
                    and isinstance(field_value, dict)
                ):
                    data[field_key] = self._replace_field_names_case_insensitively(field, field_value)
                else:
                    data[field_key] = field_value

        return data


__all__ = [
    'ConfigFileSourceMixin',
    'DefaultSettingsSource',
    'InitSettingsSource',
    'PydanticBaseEnvSettingsSource',
    'PydanticBaseSettingsSource',
    'SettingsError',
]
