class Sdk(Interface):
    """
    The SDK used to transmit this event.

    >>> {
    >>>     "name": "sentry.java",
    >>>     "version": "1.7.10",
    >>>     "integrations": ["log4j"],
    >>>     "packages": [
    >>>         {
    >>>             "name": "maven:io.sentry.sentry",
    >>>             "version": "1.7.10",
    >>>         }
    >>>     ]
    >>> }
    """

def __init__(
    self,
    data: MutableMapping[str, Any],
    version: str = "5",
    ...
):

import sentry_sdk

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    environment="production",
    release="myapp@1.0.0",
    traces_sample_rate=0.2,
    max_breadcrumbs=100,
    before_send=strip_sensitive_data,
)

def configure_sdk():
    """
    Setup and initialize the Sentry SDK.
    """
    sdk_options, dsns = _get_sdk_options()

    # 创建传输层
    if dsns.sentry4sentry:
        transport = make_transport(get_options(dsn=dsns.sentry4sentry, **sdk_options))
        sentry4sentry_transport = patch_transport_for_instrumentation(transport, "upstream")

    # 初始化 SDK
    sentry_sdk.init(
        dsn=dsns.sentry4sentry,
        transport=MultiplexingTransport,
        integrations=integrations,
        **sdk_options,
    )

def is_current_event_safe():
    scope = sentry_sdk.get_isolation_scope()
    if scope._tags.get(UNSAFE_TAG):
        return False
    project_id = scope._tags.get("processing_event_for_project")
    if project_id and project_id == settings.SENTRY_PROJECT:
        return False
    ...

from sentry_sdk import Scope, capture_exception, capture_message, isolation_scope

def patch_transport_for_instrumentation(transport, transport_name):
    _send_request = transport._send_request
    if _send_request:
        def patched_send_request(*args, **kwargs):
            metrics.incr(f"internal.sent_requests.{transport_name}.events")
            return _send_request(*args, **kwargs)
        transport._send_request = patched_send_request
    return transport

class MultiplexingTransport(sentry_sdk.transport.Transport):
    """
    Sends all envelopes and events to two Sentry instances:
    - Sentry SaaS (aka Sentry.io) and
    - Sentry4Sentry (aka S4S)
    """
    def _capture_anything(self, method_name, *args, **kwargs):
        if sentry4sentry_transport:
            ...
        if sentry_saas_transport and options.get("store.use-relay-dsn-sample-rate") == 1:
            if is_current_event_safe():
                ...

# ingest/consumer/processors.py:118
deduplication_key = f"ev:{project_id}:{event_id}"
cached_value = cache.get(deduplication_key)
if cached_value is not None:
    return  # 已被处理，跳过

job["recorded_timestamp"] = data.get("timestamp")

# event_manager.py:613-617
metrics.timing(
    "events.latency",
    job["received_timestamp"] - job["recorded_timestamp"],
    tags=metric_tags,
)

job["platform"] = event.platform

job["level"] = level = data.get("level")

job["logger_name"] = logger_name = data.get("logger")

if logger_name:
    set_tag(data, "logger", logger_name)

transaction_name = data.get("transaction")
if transaction_name:
    transaction_name = force_str(transaction_name)
job["transaction"] = transaction_name

def get_context(lineno, context_line, pre_context=None, post_context=None):
    if lineno is None:
        return []
    context = []
    start_lineno = lineno - len(pre_context or [])
    if pre_context:
        for line in pre_context:
            context.append((at_lineno, line))
            at_lineno += 1
    context.append((at_lineno, context_line))
    at_lineno += 1
    if post_context:
        for line in post_context:
            context.append((at_lineno, line))
            at_lineno += 1
    return context

@classmethod
def to_python(cls, data):
    values = []
    for index, crumb in enumerate(get_path(data, "values", filter=True, default=())):
        values.append(cls.normalize_crumb(crumb))
    return super().to_python({"values": values})

@classmethod
def to_python(cls, data):
    data = data.copy()
    for key in ("id", "email", "username", "ip_address", "name", "geo", "data"):
        data.setdefault(key, None)
    if data["geo"] is not None:
        data["geo"] = Geo.to_python(data["geo"])
    return super().to_python(data)

