from copy import copy
from dataclasses import dataclass, is_dataclass
from enum import Enum
from typing import (
    Any,
    Callable,
    Dict,
    List,
    Sequence,
    Set,
    Tuple,
    Type,
    Union,
)

from fastapi._compat import shared
from fastapi.exceptions import RequestErrorModel
from fastapi.openapi.constants import REF_PREFIX as REF_PREFIX
from fastapi.types import ModelNameMap
from pydantic.version import VERSION as PYDANTIC_VERSION
from typing_extensions import Literal

PYDANTIC_VERSION_MINOR_TUPLE = tuple(int(x) for x in PYDANTIC_VERSION.split(".")[:2])
PYDANTIC_V2 = PYDANTIC_VERSION_MINOR_TUPLE[0] == 2
# Keeping old "Required" functionality from Pydantic V1, without
# shadowing typing.Required.
RequiredParam: Any = Ellipsis  # type: ignore[no-redef]

if not PYDANTIC_V2:
    from pydantic import AnyUrl as Url  # noqa: F401
    from pydantic import (  # type: ignore[assignment]
        BaseConfig as BaseConfig,  # noqa: F401
    )
    from pydantic import BaseModel, create_model
    from pydantic import ValidationError as ValidationError  # noqa: F401
    from pydantic.class_validators import (  # type: ignore[no-redef]
        Validator as Validator,  # noqa: F401
    )
    from pydantic.color import Color as Color  # noqa: F401
    from pydantic.error_wrappers import (  # type: ignore[no-redef]
        ErrorWrapper as ErrorWrapper,  # noqa: F401
    )
    from pydantic.errors import MissingError
    from pydantic.fields import (  # type: ignore[attr-defined]
        SHAPE_FROZENSET,
        SHAPE_LIST,
        SHAPE_SEQUENCE,
        SHAPE_SET,
        SHAPE_SINGLETON,
        SHAPE_TUPLE,
        SHAPE_TUPLE_ELLIPSIS,
    )
    from pydantic.fields import FieldInfo as FieldInfo
    from pydantic.fields import (  # type: ignore[no-redef,attr-defined]
        ModelField as ModelField,  # noqa: F401
    )
    from pydantic.fields import (  # type: ignore[no-redef,attr-defined]
        Undefined as Undefined,
    )
    from pydantic.fields import (  # type: ignore[no-redef, attr-defined]
        UndefinedType as UndefinedType,  # noqa: F401
    )
    from pydantic.networks import AnyUrl as AnyUrl  # noqa: F401
    from pydantic.networks import NameEmail as NameEmail  # noqa: F401
    from pydantic.schema import TypeModelSet as TypeModelSet
    from pydantic.schema import (
        field_schema,
        get_flat_models_from_fields,
        model_process_schema,
    )
    from pydantic.schema import (  # type: ignore[no-redef]  # noqa: F401
        get_annotation_from_field_info as get_annotation_from_field_info,
    )
    from pydantic.schema import get_flat_models_from_field as get_flat_models_from_field
    from pydantic.types import SecretBytes as SecretBytes  # noqa: F401
    from pydantic.types import SecretStr as SecretStr  # noqa: F401
    from pydantic.typing import (  # type: ignore[no-redef]
        evaluate_forwardref as evaluate_forwardref,  # noqa: F401
    )
    from pydantic.utils import (  # type: ignore[no-redef]
        lenient_issubclass as lenient_issubclass,  # noqa: F401
    )

