# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import logging

from opentelemetry import metrics
from opentelemetry.configuration._common import (
    _map_compression,
    _parse_headers,
    _parse_otlp_file_output_stream,
    load_entry_point,
)
from opentelemetry.configuration._exceptions import (
    ConfigurationError,
    MissingDependencyError,
)
from opentelemetry.configuration.models import (
    Aggregation as AggregationConfig,
)
from opentelemetry.configuration.models import (
    ConsoleMetricExporter as ConsoleMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    ExemplarFilter as ExemplarFilterConfig,
)
from opentelemetry.configuration.models import (
    ExperimentalOtlpFileMetricExporter as ExperimentalOtlpFileMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    ExperimentalPrometheusMetricExporter as PrometheusMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    ExporterDefaultHistogramAggregation,
    ExporterTemporalityPreference,
    InstrumentType,
)
from opentelemetry.configuration.models import (
    MeterProvider as MeterProviderConfig,
)
from opentelemetry.configuration.models import (
    MetricReader as MetricReaderConfig,
)
from opentelemetry.configuration.models import (
    OtlpGrpcMetricExporter as OtlpGrpcMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    OtlpHttpMetricExporter as OtlpHttpMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    PeriodicMetricReader as PeriodicMetricReaderConfig,
)
from opentelemetry.configuration.models import (
    PullMetricExporter as PullMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    PullMetricReader as PullMetricReaderConfig,
)
from opentelemetry.configuration.models import (
    PushMetricExporter as PushMetricExporterConfig,
)
from opentelemetry.configuration.models import (
    View as ViewConfig,
)
from opentelemetry.sdk.metrics import (
    AlwaysOffExemplarFilter,
    AlwaysOnExemplarFilter,
    Counter,
    Histogram,
    MeterProvider,
    ObservableCounter,
    ObservableGauge,
    ObservableUpDownCounter,
    TraceBasedExemplarFilter,
    UpDownCounter,
    _Gauge,
)
from opentelemetry.sdk.metrics.export import (
    AggregationTemporality,
    ConsoleMetricExporter,
    MetricExporter,
    MetricReader,
    PeriodicExportingMetricReader,
)
from opentelemetry.sdk.metrics.view import (
    Aggregation,
    DefaultAggregation,
    DropAggregation,
    ExplicitBucketHistogramAggregation,
    ExponentialBucketHistogramAggregation,
    LastValueAggregation,
    SumAggregation,
    View,
)
from opentelemetry.sdk.resources import Resource

_logger = logging.getLogger(__name__)


# Default interval/timeout per OTel spec (milliseconds).
_DEFAULT_EXPORT_INTERVAL_MILLIS = 60000
_DEFAULT_EXPORT_TIMEOUT_MILLIS = 30000

# Instrument type → SDK instrument class mapping (for View selectors).
_INSTRUMENT_TYPE_MAP: dict[InstrumentType, type] = {
    InstrumentType.counter: Counter,
    InstrumentType.up_down_counter: UpDownCounter,
    InstrumentType.histogram: Histogram,
    InstrumentType.gauge: _Gauge,
    InstrumentType.observable_counter: ObservableCounter,
    InstrumentType.observable_gauge: ObservableGauge,
    InstrumentType.observable_up_down_counter: ObservableUpDownCounter,
}


def _map_temporality(
    pref: ExporterTemporalityPreference | None,
) -> dict[type, AggregationTemporality]:
    """Map a temporality preference to an explicit preferred_temporality dict.

    Always returns an explicit dict to suppress OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE.
    Default (None or cumulative) → all instruments CUMULATIVE.
    """
    if pref is None or pref == ExporterTemporalityPreference.cumulative:
        return {
            Counter: AggregationTemporality.CUMULATIVE,
            UpDownCounter: AggregationTemporality.CUMULATIVE,
            Histogram: AggregationTemporality.CUMULATIVE,
            ObservableCounter: AggregationTemporality.CUMULATIVE,
            ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
            ObservableGauge: AggregationTemporality.CUMULATIVE,
        }
    if pref == ExporterTemporalityPreference.delta:
        return {
            Counter: AggregationTemporality.DELTA,
            UpDownCounter: AggregationTemporality.CUMULATIVE,
            Histogram: AggregationTemporality.DELTA,
            ObservableCounter: AggregationTemporality.DELTA,
            ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
            ObservableGauge: AggregationTemporality.CUMULATIVE,
        }
    if pref == ExporterTemporalityPreference.low_memory:
        return {
            Counter: AggregationTemporality.DELTA,
            UpDownCounter: AggregationTemporality.CUMULATIVE,
            Histogram: AggregationTemporality.DELTA,
            ObservableCounter: AggregationTemporality.CUMULATIVE,
            ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
            ObservableGauge: AggregationTemporality.CUMULATIVE,
        }
    raise ConfigurationError(
        f"Unsupported temporality preference '{pref}'. Supported values: cumulative, delta, low_memory."
    )


