add ci hls play key sdk and demo

flynnzzhang · flynnzzhang · commit 1ac21f8b5c5e · 2024-08-26T19:31:25.000+08:00
diff --git a/demo/ci_media.py b/demo/ci_media.py
@@ -1905,6 +1905,23 @@ def ci_get_presigned_download_url():
     print(url)
 
 
+def ci_get_hls_play_key():
+    response, data = client.ci_get_hls_play_key(bucket_name, Accept='application/json')
+    print(response)
+    print(data)
+
+
+def ci_update_hls_play_key():
+    response, data = client.ci_update_hls_play_key(
+        Bucket=bucket_name,
+        MasterPlayKey='40c502079d484466b2e9e046ce11ae06',
+        BackupPlayKey='128d75fd2b6b4f958ccbb6fc38f60f03',
+        Accept='application/json'
+    )
+    print(response)
+    print(data)
+
+
 if __name__ == "__main__":
     # ci_get_media_queue()
     # ci_get_media_jobs()
@@ -1963,4 +1980,6 @@ def ci_get_presigned_download_url():
     # ci_create_tts_jobs()
     # ci_create_translation_jobs()
     # ci_create_words_generalize_jobs()
-    ci_get_presigned_download_url()
+    # ci_get_presigned_download_url()
+    # ci_update_hls_play_key()
+    ci_get_hls_play_key()
diff --git a/qcloud_cos/cos_client.py b/qcloud_cos/cos_client.py
@@ -10588,6 +10588,131 @@ def ci_get_ai_queue(self, Bucket, State='All', QueueIds='', PageNumber='', PageS
                                        PageNumber=PageNumber, PageSize=PageSize,
                                        UrlPath="/ai_queue", **kwargs)
 
+    def ci_get_hls_play_key(self, Bucket, **kwargs):
+        """ 获取 HLS 播放密钥 https://cloud.tencent.com/document/product/436/104292
+
+            :param Bucket(string) 存储桶名称.
+            :param kwargs:(dict) 设置上传的headers.
+            :return(dict): response header.
+            :return(dict): 请求成功返回的结果,dict类型.
+
+            .. code-block:: python
+
+                config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+                client = CosS3Client(config)
+                #  获取 HLS 播放密钥
+                response, data = client.ci_get_hls_play_key(
+                    Bucket='bucket',
+                    ObjectKey='',
+                    Body={}
+                )
+                print data
+                print response
+            """
+        headers = mapped(kwargs)
+        final_headers = {}
+        params = {}
+        for key in headers:
+            if key.startswith("response"):
+                params[key] = headers[key]
+            else:
+                final_headers[key] = headers[key]
+        headers = final_headers
+
+        params = format_values(params)
+        path = "/playKey"
+        url = self._conf.uri(bucket=Bucket, path=path, endpoint=self._conf._endpoint_ci)
+
+        logger.info(
+            "ci_get_hls_play_key result, url=:{url} ,headers=:{headers}, params=:{params}".format(
+                url=url,
+                headers=headers,
+                params=params))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            auth=CosS3Auth(self._conf, path, params=params),
+            params=params,
+            headers=headers,
+            ci_request=True)
+
+        data = rt.content
+        response = dict(**rt.headers)
+        if 'Content-Type' in response:
+            if response['Content-Type'] == 'application/xml' and 'Content-Length' in response and \
+                response['Content-Length'] != 0:
+                data = xml_to_dict(rt.content)
+                format_dict(data, ['Response'])
+            elif response['Content-Type'].startswith('application/json'):
+                data = rt.json()
+
+        return response, data
+
+    def ci_update_hls_play_key(self, Bucket, MasterPlayKey=None, BackupPlayKey=None, **kwargs):
+        """ 更新 HLS 播放密钥 https://cloud.tencent.com/document/product/436/104291
+
+            :param Bucket(string) 存储桶名称.
+            :param MasterPlayKey(string) 主播放密钥.
+            :param BackupPlayKey(string) 备播放密钥.
+            :param kwargs:(dict) 设置上传的headers.
+            :return(dict): response header.
+            :return(dict): 请求成功返回的结果,dict类型.
+
+            .. code-block:: python
+
+                config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+                client = CosS3Client(config)
+                #  获取 HLS 播放密钥
+                response, data = client.ci_get_hls_play_key(
+                    Bucket='bucket',
+                    ObjectKey='',
+                    Body={}
+                )
+                print data
+                print response
+            """
+        headers = mapped(kwargs)
+        final_headers = {}
+        params = {}
+        for key in headers:
+            if key.startswith("response"):
+                params[key] = headers[key]
+            else:
+                final_headers[key] = headers[key]
+        headers = final_headers
+
+        if MasterPlayKey is not None:
+            params['masterPlayKey'] = MasterPlayKey
+        if BackupPlayKey is not None:
+            params['backupPlayKey'] = BackupPlayKey
+        params = format_values(params)
+        path = "/playKey"
+        url = self._conf.uri(bucket=Bucket, path=path, endpoint=self._conf._endpoint_ci)
+        logger.info("ci_update_hls_play_key result, url=:{url} ,headers=:{headers}, params=:{params}".format(
+            url=url,
+            headers=headers,
+            params=params))
+        rt = self.send_request(
+            method='PUT',
+            url=url,
+            bucket=Bucket,
+            auth=CosS3Auth(self._conf, path, params=params),
+            params=params,
+            headers=headers,
+            ci_request=True)
+
+        data = rt.content
+        response = dict(**rt.headers)
+        if 'Content-Type' in response:
+            if response['Content-Type'] == 'application/xml' and 'Content-Length' in response and \
+                response['Content-Length'] != 0:
+                data = xml_to_dict(rt.content)
+                format_dict(data, ['Response'])
+            elif response['Content-Type'].startswith('application/json'):
+                data = rt.json()
+
+        return response, data
+
 
 if __name__ == "__main__":
     pass
