"""Function introspection and schema generation for FastMCP tools."""

from __future__ import annotations

import functools
import inspect
import types
from collections.abc import Callable
from dataclasses import dataclass
from typing import Annotated, Any, Generic, Union, get_args, get_origin, get_type_hints

import mcp.types
from pydantic import PydanticSchemaGenerationError
from typing_extensions import TypeVar as TypeVarExt

from fastmcp.server.dependencies import (
    transform_context_annotations,
    without_injected_parameters,
)
from fastmcp.tools.base import ToolResult
from fastmcp.utilities.docstring_parsing import ParsedDocstring, parse_docstring
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
    Audio,
    File,
    Image,
    create_function_without_params,
    get_cached_typeadapter,
    is_class_member_of_type,
    replace_type,
)

try:
    from prefab_ui.app import PrefabApp as _PrefabApp
    from prefab_ui.components.base import Component as _PrefabComponent

    _PREFAB_TYPES: tuple[type, ...] = (_PrefabApp, _PrefabComponent)
except ImportError:
    _PREFAB_TYPES = ()


def _contains_bytes_type(tp: Any) -> bool:
    """Check if *tp* is or contains bytes, recursing through unions and Annotated."""
    if tp is bytes:
        return True
    origin = get_origin(tp)
    if origin is Union or origin is types.UnionType or origin is Annotated:
        return any(_contains_bytes_type(a) for a in get_args(tp))
    return False


def _contains_prefab_type(tp: Any) -> bool:
    """Check if *tp* is or contains a prefab type, recursing through unions and Annotated."""
    if isinstance(tp, type) and issubclass(tp, _PREFAB_TYPES):
        return True
    origin = get_origin(tp)
    if origin is Union or origin is types.UnionType or origin is Annotated:
        return any(_contains_prefab_type(a) for a in get_args(tp))
    return False


T = TypeVarExt("T", default=Any)

logger = get_logger(__name__)


@dataclass
class _WrappedResult(Generic[T]):
    result: T


class _UnserializableType:
    pass


def _is_object_schema(
    schema: dict[str, Any],
    *,
    _root_schema: dict[str, Any] | None = None,
    _seen_refs: set[str] | None = None,
) -> bool:
    """Check if a JSON schema represents an object type."""
    root_schema = _root_schema or schema
    seen_refs = _seen_refs or set()

    # Direct object type
    if schema.get("type") == "object":
        return True

    # Schema with properties but no explicit type is treated as object
    if "properties" in schema:
        return True

    # Resolve local $ref definitions and recurse into the target schema.
    ref = schema.get("$ref")
    if not isinstance(ref, str) or not ref.startswith("#/"):
        return False

    if ref in seen_refs:
        return False

    # Walk the JSON Pointer path from the root schema, unescaping each
    # token per RFC 6901 (~1 → /, ~0 → ~).
    pointer = ref.removeprefix("#/")
    segments = pointer.split("/")
    target: Any = root_schema
    for segment in segments:
        unescaped = segment.replace("~1", "/").replace("~0", "~")
        if not isinstance(target, dict) or unescaped not in target:
            return False
        target = target[unescaped]

    target_schema = target
    if not isinstance(target_schema, dict):
        return False

    return _is_object_schema(
        target_schema,
        _root_schema=root_schema,
        _seen_refs=seen_refs | {ref},
    )


