Python管理LiveChannel - Object Storage Service

本文介紹Python SDK的LiveChannel常見操作，例如建立LiveChannel、列舉LiveChannel及刪除LiveChannel等。

環境準備

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))

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.'】。
已上傳推流檔案後才能產生點播列表。

Object Storage Service：Python管理LiveChannel