class ContextType:
    context_to_tag_mapping: ClassVar[dict[str, str]] = {}
    """
    This indicates which fields should be promoted into tags during event
    normalization. (See EventManager)
    """

def set_tag(data: dict[str, Any], key: str, value: Any) -> None:
    pop_tag(data, key)
    if value is not None:
        data.setdefault("tags", []).append((key, trim(value, MAX_TAG_VALUE_LENGTH)))

job["release"] = data.get("release")
job["dist"] = data.get("dist")

pop_tag(data, "release")
set_tag(data, "sentry:release", release.version)

def sdk_metadata_from_event(event: Event) -> Mapping[str, Any]:
    if not (sdk_metadata := event.data.get("sdk")):
        return {}
    try:
        return {
            "sdk": {
                "name": sdk_metadata.get("name") or "unknown",
                "name_normalized": normalized_sdk_tag_from_event(event.data),
            }
        }
    except Exception:
        logger.warning("failed to set normalized SDK name", exc_info=True)
        return {}

# SDK 内部实现（简化版）
import sys

_original_excepthook = sys.excepthook

def _sentry_excepthook(exc_type, exc_value, exc_traceback):
    # 上报到 Sentry
    sentry_sdk.capture_exception(exc_value)
    # 调用原始处理器
    _original_excepthook(exc_type, exc_value, exc_traceback)

sys.excepthook = _sentry_excepthook

from sentry_sdk import capture_exception

try:
    result = risky_operation()
except Exception as e:
    capture_exception(e)
    # 可选：添加额外上下文
    with sentry_sdk.push_scope() as scope:
        scope.set_tag("feature", "payment")
        scope.set_extra("order_id", order.id)
        capture_exception(e)

result.setdefault("meta", {})["signal"] = prune_empty_keys({
    "number": posix_signal.get("signal"),
    "code": posix_signal.get("code"),
    "name": posix_signal.get("name"),
    "code_name": posix_signal.get("code_name"),
})

@classmethod
def normalize_crumb(cls, crumb):
    crumb = dict(crumb)
    ts = parse_timestamp(crumb.get("timestamp"))
    if ts:
        crumb["timestamp"] = ts.timestamp()
    else:
        crumb["timestamp"] = None
    for key in ("type", "level", "message", "category", "event_id", "data"):
        crumb.setdefault(key, None)
    return crumb

from sentry_sdk import add_breadcrumb

# 简单面包屑
sentry_sdk.add_breadcrumb(
    category="auth",
    message="User authenticated",
    level="info",
)

# 带数据的详细面包屑
sentry_sdk.add_breadcrumb(
    category="payment",
    message="Payment initiated",
    level="info",
    data={
        "order_id": "12345",
        "amount": 99.99,
        "currency": "USD",
    },
)

# 使用 push_scope 确保面包屑仅在当前 scope 有效
with sentry_sdk.push_scope() as scope:
    scope.add_breadcrumb(
        category="checkout",
        message="Processing checkout",
        level="info",
    )
    process_checkout()  # 此函数内的错误会附带此面包屑

sentry_sdk.set_user({
    "id": "user_42",
    "email": "alice@example.com",
    "username": "alice",
    "ip_address": "",  # SDK 自动填充
})

sentry_sdk.set_tag("feature", "payment")
sentry_sdk.set_tag("shop_id", "123")
sentry_sdk.set_tag("db.instance", "replica-01")

sentry_sdk.set_extra("request_body", request.data)
sentry_sdk.set_extra("user_permissions", user.permissions)
sentry_sdk.set_extra("feature_toggles", feature_flags)

import sentry_sdk

# 1. 全局 scope（进程级别）
sentry_sdk.set_tag("service", "api-server")
sentry_sdk.set_user({"username": "system"})

# 2. 请求 scope（使用 context manager）
def handle_request(request):
    with sentry_sdk.push_scope() as scope:
        scope.set_tag("endpoint", request.path)
        scope.set_user({"id": request.user.id})
        scope.set_extra("request_id", request.id)
        # 此 scope 内的所有错误都会携带上述信息
        process(request)

# 3. 临时 scope（使用新 scope 隔离）
with sentry_sdk.new_scope() as scope:
    scope.set_tag("operation", "background_sync")
    sync_data()
    # 此 scope 在退出时自动销毁