def _map_histogram_aggregation(
    pref: ExporterDefaultHistogramAggregation | None,
) -> dict[type, Aggregation]:
    """Map a histogram aggregation preference to an explicit preferred_aggregation dict.

    Always returns an explicit dict to suppress
    OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION.
    Default (None or explicit_bucket_histogram) → ExplicitBucketHistogramAggregation.
    """
    if pref is None or pref == ExporterDefaultHistogramAggregation.explicit_bucket_histogram:
        return {Histogram: ExplicitBucketHistogramAggregation()}
    if pref == ExporterDefaultHistogramAggregation.base2_exponential_bucket_histogram:
        return {Histogram: ExponentialBucketHistogramAggregation()}
    raise ConfigurationError(
        f"Unsupported default histogram aggregation '{pref}'. "
        "Supported values: explicit_bucket_histogram, base2_exponential_bucket_histogram."
    )


def _create_aggregation(config: AggregationConfig) -> Aggregation:
    """Create an SDK Aggregation from config, passing through detail parameters."""
    if config.default is not None:
        return DefaultAggregation()
    if config.drop is not None:
        return DropAggregation()
    if config.explicit_bucket_histogram is not None:
        return ExplicitBucketHistogramAggregation(
            boundaries=config.explicit_bucket_histogram.boundaries,
            record_min_max=(
                config.explicit_bucket_histogram.record_min_max
                if config.explicit_bucket_histogram.record_min_max is not None
                else True
            ),
        )
    if config.base2_exponential_bucket_histogram is not None:
        kwargs = {}
        if config.base2_exponential_bucket_histogram.max_size is not None:
            kwargs["max_size"] = config.base2_exponential_bucket_histogram.max_size
        if config.base2_exponential_bucket_histogram.max_scale is not None:
            kwargs["max_scale"] = config.base2_exponential_bucket_histogram.max_scale
        if config.base2_exponential_bucket_histogram.record_min_max is not None:
            kwargs["record_min_max"] = config.base2_exponential_bucket_histogram.record_min_max
        return ExponentialBucketHistogramAggregation(**kwargs)
    if config.last_value is not None:
        return LastValueAggregation()
    if config.sum is not None:
        return SumAggregation()
    raise ConfigurationError(
        f"Unknown or unsupported aggregation type in config: {config!r}. "
        "Supported types: default, drop, explicit_bucket_histogram, "
        "base2_exponential_bucket_histogram, last_value, sum."
    )


def _create_view(config: ViewConfig) -> View:
    """Create an SDK View from config."""
    selector = config.selector
    stream = config.stream

    instrument_type = None
    if selector.instrument_type is not None:
        instrument_type = _INSTRUMENT_TYPE_MAP.get(selector.instrument_type)
        if instrument_type is None:
            raise ConfigurationError(f"Unknown instrument type: {selector.instrument_type!r}")

    attribute_keys: set[str] | None = None
    if stream.attribute_keys is not None:
        if stream.attribute_keys.excluded:
            _logger.warning(
                "attribute_keys.excluded is not supported by the Python SDK View; the exclusion list will be ignored."
            )
        if stream.attribute_keys.included is not None:
            attribute_keys = set(stream.attribute_keys.included)

    aggregation = None
    if stream.aggregation is not None:
        aggregation = _create_aggregation(stream.aggregation)

    return View(
        instrument_type=instrument_type,
        instrument_name=selector.instrument_name,
        meter_name=selector.meter_name,
        meter_version=selector.meter_version,
        meter_schema_url=selector.meter_schema_url,
        instrument_unit=selector.unit,
        name=stream.name,
        description=stream.description,
        attribute_keys=attribute_keys,
        aggregation=aggregation,
    )


