通过Kinesis Video Streams向Amazon Connect推送音频流问题排查

我在Amazon Connect中创建了包含Start Streaming Data块的Flow，它会向Kinesis Video Stream推送流，同时通过Lambda函数传入流信息，希望这个Lambda能把音频流推回Kinesis Video Stream，让电话里的客户听到。但编写的Python脚本运行后客户听不到声音，Connect错误日志也无报错信息，求修复方案或类似管道示例。

原代码 #

import boto3
import botocore.exceptions
import types
from functools import lru_cache
import time
import io
import wave

class KinesisVideo(object):
    _CONTROL_SERVICES = ('kinesisvideo', )
    _DATA_SERVICES = ('kinesis-video-media', 'kinesis-video-archived-media')

    def __init__(self, session=None):
        self._session = session or boto3.Session()
        self._methods = {}
        for service in self._CONTROL_SERVICES + self._DATA_SERVICES:
            prototype = self._get_client_for_service(service)
            for method in prototype.meta.method_to_api_mapping.keys():
                self._methods[method] = service

    @lru_cache()
    def _get_arn_for_stream_name(self, stream_name):
        response = self.describe_stream(StreamName=stream_name)
        return response['StreamInfo']['StreamARN']

    @lru_cache()
    def _get_endpoint_for_stream_method(self, stream_arn, method):
        response = self.get_data_endpoint(StreamARN=stream_arn, APIName=method.upper())
        return response['DataEndpoint']

    @lru_cache()
    def _get_client_for_service(self, service, endpoint_url=None):
        client = self._session.client(service, endpoint_url=endpoint_url)
        if service == 'kinesis-video-media':
            client = self._patch_kinesis_video_media(client)
        return client

    @lru_cache()
    def _get_client_by_arguments(self, method, stream_name=None, stream_arn=None):
        service = self._methods[method]
        if service not in self._DATA_SERVICES:
            return self._get_client_for_service(service)
        if not (bool(stream_name) ^ bool(stream_arn)):
            raise botocore.exceptions.ParamValidationError(report=
                'One of StreamName or StreamARN must be defined ' + \
                'to determine service endpoint'
            )
        stream_arn = self._get_arn_for_stream_name(stream_name) if stream_name else stream_arn
        endpoint_url = self._get_endpoint_for_stream_method(stream_arn, method)
        return self._get_client_for_service(service, endpoint_url)

    def __getattr__(self, method):
        if method not in self._methods:
            return getattr(super(), method)
        kwarg_map = {'StreamName': 'stream_name', 'StreamARN': 'stream_arn'}
        def _api_call(**kwargs):
            filtered_kwargs = {kwarg_map[k]: v for k, v in kwargs.items() if k in kwarg_map}
            client = self._get_client_by_arguments(method, **filtered_kwargs)
            return getattr(client, method)(**kwargs)
        return _api_call

    @staticmethod
    def _patch_kinesis_video_media(client):
        client.meta.service_model._service_description['operations']['PutMedia'] = {
            'name': 'PutMedia',
            'http': {'method': 'POST', 'requestUri': '/putMedia'},
            'input': {'shape': 'PutMediaInput'},
            'output': {'shape': 'PutMediaOutput'},
            'errors': [
                {'shape': 'ResourceNotFoundException'},
                {'shape': 'NotAuthorizedException'},
                {'shape': 'InvalidEndpointException'},
                {'shape': 'ClientLimitExceededException'},
                {'shape': 'ConnectionLimitExceededException'},
                {'shape': 'InvalidArgumentException'}
            ],
            'authtype': 'v4-unsigned-body',
        }
        client.meta.service_model._shape_resolver._shape_map['PutMediaInput'] = {
            'type': 'structure',
            'required': ['FragmentTimecodeType', 'ProducerStartTimestamp'],
            'members': {
                'FragmentTimecodeType': {
                    'shape': 'FragmentTimecodeType',
                    'location': 'header',
                    'locationName': 'x-amzn-fragment-timecode-type',
                },
                'ProducerStartTimestamp': {
                    'shape': 'Timestamp',
                    'location': 'header',
                    'locationName': 'x-amzn-producer-start-timestamp',
                },
                'StreamARN': {
                    'shape': 'ResourceARN',
                    'location': 'header',
                    'locationName': 'x-amzn-stream-arn',
                },
                'StreamName': {
                    'shape': 'StreamName',
                    'location': 'header',
                    'locationName': 'x-amzn-stream-name',
                },
                'Payload': {
                    'shape': 'Payload',
                },
            },
            'payload': 'Payload',
        }
        client.meta.service_model._shape_resolver._shape_map['PutMediaOutput'] = {
            'type': 'structure',
            'members': {'Payload': {'shape': 'Payload'}},
            'payload': 'Payload',
        }
        client.meta.service_model._shape_resolver._shape_map['FragmentTimecodeType'] = {
            'type': 'string',
            'enum': ['ABSOLUTE', 'RELATIVE'],
        }
        client.put_media = types.MethodType(
            lambda self, **kwargs: self._make_api_call('PutMedia', kwargs),
            client,
        )
        client.meta.method_to_api_mapping['put_media'] = 'PutMedia'
        return client