diff --git a/ut/test.py b/ut/test.py
@@ -5805,6 +5805,41 @@ def test_ci_asr_bucket():
     assert data['AsrBucket']['Name'] == ci_bucket_name
 
 
+def test_ci_hls_play_key():
+    kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"}
+
+    response, data = client.ci_update_hls_play_key(
+        Bucket=ci_bucket_name,
+        MasterPlayKey='40c502079d484466b2e9e046ce11ae06',
+        BackupPlayKey='128d75fd2b6b4f958ccbb6fc38f60f03',
+        **kwargs
+    )
+    assert response['Content-Type'] == 'application/xml'
+    assert data['PlayKeyList']['MasterPlayKey'] == '40c502079d484466b2e9e046ce11ae06'
+    assert data['PlayKeyList']['BackupPlayKey'] == '128d75fd2b6b4f958ccbb6fc38f60f03'
+
+    response, data = client.ci_update_hls_play_key(
+        Bucket=ci_bucket_name,
+        MasterPlayKey='40c502079d484466b2e9e046ce11ae07',
+        BackupPlayKey='128d75fd2b6b4f958ccbb6fc38f60f04',
+        Accept='application/json',
+        **kwargs
+    )
+    assert response['Content-Type'] == 'application/json'
+    assert data['PlayKeyList']['MasterPlayKey'] == '40c502079d484466b2e9e046ce11ae07'
+    assert data['PlayKeyList']['BackupPlayKey'] == '128d75fd2b6b4f958ccbb6fc38f60f04'
+
+    response, data = client.ci_get_hls_play_key(ci_bucket_name)
+    assert response['Content-Type'] == 'application/xml'
+    assert data['PlayKeyList']['MasterPlayKey'] == '40c502079d484466b2e9e046ce11ae07'
+    assert data['PlayKeyList']['BackupPlayKey'] == '128d75fd2b6b4f958ccbb6fc38f60f04'
+
+    response, data = client.ci_get_hls_play_key(ci_bucket_name, Accept='application/json')
+    assert response['Content-Type'] == 'application/json'
+    assert data['PlayKeyList']['MasterPlayKey'] == '40c502079d484466b2e9e046ce11ae07'
+    assert data['PlayKeyList']['BackupPlayKey'] == '128d75fd2b6b4f958ccbb6fc38f60f04'
+
+
 if __name__ == "__main__":
     setUp()
     test_post_bucket_inventory_configurations()