def _create_console_metric_exporter(
    config: ConsoleMetricExporterConfig,
) -> MetricExporter:
    """Create a ConsoleMetricExporter from config."""
    preferred_temporality = _map_temporality(config.temporality_preference)
    preferred_aggregation = _map_histogram_aggregation(config.default_histogram_aggregation)
    return ConsoleMetricExporter(
        preferred_temporality=preferred_temporality,
        preferred_aggregation=preferred_aggregation,
    )


def _create_otlp_http_metric_exporter(
    config: OtlpHttpMetricExporterConfig,
) -> MetricExporter:
    """Create an OTLP HTTP metric exporter from config."""
    try:
        # pylint: disable=import-outside-toplevel,no-name-in-module
        from opentelemetry.exporter.otlp.proto.http import (  # type: ignore[import-untyped]  # noqa: PLC0415
            Compression,
        )
        from opentelemetry.exporter.otlp.proto.http.metric_exporter import (  # type: ignore[import-untyped]  # noqa: PLC0415
            OTLPMetricExporter,
        )
    except ImportError as exc:
        raise MissingDependencyError(
            package="opentelemetry-exporter-otlp-proto-http",
            feature="otlp_http metric exporter",
        ) from exc

    compression = _map_compression(config.compression, Compression, allow_deflate=True)
    headers = _parse_headers(config.headers, config.headers_list)
    timeout = (config.timeout / 1000.0) if config.timeout is not None else None
    preferred_temporality = _map_temporality(config.temporality_preference)
    preferred_aggregation = _map_histogram_aggregation(config.default_histogram_aggregation)

    return OTLPMetricExporter(  # type: ignore[return-value]
        endpoint=config.endpoint,
        headers=headers,
        timeout=timeout,
        compression=compression,  # type: ignore[arg-type]
        preferred_temporality=preferred_temporality,
        preferred_aggregation=preferred_aggregation,
    )


def _create_otlp_grpc_metric_exporter(
    config: OtlpGrpcMetricExporterConfig,
) -> MetricExporter:
    """Create an OTLP gRPC metric exporter from config."""
    try:
        # pylint: disable=import-outside-toplevel,no-name-in-module
        import grpc  # type: ignore[import-untyped]  # noqa: PLC0415

        from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (  # type: ignore[import-untyped]  # noqa: PLC0415
            OTLPMetricExporter,
        )
    except ImportError as exc:
        raise MissingDependencyError(
            package="opentelemetry-exporter-otlp-proto-grpc",
            feature="otlp_grpc metric exporter",
        ) from exc

    compression = _map_compression(config.compression, grpc.Compression)
    headers = _parse_headers(config.headers, config.headers_list)
    timeout = (config.timeout / 1000.0) if config.timeout is not None else None
    preferred_temporality = _map_temporality(config.temporality_preference)
    preferred_aggregation = _map_histogram_aggregation(config.default_histogram_aggregation)

    return OTLPMetricExporter(  # type: ignore[return-value]
        endpoint=config.endpoint,
        headers=headers,
        timeout=timeout,
        compression=compression,  # type: ignore[arg-type]
        preferred_temporality=preferred_temporality,
        preferred_aggregation=preferred_aggregation,
    )