def check_tag_for_scope_bleed(
    tag_key: str, expected_value: str | int, add_to_scope: bool = True
) -> None:
    """
    Detect if the given tag has already been set to a value different than
    what we expect. If we find a mismatch, log a warning and add scope bleed
    tags to the scope.
    """
    scope = sentry_sdk.get_isolation_scope()
    current_value = scope._tags.get(tag_key)
    if not current_value:
        return
    current_value = str(current_value)
    if current_value != expected_value:
        if add_to_scope:
            scope.set_tag("possible_mistag", True)
            scope.set_tag(f"scope_bleed.{tag_key}", True)
        logger.warning("Tag already set and different (%s).", tag_key, extra=...)

def get_all_filter_specs():
    filters = [
        _localhost_filter,
        _browser_extensions_filter,
        _legacy_browsers_filter,
        _web_crawlers_filter,
        _healthcheck_filter,
    ]
    return tuple(filters)

@trace_func(name="ingest_consumer.process_event")
def process_event(
    consumer_type: str,
    message: IngestMessage,
    project: Project,
    ...
) -> None:
    """
    Perform some initial filtering and deserialize the message payload.
    """
    # 1. 提取消息字段
    payload = message["payload"]
    start_time = float(message["start_time"])
    event_id = message["event_id"]
    project_id = int(message["project_id"])
    attachments = message.get("attachments") or ()

    # 2. 去重检查
    deduplication_key = f"ev:{project_id}:{event_id}"
    cached_value = cache.get(deduplication_key)
    if cached_value is not None:
        return  # 已被处理，跳过

    # 3. Killswitch 检查
    if killswitch_matches_context("store.load-shed-pipeline-projects", ...):
        return

    # 4. 解析 JSON 载荷
    data = orjson.loads(payload)

    # 5. 根据事件类型分发
    if consumer_type == ConsumerType.Transactions or data.get("type") == "transaction":
        processing_store = transaction_processing_store
    else:
        processing_store = event_processing_store

    # 6. 将事件数据存储到处理存储（Redis）
    cache_key = processing_store.store(data)

    # 7. 按事件类型分发到不同任务
    if data.get("type") == "transaction":
        save_event_transaction.delay(cache_key=cache_key, ...)
    elif data.get("type") == "feedback":
        save_event_feedback.delay(...)
    else:
        preprocess_event(cache_key=cache_key, data=data, ...)

    # 8. 设置去重标记
    cache.set(deduplication_key, "", CACHE_TIMEOUT)

# 简化流程
def preprocess_event(cache_key, data, start_time, event_id, project, ...):
    # 从 Redis 获取事件数据
    event_data = event_processing_store.get(cache_key)

    # 应用过滤器
    if should_filter_event(event_data, project):
        return

    # 调用 EventManager 保存
    manager = EventManager(event_data)
    manager.normalize(project_id=project.id)
    event = manager.save(project_id=project.id)

    # 删除 Redis 存储中的事件数据
    event_processing_store.delete_by_key(cache_key)

def _normalize_impl(self, project_id: int | None = None) -> None:
    from sentry_relay.processing import StoreNormalizer

    rust_normalizer = StoreNormalizer(
        project_id=self._project.id if self._project else project_id,
        client_ip=self._client_ip,
        client=self._auth.client if self._auth else None,
        key_id=str(self._key.id) if self._key else None,
        grouping_config=self._grouping_config,
        protocol_version=str(self.version) if self.version is not None else None,
        is_renormalize=self._is_renormalize,
        remove_other=self._remove_other,
        normalize_user_agent=True,
        sent_at=self.sent_at.isoformat() if self.sent_at is not None else None,
        json_dumps=orjson.dumps,
        **DEFAULT_STORE_NORMALIZER_ARGS,
    )

    self._data = rust_normalizer.normalize_event(dict(self._data), json_loads=orjson.loads)

