全部产品

Python SDK

准备工作

  1. 在安装和使用阿里云SDK前,确保您已经注册阿里云账号并生成访问密钥(AccessKey)。详情请参考创建AccessKey

  2. 安装Python SDK核心库。使用依赖包工具安装(推荐)。执行以下命令,安装阿里云SDK核心库:

    pip install alibabacloud_tea_rpc

    自行下载安装。您可以使用git clone或其它手段下载aliyun-openapi-python-sdk并自行添加解决方案。

  3. 安装视频内容检索Python SDK。使用依赖包工具安装(推荐)。执行以下命令,安装视频指纹 Python SDK:最新版本1.2.0

    pip install alibabacloud_videosearch20200225

    自行下载安装。您可以使用git clone或其它手段下载aliyun-python-sdk-videosearch并自行添加解决方案。

完整代码示例

v1.2.0 版本

1.账号,Config,Client 均为标准格式,下面不做重复说明

2.参数及返回值详细说明请参考HTTP调用接口文档

3.详细出入参可查看HTTP调用接口文档

公共参数

from alibabacloud_tea_rpc.models import Config
from alibabacloud_videosearch20200225.client import Client
from alibabacloud_videosearch20200225.models import *
from alibabacloud_tea_util.models  import RuntimeOptions

config = Config(
        access_key_id="<yourAccessKeyId>",
        access_key_secret="<yourAccessKeySecret>",
        region_id="<yourRegionId>",
        endpoint="<yourEndpoint>"
    )

client = Client(config)

一丶音频指纹

查询实例信息

def getAudioInstance(client,instanceid):
  
    request = GetAudioInstanceRequest()
    request.instance_id = instanceid #实例Id

    response = client.get_audio_instance(request,runtime)
    return response

添加音频入库(url方式)

def addStorageAudioTask(client,audioUrl,audioId,decsription,audioTag=None,callback=None,sort=None):
    request = AddStorageAudioTaskRequest()
    request.instance_id = INSTANCE_ID # 实例Id
    request.audio_url = audioUrl #音频地址
    request.audio_id = audioId # 音频Id
    request.description = decsription #备注
    request.audio_tags = audioTag # 标签 
    request.callback_url = callback # 回调地址
    request.sort =sort    # (0, "低优先级"),(1, "中优先级"),(2, "高优先级"),(3, "紧急");

    response = client.add_storage_audio_task(request,runtime)

    return response

添加音频入库(文件方式)

def addStorageAudioTaskAdvance(client,audioUrl,audioId,decsription,audioTag=None,callback=None,sort=None,fileobject=None,content=None):
    request = AddStorageAudioTaskAdvanceRequest()
    request.instance_id = INSTANCE_ID
    request.audio_file_object = fileobject 
    request.content_source = content  
    request.audio_url = audioUrl 
    request.audio_id = audioId
    request.description = decsription
    request.audio_tags = audioTag
    request.callback_url = callback
    request.sort=sort


    response = client.add_storage_audio_task_advance(request,runtime)

    return response

查询音频入库信息

def addSearchAudioTask(client,audioUrl,description,queryTag=None,returnNum=10,callback=None,sort=None,needfeature=None,resourcetype=None):
    request = AddSearchAudioTaskRequest()
    request.instance_id = INSTANCE_ID
    request.audio_url = audioUrl
    request.content_source = 1  # 1. OSS导入 2. 文件上传
    request.description = description
    request.query_tags = queryTag
    request.return_audio_number = returnNum
    request.callback_url = callback
    request.sort = sort
    request.need_feature_file = needfeature
    request.resource_type = resourcetype  

    response = client.add_search_audio_task(request,runtime)
    return response

查询音频入库信息(文件方式)

def addSearchAudioTaskAdvance(client,description,queryTag=None,returnNum=10,callback=None,file=None,sort=None,needfeature=None,resourcetype=None):
    request = AddSearchAudioTaskAdvanceRequest()
    request.instance_id = INSTANCE_ID
    #request.audio_url = audioUrl
    request.audio_file_object = file
    request.content_source = 2  # 1. OSS导入 2. 文件上传
    request.description = description
    request.query_tags = queryTag
    request.return_audio_number = returnNum
    request.callback_url = callback
    request.sort =sort
    request.need_feature_file = needfeature
    request.resource_type = resourcetype  

    response = client.add_search_audio_task_advance(request,runtime)
    return response

删除音频入库

def addDeletionAudioTask(client,audioId):
    request = AddDeletionAudioTaskRequest()
    request.instance_id =INSTANCE_ID
    request.audio_id = audioId

    response = client.add_deletion_audio_task(request,runtime)
    return response

查询入库历史

def getAudioStorageHistory(client,audioId,pagenum=None,pagesize=None):
    request = GetAudioStorageHistoryRequest()
    request.instance_id =INSTANCE_ID
    request.audio_id = audioId
    request.page_number = pagenum
    request.page_size =pagesize

    response = client.get_audio_storage_history(request,runtime)
    return  response

