# -*- coding: utf-8 -*-
"""
AoqClientSdk Python API (Linux + macOS)

本模块通过 ctypes 加载本地共享库（Linux: ``libAoqClientSdk.so``；
macOS: ``libAoqClientSdk.dylib``），对外暴露与 ``AoqClientEngine.h`` 等价的 Pythonic
接口。

典型使用方式::

    from aoq_client_sdk import (
        AoqClientEngine,
        AoqCreateConfig,
        AoqConnectConfig,
        AoqTrackParam,
        AoqTrackType,
        AoqEncoderType,
    )

    class MyListener(AoqEngineEventListener):
        def on_error(self, code, message):
            print("error:", code, message)

        def on_connection_status_change(self, status):
            print("connection changed:", status)

    cfg = AoqCreateConfig(work_dir="/tmp/aoq")
    engine = AoqClientEngine.create_engine(cfg, MyListener())
    try:
        conn = AoqConnectConfig(
            token="xxx", sid="yyy", certificate="zzz",
            endpoint="1.2.3.4", port=443,
            publish_tracks=[AoqTrackParam(AoqTrackType.AUDIO, AoqEncoderType.AUDIO_OPUS)],
            subscribe_tracks=[],
        )
        engine.connect(conn)
        # ...
    finally:
        AoqClientEngine.destroy()
"""

import sys
import ctypes
import enum
import os
import threading
from dataclasses import dataclass, field
from typing import Callable, List, Optional

__all__ = [
    # 枚举
    "AoqErrorCode",
    "AoqTrackType",
    "AoqEncoderType",
    "AoqMirrorMode",
    "AoqOrientationMode",
    "AoqRenderMode",
    "AoqAudioStreamDirection",
    "AoqAudioExternalStreamToggle",
    "AoqAudioSource",
    "AoqAudioObserverMode",
    "AoqVideoSource",
    "AoqVideoObserverMode",
    "AoqVideoObserverAlignment",
    "AoqVideoPixelFormat",
    "AoqCameraDirection",
    # 数据结构
    "AoqCreateConfig",
    "AoqTrackParam",
    "AoqConnectConfig",
    "AoqAudioCaptureConfig",
    "AoqAudioPlaybackConfig",
    "AoqAudioCodecConfig",
    "AoqAudioFileMixConfig",
    "AoqAudioExternalStreamConfig",
    "AoqAudioFrameData",
    "AoqAudioObserverConfig",
    "AoqVideoObserverConfig",
    "AoqVideoCaptureConfig",
    "AoqVideoCodecConfig",
    "AoqVideoCanvas",
    "AoqVideoFrame",
    "AoqVideoEncodedFrame",
    "AoqDataMsg",
    # 引擎与回调
    "AoqEngineEventListener",
    "IAudioFrameObserver",
    "IVideoFrameObserver",
    "AoqClientEngine",
    # 工具
    "load_library",
]


# ---------------------------------------------------------------------------
# 枚举：与 AoqClientEngine.h 中保持一一对应
# ---------------------------------------------------------------------------


class AoqTrackType(enum.IntEnum):
    AUDIO = 0
    VIDEO = 1
    DATA = 2


class AoqEncoderType(enum.IntEnum):
    UNKNOWN = 0
    AUDIO_PCM = 1
    AUDIO_OPUS = 2
    VIDEO_H264 = 3
    VIDEO_JPEG = 4
    DATA_TEXT = 5


class AoqMirrorMode(enum.IntEnum):
    DISABLED = 0
    ENABLED = 1


class AoqOrientationMode(enum.IntEnum):
    AUTO = 0
    PORTRAIT = 1
    LANDSCAPE = 2


class AoqRenderMode(enum.IntEnum):
    AUTO = 0
    STRETCH = 1
    FILL = 2
    CROP = 3


class AoqErrorCode(enum.IntEnum):
    OK = 0
    PARAM_INVALID = 1
    STATE_INVALID = 2
    AUDIO = 100
    AUDIO_EXTERNAL_BUFFER_FULL = 110
    # 音频设备
    AUDIO_DEVICE = 120
    AUDIO_DEVICE_RECORDING_AUTH_FAILED = 121
    AUDIO_DEVICE_RECORDING_OCCUPIED = 122
    AUDIO_DEVICE_RECORDING_BACKGROUND_START = 123
    AUDIO_DEVICE_RECORDING_START_FAIL = 124
    AUDIO_DEVICE_PLAYOUT_OCCUPIED = 125
    AUDIO_DEVICE_PLAYOUT_BACKGROUND_START = 126
    AUDIO_DEVICE_PLAYOUT_START_FAIL = 127
    VIDEO = 200
    VIDEO_EXTERNAL_BUFFER_FULL = 210
    VIDEO_EXTERNAL_CAPTURE_NOT_ENABLED = 211
    VIDEO_EXTERNAL_ENCODER_NOT_ENABLED = 212


class AoqConnectionStatus(enum.IntEnum):
    DISCONNECTED = 0
    CONNECTING = 1
    CONNECTED = 2
    FAILED = 3


class AoqAudioStreamDirection(enum.IntEnum):
    PUBLISH = 0
    PLAYOUT = 1


class AoqAudioExternalStreamToggle(enum.IntEnum):
    NORMAL = 0
    PAUSE = 1


class AoqAudioSource(enum.IntEnum):
    """音频数据源类型，与 IAudioFrameObserver 各回调对应。"""

    CAPTURED = 0
    PROCESS_CAPTURED = 1
    PUBLISH = 2
    PLAYBACK = 3


class AoqAudioObserverMode(enum.IntEnum):
    READ_ONLY = 0
    READ_WRITE = 1


class AoqVideoSource(enum.IntEnum):
    """视频帧回调数据源（管线位置），与 IVideoFrameObserver 各回调对应。"""

    CAPTURED = 0
    PRE_ENCODE = 1
    REMOTE = 2


class AoqVideoObserverMode(enum.IntEnum):
    READ_ONLY = 0
    READ_WRITE = 1


class AoqVideoObserverAlignment(enum.IntEnum):
    DEFAULT = 0
    EVEN = 1
    ALIGN_4 = 2
    ALIGN_8 = 3
    ALIGN_16 = 4


class AoqVideoPixelFormat(enum.IntEnum):
    UNKNOWN = 0
    I420 = 1
    NV12 = 2
    NV21 = 3
    BGRA = 4
    RGBA = 5
    CV_PIXEL_BUFFER = 6  # 仅 Apple 平台有效


class AoqCameraDirection(enum.IntEnum):
    """摄像头方向；macOS 无前后置概念，Linux 一般传 FRONT 即可。"""

    FRONT = 0
    BACK = 1


# ---------------------------------------------------------------------------
# Pythonic 数据结构：使用 dataclass，对外不暴露 ctypes 细节
# ---------------------------------------------------------------------------


@dataclass
class AoqCreateConfig:
    work_dir: str = ""
    enable_dump_audio: bool = False
    extras: str = ""


@dataclass
class AoqTrackParam:
    track_type: AoqTrackType = AoqTrackType.AUDIO


@dataclass
class AoqRelayEndpoint:
    endpoint: str = ""
    port: int = 0
    # 路径序号，与其他平台 SDK 的 routeIndex 对齐；<0 时按数组下标兜底
    route_index: int = -1


@dataclass
class AoqConnectConfig:
    token: str = ""
    sid: str = ""
    certificate: str = ""
    relay_endpoints: List[AoqRelayEndpoint] = field(default_factory=list)
    workspace_id_hash: str = ""
    publish_tracks: List[AoqTrackParam] = field(default_factory=list)
    subscribe_tracks: List[AoqTrackParam] = field(default_factory=list)


@dataclass
class AoqAudioCaptureConfig:
    is_external: bool = False
    channel: int = 1


@dataclass
class AoqAudioPlaybackConfig:
    is_external: bool = False
    channel: int = 1


@dataclass
class AoqAudioCodecConfig:
    track_type: AoqTrackType = AoqTrackType.AUDIO
    codec_type: AoqEncoderType = AoqEncoderType.AUDIO_PCM
    sample_rate: int = 48000
    channel: int = 1
    bitrate: int = 32000