@trace
def save(self, project_id=None, ...):
    # 确保已规范化
    if not self._normalized:
        self.normalize(project_id=project.id)

    job = {"data": self._data, "project_id": project.id, ...}

    # 提取事件数据
    _pull_out_data([job], projects)

    # 根据事件类型分发
    event_type = self._data.get("type")
    if event_type == "transaction":
        jobs = save_transaction_events([job], projects, ...)
    elif event_type == "generic":
        jobs = save_generic_events([job], projects)
    else:
        return self.save_error_events(project, job, projects, ...)

def _get_or_create_release_many(jobs, projects):
    for job in jobs:
        if data.get("release"):
            release = Release.get_or_create(
                project=project,
                version=data["release"],
                date_added=date,
                create=create_release,
            )
            job["release"] = release
            # 防止冲突：替换原始 release tag 为 sentry:release
            pop_tag(data, "release")
            set_tag(data, "sentry:release", release.version)

job["user"] = _get_event_user(project, data)
if user:
    pop_tag(data, "user")
    set_tag(data, "sentry:user", user.tag_value)

group_info = assign_event_to_group(event=job["event"], job=job, metric_tags=metric_tags)
if not group_info:
    return job["event"]  # 被丢弃或过滤

_nodestore_save_many(jobs=jobs, app_feature="errors")

_eventstream_insert_many(jobs)

def before_send(event, hint):
    # 丢弃特定类型的错误
    if "KeyboardInterrupt" in event.get("exception", {}).get("values", [{}])[0].get("type", ""):
        return None

    # 从 HTTP cookies 中脱敏
    if "request" in event and "cookies" in event["request"]:
        for cookie_key in event["request"]["cookies"]:
            if cookie_key in ("session", "token"):
                event["request"]["cookies"][cookie_key] = "********"

    return event

sentry_sdk.init(
    dsn="...",
    before_send=before_send,
)

def before_send_transaction(event, hint):
    # 丢弃健康检查请求的性能数据
    if event.get("transaction") == "/health":
        return None

    # 速率过高的事件可能触发截断，标记此类事件
    if len(event.get("spans", [])) >= 1000:
        event["tags"]["spans_over_limit"] = "true"

    return event

sentry_sdk.init(
    dsn="...",
    before_send_transaction=before_send_transaction,
)

def my_event_processor(event, hint):
    # 添加自定义数据
    event.setdefault("extra", {})["custom_field"] = "value"
    return event

with sentry_sdk.push_scope() as scope:
    scope.add_event_processor(my_event_processor)
    # 此 scope 内的所有事件都会先经过 my_event_processor 处理

SENSITIVE_KEYS = {"password", "secret", "token", "auth", "credential", "apikey"}

def _is_sensitive(key):
    return any(sensitive in key.lower() for sensitive in SENSITIVE_KEYS)

def scrub_event(event, hint):
    # 脱敏 request data
    if "request" in event and "data" in event["request"]:
        data = event["request"]["data"]
        if isinstance(data, dict):
            for key in list(data.keys()):
                if _is_sensitive(key):
                    data[key] = "[Filtered]"

    # 脱敏 extra
    if "extra" in event:
        for key in list(event["extra"].keys()):
            if _is_sensitive(key):
                event["extra"][key] = "[Filtered]"

    # 脱敏 breadcrumbs data
    if "breadcrumbs" in event:
        for crumb in event["breadcrumbs"].get("values", []):
            if "data" in crumb and isinstance(crumb["data"], dict):
                for key in list(crumb["data"].keys()):
                    if _is_sensitive(key):
                        crumb["data"][key] = "[Filtered]"

    return event

def before_send(event: Event, hint: Hint) -> Event | None:
    if event.get("tags"):
        if settings.SILO_MODE:
            event["tags"]["silo_mode"] = str(settings.SILO_MODE)
        if settings.SENTRY_LOCAL_CELL:
            event["tags"]["sentry_region"] = settings.SENTRY_LOCAL_CELL

    event_exc = hint.get("exc_info", [None])[0]
    if event_exc == asyncio.CancelledError:
        return None  # 丢弃 CancelledError —— 这是正常的协程取消信号
    if event_exc == OperationalError:
        event["level"] = "warning"  # 数据库临时故障降级为 warning

    return event

