XCDesktop/tools/tongyi/tingwu_api.py

#!/usr/bin/env python
#coding=utf-8

import os
import json
import datetime
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkcore.auth.credentials import AccessKeyCredential

def create_common_request(domain, version, protocolType, method, uri):
    """
    创建通用请求对象
    """
    request = CommonRequest()
    request.set_accept_format('json')
    request.set_domain(domain)
    request.set_version(version)
    request.set_protocol_type(protocolType)
    request.set_method(method)
    request.set_uri_pattern(uri)
    request.add_header('Content-Type', 'application/json')
    return request

def init_parameters(app_key, file_url):
    """
    初始化请求参数
    """
    body = dict()
    body['AppKey'] = app_key

    # 基本请求参数
    input = dict()
    input['SourceLanguage'] = 'cn'
    input['TaskKey'] = 'task' + datetime.datetime.now().strftime('%Y%m%d%H%M%S')
    input['FileUrl'] = file_url
    body['Input'] = input

    # AI相关参数，按需设置即可
    parameters = dict()

    # 音视频转换相关
    transcoding = dict()
    # 将原音视频文件转成mp3文件，用以后续浏览器播放
    transcoding['TargetAudioFormat'] = 'mp3'
    transcoding['SpectrumEnabled'] = False
    parameters['Transcoding'] = transcoding

    # 语音识别控制相关
    transcription = dict()
    # 角色分离 ： 可选
    transcription['DiarizationEnabled'] = True
    diarization = dict()
    diarization['SpeakerCount'] = 2
    transcription['Diarization'] = diarization
    parameters['Transcription'] = transcription

    # 文本翻译控制相关 ： 可选
    parameters['TranslationEnabled'] = True
    translation = dict()
    translation['TargetLanguages'] = ['en']  # 假设翻译成英文
    parameters['Translation'] = translation

    # 章节速览相关 ： 可选，包括： 标题、议程摘要
    parameters['AutoChaptersEnabled'] = True

    # 智能纪要相关 ： 可选，包括： 待办、关键信息(关键词、重点内容、场景识别)
    parameters['MeetingAssistanceEnabled'] = True
    meetingAssistance = dict()
    meetingAssistance['Types'] = ['Actions', 'KeyInformation']
    parameters['MeetingAssistance'] = meetingAssistance

    # 摘要控制相关 ： 可选，包括： 全文摘要、发言人总结摘要、问答摘要(问答回顾)
    parameters['SummarizationEnabled'] = True
    summarization = dict()
    summarization['Types'] = ['Paragraph', 'Conversational', 'QuestionsAnswering', 'MindMap']
    parameters['Summarization'] = summarization

    # ppt抽取和ppt总结 ： 可选
    parameters['PptExtractionEnabled'] = True
    
    # 口语书面化 ： 可选
    parameters['TextPolishEnabled'] = True
    
    # 大模型后处理任务全局参数 ： 可选
    parameters['Model'] = 'qwq'
    parameters['LlmOutputLanguage'] = 'en'

    body['Parameters'] = parameters
    return body

def submit_tingwu_task(file_url, app_key, access_key_id, access_key_secret):
    """
    提交通义听悟任务
    """
    print(f"开始提交通义听悟任务，文件URL: {file_url}")
    
    try:
        # 初始化参数
        body = init_parameters(app_key, file_url)
        print("初始化参数完成")
        
        # 初始化客户端
        credentials = AccessKeyCredential(access_key_id, access_key_secret)
        client = AcsClient(region_id='cn-beijing', credential=credentials)
        print("初始化客户端完成")
        
        # 创建请求
        request = create_common_request('tingwu.cn-beijing.aliyuncs.com', '2023-09-30', 'https', 'PUT', '/openapi/tingwu/v2/tasks')
        request.add_query_param('type', 'offline')
        
        # 设置请求内容
        request.set_content(json.dumps(body).encode('utf-8'))
        
        # 发送请求
        response = client.do_action_with_exception(request)
        response_data = json.loads(response)
        
        print("任务提交成功")
        print(f"响应数据: {json.dumps(response_data, indent=4, ensure_ascii=False)}")
        
        # 返回任务ID
        task_id = response_data.get('Data', {}).get('TaskId')
        if task_id:
            print(f"获取到任务ID: {task_id}")
            return task_id
        else:
            print("无法获取任务ID")
            return None
            
    except Exception as e:
        print(f"任务提交失败: {e}")
        import traceback
        traceback.print_exc()
        return None

def get_task_result(task_id, access_key_id, access_key_secret):
    """
    获取任务结果
    """
    print(f"开始查询任务结果，任务ID: {task_id}")
    
    try:
        # 初始化客户端
        credentials = AccessKeyCredential(access_key_id, access_key_secret)
        client = AcsClient(region_id='cn-beijing', credential=credentials)
        
        # 创建请求
        uri = '/openapi/tingwu/v2/tasks' + '/' + task_id
        request = create_common_request('tingwu.cn-beijing.aliyuncs.com', '2023-09-30', 'https', 'GET', uri)
        
        # 发送请求
        response = client.do_action_with_exception(request)
        response_data = json.loads(response)
        
        print("任务查询成功")
        print(f"响应数据: {json.dumps(response_data, indent=4, ensure_ascii=False)}")
        
        return response_data
        
    except Exception as e:
        print(f"任务查询失败: {e}")
        import traceback
        traceback.print_exc()
        return None

def check_task_status(task_id, access_key_id, access_key_secret):
    """
    检查任务状态
    """
    result = get_task_result(task_id, access_key_id, access_key_secret)
    if result:
        task_status = result.get('Data', {}).get('TaskStatus')
        print(f"当前任务状态: {task_status}")
        return task_status
    return None