快速入门

# 按照需要的音视频处理包
pip install moviepy
pip install pydub

import hmac
import json
import sys
import time
import os
import hashlib
from datetime import datetime
if sys.version_info[0] <= 2:
    from httplib import HTTPSConnection
else:
    from http.client import HTTPSConnection


def sign(key, msg):
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()


def get_TongChuan_result(SessionUuid,Source,Target,AudioFormat,Seq,Utc,IsEnd,TranslateTime,Data):
    secret_id = "SecretId"
    secret_key = "SecretKey"
    token = ""

    service = "tsi"
    host = "tsi.tencentcloudapi.com"
    region = ""
    version = "2021-03-25"
    action = "TongChuanSync"
    payload = {
                "SessionUuid": SessionUuid,
                "Source": Source,
                "Target": Target,
                "AudioFormat": AudioFormat,
                "Seq": Seq,
                "Utc": Utc,
                "IsEnd": IsEnd,
                "TranslateTime": TranslateTime,
                "Data": Data
                }
    payload = json.dumps(payload)
    endpoint = "https://tsi.tencentcloudapi.com"
    algorithm = "TC3-HMAC-SHA256"
    timestamp = int(time.time())
    date = datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")

    # ************* 步骤 1：拼接规范请求串 *************
    http_request_method = "POST"
    canonical_uri = "/"
    canonical_querystring = ""
    ct = "application/json; charset=utf-8"
    canonical_headers = "content-type:%s\\nhost:%s\\nx-tc-action:%s\\n" % (ct, host, action.lower())
    signed_headers = "content-type;host;x-tc-action"
    hashed_request_payload = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    canonical_request = (http_request_method + "\\n" +
                        canonical_uri + "\\n" +
                        canonical_querystring + "\\n" +
                        canonical_headers + "\\n" +
                        signed_headers + "\\n" +
                        hashed_request_payload)

    # ************* 步骤 2：拼接待签名字符串 *************
    credential_scope = date + "/" + service + "/" + "tc3_request"
    hashed_canonical_request = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
    string_to_sign = (algorithm + "\\n" +
                    str(timestamp) + "\\n" +
                    credential_scope + "\\n" +
                    hashed_canonical_request)

    # ************* 步骤 3：计算签名 *************
    secret_date = sign(("TC3" + secret_key).encode("utf-8"), date)
    secret_service = sign(secret_date, service)
    secret_signing = sign(secret_service, "tc3_request")
    signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    # ************* 步骤 4：拼接 Authorization *************
    authorization = (algorithm + " " +
                    "Credential=" + secret_id + "/" + credential_scope + ", " +
                    "SignedHeaders=" + signed_headers + ", " +
                    "Signature=" + signature)

    # ************* 步骤 5：构造并发起请求 *************
    headers = {
        "Authorization": authorization,
        "Content-Type": "application/json; charset=utf-8",
        "Host": host,
        "X-TC-Action": action,
        "X-TC-Timestamp": timestamp,
        "X-TC-Version": version
    }
    if region:
        headers["X-TC-Region"] = region
    if token:
        headers["X-TC-Token"] = token

    req = HTTPSConnection(host)
    req.request("POST", "/", headers=headers, body=payload.encode("utf-8"))
    resp = req.getresponse()
    resp = json.loads(resp.read())
    return resp

  
def make_chunks(audio_segment, chunk_length_ms):
    """将音频分割成指定长度的片段"""
    return [audio_segment[i:i + chunk_length_ms] for i in range(0, len(audio_segment), chunk_length_ms)]

  
def deal_audio(video_path, Source, Target):
    '''
        video_path: 视频文件路径
        Source: 待识别的语言
        Target: 翻译的语言
    '''
    from moviepy.editor import VideoFileClip
    from pydub import AudioSegment
    import io,base64
    # 将音视频转换为pcm并提交给腾讯云进行语音识别
    # 待识别和翻译的音频文件格式是 pcm，pcm采样率要求16kHz、位深16bit、单声道、每个分片时长200ms~500ms
    SessionUuid = os.path.basename(video_path) + "_" + str(int(time.time()))
    AudioFormat = 1
    TranslateTime = 1
    IsEnd = 0
    video = VideoFileClip(video_path)
    audio = video.audio
    # 将音频写入临时文件
    temp_audio_path = "temp_audio.wav"
    audio.write_audiofile(temp_audio_path, codec='pcm_s16le', ffmpeg_params=["-ac", "1", "-ar", "16000"])
    # 使用pydub加载处理后的音频
    sound = AudioSegment.from_file(temp_audio_path, format="wav", frame_rate=16000, channels=1, sample_width=2)
    # 分片处理
    chunk_length_ms = 500  # 可以调整为200~500ms之间
    chunks = make_chunks(sound, chunk_length_ms)
    # 处理每个音频片段
    result_list = dict()
    for i, chunk in enumerate(chunks):
        # 编码为base64
        chunk_buffer = io.BytesIO()
        chunk.export(chunk_buffer, format="wav")
        base64_data = base64.b64encode(chunk_buffer.getvalue()).decode('utf-8')
        Seq = i 
        if i == len(chunks) - 1:
            IsEnd = 1
        # start_time为当前uinx时间
        start_time = int(time.time())
        result = get_TongChuan_result(SessionUuid, Source, Target, AudioFormat, Seq, start_time, IsEnd, TranslateTime, base64_data)
        if "Error" in result["Response"]:
            print(result["Response"]["Error"]["Message"])
            continue
        res_list = result["Response"]["List"]
        if len(res_list) > 0:
            for index,j in enumerate(res_list):
                SeId = j["SeId"]
                if SeId in result_list.keys():
                    continue
                # 展示同传结果
                SourceText = j["SourceText"]
                TargetText = j["TargetText"]
                print(SourceText)  # ASR结果
                print(TargetText)  # 翻译结果
                if j["IsEnd"] == True:
                    SeId = j["SeId"]
                    if SeId not in result_list.keys():
                        result_list[SeId] = {"SourceText":SourceText,"TargetText":TargetText}
        time.sleep(0.5)   # 发包间隔和包体时长保持一致
    os.remove(temp_audio_path)
    return result_list  

  
if __name__ == "__main__":
    # 本地文件路径
    path = "test.mp4"
    # 待识别和翻译的语言
    Source = "zh"
    Target = "en"
    result_list = deal_audio(path, Source, Target)
    # print(result_list)
    # 将结果写入文件
    with open("test.json","w",encoding="utf-8") as f:
        json.dump(result_list,f,ensure_ascii=False)