视频监控 > SDK 下载 > 服务端SDK >Python SDK

Python SDK

最近更新时间: 2024-08-28 15:22:08

简介

此SDK主要帮助您快速接入七牛云的视频监控平台。

主要有如下功能：

空间管理
流管理
设备管理
模板管理
录制管理

此 Python SDK 适用于 2.7 和 3.3及以上版本。

安装

直接安装:

pip install qiniu
或
easy_install qiniu

源码安装：

#从Python SDK 下载地址下载源码
tar xvzf python-sdk-$VERSION.tar.gz
cd python-sdk-$VERSION
python setup.py install

初始化鉴权

七牛 Java SDK 的所有的功能，都需要合法的授权。授权凭证的签算需要七牛账号下的一对有效的Access Key和Secret Key，这对密钥可以通过如下步骤获得：

点击注册开通七牛开发者帐号
如果已有账号，直接登录七牛开发者后台，点击这里🔗查看 Access Key 和 Secret Key

前置说明

QVS更新接口使用的是patch请求，SDK中未添加该方法，客户可以在python SDK qiniu/http.py 文件中添加下面这个方法

def _patch_with_qiniu_mac(url, data, auth):
    qn_auth = qiniu.auth.QiniuMacRequestsAuth(
        auth) if auth is not None else None
    timeout = config.get_default('connection_timeout')
    try:
        r = requests.patch(
            url,
            json=data,
            auth=qn_auth,
            timeout=timeout,
            headers=_headers)
    except Exception as e:
        return None, ResponseInfo(None, e)
    return __return_wrapper(r)

客户可在python-sdk/qiniu/services目录下创建qvs的子目录，后续的接口函数全部放置在python-sdk/qiniu/services/qvs即可。

空间管理

创建空间

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def createNamespaces(access_key, secret_key, body):
    """
    创建空间
    https://developer.qiniu.com/qvs/api/6726/create-namespace
    :param access_key: 公钥
    :param secret_key: 私钥
    :param body: 请求体
        {
            "name": 必填，空间名称(格式"^[a-zA-Z0-9_-]{1,100}$")
            "desc"：非必填，空间描述
            "accessType": 必填，接入类型(rtmp或者gb28181)
            "rtmpUrlType": accessType为“rtmp”时 必填，推拉流地址计算方式，1:static, 2:dynamic
            "domains": rtmpUrlType为1时必填，直播域名，列表格式
            "callBack"：非必填，回调地址，可用于获取空间内设备/流状态更新时的信息
            "recordTemplateId": 非必填，录制模版ID，需要录制功能时输入对应的模板ID，录制模板ID可以模板管理中获取
            "snapshotTemplateId"：非必填，截图模版ID，需要截图功能时输入对应的模板ID，截图模板ID可以模板管理中获取
            "recordTemplateApplyAll"：非必填，空间模版是否应用到全局
            "snapshotTemplateApplyAll"：非必填，截图模版是否应用到全局
        }

    :return:
        {

        }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = "http://qvs.qiniuapi.com/v1/namespaces"
    
    # 
    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, body, auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'
    # 请求体
    body = {
        "name": "test00",
        "accessType": "rtmp",
        "rtmpUrlType": 2,
        # "domains": ["tests6.com"],
        # "recordTemplateId": "xxxxxxxxxxx"
    }

    headers, result = createNamespaces(access_key, secret_key, body)
    print(f'{headers}\n{result}')

查询空间

示例

// 查询空间
# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listNamespacesInfo(access_key, secret_key, namespaceId):
    """
    查询空间信息
    https://developer.qiniu.com/qvs/api/6729/query-namespace
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}"

    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    headers, result = listNamespacesInfo(access_key, secret_key, namespaceId)
    print(f'{headers}\n{result}')

更新空间

示例


# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def updateNamespaces(access_key, secret_key, namespaceId, body):
    """
    更新空间
    https://developer.qiniu.com/qvs/api/6728/update-namespace
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param body: 请求体
            {
                "operations":[
                    {"key":"recordTemplateApplyAll", // 要修改的参数：name 空间名称;desc 空间描述;callBack 回调地址;recordTemplateId 录制模版ID;snapshotTemplateId 截图模版ID; recordTemplateApplyAll 空间模版是否应用到全局; snapshotTemplateApplyAll 截图模版是否应用到全局
                    "op":"replace", // op操作(目前支持replace和delete)
                    "value":true}， // 要修改的参数对应的value(当op为delete的时候可以忽略)
                    { },
                    .....
            ]
            }
    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}"

    # 发起PATCH请求
    ret, res = http._patch_with_qiniu_mac(url, body, auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要更新的空间ID
    namespaceId = "xxxxxxxxxxx"

    # 请求体
    body = {
        "operations": [
            {"key": "name", "op": "replace", "value": ""xxxxxxxxxxx""}
        ]
    }

    headers, result = updateNamespaces(access_key, secret_key, namespaceId, body)
    print(f'{headers}\n{result}')

获取空间列表

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listNamespaces(access_key, secret_key, offset=None, line=10, sortBy="desc:createdAt"):
    """
    获取空间列表
    https://developer.qiniu.com/qvs/api/6730/list-namespace
    :param access_key: 公钥
    :param secret_key: 私钥
    :param offset: 在全部namespace中的偏移量，时间戳
    :param line: 一次返回多少条
    :param sortBy: asc 表示升序, desc 表示降序, 默认按创建时间降序排列(可参与排序的字段有createdAt, updatedAt).asc:updatedAt表示更新时间从小到大排序, desc:updatedAt表示更新时间从大到小排序

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)
    if offset is None:
        # 请求URL
        url = f"http://qvs.qiniuapi.com/v1/namespaces?line={line}&sortBy={sortBy}"
    else:
        url = f"http://qvs.qiniuapi.com/v1/namespaces?offset={offset}&line={line}&sortBy={sortBy}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    headers, result = listNamespaces(access_key, secret_key)
    print(f'{headers}\n{result}')

禁用空间

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def disableNamespaces(access_key, secret_key, namespaceId):
    """
    禁用空间
    https://developer.qiniu.com/qvs/api/6759/disable-the-space
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/disabled"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    headers, result = disableNamespaces(access_key, secret_key, namespaceId)
    print(f'{headers}\n{result}')

启用空间

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def disableNamespaces(access_key, secret_key, namespaceId):
    """
    启用空间
    https://developer.qiniu.com/qvs/api/6760/enable-the-space
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/enabled"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    headers, result = disableNamespaces(access_key, secret_key, namespaceId)
    print(f'{headers}\n{result}')

删除空间

示例


# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def deleteNamespaces(access_key, secret_key, namespaceId):
    """
    删除空间
    https://developer.qiniu.com/qvs/api/6727/delete-namespace
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}"

    # 发起POST请求
    ret, res = http._delete_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要删除的空间ID
    namespaceId = "xxxxxxxxxxx"

    headers, result = deleteNamespaces(access_key, secret_key, namespaceId)
    print(f'{headers}\n{result}')

流管理

创建流

本接口仅用于接入类型为RTMP的空间

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def createStreams(access_key, secret_key, namespaceId, body):
    """
    创建流
    https://developer.qiniu.com/qvs/api/6734/create-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param body： 请求体
            {
                "streamId":"device009", // 必填，流名称, 流名称在空间中唯一，可包含 字母、数字、中划线、下划线；1 ~ 100 个字符长；创建后将不可修改
                "desc":"流说明信息", // 非必填，关于流的描述信息
                "recordTemplateId":"d102sns2mwhd", // 非必填，录制模版ID，配置流维度的录制模板
                "snapshotTemplateId":"截图模板ID" // 非必填，截图模版ID，配置流维度的截图模板
            }

    :return:
           200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, body, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    # 请求体
    body = {
        "streamId": "xxxxxxxxxxx",
        "desc": "创建流测试",
        # "recordTemplateId": "xxxxxxxxxxx"
    }

    headers, result = createStreams(access_key, secret_key, namespaceId, body)
    print(f'{headers}\n{result}')

