重塑银幕声音：腾讯云语音在视频中的应用

pip install ffmpeg-python requests pydub moviepy tencentcloud-sdk-python

import ffmpeg
import logging

logging.basicConfig(level=logging.INFO)


def extract_audio_from_video(input_video, output_audio):
    try:
        (
            ffmpeg
            .input(input_video)
            .output(output_audio)
            .run(overwrite_output=True)
        )
        logging.info(f"Audio extracted successfully from {input_video} to {output_audio}")
    except Exception as e:
        logging.error(f"An error occurred while extracting audio: {str(e)}")
        raise


if __name__ == '__main__':
    input_video = 'fan_hua_se01.mov'
    output_audio = 'fan_hua_se01_audio.wav'
    extract_audio_from_video(input_video, output_audio)

from qcloud_cos import CosS3Client, CosClientError, CosServiceError, CosConfig

# 将音频文件上传到 COS，并返回上传后的文件路径列表
def upload_to_cos(local_file, bucket, cos_path):
    secret_id = 'your_secret_id'
    secret_key = 'your_secret_key'
    region = 'ap-nanjing'  # 替换为你的 COS 存储桶的地域
    token = None  # 使用临时密钥需要传入 Token，默认为空，可不填写
    scheme = 'https'  # 指定使用 http/https 协议来访问 COS，默认为 https，可不填写

    # 设置 COS 配置
    config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme=scheme)
    # 获取客户端对象
    client = CosS3Client(config)

    # 上传文件
    client.upload_file(
        Bucket=bucket,  # 存储桶名称
        LocalFilePath=local_file,  # 本地文件路径
        Key=cos_path,  # COS 上的文件路径
        PartSize=10,  # 分片大小设置为10MB，默认为 20MB
        MAXThread=10,  # 最大并发上传线程数设置为 10，默认为 5
        EnableMD5=False  # 是否启用 MD5 校验，默认为 False
    )
    # logging.info(f"upload to cos resp:{response}")
    # 生成带签名的 URL，设置过期时间为1小时
    signed_url = client.get_presigned_url(
        Bucket=bucket,
        Key=cos_path,
        Method='GET',
        Expired=time.time() + 3600  # 1小时后过期
    )

    return signed_url

if __name__ == '__main__':
    input_audio = 'fan_hua_se01_audio.wav'
    bucket_name = 'your-bucket'  # COS 桶的名称
    file_url = upload_to_cos(input_audio, bucket_name, input_audio)
    logging.info(f"file_url:{file_url}")

# 初始化 client
def create_asr_client(key_id, key_secret):
    try:
        cred = credential.Credential(key_id, key_secret)
        http_profile = HttpProfile()
        http_profile.endpoint = "asr.tencentcloudapi.com"
        client_profile = ClientProfile()
        client_profile.httpProfile = http_profile
        client_profile.signMethod = "TC3-HMAC-SHA256"
        client = asr_client.AsrClient(cred, "ap-shanghai", client_profile)
        return client
    except TencentCloudSDKException as err:
        logging.error(err)
        return None

# 创建语音识别任务
def create_recon_task(engine_type, file_url):
    client = create_asr_client('your_secret_id', 'your_secret_key')
    req = models.CreateRecTaskRequest()
    params = {"ChannelNum": 1, "ResTextFormat": 3, "SourceType": 0, "ConvertNumMode": 1}
    req._deserialize(params)
    req.EngineModelType = engine_type
    req.Url = file_url
    try:
        resp = client.CreateRecTask(req)
        logging.info(resp)
        requesid = resp.RequestId
        taskid = resp.Data.TaskId
        return requesid, taskid
    except Exception as err:
        logging.error(traceback.format_exc())
        return None, None

