Object Storage Service:Manage LiveChannels (Python SDK V1)

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Create and configure the live channel.
# The channel name is test_rtmp_live. The generated M3U8 file for the live stream is named test.m3u8. This manifest contains three TS files, and each TS file has a duration of 5 seconds. This is a recommended value. The actual duration depends on the keyframes.
channel_name = "test_rtmp_live"
playlist_name = "test.m3u8"

create_result = bucket.create_live_channel(
        channel_name,
        oss2.models.LiveChannelInfo(
            status = 'enabled',
            description = 'A test live channel',
            target = oss2.models.LiveChannelInfoTarget(
                playlist_name = playlist_name,
                frag_count = 3,
                frag_duration = 5)))

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# List LiveChannels that match the rule.
# List all eligible live channels in the bucket.
# param: prefix (Type: str) specifies the prefix of the live channel names to list. If not specified, all live channels are listed.
# return: class:`ListLiveChannelResult <oss2.models.ListLiveChannelResult>`
for info in oss2.LiveChannelIterator(bucket, prefix="test"):
    print(info.name)

# Delete the LiveChannel.
bucket.delete_live_channel(info.name)

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Enable or disable the live channel.
bucket.put_live_channel_status(channel_name, 'enabled')
bucket.put_live_channel_status(channel_name, 'disabled')

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Get the ingest and playback URLs.
# After you create a live channel, you get a publish_url for stream ingest (an RTMP ingest URL) and a play_url for playback (the URL of the M3U8 file generated from the stream).
# Before you get the URLs and signature, you must create a LiveChannel and get the create_result. For more information, see the example for creating a LiveChannel.
publish_url = create_result.publish_url
play_url = create_result.play_url
print("Ingest URL:", publish_url)
print("Playback URL:", play_url)

# After you get the ingest and playback URLs, you can ingest streams to and play streams from OSS. If the bucket ACL is not public-read-write, you must sign the ingest URL. If the bucket ACL is public-read-write, you can use the publish_url directly for stream ingest.
# The expires parameter specifies a relative time in seconds. It indicates the number of seconds until the ingest URL expires.
# All parameters are included in the signature.
# After you get the signed URL, you can use a stream ingest tool to start streaming. After a connection is established with OSS, the stream is not interrupted even if the URL expires. OSS checks the validity of the expires parameter only when a new stream ingest connection is established.
signed_url = bucket.sign_rtmp_url(channel_name, playlist_name, expires=3600)
print(signed_url)

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# View the status information of the current stream.
get_status = bucket.get_live_channel_stat(channel_name)
print("Connection Time:", get_status.connected_time)
print("IP of the Ingest Client:", get_status.remote_addr)
print("Ingest Status:", get_status.status)

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', '***')

# Create a Bucket instance
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"
# Generate a VOD playlist.
# To generate a VOD playlist from the TS files of a live stream, use the post_vod_playlist method.
# This example sets the start time to 3600 seconds before the current time and the end time to the current time. This generates a playlist for the stream ingested in the last hour.
# After this operation is successfully called, a playlist file named "my_vod_list.m3u8" is generated in OSS.
bucket.post_vod_playlist(
                channel_name,
                playlist_name,
                start_time = start_time,
                end_time = end_time)

# To view the content of a playlist for a specific time period, use get_vod_playlist.
result = bucket.get_vod_playlist(channel_name, start_time=start_time, end_time=end_time)
print("playlist:", result.playlist)

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Get the configuration information of the LiveChannel.
get_result = bucket.get_live_channel(channel_name)
print("-------------------")
print("Stream Ingest Configuration")
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("-------------------")

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', '***')

# Create a Bucket instance.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# View the historical stream ingest records of a channel.
history_result = bucket.get_live_channel_history(channel_name)
print("Number of historical ingest records:",len(history_result.records))