def _create_otlp_file_development_metric_exporter(
    config: ExperimentalOtlpFileMetricExporterConfig,
) -> MetricExporter:
    """Create an OTLP file (JSON Lines) metric exporter from config."""
    try:
        # pylint: disable=import-outside-toplevel,no-name-in-module
        from opentelemetry.exporter.otlp.json.file.metric_exporter import (  # type: ignore[import-untyped]  # noqa: PLC0415
            FileMetricExporter,
        )
    except ImportError as exc:
        raise ConfigurationError(
            "otlp_file_development metric exporter requires 'opentelemetry-exporter-otlp-json-file'. "
            "Install it with: pip install opentelemetry-exporter-otlp-json-file"
        ) from exc

    path = _parse_otlp_file_output_stream(config.output_stream)
    preferred_temporality = _map_temporality(config.temporality_preference)
    preferred_aggregation = _map_histogram_aggregation(config.default_histogram_aggregation)
    return (
        FileMetricExporter(
            path,
            preferred_temporality=preferred_temporality,
            preferred_aggregation=preferred_aggregation,
        )
        if path is not None
        else FileMetricExporter(
            preferred_temporality=preferred_temporality,
            preferred_aggregation=preferred_aggregation,
        )
    )


_METRIC_EXPORTER_REGISTRY: dict = {
    "otlp_http": _create_otlp_http_metric_exporter,
    "otlp_grpc": _create_otlp_grpc_metric_exporter,
    "console": _create_console_metric_exporter,
    "otlp_file_development": _create_otlp_file_development_metric_exporter,
}


def _create_push_metric_exporter(
    config: PushMetricExporterConfig,
) -> MetricExporter:
    """Create a push metric exporter from config.

    Known exporter types are checked via typed fields on the PushMetricExporter
    dataclass. Unknown exporter names captured in additional_properties
    by the @_additional_properties decorator are loaded via the
    ``opentelemetry_metrics_exporter`` entry point group.
    """
    for name, factory in _METRIC_EXPORTER_REGISTRY.items():
        value = getattr(config, name, None)
        if value is not None:
            return factory(value)
    if config.additional_properties:
        name, plugin_config = next(iter(config.additional_properties.items()))
        return load_entry_point("opentelemetry_metrics_exporter", name)(**(plugin_config or {}))
    raise ConfigurationError(
        "No exporter type specified in push metric exporter config. "
        "Supported types: console, otlp_http, otlp_grpc, otlp_file_development."
    )


def _create_periodic_metric_reader(
    config: PeriodicMetricReaderConfig,
) -> PeriodicExportingMetricReader:
    """Create a PeriodicExportingMetricReader from config.

    Passes explicit interval/timeout defaults to suppress env var reading.
    """
    exporter = _create_push_metric_exporter(config.exporter)
    interval = config.interval if config.interval is not None else _DEFAULT_EXPORT_INTERVAL_MILLIS
    timeout = config.timeout if config.timeout is not None else _DEFAULT_EXPORT_TIMEOUT_MILLIS
    return PeriodicExportingMetricReader(
        exporter=exporter,
        export_interval_millis=float(interval),
        export_timeout_millis=float(timeout),
    )


def _create_prometheus_metric_reader(
    config: PrometheusMetricExporterConfig,
) -> MetricReader:
    """Create a PrometheusMetricReader from config.

    Dynamically imports the prometheus exporter package to avoid a hard
    dependency. Maps config fields to constructor parameters and starts
    the HTTP server.
    """
    try:
        # pylint: disable=import-outside-toplevel,no-name-in-module
        from opentelemetry.exporter.prometheus import (  # type: ignore[import-untyped]  # noqa: PLC0415
            PrometheusMetricReader,
            start_http_server,
        )
    except ImportError as exc:
        raise MissingDependencyError(
            package="opentelemetry-exporter-prometheus",
            feature="prometheus pull metric exporter",
        ) from exc

    disable_target_info = (
        config.target_info_enabled_development is not None and not config.target_info_enabled_development
    )

    if config.scope_info_enabled is not None:
        _logger.warning("scope_info_enabled is not yet supported for Prometheus metric exporter and will be ignored.")
    if config.resource_constant_labels is not None:
        _logger.warning(
            "resource_constant_labels is not yet supported for Prometheus metric exporter and will be ignored."
        )

    reader = PrometheusMetricReader(
        disable_target_info=disable_target_info,
    )

    port = config.port if config.port is not None else 9464
    host = config.host if config.host is not None else "localhost"
    start_http_server(port=port, addr=host)

    return reader