查询流

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def streamsInfo(access_key, secret_key, namespaceId, streamId):
    """
    查询流信息
    https://developer.qiniu.com/qvs/api/6736/query-information-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param streamId: 流名ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}"

    # 发起GET请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流名ID
    streamId = "xxxxxxxxxxx"

    headers, result = streamsInfo(access_key, secret_key, namespaceId, streamId)
    print(f'{headers}\n{result}')

更新流

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def updateStreams(access_key, secret_key, namespaceId, streamId, body):
    """
    更新流
    https://developer.qiniu.com/qvs/api/6758/update-the-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param streamId: 流名ID
    :param body: 请求体
                {
                    "operations":[
                        {"key":"desc", //  必填，要修改的参数：desc 流描述;recordTemplateId 录制模版ID;snapshotTemplateId 截图模版ID; disabled 流是否被禁用
                        "op":"replace", // 必填，op操作(目前支持replace和delete)
                        "value":true}， // 必填，要修改的参数对应的value(当op为delete的时候可以忽略)
                        { },
                        .....
                ]
                }

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}"

    # 发起PATCH请求
    ret, res = http._patch_with_qiniu_mac(url, body, auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流名ID
    streamId = "xxxxxxxxxxx"

    # 请求体
    body = {
        "operations": [
            {"key": "desc", "op": "replace", "value": "ceshi121212"}
        ]
    }

    headers, result = updateStreams(access_key, secret_key, namespaceId, streamId, body)
    print(f'{headers}\n{result}')

获取流列表

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listStreams(access_key, secret_key, namespaceid, offset=None, line=10, qtype=0, prefix=None,
                sortBy="desc:createdAt"):
    """
    获取流列表
    https://developer.qiniu.com/qvs/api/6737/query-list-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceid: 空间名称ID
    :param offset: 在全部namespace中的偏移量，时间戳
    :param line: 一次返回多少条
    :param qtype: 查询流类型， 0:全部,1:在线流,2:离线流
    :param prefix: 流ID 前缀，可以流ID 前缀进行检索查询
    :param sortBy: asc 表示升序, desc 表示降序, 默认按创建时间降序排列(可参与排序的字段有createdAt, updatedAt).asc:updatedAt表示更新时间从小到大排序, desc:updatedAt表示更新时间从大到小排序

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)
    if offset is None and prefix is None:
        # 请求URL
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceid}/streams?line={line}&sortBy={sortBy}&qtype={qtype}"
    else:
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceid}/streams?offset={offset}&line={line}&qtype={qtype}&prefix={prefix}&sortBy={sortBy}"
    # 发起GET请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceid = "xxxxxxxxxxx"

    headers, result = listStreams(access_key, secret_key, namespaceid, )
    print(f'{headers}\n{result}')

获取流地址

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def staticStreamDomain(access_key, secret_key, namespaceid, streamid, body):
    """
    静态模式流地址
    https://developer.qiniu.com/qvs/api/6800/static-model
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceid: 空间名称ID
    :param streamid: 流ID
    :param body: 请求体
            {
                "domain": "qvs-live-hdl.qiniu.com", // 必填，域名
                "domainType": "liveHdl", // 必填，域名类型,取值："publishRtmp":rtmp推流, "liveRtmp": rtmp播放, "liveHls": hls播放, "liveHdl": flv播放
                "urlExpireSec": // 非必填，推流地址过期时间(单位为秒)
            }
    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceid}/streams/{streamid}/domain"

    # 请求header
    h = {
        "content-type": "application/json"
    }

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac_and_headers(url, json.dumps(body), auth, headers=h)

    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