def before_send_transaction(event: Event, _: Hint) -> Event | None:
    # 丢弃通用 301 重定向的 Transaction
    if (
        event.get("tags", {}).get("http.status_code") == "301"
        and event.get("transaction_info", {}).get("source") == "url"
    ):
        return None

    # 标记 span 数量超过限制的事务
    num_of_spans = len(event["spans"])
    event["tags"]["spans_over_limit"] = str(num_of_spans >= 1000)

    # 设置 span 计数到 trace context
    data = event.setdefault("contexts", {}).setdefault("trace", {}).setdefault("data", {})
    data["num_of_spans"] = num_of_spans

    return event

import sentry_sdk

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    traces_sample_rate=1.0,
    profiles_sample_rate=1.0,
)

# 自动捕获所有未处理异常——无需额外代码
1 / 0  # 此错误会自动上报

# 手动捕获
try:
    raise ValueError("invalid input")
except Exception:
    sentry_sdk.capture_exception()

import sentry_sdk
from sentry_sdk.integrations.django import DjangoIntegration

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    integrations=[DjangoIntegration(
        transaction_style="url",
        middleware_spans=True,
        signals_spans=True,
        cache_spans=True,
    )],
    traces_sample_rate=1.0,
    send_default_pii=True,  # 发送用户 PII
)

import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration

app = Flask(__name__)

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    integrations=[FlaskIntegration(transaction_style="url")],
    traces_sample_rate=1.0,
)

import sentry_sdk
from sentry_sdk.integrations.starlette import StarletteIntegration
from sentry_sdk.integrations.fastapi import FastApiIntegration

app = FastAPI()

sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    integrations=[
        StarletteIntegration(transaction_style="url"),
        FastApiIntegration(transaction_style="url"),
    ],
    traces_sample_rate=1.0,
)

sentry_sdk.init(
    dsn="...",
    sample_rate=1.0,              # 错误事件 100% 采样
    traces_sample_rate=0.1,       # 性能追踪 10% 采样
    profiles_sample_rate=0.05,    # 性能剖析 5% 采样
)

def traces_sampler(sampling_context):
    """
    根据上下文动态决定采样率
    """
    path = sampling_context.get("wsgi_environ", {}).get("PATH_INFO", "")

    # 健康检查端点的采样率设为 0
    if path.startswith("/health"):
        return 0.0

    # 重要事务的采样率提升到 50%
    if path.startswith("/api/payment"):
        return 0.5

    # 默认 10% 采样
    return 0.1

sentry_sdk.init(
    dsn="...",
    traces_sampler=traces_sampler,
)

def traces_sampler(sampling_context):
    wsgi_path = sampling_context.get("wsgi_environ", {}).get("PATH_INFO")
    if wsgi_path and wsgi_path in SAMPLED_ROUTES:
        return SAMPLED_ROUTES[wsgi_path]

    custom_sample_rate = sampling_context.get("sample_rate")
    if custom_sample_rate is not None:
        return float(custom_sample_rate)

    if sampling_context["parent_sampled"] is not None:
        return sampling_context["parent_sampled"]

    if "taskworker" in sampling_context:
        task_name = sampling_context["taskworker"].get("task")
        if task_name in SAMPLED_TASKS:
            return SAMPLED_TASKS[task_name]

    return float(settings.SENTRY_BACKEND_APM_SAMPLING or 0)

sentry_sdk.init(
    dsn="...",
    transport_queue_size=2000,  # 队列最大容量（Sentry 内部设置）
    shutdown_timeout=5,          # 关闭时等待发送完成的超时秒数
)

sdk_options["transport_queue_size"] = 2_000

def before_send(event, hint):
    # 使用正则表达式脱敏
    import re
    def scrub(value):
        if isinstance(value, str):
            return re.sub(r'secret_\w+', '[REDACTED]', value)
        return value

    # 脱敏 request body
    if "request" in event and "data" in event["request"]:
        if isinstance(event["request"]["data"], dict):
            for k, v in event["request"]["data"].items():
                event["request"]["data"][k] = scrub(v)
    return event

sentry_sdk.init(dsn="...", max_request_body_size="small")  # 仅捕获前几千字节

def traces_sampler(ctx):
    if ctx.get("wsgi_environ", {}).get("PATH_INFO", "") == "/health":
        return 0.0
    return 0.1

def before_send(event, hint):
    if event.get("transaction") == "/health":
        return None
    return event