def _create_pull_metric_exporter(
    config: PullMetricExporterConfig,
) -> MetricReader:
    """Create a pull metric exporter (which is itself a MetricReader) from config.

    Pull metric exporters like Prometheus are combined reader+exporter objects:
    the "exporter" IS the reader. The config schema models them as separate
    exporter configs, but the factory returns a MetricReader.

    Plugin pull exporters are loaded via the ``opentelemetry_pull_metric_exporter``
    entry point group.
    """
    if config.prometheus_development is not None:
        return _create_prometheus_metric_reader(config.prometheus_development)
    if config.additional_properties:
        name, plugin_config = next(iter(config.additional_properties.items()))
        return load_entry_point("opentelemetry_pull_metric_exporter", name)(**(plugin_config or {}))
    raise ConfigurationError(
        "No exporter type specified in pull metric exporter config. Supported types: prometheus_development."
    )


def _create_pull_metric_reader(
    config: PullMetricReaderConfig,
) -> MetricReader:
    """Create a pull MetricReader from config.

    The pull reader's exporter is itself a MetricReader (combined reader+exporter).
    producers and cardinality_limits are not yet supported.
    """
    if config.producers:
        _logger.warning(
            "MetricProducer configuration is not yet supported for pull metric readers and will be ignored."
        )
    if config.cardinality_limits is not None:
        _logger.warning("cardinality_limits is not yet supported for pull metric readers and will be ignored.")
    return _create_pull_metric_exporter(config.exporter)


def _create_metric_reader(config: MetricReaderConfig) -> MetricReader:
    """Create a MetricReader from config."""
    if config.periodic is not None:
        return _create_periodic_metric_reader(config.periodic)
    if config.pull is not None:
        return _create_pull_metric_reader(config.pull)
    raise ConfigurationError("No reader type specified in metric reader config. Supported types: periodic, pull.")


def _create_exemplar_filter(
    value: ExemplarFilterConfig,
) -> object:
    """Create an SDK exemplar filter from config enum value."""
    if value == ExemplarFilterConfig.always_on:
        return AlwaysOnExemplarFilter()
    if value == ExemplarFilterConfig.always_off:
        return AlwaysOffExemplarFilter()
    if value == ExemplarFilterConfig.trace_based:
        return TraceBasedExemplarFilter()
    raise ConfigurationError(
        f"Unknown exemplar filter value: {value!r}. Supported values: always_on, always_off, trace_based."
    )


def create_meter_provider(
    config: MeterProviderConfig | None,
    resource: Resource | None = None,
) -> MeterProvider:
    """Create an SDK MeterProvider from declarative config.

    Does NOT read OTEL_METRIC_EXPORT_INTERVAL, OTEL_METRICS_EXEMPLAR_FILTER,
    or any other env vars for values explicitly controlled by the config.
    Absent config values use OTel spec defaults, matching Java SDK behavior.

    Args:
        config: MeterProvider config from the parsed config file, or None.
        resource: Resource to attach to the provider.

    Returns:
        A configured MeterProvider.
    """
    # Always pass an explicit exemplar filter to suppress env var reading.
    # Spec default is trace_based.
    exemplar_filter: object = TraceBasedExemplarFilter()
    if config is not None and config.exemplar_filter is not None:
        exemplar_filter = _create_exemplar_filter(config.exemplar_filter)

    readers: list[MetricReader] = []
    views: list[View] = []

    if config is not None:
        for reader_config in config.readers:
            readers.append(_create_metric_reader(reader_config))
        if config.views:
            for view_config in config.views:
                views.append(_create_view(view_config))

    return MeterProvider(
        resource=resource,
        metric_readers=readers,
        exemplar_filter=exemplar_filter,  # type: ignore[arg-type]
        views=views,
    )


def configure_meter_provider(
    config: MeterProviderConfig | None,
    resource: Resource | None = None,
) -> None:
    """Configure the global MeterProvider from declarative config.

    When config is None (meter_provider section absent from config file),
    the global is not set — matching Java/JS SDK behavior.

    Args:
        config: MeterProvider config from the parsed config file, or None.
        resource: Resource to attach to the provider.
    """
    if config is None:
        return
    metrics.set_meter_provider(create_meter_provider(config, resource))
