from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, Generator, List, Optional, Tuple, Union

import pandas as pd
from dacite import Config, from_dict


class HttpMethod(Enum):
    GET = "get"
    PUT = "put"
    POST = "post"
    DELETE = "delete"
    OPTIONS = "options"
    HEAD = "head"
    PATCH = "patch"
    TRACE = "trace"
    CONNECT = "connect"


@dataclass
class Info:
    title: str
    version: str
    description: Optional[str] = None
    termsOfService: Optional[str] = None
    contact: Optional["Contact"] = None
    license: Optional["License"] = None


@dataclass
class Contact:
    name: Optional[str] = None
    url: Optional[str] = None
    email: Optional[str] = None


@dataclass
class License:
    name: str
    url: Optional[str] = None


@dataclass
class ExternalDocs:
    url: str
    description: Optional[str] = None


@dataclass
class Schema:
    type: Optional[str] = None
    format: Optional[str] = None
    items: Optional[Schema] = None
    properties: Optional[Dict[str, Schema]] = None
    required: Optional[List[str]] = None
    additionalProperties: Optional[Dict] = None
    description: Optional[str] = None
    enum: Optional[List[Any]] = None
    default: Optional[Any] = None
    title: Optional[str] = None
    readOnly: Optional[bool] = None
    example: Optional[Any] = None
    pattern: Optional[Any] = None
    maximum: Optional[float] = None
    minimum: Optional[float] = None
    exclusiveMaximum: Optional[bool] = None
    exclusiveMinimum: Optional[bool] = None
    maxLength: Optional[int] = None
    minLength: Optional[int] = None
    maxItems: Optional[int] = None
    minItems: Optional[int] = None
    maxProperties: Optional[int] = None
    minProperties: Optional[int] = None
    x_nullable: Optional[bool] = None


@dataclass
class Parameter:
    name: str
    in_: str
    description: Optional[str] = None
    required: Optional[bool] = None
    schema: Optional[Schema] = None
    enum: Optional[List[Any]] = None
    type: Optional[str] = None
    format: Optional[str] = None
    items: Optional[Dict] = None
    collectionFormat: Optional[str] = None
    default: Optional[Any] = None
    pattern: Optional[str] = None
    maxItems: Optional[int] = None
    minItems: Optional[int] = None
    maximum: Optional[float] = None
    minimum: Optional[float] = None
    maxLength: Optional[int] = None
    minLength: Optional[int] = None


@dataclass
class RequestBody:
    content: Dict[str, "MediaType"]
    description: Optional[str] = None
    required: bool = False


@dataclass
class MediaType:
    schema: Schema


@dataclass
class Response:
    description: str
    content: Optional[Dict[str, MediaType]] = None
    schema: Optional[Schema] = None