else:
    from pydantic.v1 import AnyUrl as Url  # noqa: F401
    from pydantic.v1 import (  # type: ignore[assignment]
        BaseConfig as BaseConfig,  # noqa: F401
    )
    from pydantic.v1 import BaseModel, create_model
    from pydantic.v1 import ValidationError as ValidationError  # noqa: F401
    from pydantic.v1.class_validators import (  # type: ignore[no-redef]
        Validator as Validator,  # noqa: F401
    )
    from pydantic.v1.color import Color as Color  # noqa: F401
    from pydantic.v1.error_wrappers import (  # type: ignore[no-redef]
        ErrorWrapper as ErrorWrapper,  # noqa: F401
    )
    from pydantic.v1.errors import MissingError
    from pydantic.v1.fields import (  # type: ignore[attr-defined]
        SHAPE_FROZENSET,
        SHAPE_LIST,
        SHAPE_SEQUENCE,
        SHAPE_SET,
        SHAPE_SINGLETON,
        SHAPE_TUPLE,
        SHAPE_TUPLE_ELLIPSIS,
    )
    from pydantic.v1.fields import FieldInfo as FieldInfo
    from pydantic.v1.fields import (  # type: ignore[no-redef,attr-defined]
        ModelField as ModelField,  # noqa: F401
    )
    from pydantic.v1.fields import (  # type: ignore[no-redef,attr-defined]
        Undefined as Undefined,
    )
    from pydantic.v1.fields import (  # type: ignore[no-redef, attr-defined]
        UndefinedType as UndefinedType,  # noqa: F401
    )
    from pydantic.v1.networks import AnyUrl as AnyUrl  # noqa: F401
    from pydantic.v1.networks import NameEmail as NameEmail  # noqa: F401
    from pydantic.v1.schema import TypeModelSet as TypeModelSet
    from pydantic.v1.schema import (
        field_schema,
        get_flat_models_from_fields,
        model_process_schema,
    )
    from pydantic.v1.schema import (  # type: ignore[no-redef]  # noqa: F401
        get_annotation_from_field_info as get_annotation_from_field_info,
    )
    from pydantic.v1.schema import (
        get_flat_models_from_field as get_flat_models_from_field,  # noqa: F401
    )
    from pydantic.v1.types import SecretBytes as SecretBytes  # noqa: F401
    from pydantic.v1.types import SecretStr as SecretStr  # noqa: F401
    from pydantic.v1.typing import (  # type: ignore[no-redef]
        evaluate_forwardref as evaluate_forwardref,  # noqa: F401
    )
    from pydantic.v1.utils import (  # type: ignore[no-redef]
        lenient_issubclass as lenient_issubclass,  # noqa: F401
    )


GetJsonSchemaHandler = Any  # type: ignore[assignment,misc]
JsonSchemaValue = Dict[str, Any]  # type: ignore[misc]
CoreSchema = Any  # type: ignore[assignment,misc]

sequence_shapes = {
    SHAPE_LIST,
    SHAPE_SET,
    SHAPE_FROZENSET,
    SHAPE_TUPLE,
    SHAPE_SEQUENCE,
    SHAPE_TUPLE_ELLIPSIS,
}
sequence_shape_to_type = {
    SHAPE_LIST: list,
    SHAPE_SET: set,
    SHAPE_TUPLE: tuple,
    SHAPE_SEQUENCE: list,
    SHAPE_TUPLE_ELLIPSIS: list,
}


@dataclass
class GenerateJsonSchema:  # type: ignore[no-redef]
    ref_template: str


class PydanticSchemaGenerationError(Exception):  # type: ignore[no-redef]
    pass


def with_info_plain_validator_function(  # type: ignore[misc]
    function: Callable[..., Any],
    *,
    ref: Union[str, None] = None,
    metadata: Any = None,
    serialization: Any = None,
) -> Any:
    return {}


def get_model_definitions(
    *,
    flat_models: Set[Union[Type[BaseModel], Type[Enum]]],
    model_name_map: Dict[Union[Type[BaseModel], Type[Enum]], str],
) -> Dict[str, Any]:
    definitions: Dict[str, Dict[str, Any]] = {}
    for model in flat_models:
        m_schema, m_definitions, m_nested_models = model_process_schema(
            model, model_name_map=model_name_map, ref_prefix=REF_PREFIX
        )
        definitions.update(m_definitions)
        model_name = model_name_map[model]
        definitions[model_name] = m_schema
    for m_schema in definitions.values():
        if "description" in m_schema:
            m_schema["description"] = m_schema["description"].split("\f")[0]
    return definitions


def is_pv1_scalar_field(field: ModelField) -> bool:
    from fastapi import params

    field_info = field.field_info
    if not (
        field.shape == SHAPE_SINGLETON  # type: ignore[attr-defined]
        and not lenient_issubclass(field.type_, BaseModel)
        and not lenient_issubclass(field.type_, dict)
        and not shared.field_annotation_is_sequence(field.type_)
        and not is_dataclass(field.type_)
        and not isinstance(field_info, params.Body)
    ):
        return False
    if field.sub_fields:  # type: ignore[attr-defined]
        if not all(
            is_pv1_scalar_field(f)
            for f in field.sub_fields  # type: ignore[attr-defined]
        ):
            return False
    return True