def dynamicStreamDomain(access_key, secret_key, namespaceid, streamid, body):
    """
    动态模式流地址
    https://developer.qiniu.com/qvs/api/6801/dynamic-model
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceid: 空间名称ID
    :param streamid: 流ID
    :param body: 请求体
            {
                "publishIp": "", // 非必填，推流端对外IP地址
                "playIp": "", // 非必填，拉流端对外IP地址
                "urlExpireSec": // 非必填，推流地址过期时间(单位为秒)
            }
    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceid}/streams/{streamid}/urls"

    # 请求header
    h = {
        "content-type": "application/json"
    }

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac_and_headers(url, json.dumps(body), auth, headers=h)    

    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceid = "xxxxxxxxxxx"

    #  流ID
    streamid = "xxxxxxxxxxx"

    """静态模式请求体"""
    # # 请求体
    # body = {
    #     "domain": ""xxxxxxxxxxx"",
    #     "domainType": "publishRtmp"
    # }
    #
    # headers, result = staticStreamDomain(access_key, secret_key, namespaceid, streamid, body)
    # print(f'{headers}\n{result}')

    """动态模式请求体"""
    # 请求体
    body = {
        "publishIp": ""xxxxxxxxxxx"",
        "playIp": "publishRtmp"
    }

    headers, result = dynamicStreamDomain(access_key, secret_key, namespaceid, streamid, body)
    print(f'{headers}\n{result}')

查询推流历史记录

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json, time


# 日期转时间戳
def time2timestamp(datetime):
    # 转为时间数组
    timeArray = time.strptime(datetime, "%Y-%m-%d %H:%M:%S")
    timeStamp = int(time.mktime(timeArray))
    return timeStamp


def listNamespaces(access_key, secret_key, namespaceId, streamId, startDate, endDate, offset=None, line=10):
    """
    获取推流记录
    https://developer.qiniu.com/qvs/api/6742/query-flow-records
    :param access_key: 公钥
    :param secret_key: 私钥
    :param offset: 非必填，在全部namespace中的偏移量，时间戳
    :param line: 非必填，一次返回多少条
    :param start: 必填，推流开始时间(unix timestamp in second)
    :param end: 必填，推流结束时间(unix timestamp in second)

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    start = time2timestamp(startDate)

    end = time2timestamp(endDate)

    if offset is None:
        # 请求URL
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/pubhistories?start={start}&end={end}&line={line}"
    else:
        url = f"http://qvs.qiniuapi.com/v1/namespaces?start={start}&end={end}&line={line}&offset={offset}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    # 推流开始时间（Unix 时间戳）
    startDate = "2020-06-08 00:00:00"

    # 推流结束时间（Unix 时间戳）
    endDate = "2020-06-09 00:00:00"

    headers, result = listNamespaces(access_key, secret_key, namespaceId, streamId, startDate, endDate)
    print(f'{headers}\n{result}')

禁用流

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def disableStreams(access_key, secret_key, namespaceId, streamId):
    """
    禁用流
    https://developer.qiniu.com/qvs/api/6738/disable-the-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param streamId: 流ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/disabled"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    headers, result = disableStreams(access_key, secret_key, namespaceId, streamId)
    print(f'{headers}\n{result}')

启用流

示例

// 启用流
# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def enableStreams(access_key, secret_key, namespaceId, streamId):
    """
    启用流
    https://developer.qiniu.com/qvs/api/6739/enable-the-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param streamId: 流ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/enabled"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    headers, result = enableStreams(access_key, secret_key, namespaceId, streamId)
    print(f'{headers}\n{result}')

删除流

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def deleteStreams(access_key, secret_key, namespaceId, streamId):
    """
    删除流
    https://developer.qiniu.com/qvs/api/6735/delete-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param streamId: 流名ID

    :return:
           200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}"

    # 发起POST请求
    ret, res = http._delete_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 需要查询的空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流名ID
    streamId = "xxxxxxxxxxx"

    headers, result = deleteStreams(access_key, secret_key, namespaceId, streamId)
    print(f'{headers}\n{result}')

停用流

仅适用于rtmp类型的空间，用户主动断开推流。

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def stopStreams(access_key, secret_key, namespaceId, streamId):
    """
    停止推流
    https://developer.qiniu.com/qvs/api/6849/stop-using-flow
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param streamId: 流ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/stop"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    headers, result = stopStreams(access_key, secret_key, namespaceId, streamId)
    print(f'{headers}\n{result}')

设备管理

创建设备

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http, urlsafe_base64_encode
import json