@dataclass
class ParsedFunction:
    fn: Callable[..., Any]
    name: str
    description: str | None
    input_schema: dict[str, Any]
    output_schema: dict[str, Any] | None
    return_type: Any = None

    @classmethod
    def from_function(
        cls,
        fn: Callable[..., Any],
        exclude_args: list[str] | None = None,
        validate: bool = True,
        wrap_non_object_output_schema: bool = True,
    ) -> ParsedFunction:
        if validate:
            sig = inspect.signature(fn)
            # Reject functions with *args or **kwargs
            for param in sig.parameters.values():
                if param.kind == inspect.Parameter.VAR_POSITIONAL:
                    raise ValueError("Functions with *args are not supported as tools")
                if param.kind == inspect.Parameter.VAR_KEYWORD:
                    raise ValueError(
                        "Functions with **kwargs are not supported as tools"
                    )

            # Reject exclude_args that don't exist in the function or don't have a default value
            if exclude_args:
                for arg_name in exclude_args:
                    if arg_name not in sig.parameters:
                        raise ValueError(
                            f"Parameter '{arg_name}' in exclude_args does not exist in function."
                        )
                    param = sig.parameters[arg_name]
                    if param.default == inspect.Parameter.empty:
                        raise ValueError(
                            f"Parameter '{arg_name}' in exclude_args must have a default value."
                        )

        # collect name and description before we potentially modify the function
        fn_name = getattr(fn, "__name__", None) or fn.__class__.__name__
        outer_docstring = parse_docstring(fn)

        # if the fn is a callable class, we need to get the __call__ method from here out
        if not inspect.isroutine(fn) and not isinstance(fn, functools.partial):
            fn = fn.__call__
        # if the fn is a staticmethod, we need to work with the underlying function
        if isinstance(fn, staticmethod):
            fn = fn.__func__

        # For callable classes, parameter descriptions must come from
        # __call__'s docstring — where the exposed parameters are actually
        # declared. The class docstring's Args section, if any, typically
        # describes __init__, so falling back to it would risk injecting
        # constructor docs into __call__'s schema on overlapping names.
        # The description, however, comes from the class docstring (which
        # describes what the tool IS) when present.
        inner_docstring = parse_docstring(fn)
        parsed_docstring = ParsedDocstring(
            description=outer_docstring.description or inner_docstring.description,
            parameters=inner_docstring.parameters,
        )

        # Transform Context type annotations to Depends() for unified DI
        fn = transform_context_annotations(fn)

        # Handle injected parameters (Context, Docket dependencies)
        wrapper_fn = without_injected_parameters(fn)

        # Also handle exclude_args with non-serializable types (issue #2431)
        # This must happen before Pydantic tries to serialize the parameters
        if exclude_args:
            wrapper_fn = create_function_without_params(wrapper_fn, list(exclude_args))

        input_type_adapter = get_cached_typeadapter(wrapper_fn)
        input_schema = input_type_adapter.json_schema()

        # Compress and handle exclude_args
        prune_params = list(exclude_args) if exclude_args else None
        input_schema = compress_schema(
            input_schema, prune_params=prune_params, prune_titles=True
        )

        # Inject parameter descriptions from the docstring into the schema.
        # Explicit annotations (Field(description=...), Annotated[x, "..."])
        # already have a "description" key and take precedence.
        if parsed_docstring.parameters:
            properties = input_schema.get("properties", {})
            for param_name, param_desc in parsed_docstring.parameters.items():
                if (
                    param_name in properties
                    and "description" not in properties[param_name]
                ):
                    properties[param_name]["description"] = param_desc

        output_schema = None
        # Get the return annotation from the signature
        sig = inspect.signature(fn)
        output_type = sig.return_annotation

        # If the annotation is a string (from __future__ annotations), resolve it
        if isinstance(output_type, str):
            try:
                # Use get_type_hints to resolve the return type
                # include_extras=True preserves Annotated metadata
                type_hints = get_type_hints(fn, include_extras=True)
                output_type = type_hints.get("return", output_type)
            except Exception as e:
                # If resolution fails, keep the string annotation
                logger.debug("Failed to resolve type hint for return annotation: %s", e)

        # Save original for return_type before any schema-related replacement
        original_output_type = output_type

        if output_type not in (inspect._empty, None, Any, ...):
            # bytes can't be represented as structured JSON output — skip schema
            if _contains_bytes_type(output_type):
                output_type = _UnserializableType

            # Prefab component subclasses (Column, Card, etc.) shouldn't
            # produce output schemas — replace_type only does exact matching,
            # so we handle subclass matching explicitly here.  We also need
            # to handle composite types like ``Column | None`` and
            # ``Annotated[PrefabApp, ...]`` by recursing into their args.
            if _PREFAB_TYPES and _contains_prefab_type(output_type):
                output_type = _UnserializableType

            # ToolResult subclasses should suppress schema generation just
            # like ToolResult itself — replace_type only does exact matching.
            if is_class_member_of_type(output_type, ToolResult):
                output_type = _UnserializableType

            # there are a variety of types that we don't want to attempt to
            # serialize because they are either used by FastMCP internally,
            # or are MCP content types that explicitly don't form structured
            # content. By replacing them with an explicitly unserializable type,
            # we ensure that no output schema is automatically generated.
            clean_output_type = replace_type(
                output_type,
                dict.fromkeys(
                    (
                        Image,
                        Audio,
                        File,
                        ToolResult,
                        mcp.types.TextContent,
                        mcp.types.ImageContent,
                        mcp.types.AudioContent,
                        mcp.types.ResourceLink,
                        mcp.types.EmbeddedResource,
                        *_PREFAB_TYPES,
                    ),
                    _UnserializableType,
                ),
            )

            try:
                type_adapter = get_cached_typeadapter(clean_output_type)
                base_schema = type_adapter.json_schema(mode="serialization")

                # Generate schema for wrapped type if it's non-object
                # because MCP requires that output schemas are objects
                # Check if schema is an object type, resolving $ref references
                # (self-referencing types use $ref at root level)
                if wrap_non_object_output_schema and not _is_object_schema(base_schema):
                    # Use the wrapped result schema directly
                    wrapped_type = _WrappedResult[clean_output_type]
                    wrapped_adapter = get_cached_typeadapter(wrapped_type)
                    output_schema = wrapped_adapter.json_schema(mode="serialization")
                    output_schema["x-fastmcp-wrap-result"] = True
                else:
                    output_schema = base_schema

                output_schema = compress_schema(output_schema, prune_titles=True)

            except PydanticSchemaGenerationError as e:
                if "_UnserializableType" not in str(e):
                    logger.debug(f"Unable to generate schema for type {output_type!r}")

        return cls(
            fn=fn,
            name=fn_name,
            description=parsed_docstring.description,
            input_schema=input_schema,
            output_schema=output_schema or None,
            return_type=original_output_type,
        )
