diff --git a/demo/ai_recognition_demo.py b/demo/ai_recognition_demo.py new file mode 100644 index 00000000..2e69a421 --- /dev/null +++ b/demo/ai_recognition_demo.py @@ -0,0 +1,584 @@ +# -*- coding=utf-8 +import base64 + +from qcloud_cos import CosConfig +from qcloud_cos import AIRecognitionClient + +import os +import sys +import logging + +from qcloud_cos.streambody import StreamBody + +# 腾讯云COSV5Python SDK, 目前可以支持Python2.6与Python2.7以及Python3.x + +logging.basicConfig(level=logging.INFO, stream=sys.stdout) +# +# 设置用户属性, 包括 secret_id, secret_key, region等。Appid 已在CosConfig中移除，请在参数 Bucket 中带上 Appid。Bucket 由 BucketName-Appid 组成 +# 替换为用户的 SecretId，请登录访问管理控制台进行查看和管理，https://console.cloud.tencent.com/cam/capi +secret_id = os.environ["SECRETID"] +# 替换为用户的 SecretKey，请登录访问管理控制台进行查看和管理，https://console.cloud.tencent.com/cam/capi +secret_key = os.environ["SECRETKEY"] +# 替换为用户的 region，已创建桶归属的region可以在控制台查看，https://console.cloud.tencent.com/cos5/bucket +region = 'ap-chongqing' +# COS支持的所有region列表参见https://www.qcloud.com/document/product/436/6224 +token = None # 如果使用永久密钥不需要填入token，如果使用临时密钥需要填入，临时密钥生成和使用指引参见https://cloud.tencent.com/document/product/436/14048 + +config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, + Token=token) # 获取配置对象 +client = AIRecognitionClient(config) + +bucket_name = 'examplebucket-1250000000' + + +def ai_process_when_put_object(local_file, key, pic_operations): + response, data = client.ci_put_object_from_local_file( + Bucket=bucket_name, + LocalFilePath=local_file, + Key=key, + # pic operation json struct + PicOperations=pic_operations + ) + print(response['x-cos-request-id']) + print(data) + + +def ai_process_on_cloud(key, pic_operations): + response, data = client.ci_image_process( + Bucket=bucket_name, + Key=key, + # pic operation json struct + PicOperations=pic_operations + ) + print(response['x-cos-request-id']) + print(data) + + +def cos_create_ai_object_detect_job(): + # 图像主体检测 + response, data = client.cos_create_ai_object_detect_job( + Bucket=bucket_name, + ObjectKey="AIObjectDetect.jpeg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + ) + print(response) + print(data) + return response, data + + +def goods_matting(): + # 商品抠图下载时处理 + response, data = client.cos_goods_matting( + Bucket=bucket_name, + ObjectKey="GoodsMatting.jpg", + DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg", + CenterLayout=1, + PaddingLayout='300x300', + Stream=True + ) + data.get_stream_to_file('test.jpg') + print(response) + return response, data + + +def goods_matting_when_put_object(): + # 商品抠图上传时处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "test.png",' \ + '"rule": "ci-process=GoodsMatting¢er-layout=1&padding-layout=20x10" }]}' + return ai_process_when_put_object("GoodsMatting.jpg", "result.jpg", + operations) + + +def goods_matting_on_cloud(): + # 商品抠图云上处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "format.png",' \ + '"rule": "ci-process=GoodsMatting¢er-layout=1&padding-layout=20x10" }]}' + return ai_process_on_cloud("GoodsMatting.jpg", operations) + + +def cos_ai_body_recognition(): + # 人体识别 + response, data = client.cos_ai_body_recognition( + Bucket=bucket_name, + ObjectKey="test.jpg", + DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + ) + print(response) + print(data) + return response, data + + +def cos_ai_detect_face(): + # 人脸检测 + response, data = client.cos_ai_detect_face( + Bucket=bucket_name, + ObjectKey="test.jpg", + MaxFaceNum=3 + ) + print(response) + print(data) + return response, data + + +def cos_ai_detect_pet(): + # 宠物识别 + response, data = client.cos_ai_detect_pet( + Bucket=bucket_name, + ObjectKey="test.jpg", + ) + print(response) + print(data) + return response, data + + +def cos_ai_enhance_image(): + # 图像增强 + response, data = client.cos_ai_enhance_image( + Bucket=bucket_name, + ObjectKey="test.jpg", + Denoise=3, + Sharpen=3, + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg", + IgnoreError=0, + Stream=True + ) + data.get_stream_to_file('result.jpg') + print(response) + return response, data + + +def ai_enhance_image_when_put_object(): + # 图像增强上传时处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result2.png",' \ + '"rule": "ci-process=AIEnhanceImage&denoise=4&sharpen=4" }]}' + return ai_process_when_put_object("result.jpg", "result1.jpg", operations) + + +def ai_enhance_image_on_cloud(): + # 图像增强云上处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "ci-process=AIEnhanceImage&denoise=2&sharpen=2" }]}' + return ai_process_on_cloud("test.jpg", operations) + + +def cos_ai_face_effect(): + # 人脸特效 + response, data = client.cos_ai_face_effect( + Bucket=bucket_name, + ObjectKey="test.jpeg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg", + Type="face-beautify", + Whitening=30, + Smoothing=10, + FaceLifting=70, + EyeEnlarging=70, + ) + print(response) + print(data) + return response, data + + +def cos_ai_game_rec(): + # 游戏场景识别 + response, data = client.cos_ai_game_rec( + Bucket=bucket_name, + ObjectKey="test.png", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + ) + print(response) + print(data) + return response, data + + +def cos_ai_id_card_ocr(): + # 身份证识别 + response, data = client.cos_ai_id_card_ocr( + Bucket=bucket_name, + ObjectKey="test.jpeg", + # CardSide="FRONT", + Config='{"CropIdCard":true,"CropPortrait":true}' + ) + print(response) + print(data) + return response, data + + +def cos_ai_image_coloring(): + # 图片上色 + response, data = client.cos_ai_image_coloring( + Bucket=bucket_name, + ObjectKey="test.jpeg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + Stream=True + ) + data.get_stream_to_file('result.jpg') + print(response) + return response, data + + +def ai_image_coloring_when_put_object(): + # 图像增强上传时处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result2.png",' \ + '"rule": "ci-process=AIImageColoring" }]}' + return ai_process_when_put_object("test.jpeg", "result.jpg", operations) + + +def ai_image_coloring_on_cloud(): + # 图像增强云上处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "ci-process=AIImageColoring" }]}' + return ai_process_on_cloud("test.jpeg", operations) + + +def cos_ai_image_crop(): + # 图像智能裁剪 + response, data = client.cos_ai_image_crop( + Bucket=bucket_name, + ObjectKey="test.jpg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + Width=100, + Height=100, + Fixed=1, + IgnoreError=0 + ) + data.get_stream_to_file('result.jpg') + print(response) + return response, data + + +def ai_image_crop_when_put_object(): + # 图像增强上传时处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result2.png",' \ + '"rule": "ci-process=AIImageCrop&width=50&height=100&fixed=1" }]}' + return ai_process_when_put_object("result.jpg", "result.jpg", operations) + + +def ai_image_crop_on_cloud(): + # 图像增强云上处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "ci-process=AIImageCrop&width=50&height=100&fixed=1" }]}' + return ai_process_on_cloud("test.jpg", operations) + + +def cos_ai_license_rec(): + # 卡证识别 + response, data = client.cos_ai_license_rec( + Bucket=bucket_name, + ObjectKey="test.jpeg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + CardType="IDCard" + ) + print(response) + print(data) + return response, data + + +def cos_ai_pic_matting(): + # 通用抠图 + response, data = client.cos_ai_pic_matting( + Bucket=bucket_name, + ObjectKey="test.jpeg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + CenterLayout=1, + PaddingLayout="10x10", + Stream=True + ) + data.get_stream_to_file('result.jpg') + print(response) + return response, data + + +def ai_pic_matting_when_put_object(): + # 通用抠图上传时处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result2.png",' \ + '"rule": "ci-process=AIPicMatting¢er-layout=1&padding-layout=10x10" }]}' + return ai_process_when_put_object("result.jpg", "result.jpg", operations) + + +def ai_pic_matting_on_cloud(): + # 通用抠图云上处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "ci-process=AIPicMatting¢er-layout=1&padding-layout=10x10" }]}' + return ai_process_on_cloud("test.jpg", operations) + + +def cos_ai_portrait_matting(): + # 人像抠图 + response, data = client.cos_ai_portrait_matting( + Bucket=bucket_name, + ObjectKey="test.jpeg", + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg" + CenterLayout=1, + PaddingLayout="10x10", + Stream=True + ) + data.get_stream_to_file('result.jpg') + print(response) + return response, data + + +def ai_portrait_matting_when_put_object(): + # 人像抠图上传时处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result2.png",' \ + '"rule": "ci-process=AIPortraitMatting¢er-layout=1&padding-layout=10x10" }]}' + return ai_process_when_put_object("result.jpg", "result.jpg", operations) + + +def ai_portrait_matting_on_cloud(): + # 人像抠图云上处理 + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "ci-process=AIPortraitMatting¢er-layout=1&padding-layout=10x10" }]}' + return ai_process_on_cloud("test.jpg", operations) + + +def cos_auto_translation_block(): + # 实时文字翻译 + response, data = client.cos_auto_translation_block( + Bucket=bucket_name, + InputText="测试", + SourceLang="zh", + TargetLang="en", + TextDomain="general", + TextStyle="sentence" + ) + print(response) + print(data) + return response, data + + +def cos_get_action_sequence(): + # 获取动作顺序 + response, data = client.cos_get_action_sequence( + Bucket=bucket_name, + ) + print(response) + print(data) + return response, data + + +def cos_get_live_code(): + # 获取数字验证码 + response, data = client.cos_get_live_code( + Bucket=bucket_name, + ) + print(response) + print(data) + return response, data + + +def cos_image_repair(): + # 图像修复下载时处理 + mask_pic = 'https://ci-qta-cq-1251704708.cos.ap-chongqing.myqcloud.com/data/pic/identification/ImageRepair/mask.jpg' + mask_poly = '[[[100, 200], [1000, 200], [1000, 400], [100, 400]]]' + if len(mask_pic) != 0: + mask_pic = base64.b64encode(mask_pic.encode('utf-8')).decode('utf-8') + if len(mask_poly) != 0: + mask_poly = base64.b64encode(mask_poly.encode('utf-8')).decode('utf-8') + response, data = client.cos_image_repair( + Bucket=bucket_name, + ObjectKey="test.jpg", + # MaskPic=mask_pic, + MaskPoly=mask_poly + ) + data.get_stream_to_file('result.jpg') + print(response) + return response, data + + +def image_repair_when_put_object(): + # 图像修复上传时处理 + mask_pic = 'https://test-1250000000.cos.ap-chongqing.myqcloud.com/mask.jpg' + mask_poly = '[[[100, 200], [1000, 200], [1000, 400], [100, 400]]]' + rule = "ci-process=ImageRepair" + if len(mask_pic) != 0: + rule += "&MaskPic=" + base64.b64encode(mask_pic.encode('utf-8')).decode( + 'utf-8') + if len(mask_poly) != 0: + rule += "&MaskPoly=" + base64.b64encode( + mask_poly.encode('utf-8')).decode('utf-8') + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "' + rule + '" }]}' + return ai_process_when_put_object("test.jpg", "result.jpg", operations) + + +def image_repair_on_cloud(): + # 图像修复云上处理 + mask_pic = 'https://test-1250000000.cos.ap-chongqing.myqcloud.com/mask.jpg' + mask_poly = '[[[100, 200], [1000, 200], [1000, 400], [100, 400]]]' + rule = "ci-process=ImageRepair" + if len(mask_pic) != 0: + rule += "&MaskPic=" + base64.b64encode(mask_pic.encode('utf-8')).decode( + 'utf-8') + if len(mask_poly) != 0: + rule += "&MaskPoly=" + base64.b64encode( + mask_poly.encode('utf-8')).decode('utf-8') + operations = '{"is_pic_info":1,"rules":[{"fileid": "result.png",' \ + '"rule": "' + rule + '" }]}' + return ai_process_on_cloud("test.jpg", operations) + + +def cos_liveness_recognition(): + # 活体人脸核身 + response, data = client.cos_liveness_recognition( + Bucket=bucket_name, + ObjectKey="test.mp4", + IdCard="123456", + Name="测试", + LivenessType="SILENT", + ValidateData="", + BestFrameNum=5 + ) + print(response) + print(data) + return response, data + + +def ci_image_search_bucket(): + # 开通以图搜图 + body = { + # 图库容量限制 + # 是否必传：是 + 'MaxCapacity': 10, + # 图库访问限制，默认10 + # 是否必传：否 + 'MaxQps': 10, + } + response, data = client.ci_image_search_bucket( + Bucket=bucket_name, + Body=body, + ContentType="application/xml" + ) + print(response) + print(data) + return response, data + + +def cos_add_image_search(): + # 添加图库图片 + body = { + # 物品 ID，最多支持64个字符。若 EntityId 已存在，则对其追加图片 + # 是否必传：是 + 'EntityId': "test", + # 用户自定义的内容，最多支持4096个字符，查询时原样带回 + # 是否必传：否 + 'CustomContent': "custom test", + # 图片自定义标签，最多不超过10个，json 字符串，格式为 key:value （例 key1>=1 key1>='aa' ）对 + # 是否必传：否 + 'Tags': '{"key1":"val1","key2":"val2"}', + } + response, data = client.cos_add_image_search( + Bucket=bucket_name, + ObjectKey="result.png", + Body=body, + ContentType="application/xml" + ) + print(response) + print(data) + return response, data + + +def cos_get_search_image(): + # 图片搜索接口 + response, data = client.cos_get_search_image( + Bucket=bucket_name, + ObjectKey="result.png", + MatchThreshold=1, + Offset=0, + Limit=10, + # Filter="key1=val1" + ) + print(response) + print(data) + return response, data + + +def cos_delete_image_search(): + # 删除图库图片 + body = { + # 物品 ID + # 是否必传：是 + 'EntityId': "test", + } + response, data = client.cos_delete_image_search( + Bucket=bucket_name, + ObjectKey="result.png", + Body=body, + ContentType="application/xml" + ) + print(response) + print(data) + return response, data + + +def ci_get_ai_bucket(): + # 查询ai处理异步服务开通状态 + response = client.ci_get_ai_bucket( + Regions=region, + # BucketName='demo', + BucketNames=bucket_name, + PageSize="1", + PageNumber="1" + ) + print(response) + return response + + +def ci_close_ai_bucket(): + # 关闭AI内容识别服务 + response, data = client.ci_close_ai_bucket( + Bucket=bucket_name + ) + print(response) + print(data) + return response, data + + +def ci_open_ai_bucket(): + # 开通AI内容识别服务 + response, data = client.ci_open_ai_bucket( + Bucket=bucket_name + ) + print(response) + print(data) + return response, data + + +def ci_get_ai_queue(): + # 查询ai处理队列信息 + response = client.ci_get_ai_queue( + Bucket=bucket_name, + QueueIds='', + State='All', + PageNumber='', + PageSize='' + ) + print(response) + return response + + +def ci_put_ai_queue(): + # 更新ai队列信息 + body = { + 'Name': 'ai-queue', + 'QueueID': 'pa2c2afbe68xxxxxxxxxxxxxxxxxxxxxx', + 'State': 'Active', + 'NotifyConfig': { + 'Type': 'Url', + 'Url': 'http://www.demo.callback.com', + 'Event': 'TaskFinish', + 'State': 'On', + 'ResultFormat': 'JSON', + } + } + response = client.ci_update_ai_queue( + Bucket=bucket_name, + QueueId='pa2c2afbe68c44xxxxxxxxxxxxxxxxxxxx', + Request=body, + ContentType='application/xml' + ) + print(response) + return response + + +if __name__ == '__main__': + pass diff --git a/demo/ci_media.py b/demo/ci_media.py index 9fedd49a..04c3d696 100644 --- a/demo/ci_media.py +++ b/demo/ci_media.py @@ -78,39 +78,6 @@ def ci_get_ai_bucket(): return response -def ci_get_ai_queue(): - # 查询ai处理队列信息 - response = client.ci_get_ai_queue( - Bucket=bucket_name, - ) - print(response) - return response - - -def ci_put_ai_queue(): - # 更新ai队列信息 - body = { - 'Name': 'ai-queue', - 'QueueID': 'pa2c2afbe68xxxxxxxxxxxxxxxxxxxxxx', - 'State': 'Active', - 'NotifyConfig': { - 'Type': 'Url', - 'Url': 'http://www.demo.callback.com', - 'Event': 'TaskFinish', - 'State': 'On', - 'ResultFormat': 'JSON', - } - } - response = client.ci_update_ai_queue( - Bucket=bucket_name, - QueueId='pa2c2afbe68c44xxxxxxxxxxxxxxxxxxxx', - Request=body, - ContentType='application/xml' - ) - print(response) - return response - - def ci_get_media_pic_queue(): # 查询图片处理队列信息 response = client.ci_get_media_pic_queue( diff --git a/demo/ci_speech_recognition.py b/demo/ci_speech_recognition.py index 2766b08f..469fc440 100644 --- a/demo/ci_speech_recognition.py +++ b/demo/ci_speech_recognition.py @@ -27,6 +27,16 @@ bucket_name = 'examplebucket-1250000000' +def ci_open_asr_bucket(): + # 开通智能语音服务 + response, data = client.ci_open_asr_bucket( + Bucket=bucket_name + ) + print(response) + print(data) + return response, data + + def ci_get_asr_bucket(): # 查询语音识别开通状态 response = client.ci_get_asr_bucket( @@ -39,6 +49,16 @@ def ci_get_asr_bucket(): return response +def ci_close_asr_bucket(): + # 关闭智能语音服务 + response, data = client.ci_close_asr_bucket( + Bucket=bucket_name + ) + print(response) + print(data) + return response, data + + def ci_get_asr_queue(): # 查询语音识别队列信息 response = client.ci_get_asr_queue( @@ -166,10 +186,13 @@ def ci_delete_asr_template(): # ci_get_asr_bucket() # ci_get_asr_queue() # ci_put_asr_queue() - ci_create_asr_template() + # ci_create_asr_template() # ci_get_asr_template() # ci_update_asr_template() # ci_delete_asr_template() - ci_create_asr_jobs() + # ci_create_asr_jobs() # ci_get_asr_jobs() # ci_list_asr_jobs() + # ci_close_asr_bucket() + # ci_open_asr_bucket() + ci_get_asr_bucket() diff --git a/demo/ci_template.py b/demo/ci_template.py index b875330f..4b17e557 100644 --- a/demo/ci_template.py +++ b/demo/ci_template.py @@ -1730,10 +1730,95 @@ def ci_update_smart_cover_template(): return response +noise_reduction_template_config = { + # 固定值：NoiseReduction + # 是否必传：是 + 'Tag': "NoiseReduction", + # 模板名称，仅支持中文、英文、数字、_、-和*，长度不超过 64。 + # 是否必传：是 + 'Name': "noise_reduction_test", + # 降噪参数 + # 是否必传：是 + 'NoiseReduction': { + # 封装格式，支持 mp3、m4a、wav + # 是否必传：否 + 'Format': "wav", + # 采样率单位：Hz可选 8000、12000、16000、24000、32000、44100、48000 + # 是否必传：否 + 'Samplerate': "8000", + }, +} + + +def ci_create_noise_reduction_template(): + # 创建音频降噪模板 + response = client.ci_create_template( + Bucket=bucket_name, + Template=noise_reduction_template_config, + ) + print(response) + return response + + +def ci_update_noise_reduction_template(): + # 更新音频降噪模板 + response = client.ci_update_template( + Bucket=bucket_name, + TemplateId='t1ec6c1xxxxxxxxxxxxxxxxxxxxxx', + Template=noise_reduction_template_config, + ) + print(response) + return response + + +video_target_rec_template_config = { + # 模板类型：VideoTargetRec + # 是否必传：是 + 'Tag': "VideoTargetRec", + # 模板名称，仅支持中文、英文、数字、_、-和*，长度不超过 64 + # 是否必传：是 + 'Name': "video_target_rec_test", + # 视频目标检测 参数 + # 是否必传：是 + 'VideoTargetRec': { + # 是否开启人体检测，取值 true/false + # 是否必传：否 + 'Body': "true", + # 是否开启宠物检测，取值 true/false + # 是否必传：否 + 'Pet': "true", + # 是否开启车辆检测，取值 true/false + # 是否必传：否 + 'Car': "false", + }, +} + + +def ci_create_video_target_template(): + # 创建视频目标检测模板 + response = client.ci_create_template( + Bucket=bucket_name, + Template=video_target_rec_template_config, + ) + print(response) + return response + + +def ci_update_video_target_template(): + # 更新视频目标检测模板 + response = client.ci_update_template( + Bucket=bucket_name, + TemplateId='t17de5xxxxxxxxxxxxxxxxxxxxxxx', + Template=video_target_rec_template_config, + ) + print(response) + return response + + if __name__ == "__main__": # ci_create_snapshot_template() # ci_update_snapshot_template() - ci_create_transcode_template() + # ci_create_transcode_template() # ci_update_transcode_template() # ci_create_high_speed_hd_template() # ci_update_high_speed_hd_template() @@ -1756,4 +1841,8 @@ def ci_update_smart_cover_template(): # ci_update_tts_template() # ci_create_smart_cover_template() # ci_update_smart_cover_template() + # ci_create_noise_reduction_template() + # ci_update_noise_reduction_template() + # ci_create_video_target_template() + ci_update_video_target_template() diff --git a/demo/hls_decrypt_token.py b/demo/hls_decrypt_token.py new file mode 100644 index 00000000..a4899dce --- /dev/null +++ b/demo/hls_decrypt_token.py @@ -0,0 +1,100 @@ +# -*- coding=utf-8 +import base64 +import jwt +from urllib.parse import quote, quote_plus +from qcloud_cos import CosConfig +from qcloud_cos import CosS3Client +from datetime import datetime, timedelta + +import sys +import logging +import os + +# 腾讯云COSV5Python SDK, 目前可以支持Python2.6与Python2.7以及Python3.x + +# 媒体处理相关API请参考 https://cloud.tencent.com/document/product/460/84790 + +logging.basicConfig(level=logging.INFO, stream=sys.stdout) + +# 设置用户属性, 包括secret_id, secret_key, region +# appid已在配置中移除,请在参数Bucket中带上appid。Bucket由bucketname-appid组成 +# 这里秘钥是从环境变量取得，如自己测试可改成自己对应的秘钥 +secret_id = os.environ["SECRETID"] # 替换为用户的 SecretId，请登录访问管理控制台进行查看和管理，https://console.cloud.tencent.com/cam/capi +secret_key = os.environ["SECRETKEY"] # 替换为用户的 SecretKey，请登录访问管理控制台进行查看和管理，https://console.cloud.tencent.com/cam/capi +region = 'ap-chongqing' # 替换为用户的 region，已创建桶归属的region可以在控制台查看，https://console.cloud.tencent.com/cos5/bucket +# COS支持的所有region列表参见https://www.qcloud.com/document/product/436/6224 +token = None # 如果使用永久密钥不需要填入token，如果使用临时密钥需要填入，临时密钥生成和使用指引参见https://cloud.tencent.com/document/product/436/14048 + +config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme='https') # 获取配置对象 +client = CosS3Client(config) + + +bucket_name = 'demo-1250000000' +play_key = 'play_key' +object_name = 'test.m3u8' + + +def generate_token(): + now = datetime.now() + expire_time = now + timedelta(minutes=30) + path_encoded = quote(object_name) + object = quote_plus(path_encoded) + + headers = { + # 加密的算法，固定为 HS256 + "alg": "HS256", + # 类型，固定为 JWT + "typ": "JWT" + } + token_info = { + # 固定为 CosCiToken， 必填参数 + 'Type': 'CosCiToken', + # app id，必填参数 + 'AppId': '1250000000', + # 播放文件所在的BucketId， 必填参数 + 'BucketId': bucket_name, + # 播放文件名 + 'Object': object, + # 固定为client，必填参数 + 'Issuer': 'client', + # token颁发时间戳，必填参数 + 'IssuedTimeStamp': now.timestamp(), + # token过期时间戳，非必填参数，默认1天过期 + 'ExpireTimeStamp': expire_time.timestamp(), + # token使用次数限制，非必填参数，默认限制100000次 + 'UsageLimit': 20, + # 是否加密解密密钥（播放时解密ts视频流的密钥），1表示对解密密钥加密，0表示不对解密密钥加密。 + 'ProtectContentKey': 1, + } + if token_info['ProtectContentKey'] == 1: + public_key = """-----BEGIN PUBLIC KEY----- +xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxBgQCj9GNktf2yA0Mp8aCzxxxxxxxx +xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx54Jl4NVNewBLPZq1WFxxxxxxxxxx +xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxKpSCdl9hHxFZ732ixxxxxxxxxxxx +xxxxxxxxxxxxxxxxxxxx +-----END PUBLIC KEY-----""" + base64_public_key = base64.urlsafe_b64encode(public_key.encode('utf-8')).decode('utf-8') + token_info.update({ + # 保护模式，填写为 rsa1024 ，则表示使用 RSA 非对称加密的方式保护，公私钥对长度为 1024 bit + 'ProtectSchema': "rsa1024", + # 公钥。1024 bit 的 RSA 公钥，需使用 Base64 进行编码 + 'PublicKey': base64_public_key, + }) + return jwt.encode(token_info, play_key, algorithm="HS256", headers=headers) + + +def get_url(): + url = client.get_presigned_download_url( + Bucket=bucket_name, # 存储桶名称 + Key="/" + object_name, + Expired=3600, # 预签名超时时间 + ) + if token is not None: + url = url + "&x-cos-security-token=" + token + url = url + "&tokenType=JwtToken&expires=3600&ci-process=pm3u8&token=" + generate_token() + return url + + +if __name__ == '__main__': + print(generate_token()) + print(get_url()) diff --git a/demo/meta_insight_demo.py b/demo/meta_insight_demo.py index 7ac88046..43185cd2 100644 --- a/demo/meta_insight_demo.py +++ b/demo/meta_insight_demo.py @@ -248,7 +248,7 @@ def ci_describe_dataset(): # 查询数据集 response, data = client.ci_describe_dataset( - Datasetname="数据集名称", + DatasetName="数据集名称", Statistics="", ContentType="application/json" ) @@ -261,7 +261,7 @@ def ci_describe_dataset_binding(): # 查询数据集与存储桶的绑定关系 response, data = client.ci_describe_dataset_binding( - Datasetname="数据集名称", + DatasetName="数据集名称", Uri="uri", ContentType="application/json" ) @@ -274,9 +274,9 @@ def ci_describe_dataset_bindings(): # 查询绑定关系列表 response, data = client.ci_describe_dataset_bindings( - Datasetname="数据集名称", - Maxresults=100, - Nexttoken="下一页", + DatasetName="数据集名称", + MaxResults=100, + NextToken="下一页", ContentType="application/json" ) print(response) @@ -288,8 +288,8 @@ def ci_describe_datasets(): # 列出数据集 response, data = client.ci_describe_datasets( - Maxresults=100, - Nexttoken="下一页", + MaxResults=100, + NextToken="下一页", Prefix="数据集前缀", ContentType="application/json" ) @@ -302,7 +302,7 @@ def ci_describe_file_meta_index(): # 查询元数据索引 response, data = client.ci_describe_file_meta_index( - Datasetname="数据集名称", + DatasetName="数据集名称", Uri="cos://facesearch-12500000000", ContentType="application/json" ) @@ -369,7 +369,7 @@ def ci_update_file_meta_index(): 'DatasetName': "test001", # 用于建立索引的文件信息。 # 是否必传：是 - 'UpdateMetaFile': { + 'File': { # 自定义ID。该文件索引到数据集后，作为该行元数据的属性存储，用于和您的业务系统进行关联、对应。您可以根据业务需求传入该值，例如将某个URI关联到您系统内的某个ID。推荐传入全局唯一的值。在查询时，该字段支持前缀查询和排序，详情请见[字段和操作符的支持列表](https://cloud.tencent.com/document/product/460/106154)。 # 是否必传：否 'CustomId': "001", diff --git a/qcloud_cos/__init__.py b/qcloud_cos/__init__.py index 7f2b4da7..4a158b9f 100644 --- a/qcloud_cos/__init__.py +++ b/qcloud_cos/__init__.py @@ -5,6 +5,7 @@ from .cos_auth import CosS3Auth from .cos_comm import get_date from .meta_insight import MetaInsightClient +from .ai_recognition import AIRecognitionClient import logging diff --git a/qcloud_cos/ai_recognition.py b/qcloud_cos/ai_recognition.py new file mode 100644 index 00000000..d952f8f3 --- /dev/null +++ b/qcloud_cos/ai_recognition.py @@ -0,0 +1,1048 @@ +# -*- coding=utf-8 +import json + +from qcloud_cos import CosS3Auth +from qcloud_cos.cos_client import logger, CosS3Client +from .cos_comm import * + + +class AIRecognitionClient(CosS3Client): + + def cos_create_ai_object_detect_job(self, Bucket, ObjectKey="", + DetectUrl=None, **kwargs): + """ 图像主体检测 https://cloud.tencent.com/document/product/460/97979 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey。 detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg。. + :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) + # 图像主体检测 + response, data = client.cos_create_ai_object_detect_job( + Bucket='bucket', + ObjectKey='', + DetectUrl='' + ) + 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["ci-process"] = "AIObjectDetect" + if DetectUrl is not None: + params["detect-url"] = DetectUrl + + params = format_values(params) + + path = "/" + ObjectKey + url = self._conf.uri(bucket=Bucket, path=path) + + logger.info( + "cos_create_ai_object_detect_job 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=False) + + data = rt.content + response = dict(**rt.headers) + if 'Content-Type' in response: + if response['Content-Type'] == 'application/xml': + data = xml_to_dict(rt.content) + format_dict(data, ['Response']) + elif response['Content-Type'].startswith('application/json'): + data = rt.json() + + return response, data + + def cos_goods_matting(self, Bucket, ObjectKey="", DetectUrl=None, + CenterLayout=0, PaddingLayout=None, Stream=True, **kwargs): + """ 商品抠图 https://cloud.tencent.com/document/product/460/79735 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey. + :param CenterLayout(int) 抠图商品居中显示； 值为1时居中显示，值为0时不作处理，默认为0. + :param PaddingLayout(string) 将处理后的图片四边进行留白，形式为 padding-layout=x，左右两边各进行 dx 像素的留白，上下两边各进行 dy 像素的留白. + :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) + # 商品抠图 + response, data = client.cos_goods_matting( + Bucket='bucket', + ObjectKey='', + DetectUrl='' + ) + print data + print response + """ + params = {} + if DetectUrl is not None: + params["detect-url"] = DetectUrl + if CenterLayout != 0: + params["center-layout"] = CenterLayout + if PaddingLayout is not None: + params["padding-layout"] = PaddingLayout + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="GoodsMatting", Params=params, + NeedHeader=True, Stream=Stream, **kwargs) + + def cos_ai_body_recognition(self, Bucket, ObjectKey='', DetectUrl=None, + **kwargs): + """ 人体识别 https://cloud.tencent.com/document/product/460/83196 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg. + :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) + # 人体识别 + response, data = client.cos_ai_body_recognition( + Bucket='bucket', + ObjectKey='', + DetectUrl='' + ) + print data + print response + """ + + params = {} + if DetectUrl is not None: + params["detect-url"] = DetectUrl + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AIBodyRecognition", Params=params, + NeedHeader=True, **kwargs) + + def cos_ai_detect_face(self, Bucket, ObjectKey, MaxFaceNum=1, **kwargs): + """ 人脸检测 https://cloud.tencent.com/document/product/460/63223 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param MaxFaceNum(int) 最多处理的人脸数目。默认值为1（仅检测图片中面积最大的那张人脸），最大���为120。此参数用于控制处理待检测图片中的人脸个数，值越小，处理速度越快。. + :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) + # 人脸检测 + response, data = client.cos_ai_detect_face( + Bucket='bucket', + ObjectKey='', + MaxFaceNum='' + ) + print data + print response + """ + + params = {} + params["max-face-num"] = MaxFaceNum + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="DetectFace", + Params=params, NeedHeader=True, **kwargs) + + def cos_ai_detect_pet(self, Bucket, ObjectKey, **kwargs): + """ 宠物识别 https://cloud.tencent.com/document/product/460/95753 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :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) + # 宠物识别 + response, data = client.cos_ai_detect_pet( + Bucket='bucket', + ObjectKey='' + ) + print data + print response + """ + + params = {} + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="detect-pet", + Params=params, NeedHeader=True, **kwargs) + + def cos_ai_enhance_image(self, Bucket, ObjectKey='', Denoise=3, + Sharpen=3, DetectUrl=None, IgnoreError=None, Stream=True, **kwargs): + """ 图像增强 https://cloud.tencent.com/document/product/460/83792 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param Denoise(int) 去噪强度值，取值范围为 0 - 5 之间的整数，值为 0 时不进行去噪操作，默认值为3。. + :param Sharpen(int) 锐化强度值，取值范围为 0 - 5 之间的整数，值为 0 时不进行锐化操作，默认值为3。. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了detect-url 时，后台会处理 detect-url链接，无需再填写 ObjectKey ，detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为 http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg. + :param IgnoreError(int) . + :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) + # 图像增强 + response, data = client.cos_ai_enhance_image( + Bucket='bucket', + ObjectKey='', + Denoise='', + Sharpen='', + DetectUrl='', + IgnoreError='' + ) + print data + print response + """ + + params = {} + if Denoise is not None: + params["denoise"] = Denoise + if Sharpen is not None: + params["sharpen"] = Sharpen + if DetectUrl is not None: + params["detect-url"] = DetectUrl + if IgnoreError is not None: + params["ignore-error"] = IgnoreError + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AIEnhanceImage", Params=params, + NeedHeader=True, Stream=Stream, **kwargs) + + def cos_ai_face_effect(self, Bucket, Type, ObjectKey="", DetectUrl=None, + Whitening=30, Smoothing=10, FaceLifting=70, EyeEnlarging=70, + Gender=None, Age=None, **kwargs): + """ 人脸特效 https://cloud.tencent.com/document/product/460/47197 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg。. + :param Type(string) 人脸特效类型，人脸美颜：face-beautify；人脸性别转换：face-gender-transformation；人脸年龄变化：face-age-transformation；人像分割：face-segmentation. + :param Whitening(int) type为face-beautify时生效，美白程度，取值范围[0,100]。0不美白，100代表最高程度。默认值30. + :param Smoothing(int) type为face-beautify时生效，磨皮程度，取值范围[0,100]。0不磨皮，100代表最高程度。默认值10. + :param FaceLifting(int) type为face-beautify时生效，瘦脸程度，取值范围[0,100]。0不瘦脸，100代表最高程度。默认值70. + :param EyeEnlarging(int) type为face-beautify时生效，大眼程度，取值范围[0,100]。0不大眼，100代表最高程度。默认值70. + :param Gender(int) type为face-gender-transformation时生效，选择转换方向，0：男变女，1：女变男。无默认值，为必选项。限制：仅对图片中面积最大的人脸进行转换。. + :param Age(int) type为face-age-transformation时生效，变化到的人脸年龄,[10,80]。无默认值，为必选项。限制：仅对图片中面积最大的人脸进行转换。. + :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) + # 人脸特效 + response, data = client.cos_ai_face_effect( + Bucket='bucket', + ObjectKey='', + DetectUrl='', + Type='', + Whitening='', + Smoothing='', + FaceLifting='', + EyeEnlarging='', + Gender='', + Age='' + ) + print data + print response + """ + + params = {} + params["type"] = Type + if DetectUrl is not None: + params["detect-url"] = DetectUrl + if Whitening is not None: + params["whitening"] = Whitening + if Smoothing is not None: + params["smoothing"] = Smoothing + if FaceLifting is not None: + params["faceLifting"] = FaceLifting + if EyeEnlarging is not None: + params["eyeEnlarging"] = EyeEnlarging + if Gender is not None: + params["gender"] = Gender + if Age is not None: + params["age"] = Age + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="face-effect", + Params=params, NeedHeader=True, **kwargs) + + def cos_ai_game_rec(self, Bucket, ObjectKey='', DetectUrl=None, **kwargs): + """ 游戏场景识别 https://cloud.tencent.com/document/product/460/93153 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 图片地址. + :param DetectUrl(string) 您可以通过填写 detect-url 对任意公网可访问的图片进行游戏场景识别。不填写 detect-url 时，后台会默认处理 objectkey ；填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 objectkey ， detect-url 示例：http://www.example.com/abc.jpg。. + :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) + # 游戏场景识别 + response, data = client.cos_ai_game_rec( + Bucket='bucket', + ObjectKey='', + DetectUrl='' + ) + print data + print response + """ + + params = {} + if DetectUrl is not None: + params["detect-url"] = DetectUrl + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="AIGameRec", + Params=params, NeedHeader=True, **kwargs) + + def cos_ai_id_card_ocr(self, Bucket, ObjectKey, CardSide=None, Config=None, + **kwargs): + """ 身份证识别 https://cloud.tencent.com/document/product/460/48638 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param CardSide(string) FRONT：身份证有照片的一面（人像面）BACK：身份证有国徽的一面（国徽面）该参数如果不填，将为您自动判断身份证正反面. + :param Config(string) 以下可选字段均为 bool 类型，默认 false：CropIdCard，身份证照片裁剪（去掉证件外多余的边缘、自动矫正拍摄角度）CropPortrait，人像照片裁剪（自动抠取身份证头像区域）CopyWarn，复印件告警BorderCheckWarn，边框和框内遮挡告警ReshootWarn，翻拍告警DetectPsWarn，PS 检测告警TempIdWarn，临时身份证告警InvalidDateWarn，身份证有效日期不合法告警Quality，图片质量分数（评价图片的模糊程度）MultiCardDetect，是否开启多卡证检测参数设置方式参考：Config = {"CropIdCard":true,"CropPortrait":true}. + :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) + # 身份证识别 + response, data = client.cos_aiid_card_ocr( + Bucket='bucket', + ObjectKey='', + CardSide='', + Config='' + ) + print data + print response + """ + + params = {} + if CardSide is not None: + params["CardSide"] = CardSide + if Config is not None: + params["Config"] = Config + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="IDCardOCR", + Params=params, NeedHeader=True, **kwargs) + + def cos_ai_image_coloring(self, Bucket, ObjectKey="", DetectUrl=None, + Stream=True, **kwargs): + """ 图片上色 https://cloud.tencent.com/document/product/460/83794 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 待上色图片url，需要进行urlencode，与ObjectKey二选其一，如果同时存在，则默认以ObjectKey为准. + :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) + # 图片上色 + response, data = client.cos_ai_image_coloring( + Bucket='bucket', + ObjectKey='', + DetectUrl='' + ) + print data + print response + """ + + params = {} + if DetectUrl is not None: + params["detect-url"] = DetectUrl + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AIImageColoring", Params=params, + Stream=Stream, NeedHeader=True, **kwargs) + + def cos_ai_image_crop(self, Bucket, Width, Height, ObjectKey="", + DetectUrl=None, Fixed=0, IgnoreError=None, Stream=True, **kwargs): + """ 图像智能裁剪 https://cloud.tencent.com/document/product/460/83791 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg. + :param Width(int) 需要裁剪区域的宽度，与height共同组成所需裁剪的图片宽高比例；输入数字请大于0、小于图片宽度的像素值. + :param Height(int) 需要裁剪区域的高度，与width共同组成所需裁剪的图片宽高比例；输入数字请大于0、小于图片高度的像素值；width : height建议取值在[1, 2.5]之间，超过这个范围可能会影响效果. + :param Fixed(int) 是否严格按照 width 和 height 的值进行输出。取值为0时，宽高比例（width : height）会简化为最简分数，即如果width输入10、height输入20，会简化为1：2；取值为1时，输出图片的宽度等于width，高度等于height；默认值为0. + :param IgnoreError(int) 当此参数为1时，针对文件过大等导致处理失败的场景，会直接返回原图而不报错. + :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) + # 图像智能裁剪 + response, data = client.cos_ai_image_crop( + Bucket='bucket', + ObjectKey='', + DetectUrl='', + Width='', + Height='', + Fixed='', + IgnoreError='' + ) + print data + print response + """ + + params = {} + params["width"] = Width + params["height"] = Height + if DetectUrl is not None: + params["detect-url"] = DetectUrl + if Fixed is not None: + params["fixed"] = Fixed + if IgnoreError is not None: + params["ignore-error"] = IgnoreError + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="AIImageCrop", + Params=params, NeedHeader=True, Stream=Stream, + **kwargs) + + def cos_ai_license_rec(self, Bucket, CardType, ObjectKey='', DetectUrl=None, + **kwargs): + """ 卡证识别 https://cloud.tencent.com/document/product/460/96767 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg. + :param CardType(string) 卡证识别类型，有效值为IDCard，DriverLicense。
IDCard表示身份证；DriverLicense表示驾驶证，默认：DriverLicense. + :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) + # 卡证识别 + response, data = client.cos_ai_license_rec( + Bucket='bucket', + ObjectKey='', + DetectUrl='', + CardType='' + ) + print data + print response + """ + + params = {} + params["ci-process"] = "AILicenseRec" + params["CardType"] = CardType + if DetectUrl is not None: + params["detect-url"] = DetectUrl + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AILicenseRec", Params=params, + NeedHeader=True, **kwargs) + + def cos_ai_pic_matting(self, Bucket, ObjectKey='', DetectUrl=None, + CenterLayout=0, PaddingLayout=None, Stream=True, **kwargs): + """ 通用抠图 https://cloud.tencent.com/document/product/460/106750 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey detect-url 示例：http://www.example.com/abc.jpg ，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg。. + :param CenterLayout(int) 抠图主体居中显示；值为1时居中显示，值为0不做处理，默认为0. + :param PaddingLayout(string) 将处理后的图片四边进行留白，形式为 padding-layout=x，左右两边各进行 dx 像素的留白，上下两边各进行 dy 像素的留白，例如：padding-layout=20x10默认不进行留白操作，dx、dy 最大值为1000像素。. + :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) + # 通用抠图 + response, data = client.cos_ai_pic_matting( + Bucket='bucket', + ObjectKey='', + DetectUrl='', + CenterLayout='', + PaddingLayout='' + ) + print data + print response + """ + + params = {} + if DetectUrl is not None: + params["detect-url"] = DetectUrl + if CenterLayout is not None: + params["center-layout"] = CenterLayout + if PaddingLayout is not None: + params["padding-layout"] = PaddingLayout + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AIPicMatting", Params=params, + NeedHeader=True, Stream=Stream, **kwargs) + + def cos_ai_portrait_matting(self, Bucket, ObjectKey='', DetectUrl=None, + CenterLayout=0, PaddingLayout=None, Stream=True, **kwargs): + """ 人像抠图 https://cloud.tencent.com/document/product/460/106751 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param DetectUrl(string) 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接，无需再填写 ObjectKey。 detect-url 示例：http://www.example.com/abc.jpg，需要进行 UrlEncode，处理后为http%25253A%25252F%25252Fwww.example.com%25252Fabc.jpg。. + :param CenterLayout(int) 抠图主体居中显示；值为1时居中显示，值为0不做处理，默认为0. + :param PaddingLayout(string) 将处理后的图片四边进行留白，形式为 padding-layout=x，左右两边各进行 dx 像素的留白，上下两边各进行 dy 像素的留白，例如：padding-layout=20x10默认不进行留白操作，dx、dy最大值为1000像素。. + :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) + # 人像抠图 + response, data = client.cos_ai_portrait_matting( + Bucket='bucket', + ObjectKey='', + DetectUrl='', + CenterLayout='', + PaddingLayout='' + ) + print data + print response + """ + + params = {} + if DetectUrl is not None: + params["detect-url"] = DetectUrl + if CenterLayout is not None: + params["center-layout"] = CenterLayout + if PaddingLayout is not None: + params["padding-layout"] = PaddingLayout + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AIPortraitMatting", Params=params, + NeedHeader=True, Stream=Stream, **kwargs) + + def cos_auto_translation_block(self, Bucket, InputText, SourceLang, + TargetLang, TextDomain='general', TextStyle='sentence', **kwargs): + """ 实时文字翻译 https://cloud.tencent.com/document/product/460/83547 + + :param Bucket(string) 存储桶名称. + :param InputText(string) 待翻译的文本. + :param SourceLang(string) 输入语言，如 "zh". + :param TargetLang(string) 输出语言，如 "en". + :param TextDomain(string) 文本所属业务领域，如: "ecommerce", //缺省值为 general. + :param TextStyle(string) 文本类型，如: "title", //缺省值为 sentence. + :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) + # 实时文字翻译 + response, data = client.cos_auto_translation_block( + Bucket='bucket', + InputText='', + SourceLang='', + TargetLang='', + TextDomain='', + TextStyle='' + ) + print data + print response + """ + + params = {} + params["InputText"] = InputText + params["SourceLang"] = SourceLang + params["TargetLang"] = TargetLang + if TextDomain is not None: + params["TextDomain"] = TextDomain + if TextStyle is not None: + params["TextStyle"] = TextStyle + + path = "/" + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="AutoTranslationBlock", Params=params, + NeedHeader=True, **kwargs) + + def cos_get_action_sequence(self, Bucket, **kwargs): + """ 获取动作顺序 https://cloud.tencent.com/document/product/460/48648 + + :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) + # 获取动作顺序 + response, data = client.cos_get_action_sequence( + Bucket='bucket' + ) + print data + print response + """ + + params = {} + + path = "/" + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="GetActionSequence", Params=params, + NeedHeader=True, **kwargs) + + def cos_get_live_code(self, Bucket, **kwargs): + """ 获取数字验证码 https://cloud.tencent.com/document/product/460/48647 + + :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) + # 获取数字验证码 + response, data = client.cos_get_live_code( + Bucket='bucket' + ) + print data + print response + """ + + params = {} + path = "/" + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="GetLiveCode", + Params=params, NeedHeader=True, **kwargs) + + def cos_image_repair(self, Bucket, ObjectKey="", DetectUrl=None, + MaskPic=None, MaskPoly=None, Stream=True, **kwargs): + """ 图像修复 https://cloud.tencent.com/document/product/460/79042 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :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) + # 图像修复 + response, data = client.cos_image_repair( + Bucket='bucket', + ObjectKey='' + ) + print data + print response + """ + + params = {} + if DetectUrl is not None: + params['detect-url'] = DetectUrl + if MaskPic is not None: + params['MaskPic'] = MaskPic + if MaskPoly is not None: + params['MaskPoly'] = MaskPoly + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="ImageRepair", + Params=params, NeedHeader=True, Stream=Stream, + **kwargs) + + def cos_liveness_recognition(self, Bucket, ObjectKey, IdCard, Name, + LivenessType, ValidateData=None, BestFrameNum=None, **kwargs): + """ 活体人脸核身 https://cloud.tencent.com/document/product/460/48641 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param IdCard(string) 身份证号. + :param Name(string) 姓名。中文请使用 UTF-8编码. + :param LivenessType(string) 活体检测类型，取值：LIP/ACTION/SILENTLIP 为数字模式，ACTION 为动作模式，SILENT 为静默模式，三种模式选择一种传入. + :param ValidateData(string) 数字模式传参：数字验证码（1234），需先调用接口获取数字验证码动作模式传参：传动作顺序（2，1 or 1，2），需先调用接口获取动作顺序静默模式传参：空. + :param BestFrameNum(int) 需要返回多张最佳截图，取值范围1 - 10，不设置默认返回一张最佳截图. + :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) + # 活体人脸核身 + response, data = client.cos_liveness_recognition( + Bucket='bucket', + ObjectKey='', + CiProcess='', + IdCard='', + Name='', + LivenessType='', + ValidateData='', + BestFrameNum='' + ) + print data + print response + """ + + params = {} + params["IdCard"] = IdCard + params["Name"] = Name + params["LivenessType"] = LivenessType + if ValidateData is not None: + params["ValidateData"] = ValidateData + if BestFrameNum is not None: + params["BestFrameNum"] = BestFrameNum + + path = "/" + ObjectKey + return self.ci_process(Bucket=Bucket, Key=path, + CiProcess="LivenessRecognition", + Params=params, NeedHeader=True, **kwargs) + + def ci_image_search_bucket(self, Bucket, Body, **kwargs): + """ 开通以图搜图 https://cloud.tencent.com/document/product/460/63899 + + :param Bucket(string) 存储桶名称. + :param Body:(dict) 开通以图搜图配置信息. + :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) + # 开通以图搜图 + response, data = client.ci_image_search_bucket( + Bucket='bucket', + 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) + xml_config = format_xml(data=Body, root='Request') + path = "/" + "ImageSearchBucket" + url = self._conf.uri(bucket=Bucket, path=path, + endpoint=self._conf._endpoint_ci) + + logger.info( + "ci_image_search_bucket result, url=:{url} ,headers=:{headers}, params=:{params},xml_config=:{xml_config}".format( + url=url, + headers=headers, + params=params, + xml_config=xml_config)) + rt = self.send_request( + method='POST', + url=url, + data=xml_config, + 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 cos_add_image_search(self, Bucket, ObjectKey, Body, **kwargs): + """ 添加图库图片 https://cloud.tencent.com/document/product/460/63900 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param Body:(dict) 添加图库图片配置信息. + :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) + # 添加图库图片 + response, data = client.cos_add_image_search( + 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["ci-process"] = "ImageSearch" + params["action"] = "AddImage" + params = format_values(params) + + xml_config = format_xml(data=Body, root='Request') + + path = "/" + ObjectKey + url = self._conf.uri(bucket=Bucket, path=path) + + logger.info( + "cos_add_image_search result, url=:{url} ,headers=:{headers}, params=:{params},xml_config=:{xml_config}".format( + url=url, + headers=headers, + params=params, + xml_config=xml_config)) + rt = self.send_request( + method='POST', + url=url, + data=xml_config, + auth=CosS3Auth(self._conf, path, params=params), + params=params, + headers=headers, + ci_request=False) + + 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 cos_get_search_image(self, Bucket, ObjectKey, MatchThreshold=0, + Offset=0, Limit=10, Filter=None, **kwargs): + """ 图片搜索接口 https://cloud.tencent.com/document/product/460/63901 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param MatchThreshold(int) 出参 Score 中，只有超过 MatchThreshold 值的结果才会返回。默认为0. + :param Offset(int) 起始序号，默认值为0. + :param Limit(int) 返回数量，默认值为10，最大值为100. + :param Filter(string) 针对入库时提交的 Tags 信息进行条件过滤。支持>、>=、<、<=、=、!=，多个条件之间支持 AND 和 OR 进行连接. + :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) + # 图片搜索接口 + response, data = client.cos_get_search_image( + Bucket='bucket', + ObjectKey='', + MatchThreshold='', + Offset='', + Limit='', + Filter='' + ) + 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["ci-process"] = "ImageSearch" + params["action"] = "SearchImage" + if MatchThreshold is not None: + params["MatchThreshold"] = MatchThreshold + if Offset is not None: + params["Offset"] = Offset + if Limit is not None: + params["Limit"] = Limit + if Filter is not None: + params["Filter"] = Filter + + params = format_values(params) + + path = "/" + ObjectKey + url = self._conf.uri(bucket=Bucket, path=path) + + logger.info( + "cos_get_search_image 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=False) + + 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 cos_delete_image_search(self, Bucket, ObjectKey, Body, **kwargs): + """ 删除图库图片 https://cloud.tencent.com/document/product/460/63902 + + :param Bucket(string) 存储桶名称. + :param ObjectKey(string) 设置 ObjectKey. + :param Body:(dict) 删除图库图片配置信息. + :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) + # 删除图库图片 + response, data = client.cos_delete_image_search( + 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["ci-process"] = "ImageSearch" + params["action"] = "DeleteImage" + params = format_values(params) + body = format_xml(data=Body, root='Request') + path = "/" + ObjectKey + url = self._conf.uri(bucket=Bucket, path=path) + + logger.info( + "cos_delete_image_search result, url=:{url} ,headers=:{headers}, params=:{params},body=:{body}".format( + url=url, + headers=headers, + params=params, + body=body)) + rt = self.send_request( + method='POST', + url=url, + data=body, + auth=CosS3Auth(self._conf, path, params=params), + params=params, + headers=headers, + ci_request=False) + + 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 diff --git a/qcloud_cos/cos_client.py b/qcloud_cos/cos_client.py index f5de27d3..5d058324 100644 --- a/qcloud_cos/cos_client.py +++ b/qcloud_cos/cos_client.py @@ -252,7 +252,7 @@ def __init__(self, conf, retry=1, session=None): if not CosS3Client.__built_in_sessions: # 加锁后double check CosS3Client.__built_in_sessions = self.generate_built_in_connection_pool(self._conf._pool_connections, self._conf._pool_maxsize) CosS3Client.__built_in_pid = os.getpid() - + self._session = CosS3Client.__built_in_sessions self._use_built_in_pool = True logger.info("bound built-in connection pool when new client. maxsize=%d,%d" % (self._conf._pool_connections, self._conf._pool_maxsize)) @@ -267,30 +267,30 @@ def generate_built_in_connection_pool(self, PoolConnections, PoolMaxSize): built_in_sessions.mount('https://', requests.adapters.HTTPAdapter(pool_connections=PoolConnections, pool_maxsize=PoolMaxSize)) logger.info("generate built-in connection pool success. maxsize=%d,%d" % (PoolConnections, PoolMaxSize)) return built_in_sessions - + def handle_built_in_connection_pool_by_pid(self): if not CosS3Client.__built_in_sessions: return - + if not self._use_built_in_pool: - return - + return + if CosS3Client.__built_in_pid == os.getpid(): return - + with threading.Lock(): if CosS3Client.__built_in_pid == os.getpid(): # 加锁后double check return - + # 重新生成内置连接池 CosS3Client.__built_in_sessions.close() CosS3Client.__built_in_sessions = self.generate_built_in_connection_pool(self._conf._pool_connections, self._conf._pool_maxsize) CosS3Client.__built_in_pid = os.getpid() - + # 重新绑定到内置连接池 self._session = CosS3Client.__built_in_sessions logger.info("bound built-in connection pool when new processor. maxsize=%d,%d" % (self._conf._pool_connections, self._conf._pool_maxsize)) - + def get_conf(self): """获取配置""" return self._conf @@ -384,10 +384,10 @@ def send_request(self, method, url, bucket=None, timeout=30, cos_request=True, c if self._conf._allow_redirects is not None: kwargs['allow_redirects'] = self._conf._allow_redirects exception_logbuf = list() # 记录每次重试的错误日志 - + # 切换了进程需要重新生成连接池 self.handle_built_in_connection_pool_by_pid() - + for j in range(self._retry + 1): try: if j != 0: @@ -2684,7 +2684,7 @@ def put_bucket_domain_certificate(self, Bucket, DomainCertificateConfiguration, auth=CosS3Auth(self._conf), headers=headers) return None - + def get_bucket_domain_certificate(self, Bucket, DomainName, **kwargs): """获取bucket的自定义域名证书配置规则 @@ -2721,7 +2721,7 @@ def get_bucket_domain_certificate(self, Bucket, DomainName, **kwargs): params=params) data = xml_to_dict(rt.content) return data - + def delete_bucket_domain_certificate(self, Bucket, DomainName, **kwargs): """删除bucket的自定义域名证书配置规则 @@ -3575,7 +3575,7 @@ def list_buckets(self, TagKey=None, TagValue=None, Region=None, CreateTime=None, if CreateTime and Range: params['create-time'] = CreateTime params['range'] = Range - + rt = self.send_request( method='GET', url=url, @@ -5085,7 +5085,7 @@ def ci_get_image_ave_info(self, Bucket, Key, **kwargs): """ return self.ci_get_image_info(Bucket, Key, 'imageAve', **kwargs) - def ci_process(self, Bucket, Key, CiProcess, **kwargs): + def ci_process(self, Bucket, Key, CiProcess, Params={}, NeedHeader=False, Stream=False, **kwargs): """ci_process基本信息的接口 :param Bucket(string): 存储桶名称. @@ -5114,11 +5114,15 @@ def ci_process(self, Bucket, Key, CiProcess, **kwargs): else: final_headers[key] = headers[key] headers = final_headers + if Params is not None: + for key in Params: + params[key] = Params[key] params = format_values(params) url = self._conf.uri(bucket=Bucket, path=Key) - logger.info("ci_process, url=:{url} ,headers=:{headers}, params=:{params}".format( + logger.info("ci_process=:{ci_process}, url=:{url} ,headers=:{headers}, params=:{params}".format( + ci_process=CiProcess, url=url, headers=headers, params=params)) @@ -5127,11 +5131,26 @@ def ci_process(self, Bucket, Key, CiProcess, **kwargs): url=url, bucket=Bucket, auth=CosS3Auth(self._conf, Key, params=params), + stream=Stream, params=params, headers=headers) - data = xml_to_dict(rt.content) - format_dict(data, ['Response']) + data = '' + 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() + else: + if Stream: + data = StreamBody(rt) + else: + data = rt.content + + if NeedHeader: + return response, data return data def ci_image_assess_quality(self, Bucket, Key, **kwargs): @@ -5174,11 +5193,19 @@ def ci_image_detect_car(self, Bucket, Key, **kwargs): """ return self.ci_process(Bucket, Key, "DetectCar", **kwargs) - def ci_image_detect_label(self, Bucket, Key, **kwargs): - """ci_image_detect_label图片标签接口 + def ci_image_detect_label(self, Bucket, Key='', Scenes=None, DetectUrl=None, **kwargs): + """ci_image_detect_label图片标签接口 https://cloud.tencent.com/document/product/460/39082 :param Bucket(string): 存储桶名称. :param Key(string): COS路径. + :param Scenes(string): 本次调用支持的识别场景，可选值如下： + web，针对网络图片优化； + camera，针对手机摄像头拍摄图片优化； + album，针对手机相册、网盘产品优化； + news，针对新闻、资讯、广电等行业优化； + 如果不传此参数，则默认为camera。 + 支持多场景（scenes）一起检测，以，分隔。例如，使用 scenes=web，camera 即对一张图片使用两个模型同时检测，输出两套识别结果. + :param DetectUrl(string): 您可以通过填写 detect-url 处理任意公网可访问的图片链接。不填写 detect-url 时，后台会默认处理 ObjectKey ，填写了 detect-url 时，后台会处理 detect-url 链接. :param kwargs(dict): 设置下载的headers. :return(dict): 图片质量评估返回的结果,dict类型. @@ -5192,7 +5219,13 @@ def ci_image_detect_label(self, Bucket, Key, **kwargs): ) print response """ - return self.ci_process(Bucket, Key, "detect-label", **kwargs) + params = {} + if Scenes is not None: + params['scenes'] = Scenes + if DetectUrl is not None: + params['detect-url'] = DetectUrl + path = "/" + Key + return self.ci_process(Bucket=Bucket, Key=path, CiProcess="detect-label", Params=params, **kwargs) def ci_qrcode_generate(self, Bucket, QrcodeContent, Width, Mode=0, **kwargs): """二维码生成接口 @@ -8567,6 +8600,64 @@ def ci_get_doc_bucket(self, Regions='', BucketName='', BucketNames='', PageNumbe format_dict(data, ['DocBucketList']) return data + def ci_open_asr_bucket(self, Bucket, **kwargs): + """ 开通智能语音服务 https://cloud.tencent.com/document/product/460/95754 + + :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) + # 开通智能语音服务 + response, data = client.ci_open_asr_bucket( + Bucket='bucket' + ) + 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 = "/" + "asrbucket" + url = self._conf.uri(bucket=Bucket, path=path, endpoint=self._conf._endpoint_ci) + + logger.info("ci_open_asr_bucket result, url=:{url} ,headers=:{headers}, params=:{params}".format( + url=url, + headers=headers, + params=params)) + rt = self.send_request( + method='POST', + 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_get_asr_bucket(self, Regions='', BucketName='', BucketNames='', PageNumber='', PageSize='', **kwargs): """查询语音识别开通状态接口 https://cloud.tencent.com/document/product/460/46232 @@ -8632,6 +8723,64 @@ def ci_get_asr_bucket(self, Regions='', BucketName='', BucketNames='', PageNumbe format_dict(data, ['AsrBucketList']) return data + def ci_close_asr_bucket(self, Bucket, **kwargs): + """ 关闭智能语音服务 https://cloud.tencent.com/document/product/460/95755 + + :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) + # 关闭智能语音服务 + response, data = client.ci_close_asr_bucket( + Bucket='bucket' + ) + 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 = "/" + "asrbucket" + url = self._conf.uri(bucket=Bucket, path=path, endpoint=self._conf._endpoint_ci) + + logger.info("ci_close_asr_bucket result, url=:{url} ,headers=:{headers}, params=:{params}".format( + url=url, + headers=headers, + params=params)) + rt = self.send_request( + method='DELETE', + 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_get_asr_queue(self, Bucket, State='All', QueueIds='', PageNumber=1, PageSize=10, **kwargs): """查询语音识别队列接口 https://cloud.tencent.com/document/product/460/46234 @@ -10246,8 +10395,65 @@ def ci_delete_template(self, Bucket, TemplateId, **kwargs): data = xml_to_dict(rt.content) return data + def ci_open_ai_bucket(self, Bucket, **kwargs): + """ 开通AI内容识别服务 https://cloud.tencent.com/document/product/460/79593 + + :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) + # 开通AI内容识别服务 + response, data = client.ci_open_ai_bucket( + Bucket='bucket' + ) + 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 = "/" + "ai_bucket" + url = self._conf.uri(bucket=Bucket, path=path, endpoint=self._conf._endpoint_ci) + + logger.info("ci_open_ai_bucket result, url=:{url} ,headers=:{headers}, params=:{params}".format( + url=url, + headers=headers, + params=params)) + rt = self.send_request( + method='POST', + 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': + 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_get_ai_bucket(self, Regions='', BucketName='', BucketNames='', PageNumber='', PageSize='', **kwargs): - """查询ai处理开通状态接口 + """查询ai处理开通状态接口 https://cloud.tencent.com/document/product/460/79594 :param Regions(string): 地域信息，例如 ap-shanghai、ap-beijing，若查询多个地域以“,”分隔字符串 :param BucketName(string): 存储桶名称前缀，前缀搜索 @@ -10275,6 +10481,64 @@ def ci_get_ai_bucket(self, Regions='', BucketName='', BucketNames='', PageNumber BucketNames=BucketNames, PageNumber=PageNumber, PageSize=PageSize, Path='/ai_bucket', **kwargs) + def ci_close_ai_bucket(self, Bucket, **kwargs): + """ 关闭AI内容识别服务 https://cloud.tencent.com/document/product/460/95752 + + :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) + # 关闭AI内容识别服务 + response, data = client.ci_close_ai_bucket( + Bucket='bucket' + ) + 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 = "/" + "ai_bucket" + url = self._conf.uri(bucket=Bucket, path=path, endpoint=self._conf._endpoint_ci) + + logger.info("ci_close_ai_bucket result, url=:{url} ,headers=:{headers}, params=:{params}".format( + url=url, + headers=headers, + params=params)) + rt = self.send_request( + method='DELETE', + 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_ai_queue(self, Bucket, QueueId, Request={}, **kwargs): """ 更新图片处理队列接口 @@ -10300,7 +10564,7 @@ def ci_update_ai_queue(self, Bucket, QueueId, Request={}, **kwargs): Request=Request, UrlPath="/ai_queue/", **kwargs) def ci_get_ai_queue(self, Bucket, State='All', QueueIds='', PageNumber='', PageSize='', **kwargs): - """查询图片处理队列接口 + """查询ai队列接口 https://cloud.tencent.com/document/product/460/79394 :param Bucket(string): 存储桶名称. :param QueueIds(string): 队列 ID，以“,”符号分割字符串. diff --git a/qcloud_cos/cos_comm.py b/qcloud_cos/cos_comm.py index 2d83ff35..cf702497 100644 --- a/qcloud_cos/cos_comm.py +++ b/qcloud_cos/cos_comm.py @@ -63,6 +63,7 @@ 'Referer': 'Referer', 'PicOperations': 'Pic-Operations', 'TrafficLimit': 'x-cos-traffic-limit', + 'Accept': 'Accept' } diff --git a/qcloud_cos/xml2dict.py b/qcloud_cos/xml2dict.py index a9c4ab1d..0d0fba0f 100644 --- a/qcloud_cos/xml2dict.py +++ b/qcloud_cos/xml2dict.py @@ -7,6 +7,8 @@ class Xml2Dict(dict): def __init__(self, parent_node): if parent_node.items(): self.updateDict(dict(parent_node.items())) + if len(parent_node) == 0: + self.updateDict({parent_node.tag: parent_node.text}) for element in parent_node: if len(element): aDict = Xml2Dict(element) diff --git a/ut/test.py b/ut/test.py index 328fdbe3..39cf7796 100644 --- a/ut/test.py +++ b/ut/test.py @@ -11,7 +11,7 @@ import base64 import multiprocessing -from qcloud_cos import CosS3Client, MetaInsightClient +from qcloud_cos import CosS3Client, MetaInsightClient, AIRecognitionClient from qcloud_cos import CosConfig from qcloud_cos import CosServiceError, CosClientError from qcloud_cos.select_event_stream import EventStream @@ -75,6 +75,7 @@ def token(self): client = CosS3Client(conf, retry=3) meta_insight_client = MetaInsightClient(metaConf, retry=3) +ai_recognition_client = AIRecognitionClient(conf, retry=3) rsa_provider = RSAProvider() client_for_rsa = CosEncryptionClient(conf, rsa_provider) aes_provider = AESProvider() @@ -3866,7 +3867,18 @@ def test_ci_image_detect_label(): Bucket=ci_bucket_name, Key=ci_test_car_image, ) - assert response + assert len(response['Labels']) != 0 + response = client.ci_image_detect_label( + Bucket=ci_bucket_name, + Key=ci_test_car_image, + Scenes='camera', + ) + assert len(response['CameraLabels']['Labels']) != 0 + response = client.ci_image_detect_label( + Bucket=ci_bucket_name, + DetectUrl='https://' + ci_bucket_name + '.cos.' + ci_region + '.myqcloud.com/' + ci_test_car_image, + ) + assert len(response['Labels']) != 0 def test_ci_image_detect_car(): @@ -5325,6 +5337,474 @@ def test_meta_insight(): assert data["Dataset"]["DatasetName"] == mi_base_info_search_dataset_name +def test_cos_create_ai_object_detect_job(): + kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"} + # 图像主体检测 + response, data = ai_recognition_client.cos_create_ai_object_detect_job( + Bucket=ci_bucket_name, + ObjectKey="AIObjectDetect.jpeg", + **kwargs + ) + assert len(data['DetectMultiObj']) == 16 + response, data = ai_recognition_client.cos_create_ai_object_detect_job( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/AIObjectDetect.jpeg", + Accept='application/json' + ) + assert len(data['DetectMultiObj']) == 16 + + +def test_goods_matting(): + # 商品抠图下载时处理 + response, data = ai_recognition_client.cos_goods_matting( + Bucket=ci_bucket_name, + ObjectKey="GoodsMatting.jpg", + CenterLayout=1, + PaddingLayout='300x300', + Stream=True + ) + data.get_stream_to_file('test.jpg') + assert response['Content-Length'] == '830576' + + response, data = ai_recognition_client.cos_goods_matting( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/GoodsMatting.jpg", + CenterLayout=1, + PaddingLayout='300x300', + Stream=True + ) + data.get_stream_to_file('test.jpg') + assert response['Content-Length'] == '830576' + + +def test_cos_ai_body_recognition(): + # 人体识别 + response, data = ai_recognition_client.cos_ai_body_recognition( + Bucket=ci_bucket_name, + ObjectKey="renti.jpg", + ) + assert data['Status'] == '1' + assert len(data['PedestrianInfo']) == 3 + + response, data = ai_recognition_client.cos_ai_body_recognition( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/renti.jpg", + ) + assert data['Status'] == '1' + assert len(data['PedestrianInfo']) == 3 + + +def test_cos_ai_detect_face(): + # 人脸检测 + response, data = ai_recognition_client.cos_ai_detect_face( + Bucket=ci_bucket_name, + ObjectKey="relian.jpeg", + MaxFaceNum=3 + ) + assert data['Status'] == '1' + + +def test_cos_ai_detect_pet(): + # 宠物识别 + response, data = ai_recognition_client.cos_ai_detect_pet( + Bucket=ci_bucket_name, + ObjectKey="chongwu.jpg", + ) + assert data is not None + assert data['ResultInfo']['Name'] == 'cat' + + +def test_cos_ai_enhance_image(): + # 图像增强 + response, data = ai_recognition_client.cos_ai_enhance_image( + Bucket=ci_bucket_name, + ObjectKey="format.jpg", + Denoise=3, + Sharpen=3, + # DetectUrl="https://test-125000000.cos.ap-chongqing.myqcloud.com/test.jpeg", + IgnoreError=0, + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + response, data = ai_recognition_client.cos_ai_enhance_image( + Bucket=ci_bucket_name, + Denoise=3, + Sharpen=3, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/format.jpg", + IgnoreError=0, + Stream=False + ) + assert data is not None + + +def test_cos_ai_face_effect(): + # 人脸特效 + response, data = ai_recognition_client.cos_ai_face_effect( + Bucket=ci_bucket_name, + ObjectKey="relian.jpeg", + Type="face-beautify", + Whitening=30, + Smoothing=10, + FaceLifting=70, + EyeEnlarging=70, + ) + assert data is not None + + response, data = ai_recognition_client.cos_ai_face_effect( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/relian.jpeg", + Type="face-beautify", + Whitening=30, + Smoothing=10, + FaceLifting=70, + EyeEnlarging=70, + ) + assert data is not None + + +def test_cos_ai_game_rec(): + # 游戏场景识别 + response, data = ai_recognition_client.cos_ai_game_rec( + Bucket=ci_bucket_name, + ObjectKey="youxi.png", + ) + assert data['GameLabels'] is not None + + response, data = ai_recognition_client.cos_ai_game_rec( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/youxi.png", + Accept='application/json' + ) + assert data['GameLabels'] is not None + + +def test_cos_ai_id_card_ocr(): + # 身份证识别 + response, data = ai_recognition_client.cos_ai_id_card_ocr( + Bucket=ci_bucket_name, + ObjectKey="shenfenzheng.jpeg", + CardSide="FRONT", + Config='{"CropIdCard":true,"CropPortrait":true}' + ) + assert data['AdvancedInfo'] is not None + assert data['IdInfo'] is not None + + +def test_cos_ai_image_coloring(): + # 图片上色 + response, data = ai_recognition_client.cos_ai_image_coloring( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + response, data = ai_recognition_client.cos_ai_image_coloring( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/heibai.jpeg", + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + +def test_cos_ai_image_crop(): + # 图像智能裁剪 + response, data = ai_recognition_client.cos_ai_image_crop( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + Width=100, + Height=100, + Fixed=1, + IgnoreError=0 + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + response, data = ai_recognition_client.cos_ai_image_crop( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/heibai.jpeg", + Width=100, + Height=100, + Fixed=1, + IgnoreError=0 + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + +def test_cos_ai_license_rec(): + # 卡证识别 + response, data = ai_recognition_client.cos_ai_license_rec( + Bucket=ci_bucket_name, + ObjectKey="shenfenzheng.jpeg", + CardType="IDCard" + ) + assert data['Status'] == '1' + assert data['IdInfo'] is not None + + response, data = ai_recognition_client.cos_ai_license_rec( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/shenfenzheng.jpeg", + CardType="IDCard" + ) + assert data['Status'] == '1' + assert data['IdInfo'] is not None + + +def test_cos_ai_pic_matting(): + # 通用抠图 + response, data = ai_recognition_client.cos_ai_pic_matting( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + CenterLayout=1, + PaddingLayout="10x10", + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + response, data = ai_recognition_client.cos_ai_pic_matting( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/heibai.jpeg", + CenterLayout=1, + PaddingLayout="10x10", + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + +def test_cos_ai_portrait_matting(): + # 人像抠图 + response, data = ai_recognition_client.cos_ai_portrait_matting( + Bucket=ci_bucket_name, + ObjectKey="relian.jpeg", + CenterLayout=1, + PaddingLayout="10x10", + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + response, data = ai_recognition_client.cos_ai_portrait_matting( + Bucket=ci_bucket_name, + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/relian.jpeg", + CenterLayout=1, + PaddingLayout="10x10", + Stream=True + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + +def test_cos_auto_translation_block(): + # 实时文字翻译 + response, data = ai_recognition_client.cos_auto_translation_block( + Bucket=ci_bucket_name, + InputText="测试", + SourceLang="zh", + TargetLang="en", + TextDomain="general", + TextStyle="sentence" + ) + assert data['TranslationResult'] == 'test' + + +def test_cos_get_action_sequence(): + # 获取动作顺序 + response, data = ai_recognition_client.cos_get_action_sequence( + Bucket=ci_bucket_name, + ) + assert data['ActionSequence'] == '2,1' or data['ActionSequence'] == '1,2' + + +def test_cos_get_live_code(): + # 获取数字验证码 + response, data = ai_recognition_client.cos_get_live_code( + Bucket=ci_bucket_name, + ) + assert len(data['LiveCode']) != 0 + + +def test_cos_image_repair(): + # 图像修复下载时处理 + mask_pic = "https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/mask.jpg" + mask_poly = '[[[100, 200], [1000, 200], [1000, 400], [100, 400]]]' + if len(mask_pic) != 0: + mask_pic = base64.b64encode(mask_pic.encode('utf-8')).decode('utf-8') + if len(mask_poly) != 0: + mask_poly = base64.b64encode(mask_poly.encode('utf-8')).decode('utf-8') + response, data = ai_recognition_client.cos_image_repair( + Bucket=ci_bucket_name, + ObjectKey="xiufu.jpg", + MaskPic=mask_pic, + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + response, data = ai_recognition_client.cos_image_repair( + Bucket=ci_bucket_name, + ObjectKey="xiufu.jpg", + DetectUrl="https://" + ci_bucket_name + ".cos." + ci_region + ".myqcloud.com/xiufu.jpg", + MaskPoly=mask_poly + ) + data.get_stream_to_file('result.jpg') + assert data is not None + + +def test_cos_liveness_recognition(): + # 活体人脸核身 + try: + response, data = ai_recognition_client.cos_liveness_recognition( + Bucket=ci_bucket_name, + ObjectKey="silent.mp4", + IdCard="123456", + Name="测试", + LivenessType="SILENT", + ValidateData="", + BestFrameNum=5 + ) + except Exception as e: + print(e) + + +def test_ci_image_search_bucket(): + # 开通以图搜图 + body = { + # 图库容量限制 + # 是否必传：是 + 'MaxCapacity': 10, + # 图库访问限制，默认10 + # 是否必传：否 + 'MaxQps': 10, + } + try: + kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"} + response, data = ai_recognition_client.ci_image_search_bucket( + Bucket=ci_bucket_name, + Body=body, + ContentType="application/xml", + **kwargs + ) + except Exception as e: + print(e) + + +def test_cos_add_image_search(): + # 添加图库图片 + kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"} + body = { + # 物品 ID，最多支持64个字符。若 EntityId 已存在，则对其追加图片 + # 是否必传：是 + 'EntityId': "test", + # 用户自定义的内容，最多支持4096个字符，查询时原样带回 + # 是否必传：否 + 'CustomContent': "custom test", + # 图片自定义标签，最多不超过10个，json 字符串，格式为 key:value （例 key1>=1 key1>='aa' ）对 + # 是否必传：否 + 'Tags': '{"key1":"val1","key2":"val2"}', + } + response, data = ai_recognition_client.cos_add_image_search( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + Body=body, + ContentType="application/xml", + **kwargs + ) + assert data is not None + + +def test_cos_get_search_image(): + # 图片搜索接口 + kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"} + response, data = ai_recognition_client.cos_get_search_image( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + MatchThreshold=1, + Offset=0, + Limit=10, + **kwargs + # Filter="key1=val1" + ) + assert data['Count'] == '1' + assert data['ImageInfos']['PicName'] == 'heibai.jpeg' + + response, data = ai_recognition_client.cos_get_search_image( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + MatchThreshold=1, + Offset=0, + Limit=10, + Accept='application/json' + # Filter="key1=val1" + ) + assert data['Count'] == 1 + assert data['ImageInfos'][0]['PicName'] == 'heibai.jpeg' + + +def test_cos_delete_image_search(): + # 删除图库图片 + kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"} + body = { + # 物品 ID + # 是否必传：是 + 'EntityId': "test", + } + response, data = ai_recognition_client.cos_delete_image_search( + Bucket=ci_bucket_name, + ObjectKey="heibai.jpeg", + Body=body, + ContentType="application/xml", + **kwargs + ) + assert data is not None + + +def test_ci_ai_bucket(): + # 关闭AI内容识别服务 + response, data = ai_recognition_client.ci_close_ai_bucket( + Bucket=ci_bucket_name + ) + assert data['BucketName'] == ci_bucket_name + # 开通AI内容识别服务 + response, data = ai_recognition_client.ci_open_ai_bucket( + Bucket=ci_bucket_name + ) + assert data['AiBucket']['Name'] == ci_bucket_name + + response, data = ai_recognition_client.ci_close_ai_bucket( + Bucket=ci_bucket_name, + Accept="application/json" + ) + assert data['BucketName'] == ci_bucket_name + # 开通AI内容识别服务 + response, data = ai_recognition_client.ci_open_ai_bucket( + Bucket=ci_bucket_name, + Accept="application/json" + ) + assert data['AiBucket']['Name'] == ci_bucket_name + + +def test_ci_asr_bucket(): + # 关闭智能语音服务 + kwargs = {"CacheControl": "no-cache", "ResponseCacheControl": "no-cache"} + response, data = client.ci_close_asr_bucket( + Bucket=ci_bucket_name, + **kwargs + ) + assert data['BucketName'] == ci_bucket_name + + response, data = client.ci_open_asr_bucket( + Bucket=ci_bucket_name, + **kwargs + ) + assert data['AsrBucket']['Name'] == ci_bucket_name + + if __name__ == "__main__": setUp() test_post_bucket_inventory_configurations()