本文介紹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.'】。
已上傳推流文件后才能生成點播列表。