def createNamespaces(access_key, secret_key, namespaceId, body):
    """
    创建设备
    https://developer.qiniu.com/qvs/api/6896/create-device
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param body: 请求体
        {
            "name": 非必填，设备名称（可包含 字母、数字、中划线、下划线；1 ~ 100 个字符长）
            "gbId"：非必填，设备国标ID 要求全局唯一, 该字段为空时，会自动创建gbid
            "username": 非必填，用户名, 4~40位，可包含大写字母、小写字母、数字、中划线。当该字段为空时,默认值就是设备国标ID
            "password": 必填，密码, 4~40位，可包含大写字母、小写字母、数字、中划线
            "pullIfRegister": 非必填，注册成功后启动拉流, 默认关闭
            "desc"：非必填，关于设备的描述信息
        }

    :return:
        {

        }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices"

    # 请求header
    h = {
        "content-type": "application/json"
    }

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac_and_headers(url, json.dumps(body), auth, headers=h)

    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 请求体
    body = {
        "name": "device1",
        "desc": "test gb28181",
        "username": "username",
        "password": "password"
    }

    headers, result = createNamespaces(access_key, secret_key, namespaceId, body)
    print(f'{headers}\n{result}')

删除设备

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def deleteNamespaces(access_key, secret_key, namespaceId, gbId):
    """
    删除设备
    https://developer.qiniu.com/qvs/api/6898/Delete-device
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param gbId: 设备国标Id
    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices/{gbId}"

    # 发起POST请求
    ret, res = http._delete_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 设备国标Id
    gbId = "xxxxxxxxxxx"

    headers, result = deleteNamespaces(access_key, secret_key, namespaceId, gbId)
    print(f'{headers}\n{result}')