def main():
    session = boto3.Session(
        aws_access_key_id='MY_ACCESS_KEY_ID',
        aws_secret_access_key='MY_SECRET_ACCESS_KEY',
        region_name='MY_REGION'
    )

    video = KinesisVideo(session=session)

    print(video.list_streams())
    
    start_tmstp = repr(time.time()).split('.')[0]

    print(start_tmstp)

    # Ensure the audio data is correctly handled for Amazon Connect
    with open('packages/python/output.mkv', 'rb') as payload_file:
        payload_data = payload_file.read()

    # Create a PCM file from the audio data
    pcm_data = io.BytesIO(payload_data)
    with wave.open(pcm_data, 'wb') as wav_file:
        wav_file.setnchannels(1)  # Mono
        wav_file.setsampwidth(2)  # 2 bytes per sample
        wav_file.setframerate(8000)  # Sample rate
        wav_file.writeframes(payload_data)

    response = video.put_media(
        StreamName='MY_STREAM_NAME',
        Payload=pcm_data,
        FragmentTimecodeType='RELATIVE',
        ProducerStartTimestamp=start_tmstp,
    )

    import json
    for event in map(json.loads, response['Payload'].read().decode('utf-8').splitlines()):
        print(event)

if __name__ == '__main__':
    main()

问题分析与修复步骤 #

1. 音频格式处理错误 #

MKV是容器格式，不是原始PCM数据，直接当作PCM处理会导致格式完全不匹配；同时代码中用wave模块写入原MKV数据的操作完全错误，会生成无效的音频数据。

修复：
使用pydub（依赖ffmpeg）将MKV转换为Amazon Connect要求的PCM格式（单声道、8kHz采样率、16位深度）：

from pydub import AudioSegment

# 加载MKV文件并转换格式
audio = AudioSegment.from_file("packages/python/output.mkv", format="mkv")
audio = audio.set_channels(1).set_frame_rate(8000).set_sample_width(2)

# 导出为原始PCM数据
pcm_data = io.BytesIO()
audio.export(pcm_data, format="raw")
pcm_data.seek(0)  # 重置指针到数据开头

2. 时间戳格式错误 #

原代码将时间戳转为字符串传入，但ProducerStartTimestamp需要的是浮点数类型的Unix时间戳或datetime对象。

修复：
直接使用time.time()获取正确格式的时间戳：

start_tmstp = time.time()

3. Payload指针位置错误 #

处理完音频数据后没有重置BytesIO的指针位置，导致Kinesis Video读取的是空数据。

修复：
在调用put_media前执行pcm_data.seek(0)，确保从数据开头读取。

4. 权限与流配置检查 #

• 确认Lambda角色拥有kinesisvideo:PutMedia权限
• 确认Kinesis Video Stream的媒体类型支持音频流，数据保留期配置正确
• 确保Connect的Start Streaming Data块配置为双向流（需回推音频时）

修复后的完整main函数示例 #

def main():
    session = boto3.Session(
        aws_access_key_id='MY_ACCESS_KEY_ID',
        aws_secret_access_key='MY_SECRET_ACCESS_KEY',
        region_name='MY_REGION'
    )

    video = KinesisVideo(session=session)

    print(video.list_streams())
    
    start_tmstp = time.time()

    # 正确转换音频格式
    from pydub import AudioSegment
    audio = AudioSegment.from_file("packages/python/output.mkv", format="mkv")
    audio = audio.set_channels(1).set_frame_rate(8000).set_sample_width(2)
    
    pcm_data = io.BytesIO()
    audio.export(pcm_data, format="raw")
    pcm_data.seek(0)

    response = video.put_media(
        StreamName='MY_STREAM_NAME',
        Payload=pcm_data,
        FragmentTimecodeType='RELATIVE',
        ProducerStartTimestamp=start_tmstp,
    )

    import json
    for event in map(json.loads, response['Payload'].read().decode('utf-8').splitlines()):
        print(event)

额外注意事项 #

• Lambda环境需要安装pydub，并通过Lambda层添加ffmpeg运行时依赖
• 如果是实时音频推送，需要分块推送数据，而非一次性推送整个文件
• 可通过Kinesis Video的GetMedia接口验证推送的音频数据是否有效

内容的提问来源于stack exchange，提问作者Andy_ye