@dataclass
class AoqAudioFileMixConfig:
    """音频文件推流播放配置。"""

    file_name: str = ""
    cycles: int = -1            # -1 表示无限循环
    start_pos_ms: int = 0
    publish_volume: int = 100   # [0,100]
    playout_volume: int = 100   # [0,100]


@dataclass
class AoqAudioExternalStreamConfig:
    """外部音频流配置。"""

    track_type: AoqTrackType = AoqTrackType.AUDIO
    codec_type: AoqEncoderType = AoqEncoderType.AUDIO_PCM
    channels: int = 1
    sample_rate: int = 48000
    playout_volume: int = 100
    publish_volume: int = 100
    max_buffer_duration: int = 1000  # ms
    enable_3a: bool = False


@dataclass
class AoqAudioFrameData:
    """音频裸数据，用于外部输入或观察者回调。

    ``data`` 字段：
      - 调用 push_audio_external_stream_data 时传入 ``bytes``；
      - 在 IAudioFrameObserver 回调中拿到 ``bytes`` 拷贝。
    """

    data: bytes = b""
    num_of_samples: int = 0
    bytes_per_sample: int = 0
    num_of_channels: int = 0
    samples_per_sec: int = 0
    push_sequence: int = 0
    time_stamp: int = 0
    auto_gen_mute: bool = False


@dataclass
class AoqAudioObserverConfig:
    """音频帧观察者参数。"""

    sample_rate: int = 48000
    channels: int = 1
    mode: AoqAudioObserverMode = AoqAudioObserverMode.READ_ONLY


@dataclass
class AoqVideoObserverConfig:
    """视频帧观察者参数。"""

    format: AoqVideoPixelFormat = AoqVideoPixelFormat.I420
    alignment: AoqVideoObserverAlignment = AoqVideoObserverAlignment.DEFAULT
    mode: AoqVideoObserverMode = AoqVideoObserverMode.READ_ONLY
    mirror_applied: bool = False


@dataclass
class AoqVideoFrame:
    """外部视频帧。

    使用打包格式（NV12/NV21/BGRA/RGBA）时填 ``data``；
    使用 I420 三平面时填 ``data_y/u/v`` 与对应 stride。
    Apple 零拷贝场景在 Linux 用不上，``native_pixel_buffer`` 保留为 0。
    """

    format: AoqVideoPixelFormat = AoqVideoPixelFormat.UNKNOWN

    width: int = 0
    height: int = 0

    data: bytes = b""

    # I420 三平面（与打包格式互斥使用）
    data_y: bytes = b""
    data_u: bytes = b""
    data_v: bytes = b""
    stride_y: int = 0
    stride_u: int = 0
    stride_v: int = 0

    native_pixel_buffer: int = 0
    time_stamp: int = 0  # ms，0 表示由 SDK 用本地时钟补


@dataclass
class AoqVideoEncodedFrame:
    """外部已编码视频帧（如 JPEG）。

    直接走 bypass 通路送达 VideoEncoderPlugin，不经 dispense/原始帧编码链。
    要求 ``set_video_encoder_config(is_external=True, codec_type=VIDEO_JPEG)``；
    与视频采集互斥由 SDK 在编码器入口处理，无需依赖 start_video_capture。
    """

    codec: AoqEncoderType = AoqEncoderType.VIDEO_JPEG
    data: bytes = b""
    width: int = 0
    height: int = 0
    time_stamp: int = 0  # ms，0 表示由 SDK 用本地时钟补


@dataclass
class AoqVideoCaptureConfig:
    width: int = 1280
    height: int = 720
    fps: int = 15
    is_external: bool = False
    camera_direction: AoqCameraDirection = AoqCameraDirection.FRONT


@dataclass
class AoqVideoCodecConfig:
    """视频编解码参数（编解码共用）。解码时仅 track_type/codec_type/width/height/fps/bitrate 生效，其余仅编码使用。"""
    # True 时由调用方推送已编码帧，SDK 不做采集和编码；仅编码生效
    is_external: bool = False
    track_type: AoqTrackType = AoqTrackType.VIDEO
    codec_type: AoqEncoderType = AoqEncoderType.VIDEO_H264
    width: int = 540
    height: int = 960
    fps: int = 5
    bitrate: int = 500000       # bps，与 C++ 默认对齐
    min_bitrate: int = 128000   # bps，与 C++ 默认对齐
    keyframe_interval: int = 2
    mirror_mode: AoqMirrorMode = AoqMirrorMode.DISABLED
    orientation_mode: AoqOrientationMode = AoqOrientationMode.AUTO


@dataclass
class AoqVideoCanvas:
    """Linux 通常使用 X11 Window 句柄（unsigned long）或 EGL Surface 指针。

    将句柄整数值放入 ``view``；如不渲染窗口可置为 0。
    """

    view: int = 0
    render_mode: AoqRenderMode = AoqRenderMode.AUTO


@dataclass
class AoqDataMsg:
    data: bytes = b""


# ---------------------------------------------------------------------------
# 与 C 桥接层对齐的 ctypes 结构体（仅在调用 native 前临时构造）
# ---------------------------------------------------------------------------


class _CCreateConfig(ctypes.Structure):
    _fields_ = [
        ("work_dir", ctypes.c_char_p),
        ("enable_dump_audio", ctypes.c_int),
        ("extras", ctypes.c_char_p),
    ]


class _CTrackParam(ctypes.Structure):
    _fields_ = [
        ("track_type", ctypes.c_int),
    ]


class _CRelayEndpoint(ctypes.Structure):
    _fields_ = [
        ("endpoint", ctypes.c_char_p),
        ("port", ctypes.c_int),
        ("route_index", ctypes.c_int),
    ]


class _CConnectConfig(ctypes.Structure):
    _fields_ = [
        ("token", ctypes.c_char_p),
        ("sid", ctypes.c_char_p),
        ("certificate", ctypes.c_char_p),
        ("relay_endpoints", ctypes.POINTER(_CRelayEndpoint)),
        ("relay_endpoints_count", ctypes.c_size_t),
        ("workspace_id_hash", ctypes.c_char_p),
        ("publish_tracks", ctypes.POINTER(_CTrackParam)),
        ("publish_tracks_count", ctypes.c_size_t),
        ("subscribe_tracks", ctypes.POINTER(_CTrackParam)),
        ("subscribe_tracks_count", ctypes.c_size_t),
    ]


class _CAudioCaptureConfig(ctypes.Structure):
    _fields_ = [
        ("is_external", ctypes.c_int),
        ("channel", ctypes.c_int),
    ]


class _CAudioPlaybackConfig(ctypes.Structure):
    _fields_ = [
        ("is_external", ctypes.c_int),
        ("channel", ctypes.c_int),
    ]


class _CAudioCodecConfig(ctypes.Structure):
    _fields_ = [
        ("track_type", ctypes.c_int),
        ("codec_type", ctypes.c_int),
        ("sample_rate", ctypes.c_int),
        ("channel", ctypes.c_int),
        ("bitrate", ctypes.c_int),
    ]


class _CVideoCaptureConfig(ctypes.Structure):
    _fields_ = [
        ("width", ctypes.c_int),
        ("height", ctypes.c_int),
        ("fps", ctypes.c_int),
        ("is_external", ctypes.c_int),
        ("camera_direction", ctypes.c_int),
    ]


class _CAudioFileMixConfig(ctypes.Structure):
    _fields_ = [
        ("file_name", ctypes.c_char_p),
        ("cycles", ctypes.c_int),
        ("start_pos_ms", ctypes.c_int64),
        ("publish_volume", ctypes.c_int),
        ("playout_volume", ctypes.c_int),
    ]


class _CAudioExternalStreamConfig(ctypes.Structure):
    _fields_ = [
        ("track_type", ctypes.c_int),
        ("codec_type", ctypes.c_int),
        ("channels", ctypes.c_int),
        ("sample_rate", ctypes.c_int),
        ("playout_volume", ctypes.c_int),
        ("publish_volume", ctypes.c_int),
        ("max_buffer_duration", ctypes.c_int),
        ("enable_3a", ctypes.c_int),
    ]


