Python SDK
简介
此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，这对密钥可以通过如下步骤获得：
前置说明
- 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}') 文档反馈 (如有产品使用问题，请 提交工单)