@dataclass
class Operation:
    responses: Dict[str, Response]
    summary: Optional[str] = None
    description: Optional[str] = None
    parameters: Optional[List[Parameter]] = None
    requestBody: Optional[RequestBody] = None
    tags: Optional[List[str]] = None
    security: Optional[List[Dict[str, List[str]]]] = None
    operationId: Optional[str] = None
    deprecated: Optional[bool] = None

    def _get_parameters(self, in_: str) -> List[Parameter]:
        if self.parameters is None:
            return []
        return [param for param in self.parameters if param.in_ == in_]

    def get_security_type(self):
        if self.security is None:
            return None
        else:
            return list(self.security[0].keys())[0]

    def get_body_parameters(self):
        enum_values = []
        non_enum_values = []
        read_only_values = []
        leaf_nodes = []
        non_leaf_nodes = []

        def traverse_schema(schema: Schema, parent_path: str = ""):
            if schema.readOnly:
                read_only_values.append((parent_path, schema))
                # If adding the following code, `cluster.pausePlan.pauseType` will be included unexpectedly
                # if schema.properties:
                #     for prop_name, prop_schema in schema.properties.items():
                #         new_path = f"{parent_path}.{prop_name}" if parent_path else prop_name
                #         traverse_schema(prop_schema, new_path)
                # if schema.items:
                #     traverse_schema(schema.items, f"{parent_path}[]")
                return

            is_leaf = not (schema.properties or schema.items)

            if is_leaf:
                leaf_nodes.append((parent_path, schema))
                if schema.enum:
                    enum_values.append((parent_path, schema))
                else:
                    non_enum_values.append((parent_path, schema))
                return
            else:
                non_leaf_nodes.append((parent_path, schema))
                non_enum_values.append((parent_path, schema))

            if schema.properties:
                for prop_name, prop_schema in schema.properties.items():
                    new_path = (
                        f"{parent_path}.{prop_name}" if parent_path else prop_name
                    )
                    traverse_schema(prop_schema, new_path)

            if schema.items:
                traverse_schema(schema.items, f"{parent_path}[]")

        if self.requestBody:  # OpenAPI 3.0
            # not tested
            for content_type, media_type in self.requestBody.content.items():
                traverse_schema(media_type.schema, "")
        else:  # OpenAPI 2.0
            body_params = self._get_parameters("body")
            for param in body_params:
                if param.schema:
                    traverse_schema(param.schema, "body")
                elif param.readOnly:
                    read_only_values.append(("body", param))
                elif param.enum:
                    enum_values.append(("body", param))
                else:
                    non_enum_values.append(("body", param))

        # # debug
        # if self.operationId == "ClusterService_CreateCluster":
        #     print(len(enum_values), " enum_values: ", [aset[0] for aset in enum_values])
        #     print(
        #         len(non_enum_values),
        #         " non_enum_values: ",
        #         [aset[0] for aset in non_enum_values],
        #     )
        #     print(
        #         len(read_only_values),
        #         " read_only_values: ",
        #         [aset[0] for aset in read_only_values],
        #     )
        #     print(len(leaf_nodes), " leaf_nodes: ", [aset[0] for aset in leaf_nodes])
        #     print(
        #         len(non_leaf_nodes),
        #         " non_leaf_nodes: ",
        #         [aset[0] for aset in non_leaf_nodes],
        #     )
        return (
            enum_values,
            non_enum_values,
            read_only_values,
            leaf_nodes,
            non_leaf_nodes,
        )

    def get_path_parameters(self):
        return self._get_parameters("path")

    def get_query_parameters(self):
        return self._get_parameters("query")


@dataclass
class PathItem:
    get: Optional[Operation] = None
    post: Optional[Operation] = None
    put: Optional[Operation] = None
    delete: Optional[Operation] = None
    patch: Optional[Operation] = None
    options: Optional[Operation] = None
    head: Optional[Operation] = None
    trace: Optional[Operation] = None
    parameters: Optional[List[Parameter]] = None

    def iter_methods(self) -> Generator[Tuple[str, Operation], None, None]:
        for method in HttpMethod:
            operation = getattr(self, method.value, None)
            if operation is not None:
                yield method.value, operation

    def _get_parameters(self, in_: str) -> List[Parameter]:
        return [param for param in (self.parameters or []) if param.in_ == in_]

    def get_path_parameters(self):
        return self._get_parameters("path")

    def get_query_parameters(self):
        return self._get_parameters("query")


@dataclass
class SecurityScheme:
    type: str
    description: Optional[str] = None
    name: Optional[str] = None
    in_: Optional[str] = None
    scheme: Optional[str] = None
    bearerFormat: Optional[str] = None
    flows: Optional[Any] = None  # OAuth2 flows, simplified here


@dataclass
class Components:
    """OpenAPI 3.0 Components object."""

    schemas: Optional[Dict[str, Schema]] = None
    responses: Optional[Dict[str, Response]] = None
    parameters: Optional[Dict[str, Parameter]] = None
    securitySchemes: Optional[Dict[str, SecurityScheme]] = None