查询音频任务状态

def getAudioTaskStatus(client,taskId):
    
    request = GetAudioTaskStatusRequest()
    request.instance_id = INSTANCE_ID
    request.task_id = taskId

    response = client.get_audio_task_status(request, runtime)
    return response

查询检索任务列表

def listSearchAudioTasks(client,taskId=None,audioId=None,status=None,description=None,pagenum=None,pagesize=None,sortlist=None):
   
    request = ListSearchAudioTasksRequest()
    request.instance_id =INSTANCE_ID
    request.task_id = taskId     # 按照任务id查询
    request.status_list = status  # 按照状态查询
    request.audio_id = audioId    # 是否支持按照 audiid 查询?
    request.description = description  # 按照描述查询
    request.page_number = pagenum
    request.page_size =pagesize
    request.sort_list =sortlist

    response = client.list_search_audio_tasks(request,runtime)
    return  response

查询入库任务列表

def listStorageAudioTasks(client,taskId=None,audioId=None,status=None,description=None,storageInfo=None,pagenum=None,pagesize=None,sortlist=None):
    
    request = ListStorageAudioTasksRequest()
    request.instance_id = INSTANCE_ID
    request.task_id = taskId  # 按照任务id查询
    request.status_list = status  # 按照状态查询
    request.audio_id = audioId  # 按照audioId 查询
    request.description = description  # 按照描述查询
    request.page_number = pagenum
    request.page_size = pagesize
    request.storage_info_list = storageInfo  # 按照入库信息查询: 首次入库、覆盖入库、入库失败
    request.sort_list =sortlist

    response = client.list_storage_audio_tasks(request, runtime)
    return response

二丶视频指纹

查询实例信息

def getVideoInstance(client,instanceid):
    
    request = GetInstanceRequest()
    request.instance_id = instanceid

    response = client.get_instance(request,runtime)
    return response

添加视频入库(url方式)

def addStorageVideoTask(client,videoUrl,videoId,decsription,videoTag=None,callback=None,sort=None):
    request = AddStorageVideoTaskRequest()
    request.instance_id = INSTANCE_ID
    request.video_url = videoUrl
    request.video_id = videoId
    request.description = decsription
    request.video_tags = videoTag
    request.callback_url = callback
    request.sort =sort    # (0, "低优先级"),(1, "中优先级"),(2, "高优先级"),(3, "紧急");

    response = client.add_storage_video_task(request,runtime)

    return response

添加视频入库(文件方式)

def addStorageVideoTaskAdvance(client, videoUrl, videoId, decsription, videoTag=None, callback=None, sort=None,
                               fileobject=None, content=None):
    request = AddStorageVideoTaskAdvanceRequest()
    request.instance_id = INSTANCE_ID
    request.video_file_object = fileobject
    request.content_source = content  # 1是 url 方式,  2 是 文件上传方式
    request.video_url = videoUrl
    request.video_id = videoId
    request.description = decsription
    request.video_tags = videoTag
    request.callback_url = callback
    request.sort = sort

    response = client.add_storage_video_task_advance(request, runtime)

    return response

查询视频入库信息

def addSearchVideoTask(client, videoUrl, description, queryTag=None, returnNum=10, callback=None, sort=None,
                       needfeature=None, resourcetype=None):
    request = AddSearchVideoTaskRequest()
    request.instance_id = INSTANCE_ID
    request.video_url = videoUrl
    request.content_source = 1  # 1. OSS导入 2. 文件上传
    request.description = description
    request.query_tags = queryTag
    request.return_audio_number = returnNum
    request.callback_url = callback
    request.sort = sort
    request.need_feature_file = needfeature
    request.resource_type = resourcetype  

    response = client.add_search_video_task(request, runtime)
    return response

查询视频入库信息(文件方式)

    def addSearchVideoTaskAdvance(client, description, queryTag=None, returnNum=10, callback=None, file=None, sort=None,
                                  needfeature=None, resourcetype=None):
        request = AddSearchVideoTaskAdvanceRequest()
        request.instance_id = INSTANCE_ID
        # request.audio_url = audioUrl
        request.video_file_object = file
        request.content_source = 2  # 1. OSS导入 2. 文件上传
        request.description = description
        request.query_tags = queryTag
        request.return_audio_number = returnNum
        request.callback_url = callback
        request.sort = sort
        request.need_feature_file = needfeature
        request.resource_type = resourcetype  

        response = client.add_search_audio_task_advance(request, runtime)
        return response

删除视频入库

def addDeletionVideoTask(client,videoId):
    request = AddDeletionVideoTaskRequest()
    request.instance_id =INSTANCE_ID
    request.video_id = videoId

    response = client.add_deletion_video_task(request,runtime)
    return response

查询入库历史