# 查询识别结果
def query_recon_task(taskid):
    client = create_asr_client('your_secret_id', 'your_secret_key')
    req = models.DescribeTaskStatusRequest()
    params = '{"TaskId":' + str(taskid) + '}'
    req.from_json_string(params)
    result = ""
    while True:
        try:
            resp = client.DescribeTaskStatus(req)
            resp_json = resp.to_json_string()
            logging.info(resp_json)
            resp_obj = json.loads(resp_json)
            if not resp_obj["Data"]:
                return False, ""
            if resp_obj["Data"]["StatusStr"] == "success":
                result = resp_obj["Data"]["ResultDetail"]
                break
            if resp_obj["Data"]["Status"] == 3:
                return False, ""

            time.sleep(1)
        except TencentCloudSDKException as err:
            logging.info(err)
            return False, ""

    return True, result

# 创建并获取语音识别结果
def creat_audio_task(engine_type, audio_url):
    request_id, task_id = create_recon_task(engine_type, audio_url)
    ret, result = query_recon_task(task_id)
    if ret:
        return result
    return ""

if __name__ == '__main__':
    result = creat_audio_task('16k_zh', file_url)
    if result:
        logging.info(f'result:{result}')
    else:
        logging.error("result is none")

def create_tts_client(key_id, key_secret):
    try:
        cred = credential.Credential(key_id, key_secret)
        http_profile = HttpProfile()
        http_profile.endpoint = "tts.tencentcloudapi.com"
        client_profile = ClientProfile()
        client_profile.httpProfile = http_profile
        client_profile.signMethod = "TC3-HMAC-SHA256"
        client = tts_client.TtsClient(cred, "ap-shanghai", client_profile)
        return client
    except TencentCloudSDKException as err:
        logging.error(err)
        return None

# 创建语音合成任务
def create_tts_task(text):
    client = create_tts_client('your_secret_id', 'your_secret_key')
    tts_req = tts_models.CreateTtsTaskRequest()
    tts_req.Text = text
    tts_req.VoiceType = 101019
    tts_req.Codec = 'wav'
    try:
        resp = client.CreateTtsTask(tts_req)
        logging.info(resp)
        request_id = resp.RequestId
        taskid = resp.Data.TaskId
        return request_id, taskid
    except Exception as err:
        logging.error(traceback.format_exc())
        return None, None

# 查询合成结果
def query_tts_task(taskid, timeout=600):
    client = create_tts_client('your_secret_id', 'your_secret_key')
    req = tts_models.DescribeTtsTaskStatusRequest()
    req.TaskId = taskid
    start_time = time.time()

    while True:
        try:
            resp = client.DescribeTtsTaskStatus(req)
            resp_json = resp.to_json_string()
            logging.info(resp_json)
            resp_obj = json.loads(resp_json)
            if not resp_obj["Data"]:
                return False, ""
            if resp_obj["Data"]["StatusStr"] == "success":
                result = resp_obj["Data"]["ResultUrl"]
                break
            if resp_obj["Data"]["Status"] == 3:
                return False, ""

            if time.time() - start_time > timeout:
                logging.error("Timeout while waiting for TTS task to complete.")
                return False, ""

            time.sleep(1)
        except TencentCloudSDKException as err:
            logging.info(err)
            return False, ""
        
    return True, result

def new_audio_tts(text):
    request_id, task_id = create_tts_task(text)
    ret, result = query_tts_task(task_id)
    if ret:
        return result
    return ""

if __name__ == '__main__':
    audio_url = new_audio_tts(result)
    if audio_url:
        print(f"Audio URL: {audio_url}")
        try:
            # 发送 HTTP GET 请求获取文件内容
            local_path = 'fan_hua_se01_audio_new.wav'
            response = requests.get(audio_url, stream=True)
            response.raise_for_status()  # 检查请求是否成功

            # 将内容写入本地文件
            with open(local_path, 'wb') as file:
                for chunk in response.iter_content(chunk_size=8192):
                    file.write(chunk)

            print(f"File downloaded successfully: {local_path}")
        except requests.exceptions.RequestException as e:
            print(f"An error occurred while downloading the file: {e}")
    else:
        print("Failed to retrieve TTS task result.")