class _CAudioFrameData(ctypes.Structure):
    _fields_ = [
        ("data_ptr", ctypes.c_void_p),
        ("num_of_samples", ctypes.c_int),
        ("bytes_per_sample", ctypes.c_int),
        ("num_of_channels", ctypes.c_int),
        ("samples_per_sec", ctypes.c_int),
        ("push_sequence", ctypes.c_int32),
        ("time_stamp", ctypes.c_int64),
        ("data_size", ctypes.c_int),
        ("auto_gen_mute", ctypes.c_bool),
    ]


class _CAudioObserverConfig(ctypes.Structure):
    _fields_ = [
        ("sample_rate", ctypes.c_int),
        ("channels", ctypes.c_int),
        ("mode", ctypes.c_int),
    ]


class _CVideoObserverConfig(ctypes.Structure):
    _fields_ = [
        ("format", ctypes.c_int),
        ("alignment", ctypes.c_int),
        ("mode", ctypes.c_int),
        ("mirror_applied", ctypes.c_int),
    ]


class _CVideoFrame(ctypes.Structure):
    _fields_ = [
        ("format", ctypes.c_int),
        ("width", ctypes.c_int),
        ("height", ctypes.c_int),
        ("data_ptr", ctypes.c_void_p),
        ("data_size", ctypes.c_int),
        ("data_y", ctypes.c_void_p),
        ("data_u", ctypes.c_void_p),
        ("data_v", ctypes.c_void_p),
        ("stride_y", ctypes.c_int),
        ("stride_u", ctypes.c_int),
        ("stride_v", ctypes.c_int),
        ("native_pixel_buffer", ctypes.c_void_p),
        ("time_stamp", ctypes.c_int64),
    ]


class _CVideoEncodedFrame(ctypes.Structure):
    _fields_ = [
        ("codec", ctypes.c_int),
        ("data", ctypes.c_void_p),
        ("data_size", ctypes.c_int),
        ("width", ctypes.c_int),
        ("height", ctypes.c_int),
        ("time_stamp", ctypes.c_int64),
    ]


class _CVideoCodecConfig(ctypes.Structure):
    _fields_ = [
        ("is_external", ctypes.c_int),
        ("track_type", ctypes.c_int),
        ("codec_type", ctypes.c_int),
        ("width", ctypes.c_int),
        ("height", ctypes.c_int),
        ("fps", ctypes.c_int),
        ("bitrate", ctypes.c_int),
        ("min_bitrate", ctypes.c_int),
        ("keyframe_interval", ctypes.c_int),
        ("mirror_mode", ctypes.c_int),
        ("orientation_mode", ctypes.c_int),
    ]


class _CVideoCanvas(ctypes.Structure):
    _fields_ = [
        ("view", ctypes.c_void_p),
        ("render_mode", ctypes.c_int),
    ]


class _CDataMsg(ctypes.Structure):
    _fields_ = [
        ("data", ctypes.POINTER(ctypes.c_uint8)),
        ("data_size", ctypes.c_size_t),
    ]


# 回调函数类型，签名对齐 C 头文件中的 aoq_event_callbacks_t
_OnErrorFn = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int, ctypes.c_char_p)
# on_warning 与 on_error 的 C 签名完全一致，这里仅作语义别名以便阅读
_OnWarningFn = _OnErrorFn
_OnConnFn = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int)
_OnElapsedFn = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int64)
_OnVideoFrameFn = ctypes.CFUNCTYPE(
    None, ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_int64
)
_OnDataMsgFn = ctypes.CFUNCTYPE(
    None, ctypes.c_void_p, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t
)


# 音频帧观察者回调，签名对齐 C 头 aoq_audio_frame_observer_t
_OnAudioFrameFn = ctypes.CFUNCTYPE(
    None, ctypes.c_void_p, ctypes.POINTER(_CAudioFrameData)
)
_OnTrackedAudioFrameFn = ctypes.CFUNCTYPE(
    None, ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(_CAudioFrameData)
)


class _CAudioFrameObserver(ctypes.Structure):
    _fields_ = [
        ("user_data", ctypes.c_void_p),
        ("on_captured", _OnAudioFrameFn),
        ("on_process_captured", _OnAudioFrameFn),
        ("on_publish", _OnTrackedAudioFrameFn),
        ("on_playback", _OnAudioFrameFn),
        ("on_all", _OnAudioFrameFn),
        ("on_remote", _OnTrackedAudioFrameFn),
    ]


# 视频帧观察者回调，签名对齐 C 头 aoq_video_frame_observer_t；
# 返回非 0 表示数据已修改需写回 SDK（仅 I420 读写模式生效）
_OnObservedVideoFrameFn = ctypes.CFUNCTYPE(
    ctypes.c_int, ctypes.c_void_p, ctypes.POINTER(_CVideoFrame)
)
_OnTrackedObservedVideoFrameFn = ctypes.CFUNCTYPE(
    ctypes.c_int, ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(_CVideoFrame)
)


class _CVideoFrameObserver(ctypes.Structure):
    _fields_ = [
        ("user_data", ctypes.c_void_p),
        ("on_captured", _OnObservedVideoFrameFn),
        ("on_pre_encode", _OnTrackedObservedVideoFrameFn),
        ("on_remote", _OnTrackedObservedVideoFrameFn),
    ]


class _CEventCallbacks(ctypes.Structure):
    _fields_ = [
        ("user_data", ctypes.c_void_p),
        ("on_error", _OnErrorFn),
        ("on_warning", _OnWarningFn),
        ("on_connection_status_change", _OnConnFn),
        ("on_data_msg", _OnDataMsgFn),
    ]


# ---------------------------------------------------------------------------
# 共享库加载
# ---------------------------------------------------------------------------


# 主库文件名按平台不同：
#   Linux:  libAoqClientSdk.so
#   macOS:  libAoqClientSdk.dylib
# 其它/未知平台先按 .so 兜底（极少见，主要避免 NameError）。
if sys.platform == "darwin":
    _LIB_NAME = "libAoqClientSdk.dylib"
else:
    _LIB_NAME = "libAoqClientSdk.so"
# 同目录下若同时存在两份命名，作为额外兜底候选，方便交叉部署
_LIB_NAME_FALLBACKS = [
    name for name in ("libAoqClientSdk.dylib", "libAoqClientSdk.so")
    if name != _LIB_NAME
]
_lib: Optional[ctypes.CDLL] = None
_lib_lock = threading.Lock()

# 已注销的 ctypes 回调闭包保活列表：旧闭包不立即让 Python GC，避免
# native 线程上 in-flight 的回调跳板访问到已释放的函数指针（UAF
# 段错误）。进程退出时由 OS 一并回收，几 KB/case 的内存换稳定性。
_AUDIO_OBSERVER_KEEPALIVE: List = []
_VIDEO_OBSERVER_KEEPALIVE: List = []
_LISTENER_KEEPALIVE: List = []


def load_library(path: Optional[str] = None) -> ctypes.CDLL:
    """加载 native 共享库。

    :param path: 共享库绝对路径；为空时按以下顺序查找：
        ``AOQ_CLIENT_SDK_LIB`` 环境变量、与本文件同目录、系统默认路径。
    """
    global _lib
    with _lib_lock:
        if _lib is not None:
            return _lib

        candidates: List[str] = []
        if path:
            candidates.append(path)
        env_path = os.environ.get("AOQ_CLIENT_SDK_LIB")
        if env_path:
            candidates.append(env_path)
        here = os.path.dirname(os.path.abspath(__file__))
        # 优先平台匹配的命名
        candidates.append(os.path.join(here, _LIB_NAME))
        candidates.append(os.path.join(here, "..", "lib", _LIB_NAME))
        # 兜底：另一个平台命名（极少触发，应对交叉拷贝场景）
        for fb in _LIB_NAME_FALLBACKS:
            candidates.append(os.path.join(here, fb))
            candidates.append(os.path.join(here, "..", "lib", fb))
        candidates.append(_LIB_NAME)

        last_err: Optional[OSError] = None
        attempts = []  # type: List[str]
        for cand in candidates:
            # 仅当文件存在/或为简短库名时再尝试，避免不必要的 OSError
            try:
                _lib = ctypes.CDLL(cand)
                _bind_symbols(_lib)
                return _lib
            except OSError as exc:
                last_err = exc
                exists = os.path.isfile(cand) if os.path.isabs(cand) else "n/a"
                attempts.append(f"  - {cand!r} (exists={exists}): {exc}")
                continue
        # 把每个候选路径的失败原因都暴露出来，便于排查
        # 典型场景：主 .so 文件存在但其依赖（如 libstdc++.so.6 / libcurl.so.4）
        # 不在 LD_LIBRARY_PATH 中，dlerror 会用依赖名作为前缀。
        raise OSError(
            "Failed to load {0}; attempts:\n{1}".format(
                _LIB_NAME, "\n".join(attempts) if attempts else "<none>")
        )


