import os
import sys
import warnings
from copy import copy, deepcopy
from collections import deque
from contextlib import contextmanager
from enum import Enum
from datetime import datetime, timezone
from functools import wraps
from itertools import chain

import sentry_sdk
from sentry_sdk._types import AnnotatedValue
from sentry_sdk.attachments import Attachment
from sentry_sdk.consts import (
    DEFAULT_MAX_BREADCRUMBS,
    FALSE_VALUES,
    INSTRUMENTER,
    SPANDATA,
)
from sentry_sdk.feature_flags import FlagBuffer, DEFAULT_FLAG_CAPACITY
from sentry_sdk.profiler.continuous_profiler import (
    get_profiler_id,
    try_autostart_continuous_profiler,
    try_profile_lifecycle_trace_start,
)
from sentry_sdk.profiler.transaction_profiler import Profile
from sentry_sdk.session import Session
from sentry_sdk.tracing_utils import (
    Baggage,
    has_tracing_enabled,
    has_span_streaming_enabled,
    is_ignored_span,
    _make_sampling_decision,
    PropagationContext,
)
from sentry_sdk.traces import _DEFAULT_PARENT_SPAN, StreamedSpan, NoOpStreamedSpan
from sentry_sdk.tracing import (
    BAGGAGE_HEADER_NAME,
    SENTRY_TRACE_HEADER_NAME,
    NoOpSpan,
    Span,
    Transaction,
)
from sentry_sdk.utils import (
    capture_internal_exception,
    capture_internal_exceptions,
    ContextVar,
    datetime_from_isoformat,
    disable_capture_event,
    event_from_exception,
    exc_info_from_error,
    format_attribute,
    logger,
    has_logs_enabled,
    has_metrics_enabled,
)

from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
    from collections.abc import Mapping

    from typing import Any
    from typing import Callable
    from typing import Deque
    from typing import Dict
    from typing import Generator
    from typing import Iterator
    from typing import List
    from typing import Optional
    from typing import ParamSpec
    from typing import Tuple
    from typing import TypeVar
    from typing import Union

    from typing_extensions import Unpack

    from sentry_sdk._types import (
        Attributes,
        AttributeValue,
        Breadcrumb,
        BreadcrumbHint,
        ErrorProcessor,
        Event,
        EventProcessor,
        ExcInfo,
        Hint,
        Log,
        LogLevelStr,
        Metric,
        SamplingContext,
        Type,
    )

    from sentry_sdk.tracing import TransactionKwargs

    import sentry_sdk

    P = ParamSpec("P")
    R = TypeVar("R")

    F = TypeVar("F", bound=Callable[..., Any])
    T = TypeVar("T")


# Holds data that will be added to **all** events sent by this process.
# In case this is a http server (think web framework) with multiple users
# the data will be added to events of all users.
# Typically this is used for process wide data such as the release.
_global_scope: "Optional[Scope]" = None

# Holds data for the active request.
# This is used to isolate data for different requests or users.
# The isolation scope is usually created by integrations, but may also
# be created manually
_isolation_scope = ContextVar("isolation_scope", default=None)

# Holds data for the active span.
# This can be used to manually add additional data to a span.
_current_scope = ContextVar("current_scope", default=None)

global_event_processors: "List[EventProcessor]" = []

# A function returning a (trace_id, span_id) tuple
# from an external tracing source (such as otel)
_external_propagation_context_fn: "Optional[Callable[[], Optional[Tuple[str, str]]]]" = None


class ScopeType(Enum):
    CURRENT = "current"
    ISOLATION = "isolation"
    GLOBAL = "global"
    MERGED = "merged"


class _ScopeManager:
    def __init__(self, hub: "Optional[Any]" = None) -> None:
        self._old_scopes: "List[Scope]" = []

    def __enter__(self) -> "Scope":
        isolation_scope = Scope.get_isolation_scope()

        self._old_scopes.append(isolation_scope)

        forked_scope = isolation_scope.fork()
        _isolation_scope.set(forked_scope)

        return forked_scope

    def __exit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
        old_scope = self._old_scopes.pop()
        _isolation_scope.set(old_scope)


def add_global_event_processor(processor: "EventProcessor") -> None:
    global_event_processors.append(processor)


def register_external_propagation_context(
    fn: "Callable[[], Optional[Tuple[str, str]]]",
) -> None:
    global _external_propagation_context_fn
    _external_propagation_context_fn = fn