查询设备

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listNamespacesInfo(access_key, secret_key, namespaceId, gbId):
    """
    查询设备信息
	https://developer.qiniu.com/qvs/api/6901/Query-device-information
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param gbId: 设备国标Id
    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices/{gbId}"

    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 设备国标Id
    gbId = "xxxxxxxxxxx"

    headers, result = listNamespacesInfo(access_key, secret_key, namespaceId, gbId)
    print(f'{headers}\n{result}')

更新设备链接

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def updateNamespaces(access_key, secret_key, namespaceId, gbId, body):
    """
    更新设备
    https://developer.qiniu.com/qvs/api/6899/update-the-device
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param gbId: 设备国标Id
    :param body: 请求体
            {
                "operations":[
                    {"key":"desc", // 要修改的参数：name 设备名称;desc 设备描述;gbId 设备国标ID;username 用户名;password 密码; pullIfRegister 注册成功后启动拉流
                    "op":"replace", // op操作(目前支持replace和delete)
                    "value":gb patch test}， // 要修改的参数对应的value(当op为delete的时候可以忽略)
                    { },
                    .....
            ]
            }
    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices/{gbId}"

    # 发起PATCH请求
    ret, res = http._patch_with_qiniu_mac(url, body, auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 设备国标Id
    gbId = "xxxxxxxxxxx"

    # 请求体
    body = {
        "operations": [
            {"key": "desc",
             "op": "replace",
             "value": "gb patch test"}
        ]
    }

    headers, result = updateNamespaces(access_key, secret_key, namespaceId, gbId, body)
    print(f'{headers}\n{result}')

获取设备列表

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listNamespaces(access_key, secret_key, namespaceId, offset=0, line=10, prefix=None, state="notReg",
                   qtype=0):
    """
    获取设备列表
    https://developer.qiniu.com/qvs/api/6902/Query-device-list
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param offset: 在全部namespace中的偏移量，integer
    :param line: 一次返回多少条
    :param prefix: 可以通过gbid前缀进行检索查询
    :param state: 按设备状态查询，offline: 离线, online: 在线, notReg: 未注册, locked: 锁定
    :param qtype: 按设备类型查询，0:全部, 1:摄像头, 2:平台

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices?offset={offset}&line={line}&prefix={prefix}&state={state}&qtype={qtype}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    headers, result = listNamespaces(access_key, secret_key, namespaceId)
    print(f'{headers}\n{result}')

获取通道列表

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listNamespaces(access_key, secret_key, namespaceId, gbId, prefix=None):
    """
    获取通道列表
    https://developer.qiniu.com/qvs/api/6906/query-channel-list
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param gbId: 设备国标Id
    :param prefix: 通道国标ID前缀

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices/{gbId}/channels?prefix={prefix}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 设备国标Id
    gbId = "xxxxxxxxxxx"

    headers, result = listNamespaces(access_key, secret_key, namespaceId, gbId)
    print(f'{headers}\n{result}')

启动设备拉流

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def startDevice(access_key, secret_key, namespaceId, gbId):
    """
    启动设备拉流
    https://developer.qiniu.com/qvs/api/6907/start-device
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param gbId: 设备国标Id

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices/{gbId}/start"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 设备国标Id
    gbId = "xxxxxxxxxxx"

    headers, result = startDevice(access_key, secret_key, namespaceId, gbId)
    print(f'{headers}\n{result}')

停止设备拉流

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def stopDevice(access_key, secret_key, namespaceId, gbId):
    """
    停止设备拉流
    https://developer.qiniu.com/qvs/api/6908/stop-device
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 空间ID
    :param gbId: 设备国标Id

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/devices/{gbId}/stop"

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac(url, None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
   	# 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 设备国标Id
    gbId = "xxxxxxxxxxx"

    headers, result = stopDevice(access_key, secret_key, namespaceId, gbId)
    print(f'{headers}\n{result}')

模板管理

创建模板

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http, urlsafe_base64_encode
import json


def createTemplate(access_key, secret_key, body):
    """
    创建模板
    https://developer.qiniu.com/qvs/api/6721/create-template
    :param access_key: 公钥
    :param secret_key: 私钥
    :param body: 请求体
        {
            "name": 必填，模版名称，格式为 1 ~ 100个字符，可包含小写字母、数字、中划线
            "desc"：非必填，模板描述
            "bucket": 必填，模版对应的对象存储的bucket
            "deleteAfterDays": 必填，存储过期时间,默认永久不过期
            "fileType": 必填，文件存储类型,取值：0（普通存储）,1（低频存储）
            "recordFileFormat": 非必填，录制文件存储格式,取值：0（m3u8格式存储）
            "templateType": 必填，模板类型,取值：0（录制模版）, 1（截图模版）
            "recordType"：templateType为0时须指定，录制模式, 0（不录制）,1（实时录制）2 (按需录制)
            "jpgOverwriteStatus": templateType为1时须指定，开启覆盖式截图(一般用于流封面)
            "jpgSequenceStatus"：templateType为1时须指定，开启序列式截图
            "jpgOnDemandStatus"：templateType为1时须指定，开启按需截图
            "recordInterval"：非必填，录制文件时长 单位为秒,600~3600
            "snapInterval": 非必填，截图间隔, 单位为秒, 10~600
        }

    :return:
        {

        }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = "http://qvs.qiniuapi.com/v1/templates"

    # 请求header
    h = {
        "content-type": "application/json"
    }

    # 发起POST请求
    ret, res = http._post_with_qiniu_mac_and_headers(url, json.dumps(body), auth, headers=h)

    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 请求体
    body = {
        "name": "test0013",
        "desc": "this is a test",
        "delateAfterDays": 7,
        "interval": 5,
        "templateType": 1,
        "bucket": "xxxxxxxxxxx",
        "jpgOverwriteStatus": True,
        "jpgSequenceStatus": True,
        "jpgOnDemandStatus": True
    }

    headers, result = createTemplate(access_key, secret_key, body)
    print(f'{headers}\n{result}')

查询模板

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listNamespacesInfo(access_key, secret_key, templateId):
    """
    查询模板信息
    https://developer.qiniu.com/qvs/api/6724/template-information
    :param access_key: 公钥
    :param secret_key: 私钥
    :param templateId: 模板ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/templates/{templateId}"

    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 模板ID
    templateId = "xxxxxxxxxxx"

    headers, result = listNamespacesInfo(access_key, secret_key, templateId)
    print(f'{headers}\n{result}')

更新模板

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def updateTemplate(access_key, secret_key, templateId, body):
    """
    更新模板
    https://developer.qiniu.com/qvs/api/6723/modify-the-template
    :param access_key: 公钥
    :param secret_key: 私钥
    :param templateId: 模板ID
    :param body: 请求体
            {
                "operations":[
                    {"key":"recordTemplateApplyAll", // 要修改的参数：name 模板名称;desc 模板描述;bucket 模版对应的对象存储的bucket;deleteAfterDays 存储过期时间,默认永久不过期;fileType 文件存储类型,取值：0（普通存储）,1（低频存储）; recordType 录制模式, 0（不录制）,1（实时录制）; jpgOverwriteStatus 开启覆盖式截图(一般用于流封面); recordInterval 录制文件时长 单位为秒, 600~3600; snapInterval 截图间隔, 单位为秒, 10~600; jpgSequenceStatus 开启序列式截图
                    "op":"replace", // op操作(目前支持replace和delete)
                    "value":true}， // 要修改的参数对应的value(当op为delete的时候可以忽略)
                    { },
                    .....
            ]
            }
    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/templates/{templateId}"

    # 发起PATCH请求
    ret, res = http._patch_with_qiniu_mac(url, body, auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 模板ID
    templateId = "xxxxxxxxxxx"

    # 请求体
    body = {
        "operations": [
            {"key": "deleteAfterDays", "op": "replace", "value": 0}
        ]
    }

    headers, result = updateTemplate(access_key, secret_key, templateId, body)
    print(f'{headers}\n{result}')

获取模板列表

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def listTemplates(access_key, secret_key, offset=None, templateType=None, match=None, line=10,
                  sortBy="desc:createdAt"):
    """
    获取模板列表
    https://developer.qiniu.com/qvs/api/6725/list-template
    :param access_key: 公钥
    :param secret_key: 私钥
    :param offset: 非必填，在全部namespace中的偏移量，时间戳
    :param line: 非必填，一次返回多少条
    :param sortBy: 非必填，asc 表示升序, desc 表示降序, 默认按创建时间降序排列(可参与排序的字段有createdAt, updatedAt).asc:updatedAt表示更新时间从小到大排序, desc:updatedAt表示更新时间从大到小排序
    :param templateType: 非必填，模板类型，取值：0（录制模版）, 1（截图模版）
    :param match: 非必填，模糊匹配查询(模版名称包含match串时返回)

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)
    if offset is None and match is None and templateType is None:
        # 请求URL
        url = f"http://qvs.qiniuapi.com/v1/templates?line={line}&sortBy={sortBy}"
    else:
        url = f"http://qvs.qiniuapi.com/v1/templates?offset={offset}&line={line}&sortBy={sortBy}&templateType={templateType}&match={match}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    headers, result = listTemplates(access_key, secret_key)
    print(f'{headers}\n{result}')

删除模板

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json


def deleteTemplate(access_key, secret_key, templateId):
    """
    删除模板
    https://developer.qiniu.com/qvs/api/6722/delete-template
    :param access_key: 公钥
    :param secret_key: 私钥
    :param templateId: 模板ID

    :return:
            {
                "code": 200
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/templates/{templateId}"

    # 发起POST请求
    ret, res = http._delete_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result


if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 模板ID
    templateId = "xxxxxxxxxxx"

    headers, result = deleteTemplate(access_key, secret_key, templateId)
    print(f'{headers}\n{result}')

录制管理

查询云端录制记录

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json, time


# 日期转时间戳
def time2timestamp(datetime):
    # 转为时间数组
    timeArray = time.strptime(datetime, "%Y-%m-%d %H:%M:%S")
    timeStamp = int(time.mktime(timeArray))
    return timeStamp


def recordHistories(access_key, secret_key, namespaceId, streamId, start, end, line=10, marker=None):
    """
    查询录制记录
    https://developer.qiniu.com/qvs/api/6745/query-recordhistories
    :param access_key: 公钥
    :param secret_key: 私钥
    :param offset: 非必填，在全部namespace中的偏移量，时间戳
    :param line: 非必填，一次返回多少条
    :param sortBy: 非必填，asc 表示升序, desc 表示降序, 默认按创建时间降序排列(可参与排序的字段有createdAt, updatedAt).asc:updatedAt表示更新时间从小到大排序, desc:updatedAt表示更新时间从大到小排序
    :param templateType: 非必填，模板类型，取值：0（录制模版）, 1（截图模版）
    :param match: 非必填，模糊匹配查询(模版名称包含match串时返回)

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    start = time2timestamp(startDate)

    end = time2timestamp(endDate)

    if marker is None:
        # 请求URL
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/recordhistories?start={start}&end={end}&line={line}"
    else:
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/recordhistories?start={start}&end={end}&marker={marker}&line={line}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    headers = {"code": res.status_code, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    # 推流开始时间（Unix 时间戳）
    startDate = "2020-06-08 00:00:00"

    # 推流结束时间（Unix 时间戳）
    endDate = "2020-06-09 00:00:00"

    headers, result = recordHistories(access_key, secret_key, namespaceId, streamId, startDate, endDate)
    print(f'{headers}\n{result}')

查询流封面

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json, time


# 日期转时间戳
def time2timestamp(datetime):
    # 转为时间数组
    timeArray = time.strptime(datetime, "%Y-%m-%d %H:%M:%S")
    timeStamp = int(time.mktime(timeArray))
    return timeStamp


def streamSnapshotss(access_key, secret_key, namespaceId, streamId):
    """
    获取直播封面截图
    https://developer.qiniu.com/qvs/api/6814/the-cover-for-screenshots
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 必填项，空间ID
    :param streamId: 必填项，流ID

    :return:
        200
            {
                "url": "http://prlghu509.test.bkt.clouddn.com/snapshot/jpg/2akrarrzns76w/t0.jpg?e=1588124787&token=Ves3WTXC8XnEHT0I_vacEQQz-9jrJZxNExcmarzQ:bH8s5m5N5Ugp2wo6ACRUVeIK280"
            }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    # 请求URL
    url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/cover"

    # 发起GET请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    result = json.loads(res.text_body)
    if res.status_code == 401:
        headers = {"code": res.status_code, "reqid": res.req_id, "error": result["error"]}
    elif res.status_code != 401 and res.text_body:
        headers = {"code": result["code"], "reqid": res.req_id, "xlog": result["error"]}
    else:
        headers = {"code": res.text_body, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    headers, result = streamSnapshotss(access_key, secret_key, namespaceId, streamId)
    print(f'{headers}\n{result}')

获取截图列表

示例

# -*- coding: utf-8 -*-
from qiniu import QiniuMacAuth, http
import json, time


# 日期转时间戳
def time2timestamp(datetime):
    # 转为时间数组
    timeArray = time.strptime(datetime, "%Y-%m-%d %H:%M:%S")
    timeStamp = int(time.mktime(timeArray))
    return timeStamp


def streamSnapshots(access_key, secret_key, namespaceId, streamId, startDate, endDate, type, line=30, marker=None):
    """
    获取截图列表
    https://developer.qiniu.com/qvs/api/6749/list-stream-snapshots
    :param access_key: 公钥
    :param secret_key: 私钥
    :param namespaceId: 必填项，空间ID
    :param streamId: 必填项，流ID
    :param type: 必填项，1:实时截图对应的图片列表
    :param line: 非必填，限定返回截图的个数，只能输入1-100的整数，不指定默认返回30个
    :param startDate:  必填项，查询开始时间(unix时间戳,单位为秒)
    :param endDate: 必填项，查询结束时间(unix时间戳,单位为秒)
    :param marker: 非必填，上一次查询返回的标记，用于提示服务端从上一次查到的位置继续查询，不指定表示从头查询

    :return:
        200 { }
    """
    auth = QiniuMacAuth(access_key, secret_key)

    start = time2timestamp(startDate)

    end = time2timestamp(endDate)

    if marker is None:
        # 请求URL
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/snapshots?type={type}&start={start}&end={end}&line={line}"
    else:
        url = f"http://qvs.qiniuapi.com/v1/namespaces/{namespaceId}/streams/{streamId}/snapshots?type={type}&start={end}&end={end}&line={line}&marker={marker}"
    # 发起POST请求
    ret, res = http._get_with_qiniu_mac(url, params=None, auth=auth)
    if res.text_body:
        result = json.loads(res.text_body)
        headers = {"code": result["code"], "reqid": res.req_id, "xlog": result["error"]}
    else:
        headers = {"code": res.text_body, "reqid": res.req_id, "xlog": res.x_log, "text_body": res.text_body}

    # 格式化响应体
    Headers = json.dumps(headers, indent=4, ensure_ascii=False)
    result = json.dumps(ret, indent=4, ensure_ascii=False)
    return Headers, result

if __name__ == '__main__':
    # 七牛账号 AK、SK
    access_key = '<access_key>'
    secret_key = '<secret_key>'

    # 空间ID
    namespaceId = "xxxxxxxxxxx"

    # 流ID
    streamId = "xxxxxxxxxxx"

    # 1:实时截图对应的图片列表
    type = 1

    # 推流开始时间（Unix 时间戳）
    startDate = "2020-06-08 00:00:00"

    # 推流结束时间（Unix 时间戳）
    endDate = "2020-06-09 00:00:00"

    headers, result = streamSnapshots(access_key, secret_key, namespaceId, streamId, startDate, endDate, type)
    print(f'{headers}\n{result}')

以上内容是否对您有帮助？