def _bind_symbols(lib: ctypes.CDLL) -> None:
    """声明 native 函数的参数与返回值类型。"""

    lib.aoq_get_version.restype = ctypes.c_char_p
    lib.aoq_get_version.argtypes = []

    lib.aoq_create_engine.restype = ctypes.c_void_p
    lib.aoq_create_engine.argtypes = [
        ctypes.POINTER(_CCreateConfig),
        ctypes.POINTER(_CEventCallbacks),
    ]

    lib.aoq_destroy.restype = ctypes.c_int
    lib.aoq_destroy.argtypes = []

    lib.aoq_connect.restype = ctypes.c_int
    lib.aoq_connect.argtypes = [ctypes.c_void_p, ctypes.POINTER(_CConnectConfig)]

    lib.aoq_disconnect.restype = ctypes.c_int
    lib.aoq_disconnect.argtypes = [ctypes.c_void_p]

    lib.aoq_start_audio_capture.restype = ctypes.c_int
    lib.aoq_start_audio_capture.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CAudioCaptureConfig),
    ]
    lib.aoq_stop_audio_capture.restype = ctypes.c_int
    lib.aoq_stop_audio_capture.argtypes = [ctypes.c_void_p]

    lib.aoq_mute_audio_capture.restype = ctypes.c_int
    lib.aoq_mute_audio_capture.argtypes = [ctypes.c_void_p, ctypes.c_int]

    lib.aoq_start_audio_player.restype = ctypes.c_int
    lib.aoq_start_audio_player.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CAudioPlaybackConfig),
    ]
    lib.aoq_stop_audio_player.restype = ctypes.c_int
    lib.aoq_stop_audio_player.argtypes = [ctypes.c_void_p]
    lib.aoq_pause_audio_player.restype = ctypes.c_int
    lib.aoq_pause_audio_player.argtypes = [ctypes.c_void_p, ctypes.c_int]
    lib.aoq_resume_audio_player.restype = ctypes.c_int
    lib.aoq_resume_audio_player.argtypes = [ctypes.c_void_p, ctypes.c_int]
    lib.aoq_interrupt_audio_player.restype = ctypes.c_int
    lib.aoq_interrupt_audio_player.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int]

    lib.aoq_set_audio_encoder_config.restype = ctypes.c_int
    lib.aoq_set_audio_encoder_config.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CAudioCodecConfig),
    ]

    lib.aoq_set_audio_decoder_config.restype = ctypes.c_int
    lib.aoq_set_audio_decoder_config.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CAudioCodecConfig),
    ]

    lib.aoq_start_video_capture.restype = ctypes.c_int
    lib.aoq_start_video_capture.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CVideoCaptureConfig),
    ]
    lib.aoq_stop_video_capture.restype = ctypes.c_int
    lib.aoq_stop_video_capture.argtypes = [ctypes.c_void_p]

    lib.aoq_set_local_view.restype = ctypes.c_int
    lib.aoq_set_local_view.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(_CVideoCanvas)]
    lib.aoq_set_remote_view.restype = ctypes.c_int
    lib.aoq_set_remote_view.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(_CVideoCanvas)]

    lib.aoq_set_video_encoder_config.restype = ctypes.c_int
    lib.aoq_set_video_encoder_config.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CVideoCodecConfig),
    ]
    lib.aoq_set_video_decoder_config.restype = ctypes.c_int
    lib.aoq_set_video_decoder_config.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CVideoCodecConfig),
    ]

    # ---- 视频外部帧 / 切换摄像头 ----
    lib.aoq_push_external_video_frame.restype = ctypes.c_int
    lib.aoq_push_external_video_frame.argtypes = [
        ctypes.c_void_p,
        ctypes.c_int,  # track_type
        ctypes.POINTER(_CVideoFrame),
    ]
    lib.aoq_push_external_video_encoded_frame.restype = ctypes.c_int
    lib.aoq_push_external_video_encoded_frame.argtypes = [
        ctypes.c_void_p,
        ctypes.c_int,
        ctypes.POINTER(_CVideoEncodedFrame),
    ]
    if hasattr(lib, "aoq_switch_camera"):
        lib.aoq_switch_camera.restype = ctypes.c_int
        lib.aoq_switch_camera.argtypes = [ctypes.c_void_p, ctypes.c_int]

    # ---- 音频文件混音 ----
    lib.aoq_start_audio_file.restype = ctypes.c_int
    lib.aoq_start_audio_file.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.POINTER(_CAudioFileMixConfig),
    ]
    for sym in (
        "aoq_stop_audio_file",
        "aoq_pause_audio_file",
        "aoq_resume_audio_file",
    ):
        getattr(lib, sym).restype = ctypes.c_int
        getattr(lib, sym).argtypes = [ctypes.c_void_p, ctypes.c_char_p]

    lib.aoq_get_audio_file_duration.restype = ctypes.c_int64
    lib.aoq_get_audio_file_duration.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
    lib.aoq_get_audio_file_current_position.restype = ctypes.c_int64
    lib.aoq_get_audio_file_current_position.argtypes = [ctypes.c_void_p, ctypes.c_char_p]

    lib.aoq_set_audio_file_position_millis.restype = ctypes.c_int
    lib.aoq_set_audio_file_position_millis.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_int64,
    ]

    lib.aoq_set_audio_file_volume.restype = ctypes.c_int
    lib.aoq_set_audio_file_volume.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_int,
        ctypes.c_int,
    ]
    lib.aoq_get_audio_file_volume.restype = ctypes.c_int
    lib.aoq_get_audio_file_volume.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_int,
    ]

    # ---- 外部音频流 ----
    lib.aoq_add_audio_external_stream.restype = ctypes.c_int
    lib.aoq_add_audio_external_stream.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.POINTER(_CAudioExternalStreamConfig),
    ]
    lib.aoq_push_audio_external_stream_data.restype = ctypes.c_int
    lib.aoq_push_audio_external_stream_data.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.POINTER(_CAudioFrameData),
    ]
    lib.aoq_set_audio_external_stream_volume.restype = ctypes.c_int
    lib.aoq_set_audio_external_stream_volume.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_int,
        ctypes.c_int,
    ]
    lib.aoq_get_audio_external_stream_volume.restype = ctypes.c_int
    lib.aoq_get_audio_external_stream_volume.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_int,
    ]
    lib.aoq_clear_audio_external_stream_buffer.restype = None
    lib.aoq_clear_audio_external_stream_buffer.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_int,
    ]
    lib.aoq_remove_audio_external_stream.restype = ctypes.c_int
    lib.aoq_remove_audio_external_stream.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
    ]

    # ---- 音频帧观察者 ----
    lib.aoq_set_audio_frame_observer.restype = ctypes.c_int
    lib.aoq_set_audio_frame_observer.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CAudioFrameObserver),
    ]
    lib.aoq_enable_audio_frame_observer.restype = ctypes.c_int
    lib.aoq_enable_audio_frame_observer.argtypes = [
        ctypes.c_void_p,
        ctypes.c_int,
        ctypes.c_int,
        ctypes.POINTER(_CAudioObserverConfig),
    ]

    # ---- 视频帧观察者 ----
    lib.aoq_set_video_frame_observer.restype = ctypes.c_int
    lib.aoq_set_video_frame_observer.argtypes = [
        ctypes.c_void_p,
        ctypes.POINTER(_CVideoFrameObserver),
    ]
    lib.aoq_enable_video_frame_observer.restype = ctypes.c_int
    lib.aoq_enable_video_frame_observer.argtypes = [
        ctypes.c_void_p,
        ctypes.c_int,
        ctypes.c_int,
        ctypes.POINTER(_CVideoObserverConfig),
    ]

    lib.aoq_send_data_msg.restype = ctypes.c_int
    lib.aoq_send_data_msg.argtypes = [ctypes.c_void_p, ctypes.POINTER(_CDataMsg)]

    lib.aoq_enable_send_media_stream.restype = ctypes.c_int
    lib.aoq_enable_send_media_stream.argtypes = [
        ctypes.c_void_p,
        ctypes.c_int,
        ctypes.c_int,
    ]