def remove_external_propagation_context() -> None:
    global _external_propagation_context_fn
    _external_propagation_context_fn = None


def get_external_propagation_context() -> "Optional[Tuple[str, str]]":
    return (
        _external_propagation_context_fn() if _external_propagation_context_fn else None
    )


def has_external_propagation_context() -> bool:
    return _external_propagation_context_fn is not None


def _attr_setter(fn: "Any") -> "Any":
    return property(fset=fn, doc=fn.__doc__)


def _disable_capture(fn: "F") -> "F":
    @wraps(fn)
    def wrapper(self: "Any", *args: "Dict[str, Any]", **kwargs: "Any") -> "Any":
        if not self._should_capture:
            return
        try:
            self._should_capture = False
            return fn(self, *args, **kwargs)
        finally:
            self._should_capture = True

    return wrapper  # type: ignore


class Scope:
    """The scope holds extra information that should be sent with all
    events that belong to it.
    """

    # NOTE: Even though it should not happen, the scope needs to not crash when
    # accessed by multiple threads. It's fine if it's full of races, but those
    # races should never make the user application crash.
    #
    # The same needs to hold for any accesses of the scope the SDK makes.

    __slots__ = (
        "_level",
        "_name",
        "_fingerprint",
        # note that for legacy reasons, _transaction is the transaction *name*,
        # not a Transaction object (the object is stored in _span)
        "_transaction",
        "_transaction_info",
        "_user",
        "_tags",
        "_contexts",
        "_extras",
        "_breadcrumbs",
        "_n_breadcrumbs_truncated",
        "_gen_ai_original_message_count",
        "_gen_ai_conversation_id",
        "_event_processors",
        "_error_processors",
        "_should_capture",
        "_span",
        "_session",
        "_attachments",
        "_force_auto_session_tracking",
        "_profile",
        "_propagation_context",
        "client",
        "_type",
        "_last_event_id",
        "_flags",
        "_attributes",
    )

    def __init__(
        self,
        ty: "Optional[ScopeType]" = None,
        client: "Optional[sentry_sdk.Client]" = None,
    ) -> None:
        self._type = ty

        self._event_processors: "List[EventProcessor]" = []
        self._error_processors: "List[ErrorProcessor]" = []

        self._name: "Optional[str]" = None
        self._propagation_context: "Optional[PropagationContext]" = None
        self._n_breadcrumbs_truncated: int = 0
        self._gen_ai_original_message_count: "Dict[str, int]" = {}

        self.client: "sentry_sdk.client.BaseClient" = NonRecordingClient()

        if client is not None:
            self.set_client(client)

        self.clear()

        incoming_trace_information = self._load_trace_data_from_env()
        self.generate_propagation_context(incoming_data=incoming_trace_information)

    def __copy__(self) -> "Scope":
        """
        Returns a copy of this scope.
        This also creates a copy of all referenced data structures.
        """
        rv: "Scope" = object.__new__(self.__class__)

        rv._type = self._type
        rv.client = self.client
        rv._level = self._level
        rv._name = self._name
        rv._fingerprint = self._fingerprint
        rv._transaction = self._transaction
        rv._transaction_info = self._transaction_info.copy()
        rv._user = self._user

        rv._tags = self._tags.copy()
        rv._contexts = self._contexts.copy()
        rv._extras = self._extras.copy()

        rv._breadcrumbs = copy(self._breadcrumbs)
        rv._n_breadcrumbs_truncated = self._n_breadcrumbs_truncated
        rv._gen_ai_original_message_count = self._gen_ai_original_message_count.copy()
        rv._event_processors = self._event_processors.copy()
        rv._error_processors = self._error_processors.copy()
        rv._propagation_context = self._propagation_context

        rv._should_capture = self._should_capture
        rv._span = self._span
        rv._session = self._session
        rv._force_auto_session_tracking = self._force_auto_session_tracking
        rv._attachments = self._attachments.copy()

        rv._profile = self._profile

        rv._last_event_id = self._last_event_id

        rv._flags = deepcopy(self._flags)

        rv._attributes = self._attributes.copy()

        rv._gen_ai_conversation_id = self._gen_ai_conversation_id

        return rv

    @classmethod
    def get_current_scope(cls) -> "Scope":
        """
        .. versionadded:: 2.0.0

        Returns the current scope.
        """
        current_scope = _current_scope.get()
        if current_scope is None:
            current_scope = Scope(ty=ScopeType.CURRENT)
            _current_scope.set(current_scope)

        return current_scope