A script for managing Aliyun OSS live channels

config.yml

access_key_id:
access_key_secret:
endpoint:
bucket_name:

osslive.py

#!/usr/bin/env python3

import argparse
import arrow
import json
import oss2
import pathlib
import sys
import yaml

def init_config(conf):
    return yaml.load(open(conf,'r').read())

def init_bucket(conf):
    config=init_config(conf)
    if config['endpoint'].startswith('https://'):
        endpoint=config['endpoint']
    elif config['endpoint'].startswith('http://'):
        endpoint=config['endpoint'].replace('http://','https://')
    else:
        endpoint='https://%s'%config['endpoint']
    return oss2.Bucket(oss2.Auth(config['access_key_id'],config['access_key_secret']),endpoint,config['bucket_name'])

def to_datetime(timestamp):
    if timestamp:
        return arrow.get(timestamp).to('Asia/Shanghai').strftime('%Y-%m-%dT%H:%M:%SZ')
    else:
        return

def create_channel(conf,create,status,desc,playlist_name,frag_count,frag_duration):
    bucket=init_bucket(conf)
    if playlist_name is None:
        playlist_name='%s.m3u8'%create
    bucket.create_live_channel(create,oss2.models.LiveChannelInfo(status,desc,oss2.models.LiveChannelInfoTarget(playlist_name=playlist_name,frag_count=frag_count,frag_duration=frag_duration)))

def get_channels_info(conf,info,has_history):
    bucket=init_bucket(conf)
    data={}
    for channel in oss2.LiveChannelIterator(bucket,'',1000):
        if info=='all' or info==channel.name:
            result=bucket.get_live_channel(channel.name)
            status=bucket.get_live_channel_stat(channel.name)
            obj={}
            obj['description']=result.description
            obj['status']=result.status
            obj['type']=result.target.type
            obj['fragmentCount']=int(result.target.frag_count)
            obj['fragmentDuration']=int(result.target.frag_duration)
            obj['playlistName']=result.target.playlist_name
            obj['remoteAddress']=status.remote_addr
            obj['connectedTime']=to_datetime(status.connected_time)
            if status.video:
                video={}
                video['width']=status.video.width
                video['height']=status.video.height
                video['frameRate']=status.video.frame_rate
                video['bandwidth']=status.video.bandwidth
                video['codec']=status.video.codec
                obj['video']=video
            if status.audio:
                audio={}
                audio['bandwidth']=status.audio.bandwidth
                audio['sampleRate']=status.audio.sample_rate
                audio['codec']=status.audio.codec
                obj['audio']=audio
            if has_history:
                history=[]
                for record in bucket.get_live_channel_history(channel.name).records:
                    value={}
                    value['startTime']=to_datetime(record.start_time)
                    value['endTime']=to_datetime(record.end_time)
                    value['remoteAddress']=record.remote_addr
                    history.append(value)
                obj['history']=history
            data[channel.name]=obj
    print(json.dumps(data,indent=2))

def put_channel_status(conf,put,status):
    bucket=init_bucket(conf)
    bucket.put_live_channel_status(put,status)

def list_channels(conf,list,expires,protocol):
    config=init_config(conf)
    bucket=init_bucket(conf)
    if config['endpoint'].startswith('https://'):
        domain_name=config['endpoint'].replace('https://','')
    elif config['endpoint'].startswith('http://'):
        domain_name=config['endpoint'].replace('http://','')
    else:
        domain_name=config['endpoint']
    data={}
    for channel in oss2.LiveChannelIterator(bucket,'',1000):
        if list=='all' or list==channel.name:
            playlist_name=bucket.get_live_channel(channel.name).target.playlist_name
            obj={}
            if protocol=='all' or protocol=='rtmp':
                obj['rtmpPath']=bucket.sign_rtmp_url(channel.name,playlist_name,expires)
            if protocol=='all' or protocol=='hls':
                obj['hlsPath']='https://%s.%s/%s/%s'%(config['bucket_name'],domain_name,channel.name,playlist_name)
            data[channel.name]=obj
    print(json.dumps(data,indent=2))

def delete_channel(conf,delete):
    bucket=init_bucket(conf)
    bucket.delete_live_channel(delete)

def main():
    parser=argparse.ArgumentParser()
    add=parser.add_argument
    add('--config',type=pathlib.Path,default='config.yml')
    add('-c','--create')
    add('--status',choices=['enabled','disabled'],default='enabled')
    add('--description')
    add('--playlist-name')
    add('--frag-count',type=int,default=2)
    add('--frag-duration',type=int,default=5)
    add('-i','--info',nargs='?',const='all')
    add('--history',action='store_true')
    add('-p','--put')
    add('-l','--list',nargs='?',const='all')
    add('--expires',type=int,default=14400)
    add('--protocol',choices=['all','rtmp','hls'],default='all')
    add('-d','--delete')
    args=parser.parse_args()
    if args.create:
        sys.exit(create_channel(args.config,args.create,args.status,args.description,args.playlist_name,args.frag_count,args.frag_duration))
    elif args.info:
        sys.exit(get_channels_info(args.config,args.info,args.history))
    elif args.put:
        sys.exit(put_channel_status(args.config,args.put,args.status))
    elif args.list:
        sys.exit(list_channels(args.config,args.list,args.expires,args.protocol))
    elif args.delete:
        sys.exit(delete_channel(args.config,args.delete))

if __name__=='__main__':
    main()

requirements.txt

arrow
oss2
PyYAML