# ---------------------------------------------------------------------------
# 事件监听基类
# ---------------------------------------------------------------------------


class AoqEngineEventListener:
    """引擎回调基类。所有方法均为可选 override，默认空实现。

    回调在 native 线程触发，使用者需自行保证线程安全。
    """

    def on_error(self, code: int, message: str) -> None: ...
    def on_connection_status_change(self, status: AoqConnectionStatus) -> None: ...

    def on_data_msg(self, msg: AoqDataMsg) -> None: ...


class IAudioFrameObserver:
    """音频数据监听基类。所有方法均为可选 override，默认空实现。

    回调在 native 音频线程触发，禁止在其中做任何耗时操作。
    数据生命周期仅限本次回调内有效；如需异步使用请自行拷贝
    （``AoqAudioFrameData.data`` 已经是 bytes 拷贝）。
    """

    def on_captured_audio_frame(self, data: AoqAudioFrameData) -> None: ...
    def on_process_captured_audio_frame(self, data: AoqAudioFrameData) -> None: ...
    def on_publish_audio_frame(self, track_type: AoqTrackType, data: AoqAudioFrameData) -> None: ...
    def on_playback_audio_frame(self, data: AoqAudioFrameData) -> None: ...


class IVideoFrameObserver:
    """视频数据监听基类。所有方法均为可选 override，默认返回 False。

    回调在 native 视频线程触发，禁止在其中做任何耗时操作。
    ``AoqVideoFrame`` 中的像素数据已是 bytes 拷贝；返回 True 表示
    数据已修改需写回 SDK（仅 I420 读写模式生效，当前 Python 层
    为拷贝语义，写回暂不支持，建议始终返回 False）。
    """

    def on_captured_video_frame(self, frame: AoqVideoFrame) -> bool:
        return False

    def on_pre_encode_video_frame(self, track_type: AoqTrackType, frame: AoqVideoFrame) -> bool:
        return False

    def on_remote_video_frame(self, track_type: AoqTrackType, frame: AoqVideoFrame) -> bool:
        return False

# ---------------------------------------------------------------------------
# 引擎类（单例语义，与 C++ 基类一致）
# ---------------------------------------------------------------------------


def _to_bytes(s: Optional[str]) -> Optional[bytes]:
    if s is None:
        return None
    return s.encode("utf-8")


def _to_str(b: Optional[bytes]) -> str:
    if not b:
        return ""
    return b.decode("utf-8", errors="replace")


class _TracksBuf:
    """临时持有 ctypes track 数组，保证 native 调用期间不被回收。"""

    def __init__(self, tracks: List[AoqTrackParam]):
        n = len(tracks)
        arr_t = _CTrackParam * n if n > 0 else _CTrackParam * 0
        arr = arr_t()
        for i, t in enumerate(tracks):
            arr[i].track_type = int(t.track_type)
        self.array = arr
        self.count = n

    def ptr(self):
        if self.count == 0:
            return ctypes.cast(0, ctypes.POINTER(_CTrackParam))
        return ctypes.cast(self.array, ctypes.POINTER(_CTrackParam))


class _RelayBuf:
    """临时持有 ctypes relay 数组，保证 native 调用期间不被回收。

    ctypes c_char_p 不持有字符串所有权，需额外保存 endpoint 的 bytes
    引用，避免在 native 调用前被 GC。
    """

    def __init__(self, endpoints: List[AoqRelayEndpoint]):
        n = len(endpoints)
        arr_t = _CRelayEndpoint * n if n > 0 else _CRelayEndpoint * 0
        arr = arr_t()
        self._keepalive = []  # 持有 endpoint bytes，防止 GC
        for i, ep in enumerate(endpoints):
            ep_bytes = _to_bytes(ep.endpoint)
            self._keepalive.append(ep_bytes)
            arr[i].endpoint = ep_bytes
            arr[i].port = int(ep.port)
            # route_index 负值按数组下标兜底，与其他平台 SDK 行为一致
            route_index = int(ep.route_index)
            arr[i].route_index = route_index if route_index >= 0 else i
        self.array = arr
        self.count = n

    def ptr(self):
        if self.count == 0:
            return ctypes.cast(0, ctypes.POINTER(_CRelayEndpoint))
        return ctypes.cast(self.array, ctypes.POINTER(_CRelayEndpoint))