def getVideoStorageHistory(client,videoId,pagenum=None,pagesize=None):
    # 查询入库历史

    request = GetStorageHistoryRequest()
    request.instance_id =INSTANCE_ID
    request.video_id = videoId
    request.page_number = pagenum
    request.page_size =pagesize

    response = client.get_storage_history(request,runtime)
    return  response

查询视频任务状态

def getVideoTaskStatus(client,taskId):
    # 查询 taskid状态
    request = GetTaskStatusRequest()
    request.instance_id = INSTANCE_ID
    request.task_id = taskId

    response = client.get_task_status(request, runtime)
    return response

查询检索任务列表

def listSearchVideoTasks(client,taskId=None,videoId=None,status=None,description=None,pagenum=None,pagesize=None,sortlist=None):
    # 查询 检索任务列表
    request = ListSearchVideoTasksRequest()
    request.instance_id =INSTANCE_ID
    request.task_id = taskId     # 按照任务id查询
    request.status_list = status  # 按照状态查询
    request.video_id = videoId    
    request.description = description  # 按照描述查询
    request.page_number = pagenum
    request.page_size =pagesize
    request.sort_list =sortlist

    response = client.list_search_video_tasks(request,runtime)
    return  response

查询入库任务列表

def listStorageVideoTasks(client,taskId=None,videoId=None,status=None,description=None,storageInfo=None,pagenum=None,pagesize=None,sortlist=None):
    # 查询 入库任务列表
    request = ListStorageVideoTasksRequest()
    request.instance_id = INSTANCE_ID
    request.task_id = taskId  # 按照任务id查询
    request.status_list = status  # 按照状态查询
    request.video_id = videoId  
    request.description = description  # 按照描述查询
    request.page_number = pagenum
    request.page_size = pagesize
    request.storage_info_list = storageInfo  # 按照入库信息查询: 首次入库、覆盖入库、入库失败
    request.sort_list =sortlist

    response = client.list_storage_video_tasks(request, runtime)
    return response

三丶通用示例

创建批量任务

def createBatchTask(client,metafile,callback=None):
    request = CreateBatchTaskRequest()
    request.instance_id =INSTANCE_ID ## 实例ID
    request.batch_task_type = 1   #批量任务类型, 必填(1. OSS导入 2. 文件上传)
    request.role_arn = ROLE_ARN
    request.oss_bucket_name = OSS_BUCKET_NAME
    request.oss_data_path = OSS_DATA_PATH
    request.oss_meta_file = metafile
    request.callback_url = callback

    respone = client.create_batch_task(request,runtime)
    return  respone

创建批量任务(文件方式)

def createBatchTaskAdvance(client,file,callback=None):
    request = CreateBatchTaskAdvanceRequest()
    request.instance_id =INSTANCE_ID
    request.batch_task_type = 2   #批量任务类型, 必填(1. OSS导入 2. 文件上传)
    request.file_url_object = file
    request.callback_url = callback

    respone = client.create_batch_task_advance(request,runtime)
    return  respone

查询批量任务

def getBatchTask(client,batchtaskId):
    request = GetBatchTaskRequest()
    request.instance_id =INSTANCE_ID
    request.batch_task_id = batchtaskId

    response  =client.get_batch_task(request,runtime)
    return  response

查询任务状态

def getTaskStatus(client,taskid):
    request = GetTaskStatusRequest()
    request.instance_id = INSTANCE_ID
    request.task_id = taskid
    response = client.get_task_status(request,runtime)
    return response

查询批量任务列表

def listBatchTask(client,bucketname=None,datapath=None,status=None,tasktype=None,pagenum=None,pagesize=None):
    
    request = ListBatchTaskRequest()
    request.instance_id = INSTANCE_ID
    request.bucket_name = bucketname  #按照bucket 查询
    request.data_path = datapath   # 按照数据路径 查询
    request.status =status
    # 0. 排队中 1. 处理中(meta文件解析中) 2. 已完成 3. 失败 4. 部分失败 5. 处理中(子任务处理中)
    request.batch_task_type = tasktype  # 1. OSS 2. 文件上传  '[2]'
    request.page_number =pagenum
    request.page_size =pagesize

    response = client.list_batch_task(request,runtime)
    return  response

查询实例列表

def listInstances(client,instanceName=None,status=None,instanceType=None,tags=None,pagenum=None,pagesize=None):
    
    request = ListInstancesRequest()
    request.instance_name = instanceName
    request.status = status
    request.instance_type = instanceType
    request.tags = tags
    request.page_number = pagenum
    request.page_size = pagesize

    response = client.list_instances(request,runtime)
    return response

更改优先级

def modifyPriority(client,taskId,sort):
    request = ModifyPriorityRequest()
    request.instance_id =INSTANCE_ID
    request.task_id = taskId
    request.sort = sort

    response = client.modify_priority(request,runtime)
    return response