import logging
from abc import ABC, abstractmethod
from collections.abc import Generator, Mapping
from typing import Any, Union, cast

from pydantic import JsonValue

from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
from graphon.model_runtime.errors.invoke import InvokeError

logger = logging.getLogger(__name__)


class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
    @classmethod
    def _cast_blocking_response(cls, response: AppBlockingResponse) -> TBlockingResponse:
        return cast(TBlockingResponse, response)

    @classmethod
    def convert(
        cls, response: Union[AppBlockingResponse, Generator[AppStreamResponse, Any, None]], invoke_from: InvokeFrom
    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
        if invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.SERVICE_API}:
            if isinstance(response, AppBlockingResponse):
                return cls.convert_blocking_full_response(cls._cast_blocking_response(response))
            else:

                def _generate_full_response() -> Generator[dict[str, Any] | str, Any, None]:
                    yield from cls.convert_stream_full_response(response)

                return _generate_full_response()
        else:
            if isinstance(response, AppBlockingResponse):
                return cls.convert_blocking_simple_response(cls._cast_blocking_response(response))
            else:

                def _generate_simple_response() -> Generator[dict[str, Any] | str, Any, None]:
                    yield from cls.convert_stream_simple_response(response)

                return _generate_simple_response()

    @classmethod
    @abstractmethod
    def convert_blocking_full_response(cls, blocking_response: TBlockingResponse) -> dict[str, Any]:
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def convert_blocking_simple_response(cls, blocking_response: TBlockingResponse) -> dict[str, Any]:
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def convert_stream_full_response(
        cls, stream_response: Generator[AppStreamResponse, None, None]
    ) -> Generator[dict[str, Any] | str, None, None]:
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def convert_stream_simple_response(
        cls, stream_response: Generator[AppStreamResponse, None, None]
    ) -> Generator[dict[str, Any] | str, None, None]:
        raise NotImplementedError

    @classmethod
    def _get_simple_metadata(cls, metadata: dict[str, Any]):
        """
        Get simple metadata.
        :param metadata: metadata
        :return:
        """
        # show_retrieve_source
        updated_resources = []
        if "retriever_resources" in metadata:
            for resource in metadata["retriever_resources"]:
                updated_resources.append(
                    {
                        "dataset_id": resource.get("dataset_id"),
                        "dataset_name": resource.get("dataset_name"),
                        "document_id": resource.get("document_id"),
                        "segment_id": resource.get("segment_id", ""),
                        "position": resource["position"],
                        "data_source_type": resource.get("data_source_type"),
                        "document_name": resource["document_name"],
                        "score": resource["score"],
                        "hit_count": resource.get("hit_count"),
                        "word_count": resource.get("word_count"),
                        "segment_position": resource.get("segment_position"),
                        "index_node_hash": resource.get("index_node_hash"),
                        "content": resource["content"],
                        "page": resource.get("page"),
                        "title": resource.get("title"),
                        "files": resource.get("files"),
                        "summary": resource.get("summary"),
                    }
                )
            metadata["retriever_resources"] = updated_resources

        # show annotation reply
        if "annotation_reply" in metadata:
            del metadata["annotation_reply"]

        # show usage
        if "usage" in metadata:
            del metadata["usage"]

        return metadata

    @classmethod
    def _error_to_stream_response(cls, e: Exception) -> dict[str, JsonValue]:
        """
        Error to stream response.
        :param e: exception
        :return:
        """
        error_responses: dict[type[Exception], dict[str, JsonValue]] = {
            ValueError: {"code": "invalid_param", "status": 400},
            ProviderTokenNotInitError: {"code": "provider_not_initialize", "status": 400},
            QuotaExceededError: {
                "code": "provider_quota_exceeded",
                "message": "Your quota for Dify Hosted Model Provider has been exhausted. "
                "Please go to Settings -> Model Provider to complete your own provider credentials.",
                "status": 400,
            },
            ModelCurrentlyNotSupportError: {"code": "model_currently_not_support", "status": 400},
            InvokeError: {"code": "completion_request_error", "status": 400},
        }

        # Determine the response based on the type of exception
        data: dict[str, JsonValue] | None = None
        for k, v in error_responses.items():
            if isinstance(e, k):
                data = v

        if data:
            data.setdefault("message", getattr(e, "description", str(e)))
        else:
            logger.error(e)
            data = {
                "code": "internal_server_error",
                "message": "Internal Server Error, please contact support.",
                "status": 500,
            }

        return data