AI 助理
备案控制台
文档
输入文档关键字查找
首页 对象存储 开发参考 SDK参考 Python LiveChannel管理

Python管理LiveChannel

更新时间：
产品详情
相关技术圈
我的收藏

本文介绍Python SDK的LiveChannel常见操作，例如创建LiveChannel、列举LiveChannel及删除LiveChannel等。

说明

本文示例由阿里云用户fralychen提供，仅供参考。

环境准备

  • Python 3.6

    说明

    本文档示例基于Python 3.6版本编写，同样适用于Python 2.6、2.7、3.3、3.4、3.5版本。

  • aliyun-oss-python-sdk 2.9.0

  • OBS Studio推流工具

  • IDE

创建 LiveChannel

通过RTMP协议上传音视频数据前，必须先调用该接口创建一个LiveChannel。调用PutLiveChannel接口会返回RTMP推流地址，以及对应的播放地址。

说明

您可以使用返回的地址进行推流、播放，您还可以根据该LiveChannel的名称来发起相关的操作，如查询推流状态、查询推流记录、禁止推流等。

创建时如果已存在同名的LiveChannel，新创建的LiveChannel会覆盖之前的，且新建的LiveChannel设置和状态也会变成初始默认。

以下代码用于创建LiveChannel。

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 创建并配置流频道。
# 频道的名称是test_rtmp_live。直播生成的m3u8文件叫做test.m3u8，该索引文件包含3片ts文件，每片ts文件的时长为5秒（这只是一个建议值，具体的时长取决于关键帧）。
channel_name = "test_rtmp_live"
playlist_name = "test.m3u8"

create_result = bucket.create_live_channel(
        channel_name,
        oss2.models.LiveChannelInfo(
            status = 'enabled',
            description = '测试使用的直播频道',
            target = oss2.models.LiveChannelInfoTarget(
                playlist_name = playlist_name,
                frag_count = 3,
                frag_duration = 5)))

列举和删除LiveChannel

以下代码用于列举和删除LiveChannel：

说明

如果存在多个LiveChannel，列举删除时会根据前缀只删除单个生成时间最晚的流，所以需要删除特定流时建议“prefix”填入流的全称。由于逻辑是先列举后删除，执行时会列举出包含本次被删除流在内的所有包含这个前缀的LiveChannel，删除成功没有返回值，不存在前缀的流时，调用会报错。

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 列举符合规则的LiveChannel。
# 列举出Bucket下所有符合条件的livechannel。
# param: prefix (类型： str)表示要列举的livechannel名称的前缀，不指定则列举所有的livechannel。
# return: class:`ListLiveChannelResult <oss2.models.ListLiveChannelResult>`
for info in oss2.LiveChannelIterator(bucket, prefix="test"):
    print(info.name)

# 删除LiveChannel。
bucket.delete_live_channel(info.name)

设置LiveChannel状态

以下代码用于设置LiveChannel状态，如果运行完毕为出现错误信息则表示设置成功。

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 打开或关闭流频道。
bucket.put_live_channel_status(channel_name, 'enabled')
bucket.put_live_channel_status(channel_name, 'disabled')

获取RTMP推流地址及签名（仅支持V1签名）

以下代码用于获取RTMP推流地址及签名：

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 获取推流和观流地址。
# 创建直播频道之后拿到推流用的play_url（rtmp推流的url，如果Bucket ACL为非公共读写，则需要带上签名，见下文示例）和观流用的publish_url（推流产生的m3u8文件的url）。
# 执行下面获取地址及签名前需要先创建LiveChannel并获取create_result，具体参考创建LiveChannel的示例。
publish_url = create_result.publish_url
play_url = create_result.play_url
print("推流地址:", publish_url)
print("观流地址:", play_url)

# 拿到推流地址和观流地址之后就可以向OSS推流和观流。如果Bucket ACL为非公共读写，则需要对推流做签名，如果Bucket ACL为公共读写，则可以直接用publish_url推流。
# 这里的expires是一个相对时间，表示从现在开始此次推流过期的秒数。
# 所有的参数都会参与签名。
# 拿到签名后的signed_url就可以使用推流工具直接进行推流。一旦连接上OSS之后即使超出上面设置的expires也不会断流，OSS仅在每次推流连接的时候检查expires是否合法。
signed_url = bucket.sign_rtmp_url(channel_name, playlist_name, expires=3600)
print(signed_url)

获取LiveChannel状态信息

以下代码用于获取指定LiveChannel的推流状态信息。 

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 查看当前流的状态信息。
get_status = bucket.get_live_channel_stat(channel_name)
print("连接时间:", get_status.connected_time)
print("推流客户端的IP:", get_status.remote_addr)
print("推流状态:", get_status.status)

生成并查看播放列表

PostVodPlaylist接口用于为指定的LiveChannel生成一个点播用的播放列表。OSS会查询指定时间范围内由该LiveChannel推流生成的ts文件，并将其拼装为一个m3u8播放列表。

以下代码用于生成并查看播放列表：

import os
import oss2
import time

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

end_time = int(time.time())
start_time = end_time - 3600
generate_playlist = "my_vod_list.m3u8"
# 生成点播列表。
# 如果希望利用直播推流产生的ts文件生成一个点播列表，可以使用post_vod_playlist方法。
# 本示例指定起始时间为当前时间减去3600秒，结束时间为当前时间，即表示将最近一个小时的推流生成一个播放列表。
# 这个接口调用成功之后会在OSS上生成一个名为“my_vod_list.m3u8”的播放列表文件。
bucket.post_vod_playlist(
                channel_name,
                playlist_name,
                start_time = start_time,
                end_time = end_time)

# 如果想查看指定时间段内的播放列表内容，可以使用get_vod_playlist。
result = bucket.get_vod_playlist(channel_name, start_time=start_time, end_time=end_time)
print("playlist:", result.playlist)

获取LiveChannel配置信息

以下代码用于获取指定LiveChannel的配置信息。

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 获取LiveChannel的配置信息。
get_result = bucket.get_live_channel(channel_name)
print("-------------------")
print("推流配置信息")
print(get_result.description)
print(get_result.status)
print(get_result.target.type)
print(get_result.target.frag_count)
print(get_result.target.frag_duration)
print(get_result.target.playlist_name)
print("-------------------")

获取LiveChannel推流记录

使用GetLiveChannelHistory接口最多会返回指定LiveChannel最近的10次推流记录。以下代码用于获取 LiveChannel推流记录：

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# 创建Bucket实例。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# 查看一个频道历史推流记录。
history_result = bucket.get_live_channel_history(channel_name)
print("推流历史次数:",len(history_result.records))

相关文档

关于LiveChannel的更多信息，请参见LiveChannel简介、live_channel.py以及api.py。

FAQ

  • 推流状态、客户端 IP、连接时间等信息为何获取不了？

    get_live_channel_stat 获取推流状态信息要求对应的频道（channel_name）处于 Live 状态，即连接上推流地址后客户端正处于推流状态。

  • .get_live_channel_history能否获取历史推流的起止时间和远端地址？

    可以。详情请参见GetLiveChannelHistory。

  • 通过 list_live_channel 获取到的频道信息是什么类型的？

    字符串。详情请参见ListLiveChannel。

  • 生成点播列表 post_vod_playlist 函数中end_time参数所需要的格式是什么？

    整数。详情请参见PostVodPlaylist。

  • 报错【'Code': 'InvalidArgument', 'Message': 'No ts file found in specified time span.'】。

    已上传推流文件后才能生成点播列表。

上一篇：视频转码下一篇：中文和时间
文档推荐