def is_pv1_scalar_sequence_field(field: ModelField) -> bool:
    if (field.shape in sequence_shapes) and not lenient_issubclass(  # type: ignore[attr-defined]
        field.type_, BaseModel
    ):
        if field.sub_fields is not None:  # type: ignore[attr-defined]
            for sub_field in field.sub_fields:  # type: ignore[attr-defined]
                if not is_pv1_scalar_field(sub_field):
                    return False
        return True
    if shared._annotation_is_sequence(field.type_):
        return True
    return False


def _normalize_errors(errors: Sequence[Any]) -> List[Dict[str, Any]]:
    use_errors: List[Any] = []
    for error in errors:
        if isinstance(error, ErrorWrapper):
            new_errors = ValidationError(  # type: ignore[call-arg]
                errors=[error], model=RequestErrorModel
            ).errors()
            use_errors.extend(new_errors)
        elif isinstance(error, list):
            use_errors.extend(_normalize_errors(error))
        else:
            use_errors.append(error)
    return use_errors


def _regenerate_error_with_loc(
    *, errors: Sequence[Any], loc_prefix: Tuple[Union[str, int], ...]
) -> List[Dict[str, Any]]:
    updated_loc_errors: List[Any] = [
        {**err, "loc": loc_prefix + err.get("loc", ())}
        for err in _normalize_errors(errors)
    ]

    return updated_loc_errors


def _model_rebuild(model: Type[BaseModel]) -> None:
    model.update_forward_refs()


def _model_dump(
    model: BaseModel, mode: Literal["json", "python"] = "json", **kwargs: Any
) -> Any:
    return model.dict(**kwargs)


def _get_model_config(model: BaseModel) -> Any:
    return model.__config__  # type: ignore[attr-defined]


def get_schema_from_model_field(
    *,
    field: ModelField,
    model_name_map: ModelNameMap,
    field_mapping: Dict[
        Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue
    ],
    separate_input_output_schemas: bool = True,
) -> Dict[str, Any]:
    # This expects that GenerateJsonSchema was already used to generate the definitions
    return field_schema(  # type: ignore[no-any-return]
        field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
    )[0]


# def get_compat_model_name_map(fields: List[ModelField]) -> ModelNameMap:
#     models = get_flat_models_from_fields(fields, known_models=set())
#     return get_model_name_map(models)  # type: ignore[no-any-return]


def get_definitions(
    *,
    fields: List[ModelField],
    model_name_map: ModelNameMap,
    separate_input_output_schemas: bool = True,
) -> Tuple[
    Dict[Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue],
    Dict[str, Dict[str, Any]],
]:
    models = get_flat_models_from_fields(fields, known_models=set())
    return {}, get_model_definitions(flat_models=models, model_name_map=model_name_map)


def is_scalar_field(field: ModelField) -> bool:
    return is_pv1_scalar_field(field)


def is_sequence_field(field: ModelField) -> bool:
    return field.shape in sequence_shapes or shared._annotation_is_sequence(field.type_)  # type: ignore[attr-defined]


def is_scalar_sequence_field(field: ModelField) -> bool:
    return is_pv1_scalar_sequence_field(field)


def is_bytes_field(field: ModelField) -> bool:
    return lenient_issubclass(field.type_, bytes)


def is_bytes_sequence_field(field: ModelField) -> bool:
    return field.shape in sequence_shapes and lenient_issubclass(field.type_, bytes)  # type: ignore[attr-defined]


def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
    return copy(field_info)


def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]:
    return sequence_shape_to_type[field.shape](value)  # type: ignore[no-any-return,attr-defined]


def get_missing_field_error(loc: Tuple[str, ...]) -> Dict[str, Any]:
    missing_field_error = ErrorWrapper(MissingError(), loc=loc)  # type: ignore[call-arg]
    new_error = ValidationError([missing_field_error], RequestErrorModel)
    return new_error.errors()[0]  # type: ignore[return-value]


def create_body_model(
    *, fields: Sequence[ModelField], model_name: str
) -> Type[BaseModel]:
    BodyModel = create_model(model_name)
    for f in fields:
        BodyModel.__fields__[f.name] = f  # type: ignore[index]
    return BodyModel


def get_model_fields(model: Type[BaseModel]) -> List[ModelField]:
    return list(model.__fields__.values())  # type: ignore[attr-defined]