阿里云官方对地址参数的解释说明
在这里插入图片描述

代码

以下都是按照RTMP协议构建,如需其他协议或其他开发语言可以参考阿里官方文档:

https://help.aliyun.com/zh/live/user-guide/ingest-and-streaming-urls?spm=a2c4g.11186623.0.0.5b1c3b4040CpuK#04753bf543bde

官方给出的代码示例

import re
import time
import hashlib
import datetime
def md5sum(src):
    m = hashlib.md5()
    m.update(src)
    return m.hexdigest()
def a_auth(uri, key, exp):
    p = re.compile("^(rtmp://)?([^/?]+)(/[^?]*)?(\\?.*)?$")
    if not p:
        return None
    m = p.match(uri)
    scheme, host, path, args = m.groups()
    if not scheme: scheme = "rtmp://"
    if not path: path = "/"
    if not args: args = ""
    rand = "0"      # "0" by default, other value is ok
    uid = "0"       # "0" by default, other value is ok
    sstring = "%s-%s-%s-%s-%s" %(path, exp, rand, uid, key)
    hashvalue = md5sum(sstring.encode('utf-8'))
    auth_key = "%s-%s-%s-%s" %(exp, rand, uid, hashvalue)
    if args:
        return "%s%s%s%s&auth_key=%s" %(scheme, host, path, args, auth_key)
    else:
        return "%s%s%s%s?auth_key=%s" %(scheme, host, path, args, auth_key)
def main():
    # 待加密的推/播流地址,example.aliyundoc.com为推/播流域名,liveApp为AppName,liveStream为StreamName
    # 推流和播流URL采用同样的加密方法
    # AppName或StreamName不超过256字符,支持数字、大小写字母、短划线(-)、下划线(_)、等号(=)。
    uri = "rtmp://example.aliyundoc.com/liveApp/liveStream"  
    # 鉴权Key,推流地址使用推流域名URL鉴权Key,播流地址使用播流域名URL鉴权Key
    key = "<input private key>"                         
    # exp值为UNIX时间戳,单位秒,最终失效时间由该值加上域名URL鉴权有效时长决定
    # 比如您在此处设置exp为:当前时间+3600秒,那最终失效时间为:当前时间+3600秒+域名URL鉴权有效时长。如果设置exp为:当前时间,那最终失效时间为:当前时间+域名URL鉴权有效时长
    exp = int(time.time()) + 1 * 3600                   
    authuri = a_auth(uri, key, exp)                     
    print("URL : %s\nAUTH: %s" %(uri, authuri))
if __name__ == "__main__":
    main()

当时官方好像并没有给出Python代码的示例所以本人按照官方标准完成的下面代码,仅作参考一切以官方为准

import hashlib
import random
import string
import time


def md5(param):
    if param is None or len(param) == 0:
        return None
    md5_hash = hashlib.md5()
    md5_hash.update(param.encode('utf-8'))
    return md5_hash.hexdigest()


def generate_push_url(push_domain, push_key, app_name, stream_name, expire_time):
    push_url = ""
    if push_key == "":
        push_url = f"rtmp://{push_domain}/{app_name}/{stream_name}"
    else:
        timestamp = int(time.time()) + expire_time
        string_to_md5 = f"/{app_name}/{stream_name}-{timestamp}-0-0-{push_key}"
        auth_key = md5(string_to_md5)
        push_url = f"rtmp://{push_domain}/{app_name}/{stream_name}?auth_key={timestamp}-0-0-{auth_key}"
    print(f"Push URL: {push_url}")


def general_pull_url(pull_domain, pull_key, app_name, stream_name, expire_time):
    rtmp_url = ""
    hls_url = ""
    flv_url = ""
    if pull_key == "":
        rtmp_url = f"rtmp://{pull_domain}/{app_name}/{stream_name}"
        hls_url = f"http://{pull_domain}/{app_name}/{stream_name}.m3u8"
        flv_url = f"http://{pull_domain}/{app_name}/{stream_name}.flv"
    else:
        timestamp = int(time.time()) + expire_time

        rtmp_to_md5 = f"/{app_name}/{stream_name}-{timestamp}-0-0-{pull_key}"
        rtmp_auth_key = md5(rtmp_to_md5)
        rtmp_url = f"rtmp://{pull_domain}/{app_name}/{stream_name}?auth_key={timestamp}-0-0-{rtmp_auth_key}"

        hls_to_md5 = f"/{app_name}/{stream_name}.m3u8-{timestamp}-0-0-{pull_key}"
        hls_auth_key = md5(hls_to_md5)
        hls_url = f"http://{pull_domain}/{app_name}/{stream_name}.m3u8?auth_key={timestamp}-0-0-{hls_auth_key}"

        flv_to_md5 = f"/{app_name}/{stream_name}.flv-{timestamp}-0-0-{pull_key}"
        flv_auth_key = md5(flv_to_md5)
        flv_url = f"http://{pull_domain}/{app_name}/{stream_name}.flv?auth_key={timestamp}-0-0-{flv_auth_key}"

    print(f"RTMP URL: {rtmp_url}")
    print(f"HLS URL: {hls_url}")
    print(f"FLV URL: {flv_url}")


def main():
    app_name = ''.join(random.choices(string.ascii_letters + string.digits, k=5))
    stream_name = ''.join(random.choices(string.ascii_letters + string.digits, k=5))

    expire_time = 3600  # 过期时间
    pull_domain = "拉流域名"
    pull_key = "拉流URL鉴权Key"

    push_domain = "推流域名"
    push_key = "推流URL鉴权Key"

    general_pull_url(pull_domain, pull_key, app_name, stream_name, expire_time)
    generate_push_url(push_domain, push_key, app_name, stream_name, expire_time)


if __name__ == '__main__':
    main()

打印结果

RTMP URL: rtmp://拉流域名/uQU2G/ZOOzm?auth_key=1719481549-0-0-b07661172c6e730a735fbb624b9e0b2f
HLS URL: http://拉流域名/uQU2G/ZOOzm.m3u8?auth_key=1719481549-0-0-56a23c4b85f9f83fc3b973e532821dac
FLV URL: http://拉流域名/uQU2G/ZOOzm.flv?auth_key=1719481549-0-0-831a487d70ef905958a5e9baa39ae40c
Push URL: rtmp://推流域名/uQU2G/ZOOzm?auth_key=1719481549-0-0-b6bf523e3ac9658d297e30aa9dd8db30
Logo

火山引擎视频云技术社区,是面向 AI 音视频开发者的技术交流平台。这里汇聚源自抖音、豆包等亿级 DAU 产品的 RTC、直播、点播、AI 媒体处理、音视频互动技术,提供接入指南、最佳实践、性能调优、场景案例、Demo 代码、开源项目、白皮书和 API 文档。社区汇聚官方工程师与一线开发者,为 AI 视频通话、数字人、AI 视频处理等应用的开发与落地提供技术支持。

更多推荐