class AoqClientEngine:
    """全局单例。``create_engine`` 重复调用返回同一实例，``destroy``
    后需再次 ``create_engine`` 方可继续使用。
    """

    _instance: Optional["AoqClientEngine"] = None
    _instance_lock = threading.Lock()

    def __init__(self, handle: int, listener: AoqEngineEventListener,
                 callbacks: _CEventCallbacks):
        self._handle = ctypes.c_void_p(handle)
        # 保持引用，避免 ctypes 回调结构体被 GC
        self._listener = listener
        self._callbacks = callbacks

    # -------------------- 工厂方法 --------------------

    @classmethod
    def create_engine(
        cls,
        config: AoqCreateConfig,
        listener: AoqEngineEventListener,
    ) -> "AoqClientEngine":
        lib = load_library()
        with cls._instance_lock:
            if cls._instance is not None:
                return cls._instance

            c_cfg = _CCreateConfig(
                work_dir=_to_bytes(config.work_dir),
                enable_dump_audio=int(bool(config.enable_dump_audio)),
                extras=_to_bytes(config.extras),
            )
            cbs = cls._build_callbacks(listener)
            handle = lib.aoq_create_engine(ctypes.byref(c_cfg), ctypes.byref(cbs))
            if not handle:
                raise RuntimeError("aoq_create_engine returned NULL")
            cls._instance = cls(handle, listener, cbs)
            return cls._instance

    @classmethod
    def destroy(cls) -> int:
        with cls._instance_lock:
            inst = cls._instance
            if inst is None:
                return 0
            lib = load_library()
            ret = lib.aoq_destroy()
            # native aoq_destroy 已对回调通路 Detach 并保证不再回到 Python。
            # 但 SDK 内部仍可能有 in-flight 回调还在跑跳板，所以这里把旧
            # ctypes 闭包推入模块级 keepalive 列表保活，而不是立刻置 None
            # 让 GC 释放；否则 native 线程拿着野指针跳进来就是段错误。
            prev_listener = inst._listener, inst._callbacks
            inst._listener = None
            inst._callbacks = None
            _LISTENER_KEEPALIVE.append(prev_listener)
            prev_audio_observer = getattr(inst, "_audio_observer", None)
            if prev_audio_observer is not None:
                inst._audio_observer = None
                _AUDIO_OBSERVER_KEEPALIVE.append(prev_audio_observer)
            cls._instance = None
            return int(ret)

    @staticmethod
    def get_version() -> str:
        lib = load_library()
        return _to_str(lib.aoq_get_version())

    # -------------------- 连接 --------------------

    def connect(self, config: AoqConnectConfig) -> int:
        pub = _TracksBuf(config.publish_tracks)
        sub = _TracksBuf(config.subscribe_tracks)
        relay = _RelayBuf(config.relay_endpoints)
        c_cfg = _CConnectConfig(
            token=_to_bytes(config.token),
            sid=_to_bytes(config.sid),
            certificate=_to_bytes(config.certificate),
            relay_endpoints=relay.ptr(),
            relay_endpoints_count=relay.count,
            workspace_id_hash=_to_bytes(config.workspace_id_hash),
            publish_tracks=pub.ptr(),
            publish_tracks_count=pub.count,
            subscribe_tracks=sub.ptr(),
            subscribe_tracks_count=sub.count,
        )
        return int(load_library().aoq_connect(self._handle, ctypes.byref(c_cfg)))

    def disconnect(self) -> int:
        return int(load_library().aoq_disconnect(self._handle))

    # -------------------- 音频 --------------------

    def start_audio_capture(self, config: AoqAudioCaptureConfig) -> int:
        c = _CAudioCaptureConfig(
            is_external=int(bool(config.is_external)),
            channel=int(config.channel),
        )
        return int(load_library().aoq_start_audio_capture(self._handle, ctypes.byref(c)))

    def stop_audio_capture(self) -> int:
        return int(load_library().aoq_stop_audio_capture(self._handle))

    def mute_audio_capture(self, mute: bool) -> int:
        return int(load_library().aoq_mute_audio_capture(self._handle, int(bool(mute))))

    def start_audio_player(self, config: AoqAudioPlaybackConfig) -> int:
        c = _CAudioPlaybackConfig(
            is_external=int(bool(config.is_external)),
            channel=int(config.channel),
        )
        return int(load_library().aoq_start_audio_player(self._handle, ctypes.byref(c)))

    def stop_audio_player(self) -> int:
        return int(load_library().aoq_stop_audio_player(self._handle))

    def pause_audio_player(self, fade_ms: int = 0) -> int:
        return int(load_library().aoq_pause_audio_player(self._handle, fade_ms))

    def resume_audio_player(self, fade_ms: int = 0) -> int:
        return int(load_library().aoq_resume_audio_player(self._handle, fade_ms))

    def interrupt_audio_player(self, track_type: int, fade_ms: int = 0) -> int:
        return int(load_library().aoq_interrupt_audio_player(self._handle, track_type, fade_ms))

    def set_audio_encoder_config(self, config: AoqAudioCodecConfig) -> int:
        c = _CAudioCodecConfig(
            track_type=int(config.track_type),
            codec_type=int(config.codec_type),
            sample_rate=int(config.sample_rate),
            channel=int(config.channel),
            bitrate=int(config.bitrate),
        )
        return int(load_library().aoq_set_audio_encoder_config(self._handle, ctypes.byref(c)))

    def set_audio_decoder_config(self, config: AoqAudioCodecConfig) -> int:
        c = _CAudioCodecConfig(
            track_type=int(config.track_type),
            codec_type=int(config.codec_type),
            sample_rate=int(config.sample_rate),
            channel=int(config.channel),
            bitrate=int(config.bitrate),
        )
        return int(load_library().aoq_set_audio_decoder_config(self._handle, ctypes.byref(c)))

    # -------------------- 视频 --------------------

    def start_video_capture(self, config: AoqVideoCaptureConfig) -> int:
        c = _CVideoCaptureConfig(
            width=int(config.width),
            height=int(config.height),
            fps=int(config.fps),
            is_external=int(bool(config.is_external)),
            camera_direction=int(config.camera_direction),
        )
        return int(load_library().aoq_start_video_capture(self._handle, ctypes.byref(c)))

    def stop_video_capture(self) -> int:
        return int(load_library().aoq_stop_video_capture(self._handle))

    def switch_camera(self, direction: AoqCameraDirection) -> int:
        """切换前后置摄像头（仅移动端 native 库导出符号；Linux 通常返回 -1）。"""
        lib = load_library()
        if not hasattr(lib, "aoq_switch_camera"):
            return -1
        return int(lib.aoq_switch_camera(self._handle, int(direction)))

    def set_local_view(self, track_type: int, canvas: Optional[AoqVideoCanvas]) -> int:
        c = _CVideoCanvas(
            view=ctypes.c_void_p(canvas.view if canvas else 0),
            render_mode=int(canvas.render_mode) if canvas else int(AoqRenderMode.AUTO),
        )
        return int(load_library().aoq_set_local_view(self._handle, track_type, ctypes.byref(c)))

    def set_remote_view(self, track_type: int, canvas: Optional[AoqVideoCanvas]) -> int:
        c = _CVideoCanvas(
            view=ctypes.c_void_p(canvas.view if canvas else 0),
            render_mode=int(canvas.render_mode) if canvas else int(AoqRenderMode.AUTO),
        )
        return int(load_library().aoq_set_remote_view(self._handle, track_type, ctypes.byref(c)))

    def set_video_encoder_config(self, config: AoqVideoCodecConfig) -> int:
        c = _CVideoCodecConfig(
            is_external=int(bool(config.is_external)),
            track_type=int(config.track_type),
            codec_type=int(config.codec_type),
            width=int(config.width),
            height=int(config.height),
            fps=int(config.fps),
            bitrate=int(config.bitrate),
            min_bitrate=int(config.min_bitrate),
            keyframe_interval=int(config.keyframe_interval),
            mirror_mode=int(config.mirror_mode),
            orientation_mode=int(config.orientation_mode),
        )
        return int(load_library().aoq_set_video_encoder_config(self._handle, ctypes.byref(c)))

    def set_video_decoder_config(self, config: AoqVideoCodecConfig) -> int:
        c = _CVideoCodecConfig(
            track_type=int(config.track_type),
            codec_type=int(config.codec_type),
            width=int(config.width),
            height=int(config.height),
            fps=int(config.fps),
            bitrate=int(config.bitrate),
        )
        return int(load_library().aoq_set_video_decoder_config(self._handle, ctypes.byref(c)))

    def push_external_video_frame(
        self, frame: AoqVideoFrame, track_type: AoqTrackType = AoqTrackType.VIDEO
    ) -> int:
        """推送外部视频帧。要求对应 startVideoCapture 时 isExternal=True。

        - 打包格式（NV12/NV21/BGRA/RGBA）填 ``frame.data``；
        - I420 三平面填 ``frame.data_y/u/v`` 与 stride。
        - ``track_type`` 与 C 端 aoq_push_external_video_frame 的 track_type
          参数对齐，默认 VIDEO（与 start_video_capture 主流 track 对齐）。
        Buffer 仅在本次调用期间使用，调用方持有引用即可。
        """

        # 保留 buffer 引用，防止 GC
        keep: List[bytes] = []

        def _addr(buf: bytes) -> int:
            if not buf:
                return 0
            keep.append(buf)
            return ctypes.cast(
                (ctypes.c_uint8 * len(buf)).from_buffer_copy(buf), ctypes.c_void_p
            ).value or 0

        # 注：对于 I420 / 打包格式，分别取得地址
        c = _CVideoFrame(
            format=int(frame.format),
            width=int(frame.width),
            height=int(frame.height),
            data_ptr=_addr(frame.data),
            data_size=len(frame.data),
            data_y=_addr(frame.data_y),
            data_u=_addr(frame.data_u),
            data_v=_addr(frame.data_v),
            stride_y=int(frame.stride_y),
            stride_u=int(frame.stride_u),
            stride_v=int(frame.stride_v),
            native_pixel_buffer=int(frame.native_pixel_buffer),
            time_stamp=int(frame.time_stamp),
        )
        ret = int(load_library().aoq_push_external_video_frame(
            self._handle, int(track_type), ctypes.byref(c)))
        # keep 在此处仍在作用域内，确保 native 调用过程中不会被回收
        del keep
        return ret

    def push_external_video_encoded_frame(
        self, track_type: AoqTrackType, frame: AoqVideoEncodedFrame
    ) -> int:
        """推送已编码视频帧（如 JPEG），走 bypass 通路直达 VideoEncoderPlugin。

        要求 ``set_video_encoder_config(is_external=True, codec_type=VIDEO_JPEG)``
        已设置，否则返回 ``VIDEO_EXTERNAL_ENCODER_NOT_ENABLED``。
        """
        keep: List[bytes] = []
        buf = None
        if frame.data:
            buf = (ctypes.c_uint8 * len(frame.data)).from_buffer_copy(frame.data)
            keep.append(frame.data)
        c = _CVideoEncodedFrame(
            codec=int(frame.codec),
            data=ctypes.cast(buf, ctypes.c_void_p) if buf else None,
            data_size=len(frame.data),
            width=int(frame.width),
            height=int(frame.height),
            time_stamp=int(frame.time_stamp),
        )
        ret = int(load_library().aoq_push_external_video_encoded_frame(
            self._handle, int(track_type), ctypes.byref(c)))
        del keep
        return ret

    # -------------------- 音频文件混音 --------------------

    def start_audio_file(self, file_id: str, config: AoqAudioFileMixConfig) -> int:
        c = _CAudioFileMixConfig(
            file_name=_to_bytes(config.file_name),
            cycles=int(config.cycles),
            start_pos_ms=int(config.start_pos_ms),
            publish_volume=int(config.publish_volume),
            playout_volume=int(config.playout_volume),
        )
        return int(load_library().aoq_start_audio_file(
            self._handle, _to_bytes(file_id), ctypes.byref(c)))

    def stop_audio_file(self, file_id: str) -> int:
        return int(load_library().aoq_stop_audio_file(self._handle, _to_bytes(file_id)))

    def pause_audio_file(self, file_id: str) -> int:
        return int(load_library().aoq_pause_audio_file(self._handle, _to_bytes(file_id)))

    def resume_audio_file(self, file_id: str) -> int:
        return int(load_library().aoq_resume_audio_file(self._handle, _to_bytes(file_id)))

    def get_audio_file_duration(self, file_id: str) -> int:
        return int(load_library().aoq_get_audio_file_duration(
            self._handle, _to_bytes(file_id)))

    def get_audio_file_current_position(self, file_id: str) -> int:
        return int(load_library().aoq_get_audio_file_current_position(
            self._handle, _to_bytes(file_id)))

    def set_audio_file_position_millis(self, file_id: str, position_ms: int) -> int:
        return int(load_library().aoq_set_audio_file_position_millis(
            self._handle, _to_bytes(file_id), int(position_ms)))

    def set_audio_file_volume(
        self, file_id: str, type_: AoqAudioStreamDirection, volume: int
    ) -> int:
        return int(load_library().aoq_set_audio_file_volume(
            self._handle, _to_bytes(file_id), int(type_), int(volume)))

    def get_audio_file_volume(
        self, file_id: str, type_: AoqAudioStreamDirection
    ) -> int:
        return int(load_library().aoq_get_audio_file_volume(
            self._handle, _to_bytes(file_id), int(type_)))

    # -------------------- 外部音频流 --------------------

    def add_audio_external_stream(
        self, stream_id: str, config: AoqAudioExternalStreamConfig
    ) -> int:
        c = _CAudioExternalStreamConfig(
            track_type=int(config.track_type),
            codec_type=int(config.codec_type),
            channels=int(config.channels),
            sample_rate=int(config.sample_rate),
            playout_volume=int(config.playout_volume),
            publish_volume=int(config.publish_volume),
            max_buffer_duration=int(config.max_buffer_duration),
            enable_3a=int(bool(config.enable_3a)),
        )
        return int(load_library().aoq_add_audio_external_stream(
            self._handle, _to_bytes(stream_id), ctypes.byref(c)))

    def push_audio_external_stream_data(
        self, stream_id: str, data: AoqAudioFrameData
    ) -> int:
        """推送外部音频流 PCM 数据。返回 ``AoqErrorCode.AUDIO_EXTERNAL_BUFFER_FULL``
        时建议等待 ~20ms 后再重试同一帧。
        """
        buf = (ctypes.c_uint8 * len(data.data)).from_buffer_copy(data.data) \
            if data.data else (ctypes.c_uint8 * 0)()
        c = _CAudioFrameData(
            data_ptr=ctypes.cast(buf, ctypes.c_void_p),
            num_of_samples=int(data.num_of_samples),
            bytes_per_sample=int(data.bytes_per_sample),
            num_of_channels=int(data.num_of_channels),
            samples_per_sec=int(data.samples_per_sec),
            push_sequence=int(data.push_sequence),
            time_stamp=int(data.time_stamp),
            data_size=len(data.data),
        )
        return int(load_library().aoq_push_audio_external_stream_data(
            self._handle, _to_bytes(stream_id), ctypes.byref(c)))

    def set_audio_external_stream_volume(
        self, stream_id: str, type_: AoqAudioStreamDirection, volume: int
    ) -> int:
        return int(load_library().aoq_set_audio_external_stream_volume(
            self._handle, _to_bytes(stream_id), int(type_), int(volume)))

    def get_audio_external_stream_volume(
        self, stream_id: str, type_: AoqAudioStreamDirection
    ) -> int:
        return int(load_library().aoq_get_audio_external_stream_volume(
            self._handle, _to_bytes(stream_id), int(type_)))

    def clear_audio_external_stream_buffer(
        self, stream_id: str, fadeout_ms: int = -1
    ) -> None:
        load_library().aoq_clear_audio_external_stream_buffer(
            self._handle, _to_bytes(stream_id), int(fadeout_ms))

    def remove_audio_external_stream(self, stream_id: str) -> int:
        return int(load_library().aoq_remove_audio_external_stream(
            self._handle, _to_bytes(stream_id)))

    # -------------------- 音频帧观察者 --------------------

    def set_audio_frame_observer(
        self, observer: Optional[IAudioFrameObserver]
    ) -> int:
        """注册或注销音频帧观察者。``observer=None`` 表示注销。"""
        if observer is None:
            # 必须先调 native：让 SDK 内部解绑、Detach、delete adapter。
            # 在 native 返回前若先把 self._audio_observer = None，CFUNCTYPE
            # 闭包可能被 GC；与此同时 native 端 A-Play / audio 线程上
            # in-flight 的回调仍持有这个函数指针，跳进去就是 UAF。
            ret = int(load_library().aoq_set_audio_frame_observer(
                self._handle, ctypes.cast(0, ctypes.POINTER(_CAudioFrameObserver))))
            prev = getattr(self, "_audio_observer", None)
            self._audio_observer = None  # type: ignore[attr-defined]
            # 旧闭包再多活一段时间：即便 SDK 内部还排着观察者引用相关任务，
            # 跳板函数指针也不会指向已释放的内存。模块级保活，进程退出时回收。
            if prev is not None:
                _AUDIO_OBSERVER_KEEPALIVE.append(prev)
            return ret

        cbs = self._build_audio_observer(observer)
        # 保持引用，防止回调被 GC
        self._audio_observer = (observer, cbs)  # type: ignore[attr-defined]
        return int(load_library().aoq_set_audio_frame_observer(
            self._handle, ctypes.byref(cbs)))

    def enable_audio_frame_observer(
        self,
        enabled: bool,
        audio_source: AoqAudioSource,
        config: AoqAudioObserverConfig,
    ) -> int:
        c = _CAudioObserverConfig(
            sample_rate=int(config.sample_rate),
            channels=int(config.channels),
            mode=int(config.mode),
        )
        return int(load_library().aoq_enable_audio_frame_observer(
            self._handle, int(bool(enabled)), int(audio_source), ctypes.byref(c)))

    # -------------------- 视频帧观察者 --------------------

    def set_video_frame_observer(
        self, observer: Optional[IVideoFrameObserver]
    ) -> int:
        """注册或注销视频帧观察者。``observer=None`` 表示注销。"""
        if observer is None:
            # 与音频观察者同样的顺序约束：先调 native 解绑，再释放
            # Python 侧引用，旧闭包进保活列表防止 in-flight 回调 UAF。
            ret = int(load_library().aoq_set_video_frame_observer(
                self._handle, ctypes.cast(0, ctypes.POINTER(_CVideoFrameObserver))))
            prev = getattr(self, "_video_observer", None)
            self._video_observer = None  # type: ignore[attr-defined]
            if prev is not None:
                _VIDEO_OBSERVER_KEEPALIVE.append(prev)
            return ret

        cbs = self._build_video_observer(observer)
        # 保持引用，防止回调被 GC
        self._video_observer = (observer, cbs)  # type: ignore[attr-defined]
        return int(load_library().aoq_set_video_frame_observer(
            self._handle, ctypes.byref(cbs)))

    def enable_video_frame_observer(
        self,
        enabled: bool,
        video_source: AoqVideoSource,
        config: AoqVideoObserverConfig,
    ) -> int:
        c = _CVideoObserverConfig(
            format=int(config.format),
            alignment=int(config.alignment),
            mode=int(config.mode),
            mirror_applied=int(bool(config.mirror_applied)),
        )
        return int(load_library().aoq_enable_video_frame_observer(
            self._handle, int(bool(enabled)), int(video_source), ctypes.byref(c)))

    # -------------------- 实时消息 --------------------

    def send_data_msg(self, msg: AoqDataMsg) -> int:
        buf = (ctypes.c_uint8 * len(msg.data)).from_buffer_copy(msg.data) \
            if msg.data else (ctypes.c_uint8 * 0)()
        c = _CDataMsg(
            data=ctypes.cast(buf, ctypes.POINTER(ctypes.c_uint8)),
            data_size=len(msg.data),
        )
        return int(load_library().aoq_send_data_msg(self._handle, ctypes.byref(c)))

    # -------------------- 媒体流发送控制 --------------------

    def enable_send_media_stream(
        self, track_type: AoqTrackType, enable: bool
    ) -> int:
        """Controls whether local media stream is sent for the given track.

        Args:
            track_type: AoqTrackType.AUDIO / VIDEO
            enable: True to enable sending, False to disable.

        Returns:
            0 on success (async); non-zero on failure.
        """
        return int(load_library().aoq_enable_send_media_stream(
            self._handle, int(track_type), int(bool(enable))))

    # -------------------- 内部：回调桥接 --------------------

    @classmethod
    def _build_callbacks(cls, listener: AoqEngineEventListener) -> _CEventCallbacks:
        # 在 native 线程被回调时使用：避免 traceback 在子线程触发任何 ctypes 路径，
        # 仅以最朴素的 sys.stderr.write 报告异常，最大限度避免二次崩溃。
        def _safe(fn: Callable, *args):
            try:
                fn(*args)
            except BaseException as exc:  # noqa: BLE001 - 防止异常穿越 native
                try:
                    import sys as _sys
                    _sys.stderr.write(
                        f"[aoq_client_sdk] callback {getattr(fn, '__name__', fn)!r} "
                        f"raised {type(exc).__name__}: {exc}\n"
                    )
                    _sys.stderr.flush()
                except BaseException:  # noqa: BLE001
                    pass

        @_OnErrorFn
        def on_error(_ud, code, msg):
            _safe(listener.on_error, int(code), _to_str(msg))


        @_OnWarningFn
        def on_warning(_ud, code, msg):
            # AoqEngineEventListener 基类未定义 on_warning，转发到 on_error 同名（如有），
            # 否则忽略；保持与 C 端 aoq_event_callbacks_t.on_warning 槽位对齐。
            handler = getattr(listener, "on_warning", None)
            if callable(handler):
                _safe(handler, int(code), _to_str(msg))

        @_OnConnFn
        def on_conn(_ud, status):
            # 诊断打印：第一次回调时 dump listener 真实类型，便于排查 use-after-free
            try:
                import sys as _sys
                _sys.stderr.write(
                    f"[aoq_client_sdk][diag] on_conn fired status={int(status)} "
                    f"listener_type={type(listener).__name__} "
                    f"listener_id=0x{id(listener):x} "
                    f"attr_type={type(getattr(listener, 'on_connection_status_change', None)).__name__}\n"
                )
                _sys.stderr.flush()
            except BaseException:
                pass
            _safe(listener.on_connection_status_change, AoqConnectionStatus(int(status)))

        @_OnDataMsgFn
        def on_data(_ud, data, size):
            raw = bytes(ctypes.string_at(data, size)) if data and size else b""
            _safe(listener.on_data_msg, AoqDataMsg(data=raw))

        cbs = _CEventCallbacks(
            user_data=None,
            on_error=on_error,
            on_warning=on_warning,
            on_connection_status_change=on_conn,
            on_data_msg=on_data,
        )
        # 把闭包绑定到结构体对象上，避免被 GC 后回调失效
        cbs._trampolines = (  # type: ignore[attr-defined]
            on_error, on_warning, on_conn,
            on_data,
        )
        return cbs

    @staticmethod
    def _build_audio_observer(observer: IAudioFrameObserver) -> _CAudioFrameObserver:
        """构造 native 音频帧观察者函数表，闭包引用保持在结构体上避免被 GC。"""

        def _safe(fn: Callable, *args):
            try:
                fn(*args)
            except Exception:  # noqa: BLE001
                import traceback
                traceback.print_exc()

        def _to_frame(p_frame) -> AoqAudioFrameData:
            if not p_frame:
                return AoqAudioFrameData()
            f = p_frame.contents
            data_size = max(int(f.data_size), 0)
            raw = ctypes.string_at(f.data_ptr, data_size) if (f.data_ptr and data_size) else b""
            return AoqAudioFrameData(
                data=bytes(raw),
                num_of_samples=int(f.num_of_samples),
                bytes_per_sample=int(f.bytes_per_sample),
                num_of_channels=int(f.num_of_channels),
                samples_per_sec=int(f.samples_per_sec),
                push_sequence=int(f.push_sequence),
                time_stamp=int(f.time_stamp),
                auto_gen_mute=bool(f.auto_gen_mute),
            )

        @_OnAudioFrameFn
        def on_captured(_ud, p):
            _safe(observer.on_captured_audio_frame, _to_frame(p))

        @_OnAudioFrameFn
        def on_proc(_ud, p):
            _safe(observer.on_process_captured_audio_frame, _to_frame(p))

        @_OnTrackedAudioFrameFn
        def on_pub(_ud, tt, p):
            _safe(observer.on_publish_audio_frame, AoqTrackType(int(tt)), _to_frame(p))

        @_OnAudioFrameFn
        def on_play(_ud, p):
            _safe(observer.on_playback_audio_frame, _to_frame(p))

        cbs = _CAudioFrameObserver(
            user_data=None,
            on_captured=on_captured,
            on_process_captured=on_proc,
            on_publish=on_pub,
            on_playback=on_play,
        )
        cbs._trampolines = (  # type: ignore[attr-defined]
            on_captured, on_proc, on_pub, on_play,
        )
        return cbs

    @staticmethod
    def _build_video_observer(observer: IVideoFrameObserver) -> _CVideoFrameObserver:
        """构造 native 视频帧观察者函数表，闭包引用保持在结构体上避免被 GC。"""

        def _safe(fn: Callable, *args) -> int:
            try:
                return 1 if fn(*args) else 0
            except Exception:  # noqa: BLE001
                import traceback
                traceback.print_exc()
                return 0

        def _to_frame(p_frame) -> AoqVideoFrame:
            if not p_frame:
                return AoqVideoFrame()
            f = p_frame.contents
            fmt = AoqVideoPixelFormat(int(f.format)) \
                if int(f.format) in AoqVideoPixelFormat._value2member_map_ \
                else AoqVideoPixelFormat.UNKNOWN
            width = int(f.width)
            height = int(f.height)
            data = b""
            data_y = b""
            data_u = b""
            data_v = b""
            if fmt == AoqVideoPixelFormat.I420 and f.data_y:
                # I420 三平面按 stride*rows 拷贝，UV 平面行数为高度一半
                y_rows = height
                uv_rows = (height + 1) // 2
                if f.stride_y > 0:
                    data_y = bytes(ctypes.string_at(f.data_y, int(f.stride_y) * y_rows))
                if f.data_u and f.stride_u > 0:
                    data_u = bytes(ctypes.string_at(f.data_u, int(f.stride_u) * uv_rows))
                if f.data_v and f.stride_v > 0:
                    data_v = bytes(ctypes.string_at(f.data_v, int(f.stride_v) * uv_rows))
            elif f.data_ptr and f.data_size > 0:
                data = bytes(ctypes.string_at(f.data_ptr, int(f.data_size)))
            return AoqVideoFrame(
                format=fmt,
                width=width,
                height=height,
                data=data,
                data_y=data_y,
                data_u=data_u,
                data_v=data_v,
                stride_y=int(f.stride_y),
                stride_u=int(f.stride_u),
                stride_v=int(f.stride_v),
                time_stamp=int(f.time_stamp),
            )

        @_OnObservedVideoFrameFn
        def on_captured(_ud, p):
            return _safe(observer.on_captured_video_frame, _to_frame(p))

        @_OnTrackedObservedVideoFrameFn
        def on_pre_encode(_ud, tt, p):
            return _safe(observer.on_pre_encode_video_frame,
                         AoqTrackType(int(tt)), _to_frame(p))

        @_OnTrackedObservedVideoFrameFn
        def on_remote(_ud, tt, p):
            return _safe(observer.on_remote_video_frame,
                         AoqTrackType(int(tt)), _to_frame(p))

        cbs = _CVideoFrameObserver(
            user_data=None,
            on_captured=on_captured,
            on_pre_encode=on_pre_encode,
            on_remote=on_remote,
        )
        cbs._trampolines = (  # type: ignore[attr-defined]
            on_captured, on_pre_encode, on_remote,
        )
        return cbs
