# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import json
import os
from pathlib import Path
from typing import Any

import yaml

from haystack import component, logging
from haystack.dataclasses.byte_stream import ByteStream
from haystack.lazy_imports import LazyImport

logger = logging.getLogger(__name__)

with LazyImport("Run 'pip install jsonref'") as openapi_imports:
    import jsonref


@component
class OpenAPIServiceToFunctions:
    """
    Converts OpenAPI service definitions to a format suitable for OpenAI function calling.

    The definition must respect OpenAPI specification 3.0.0 or higher.
    It can be specified in JSON or YAML format.
    Each function must have:
        - unique operationId
        - description
        - requestBody and/or parameters
        - schema for the requestBody and/or parameters
    For more details on OpenAPI specification see the [official documentation](https://github.com/OAI/OpenAPI-Specification).
    For more details on OpenAI function calling see the [official documentation](https://platform.openai.com/docs/guides/function-calling).

    Usage example:
    ```python
    from haystack.components.converters import OpenAPIServiceToFunctions

    converter = OpenAPIServiceToFunctions()
    result = converter.run(sources=["path/to/openapi_definition.yaml"])
    assert result["functions"]
    ```
    """

    MIN_REQUIRED_OPENAPI_SPEC_VERSION = 3

    def __init__(self):
        """
        Create an OpenAPIServiceToFunctions component.
        """
        openapi_imports.check()

    @component.output_types(functions=list[dict[str, Any]], openapi_specs=list[dict[str, Any]])
    def run(self, sources: list[str | Path | ByteStream]) -> dict[str, Any]:
        """
        Converts OpenAPI definitions in OpenAI function calling format.

        :param sources:
            File paths or ByteStream objects of OpenAPI definitions (in JSON or YAML format).

        :returns:
            A dictionary with the following keys:
            - functions: Function definitions in JSON object format
            - openapi_specs: OpenAPI specs in JSON/YAML object format with resolved references

        :raises RuntimeError:
            If the OpenAPI definitions cannot be downloaded or processed.
        :raises ValueError:
            If the source type is not recognized or no functions are found in the OpenAPI definitions.
        """
        all_extracted_fc_definitions: list[dict[str, Any]] = []
        all_openapi_specs = []
        for source in sources:
            openapi_spec_content = None
            if isinstance(source, (str, Path)):
                if os.path.exists(source):
                    try:
                        with open(source, "r") as f:
                            openapi_spec_content = f.read()
                    except IOError as e:
                        logger.warning(
                            "IO error reading OpenAPI specification file: {source}. Error: {e}", source=source, e=e
                        )
                else:
                    logger.warning("OpenAPI specification file not found: {source}", source=source)
            elif isinstance(source, ByteStream):
                openapi_spec_content = source.data.decode("utf-8")
                if not openapi_spec_content:
                    logger.warning(
                        "Invalid OpenAPI specification content provided: {openapi_spec_content}",
                        openapi_spec_content=openapi_spec_content,
                    )
            else:
                logger.warning(
                    "Invalid source type {source}. Only str, Path, and ByteStream are supported.", source=type(source)
                )
                continue

            if openapi_spec_content:
                try:
                    service_openapi_spec = self._parse_openapi_spec(openapi_spec_content)
                    functions: list[dict[str, Any]] = self._openapi_to_functions(service_openapi_spec)
                    all_extracted_fc_definitions.extend(functions)
                    all_openapi_specs.append(service_openapi_spec)
                except Exception as e:
                    logger.error(
                        "Error processing OpenAPI specification from source {source}: {error}", source=source, error=e
                    )

        if not all_extracted_fc_definitions:
            logger.warning("No OpenAI function definitions extracted from the provided OpenAPI specification sources.")

        return {"functions": all_extracted_fc_definitions, "openapi_specs": all_openapi_specs}

    def _openapi_to_functions(self, service_openapi_spec: dict[str, Any]) -> list[dict[str, Any]]:
        """
        OpenAPI to OpenAI function conversion.

        Extracts functions from the OpenAPI specification of the service and converts them into a format
        suitable for OpenAI function calling.

        :param service_openapi_spec: The OpenAPI specification from which functions are to be extracted.
        :type service_openapi_spec: dict[str, Any]
        :return: A list of dictionaries, each representing a function. Each dictionary includes the function's
                 name, description, and a schema of its parameters.
        :rtype: list[dict[str, Any]]
        """

        # Doesn't enforce rigid spec validation because that would require a lot of dependencies
        # We check the version and require minimal fields to be present, so we can extract functions
        spec_version = service_openapi_spec.get("openapi")
        if not spec_version:
            raise ValueError(f"Invalid OpenAPI spec provided. Could not extract version from {service_openapi_spec}")
        service_openapi_spec_version = int(spec_version.split(".")[0])

        # Compare the versions
        if service_openapi_spec_version < OpenAPIServiceToFunctions.MIN_REQUIRED_OPENAPI_SPEC_VERSION:
            raise ValueError(
                f"Invalid OpenAPI spec version {service_openapi_spec_version}. Must be "
                f"at least {OpenAPIServiceToFunctions.MIN_REQUIRED_OPENAPI_SPEC_VERSION}."
            )

        functions: list[dict[str, Any]] = []
        for paths in service_openapi_spec["paths"].values():
            for path_spec in paths.values():
                function_dict = self._parse_endpoint_spec(path_spec)
                if function_dict:
                    functions.append(function_dict)
        return functions

    def _parse_endpoint_spec(self, resolved_spec: dict[str, Any]) -> dict[str, Any] | None:
        if not isinstance(resolved_spec, dict):
            logger.warning("Invalid OpenAPI spec format provided. Could not extract function.")
            return {}

        function_name = resolved_spec.get("operationId")
        description = resolved_spec.get("description") or resolved_spec.get("summary", "")

        schema: dict[str, Any] = {"type": "object", "properties": {}}

        # requestBody section
        req_body_schema = (
            resolved_spec.get("requestBody", {}).get("content", {}).get("application/json", {}).get("schema", {})
        )
        if "properties" in req_body_schema:
            for prop_name, prop_schema in req_body_schema["properties"].items():
                schema["properties"][prop_name] = self._parse_property_attributes(prop_schema)

            if "required" in req_body_schema:
                schema.setdefault("required", []).extend(req_body_schema["required"])

        # parameters section
        for param in resolved_spec.get("parameters", []):
            if "schema" in param:
                schema_dict = self._parse_property_attributes(param["schema"])
                # these attributes are not in param[schema] level but on param level
                useful_attributes = ["description", "pattern", "enum"]
                schema_dict.update({key: param[key] for key in useful_attributes if param.get(key)})
                schema["properties"][param["name"]] = schema_dict
                if param.get("required", False):
                    schema.setdefault("required", []).append(param["name"])

        if function_name and description and schema["properties"]:
            return {"name": function_name, "description": description, "parameters": schema}
        else:
            logger.warning(
                "Invalid OpenAPI spec format provided. Could not extract function from {spec}", spec=resolved_spec
            )
            return {}

    def _parse_property_attributes(
        self, property_schema: dict[str, Any], include_attributes: list[str] | None = None
    ) -> dict[str, Any]:
        """
        Parses the attributes of a property schema.

        Recursively parses the attributes of a property schema, including nested objects and arrays,
        and includes specified attributes like description, pattern, etc.

        :param property_schema: The schema of the property to parse.
        :param include_attributes: The list of attributes to include in the parsed schema.
        :return: The parsed schema of the property including the specified attributes.
        """
        include_attributes = include_attributes or ["description", "pattern", "enum"]

        schema_type = property_schema.get("type")

        parsed_schema = {"type": schema_type} if schema_type else {}
        for attr in include_attributes:
            if attr in property_schema:
                parsed_schema[attr] = property_schema[attr]

        if schema_type == "object":
            properties = property_schema.get("properties", {})
            parsed_properties = {
                prop_name: self._parse_property_attributes(prop, include_attributes)
                for prop_name, prop in properties.items()
            }
            parsed_schema["properties"] = parsed_properties

            if "required" in property_schema:
                parsed_schema["required"] = property_schema["required"]

        elif schema_type == "array":
            items = property_schema.get("items", {})
            parsed_schema["items"] = self._parse_property_attributes(items, include_attributes)

        return parsed_schema

    def _parse_openapi_spec(self, content: str) -> dict[str, Any]:
        """
        Parses OpenAPI specification content, supporting both JSON and YAML formats.

        :param content: The content of the OpenAPI specification.
        :return: The parsed OpenAPI specification.
        """
        open_api_spec_content = None
        try:
            open_api_spec_content = json.loads(content)
            return jsonref.replace_refs(open_api_spec_content)
        except json.JSONDecodeError as json_error:
            # heuristic to confirm that the content is likely malformed JSON
            if content.strip().startswith(("{", "[")):
                raise json_error

        try:
            open_api_spec_content = yaml.safe_load(content)
        except yaml.YAMLError:
            error_message = (
                "Failed to parse the OpenAPI specification. The content does not appear to be valid JSON or YAML.\n\n"
            )
            raise RuntimeError(error_message, content)

        # Replace references in the object with their resolved values, if any
        return jsonref.replace_refs(open_api_spec_content)
