video_rewriting/common/tts_help.py

import subprocess

import requests
import json
import random
import re
import time
from pydub import AudioSegment

class TTS:

    @classmethod
    def get_pw_zm(cls, text):
        max_retries = 3
        for attempt in range(max_retries):
            url = "https://zh.api.guiji.cn/avatar2c/tool/sec_tts"

            payload = json.dumps({
                "text": text,
                "speaker_id": "160"
            })
            headers = {
                'accept': 'application/json, text/plain, */*',
                'accept-language': 'zh-CN',
                'cache-control': 'no-cache',
                'content-type': 'application/json',
                'cookie': 'anylangIsLogin=true',
                'origin': 'https://app.guiji.cn',
                'pragma': 'no-cache',
                'priority': 'u=1, i',
                'referer': 'https://app.guiji.cn/',
                'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
                'sec-ch-ua-mobile': '?0',
                'sec-ch-ua-platform': '"macOS"',
                'sec-fetch-dest': 'empty',
                'sec-fetch-mode': 'cors',
                'sec-fetch-site': 'same-site',
                'token': '72912b4ac30b412cb19d2e835a678854'
            }
            wait_time = random.uniform(5, 20)
            time.sleep(wait_time)
            response = requests.request("POST", url, headers=headers, data=payload)
            try:
                response = response.json()
                code = response["code"]
                if code == 200:
                    mp3 = response["data"]
                    return mp3
                else:
                    if attempt == max_retries - 1:
                        return None
            except Exception:
                if attempt == max_retries - 1:
                    return None

    """
    音频下载到本地
    """
    @classmethod
    def download_mp3(cls,  video_file, video_path_url, pw_random_id):
        pw_mp3_path = video_path_url + str(pw_random_id) +'pw_video.mp3'
        for i in range(3):
            payload = {}
            headers = {}
            response = requests.request("GET", video_file, headers=headers, data=payload)
            if response.status_code == 200:
                # 以二进制写入模式打开文件
                with open(f"{pw_mp3_path}", "wb") as file:
                    # 将响应内容写入文件
                    file.write(response.content)
                # 增加音频音量
                audio = AudioSegment.from_file(pw_mp3_path)
                louder_audio = audio + 10
                louder_audio.export(pw_mp3_path, format="mp3")
                time.sleep(5)
                return pw_mp3_path
        return ''

    @classmethod
    def get_srt_format(cls, pw_srt_text, pw_url_sec):
        segments = re.split(r'(，|。|！|？)', pw_srt_text)
        segments = [segments[i] + segments[i + 1] for i in range(0, len(segments) - 1, 2)]
        pw_url_sec = int(pw_url_sec) + 1
        # 确定每段显示时间
        num_segments = len(segments)
        duration_per_segment = pw_url_sec / num_segments
        srt_content = ""
        start_time = 0.0
        for i, segment in enumerate(segments):
            end_time = start_time + duration_per_segment
            srt_content += f"{i + 1}\n"
            srt_content += f"{int(start_time // 3600):02}:{int((start_time % 3600) // 60):02}:{int(start_time % 60):02},{int((start_time % 1) * 1000):03} --> "
            srt_content += f"{int(end_time // 3600):02}:{int((end_time % 3600) // 60):02}:{int(end_time % 60):02},{int((end_time % 1) * 1000):03}\n"
            srt_content += f"{segment.strip()}\n\n"
            start_time = end_time

        print(srt_content)
        return srt_content

    @classmethod
    def process_srt(cls, srt):
        lines = srt.strip().split('\n')
        processed_lines = []

        for line in lines:
            if re.match(r'^\d+$', line):
                processed_lines.append(line)
            elif re.match(r'^\d{2}:\d{2}:\d{2}\.\d{1,3}-->\d{2}:\d{2}:\d{2}\.\d{1,3}$', line):
                processed_lines.append(line.replace('-->', ' --> '))
            else:
                line = re.sub(r'[，。！？；、]$', '', line)
                # 添加换行符
                processed_lines.append(line + '\n')

        return '\n'.join(processed_lines)

    @classmethod
    def getSrt(cls, mp3_id):
        url = "http://api-internal.piaoquantv.com/produce-center/srt/get/content"

        payload = json.dumps({
            "params": {
                "resourceChannel": "outer",
                "videoPath": mp3_id
            }
        })
        headers = {
            'User-Agent': 'Apifox/1.0.0 (https://apifox.com)',
            'Content-Type': 'application/json',
            'Accept': '*/*',
            'Host': 'api-internal.piaoquantv.com',
            'Connection': 'keep-alive'
        }

        response = requests.request("POST", url, headers=headers, data=payload)
        time.sleep(1)
        data_list = response.json()
        code = data_list["code"]
        if code == 0:
            srt = data_list["data"]
            if srt:
                srt = srt.replace("/n", "\n")
                # srt = re.sub(r'(\w+)([,。!?])', r'\n\n', srt)
                new_srt = cls.process_srt(srt)
                return new_srt
            else:
                return None
        else:
            return None


if __name__ == '__main__':
    text = "真是太实用了，分享给身边的准妈妈们吧！这些孕期禁忌一定要记住，赶紧转发给更多人，帮助更多的宝妈们。一起为宝宝的健康加油！"
    mp3 = TTS.get_pw_zm(text)
    print(mp3)
    # command = [
    #   'ffmpeg',
    #   '-i', mp3,
    #   '-q:a', '0',
    #   '-map', 'a',
    #   # '-codec:a', 'libmp3lame',  # 指定 MP3 编码器
    #   "/Users/tzld/Desktop/video_rewriting/path/pw_video.mp3"
    # ]
    # subprocess.run(command)
    # print("完成")
    # video_file = 'https://digital-public.obs.myhuaweicloud.com/vcm_server/20240715/KnOgRTYKvZbWD2EX_ms/FwA2SEh7DZQGlwAe.wav'
    # result = subprocess.run(
    #     ["ffprobe", "-v", "error", "-show_entries", "format=duration",
    #      "-of", "default=noprint_wrappers=1:nokey=1", video_file],
    #     capture_output=True, text=True)
    # print(float(result.stdout))