@dataclass
class OpenAPISpec2:
    swagger: str
    info: Info
    paths: Dict[str, PathItem] = field(default_factory=dict)
    basePath: Optional[str] = None
    host: Optional[str] = None
    schemes: Optional[List[str]] = None
    consumes: Optional[List[str]] = None
    produces: Optional[List[str]] = None
    definitions: Optional[Dict[str, Schema]] = None
    parameters: Optional[Dict[str, Parameter]] = None
    responses: Optional[Dict[str, Response]] = None
    securityDefinitions: Optional[Dict[str, SecurityScheme]] = None
    security: Optional[List[Dict[str, List[str]]]] = None
    tags: Optional[List[Dict[str, str]]] = None
    externalDocs: Optional[ExternalDocs] = None


@dataclass
class OpenAPISpec3:
    openapi: str
    info: Info
    paths: Dict[str, PathItem] = field(default_factory=dict)
    components: Optional[Components] = None
    servers: Optional[List[Dict[str, str]]] = None
    security: Optional[List[Dict[str, List[str]]]] = None
    tags: Optional[List[Dict[str, str]]] = None
    externalDocs: Optional[ExternalDocs] = None


@dataclass
class OpenAPISpec:
    spec: Union[OpenAPISpec2, OpenAPISpec3]
    coefficient_table: Optional[Dict[str, Dict[str, int]]] = None
    pre_defined_cases: Optional[Dict[str, Dict[str, List]]] = None

    @classmethod
    def from_dict(cls, data: dict) -> "OpenAPISpec":
        config = Config(
            type_hooks={
                Dict: dict,
                Parameter: parameter_hook,
                Schema: schema_hook,
            },
        )
        if "swagger" in data:
            spec = from_dict(data_class=OpenAPISpec2, data=data, config=config)  # type: ignore
        elif "openapi" in data:
            spec = from_dict(data_class=OpenAPISpec3, data=data, config=config)  # type: ignore
        else:
            raise ValueError("Invalid OpenAPI specification")

        return cls(spec=spec)

    def add_coefficient_table(self, coeff_dict: Dict[str, Dict[str, int]]) -> None:
        self.coefficient_table = coeff_dict

    def add_cases(self, cases: Dict[str, dict]) -> None:
        # TODO: support multi-step cases
        self.pre_defined_cases = {}
        for operation_id, case in cases.items():
            case["operation_id"] = operation_id
            if operation_id not in self.pre_defined_cases:
                self.pre_defined_cases[operation_id] = {
                    "cases": [],
                    "multi_step_cases": [],
                }
            self.pre_defined_cases[operation_id]["cases"].append(case)


def parameter_hook(data: Dict[str, Any]) -> Parameter:
    if "in" in data:
        data["in_"] = data.pop("in")
    if "schema" in data and isinstance(data["schema"], dict):
        data["schema"] = schema_hook(data["schema"])
    if "items" in data and isinstance(data["items"], dict):
        data["items"] = schema_hook(data["items"])
    return Parameter(**data)


def schema_hook(data: Dict[str, Any]) -> Schema:
    if "items" in data and isinstance(data["items"], dict):
        data["items"] = schema_hook(data["items"])
    if "properties" in data:
        data["properties"] = {
            k: schema_hook(v) if isinstance(v, dict) else v
            for k, v in data["properties"].items()
        }
    if "x-nullable" in data:
        data["x_nullable"] = data.pop("x-nullable")
    return Schema(**data)


def generate_coefficient_table(csv_file: str) -> Dict[str, Dict[str, int]]:
    df = pd.read_csv(csv_file)
    df = df.set_index("operationId")
    result = {}
    for operation_id, row in df.iterrows():
        zeros = row[row == 0]
        if not zeros.empty:
            result[str(operation_id)] = zeros.astype(int).to_dict()
    return result