carry_efficiency_tool/utils/tts_help.py

from datetime import timedelta

import requests
import json
import random
import re
import time
from utils.long_tts_client import AliyunTTS
from utils.aliyun_oss import Oss
from loguru import logger

class TTS:
    @classmethod
    def get_pw_zm(cls, text, voice):
        max_retries = 3
        for attempt in range(max_retries):
            url = "http://api.piaoquantv.com/produce-center/speechSynthesis"
            payload = json.dumps({
                "params": {
                    "text": text,
                    "voice": voice,
                    # "vocie": "zhiyuan",
                    "format": "pcm",
                    "volume": 90,
                    "speechRate": 80,
                    "pitchRate": 0
                }
            })

            headers = {
                'Content-Type': 'application/json'
            }
            wait_time = random.uniform(1, 10)
            time.sleep(wait_time)
            try:
                response = requests.request("POST", url, headers=headers, data=payload, timeout=60)
                response = response.json()
                print(response)
                code = response["code"]
                if code == 0:
                    mp3 = response["data"]
                    return mp3
            except Exception:
                if attempt == max_retries - 1:
                    return None
        return None

    """
    音频下载到本地
    """
    @classmethod
    def download_mp3(cls,  pw_url, file_path):
        pw_mp3_path = file_path +'pw_video.mp3'
        for i in range(3):
            payload = {}
            headers = {}
            response = requests.request("GET", pw_url, headers=headers, data=payload, timeout= 30)
            if response.status_code == 200:
                # 以二进制写入模式打开文件
                with open(f"{pw_mp3_path}", "wb") as file:
                    # 将响应内容写入文件
                    file.write(response.content)
                return pw_mp3_path
        return None

    @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 parse_timecode(cls, timecode):
        h, m, s = map(float, timecode.replace(',', '.').split(':'))
        return timedelta(hours=h, minutes=m, seconds=s)

    @classmethod
    def format_timecode(cls, delta):
        total_seconds = delta.total_seconds()
        hours, remainder = divmod(total_seconds, 3600)
        minutes, seconds = divmod(remainder, 60)
        return f"{int(hours):02}:{int(minutes):02}:{seconds:06.3f}".replace('.', ',')

    @classmethod
    def split_subtitle(cls, subtitle_string):
        max_len = 14
        lines = subtitle_string.strip().split('\n')
        subtitles = []
        for i in range(0, len(lines), 4):
            sub_id = int(lines[i].strip())
            timecode_line = lines[i + 1].strip()
            start_time, end_time = timecode_line.split(' --> ')
            text = lines[i + 2].strip()
            if re.search(r'[a-zA-Z]', text):
                text = re.sub(r'[a-zA-Z]', '', text)
            start_delta = cls.parse_timecode(start_time)
            end_delta = cls.parse_timecode(end_time)
            total_duration = (end_delta - start_delta).total_seconds()
            char_duration = total_duration / len(text)

            current_start = start_delta
            for j in range(0, len(text), max_len):
                segment = text[j:j + max_len]
                current_end = current_start + timedelta(seconds=char_duration * len(segment))
                subtitles.append((sub_id, current_start, current_end, segment))
                current_start = current_end
                sub_id += 1

        return subtitles

    @classmethod
    def generate_srt(cls, subtitles):
        srt_content = ''
        for idx, sub in enumerate(subtitles, start=1):
            srt_content += f"{idx}\n"
            srt_content += f"{cls.format_timecode(sub[1])} --> {cls.format_timecode(sub[2])}\n"
            srt_content += f"{sub[3]}\n\n"
        return srt_content.strip()

    @classmethod
    def getSrt(cls, mp3_id):
        url = "http://api.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, timeout=30)
        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")
                new_srt = cls.process_srt(srt)
                result = cls.split_subtitle(new_srt)
                # 生成SRT格式内容
                srt_content = cls.generate_srt(result)
                return srt_content
            else:
                return None
        else:
            return None


    # 长文本语音生成
    @classmethod
    def get_lone_pw_zm(cls, text, voice, file_path):
        # 阿里云根据文本生成语音
        mps_url = AliyunTTS().synthesize(text, voice)
        if not mps_url:
            return
        pw_mp3_path = TTS.download_mp3(mps_url, file_path)
        if not pw_mp3_path:
            return
        return Oss.upload_to_aliyun(pw_mp3_path)

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 = 'http://clipres.yishihui.com/longvideo/crawler/voice/pre/20240821/37fbb8cfc7f1439b8d8a032a1d01d37f1724219959925.mp3'
    # TTS.getSrt(video_file)
    # 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))
    voice = "zhifeng_emo"
    localpath = "/Users/zhangliang/Downloads/47f26a"
    # pw_srt_text = "讲述张鹏程校长爷爷的故事，不仅让人感动，更让我们看到教育的力量和责任。这样的故事值得让更多人知道，您可以通过下方黄色按钮，把这段感人的视频分享至您的家庭群、老乡群、朋友群，让身边的人一同感受这份温暖和感动。每一次转发，都是在传递这份爱与责任。也请您点击下方绿色按钮，发给三五个好友，尤其是那些有孩子的朋友们，让他们也了解到这位校长爷爷的无私奉献。张鹏程用自己的行动告诉我们，教育不仅仅是知识的传授，更是心灵的关怀，让孩子们在爱的环境中健康成长。您分享一次，可能会启发更多人，让我们共同传递这份爱心，让更多孩子在爱的阳光下茁壮成长。请立即行动起来，分享这个故事，让更多的人加入到关心教育、关爱孩子的行列中来。每一次转发，都是在为孩子们的未来贡献一份力量。让我们一起把这份美好传播开去！"
    pw_srt_text = "生活中总有一些故事能让我们感受到温暖和智慧，赵元任的传奇经历就是这样一个值得分享的好故事。各位朋友，看到这里请您立即点击下方的黄色按钮，把这个视频分享至您所在的家庭群、邻里群和兴趣群，让更多人了解这位天才的事迹，激励我们每一个人追求更好的生活。同时，也请您点击绿色按钮，直接发给您身边的朋友们，让他们一起感受这份智慧。每一次分享，都是在传播正能量，鼓励更多人去探索和学习。赵元任不仅是一位语言天才，更是教育的先行者，他用自己的方式告诉我们，学习不仅要有趣，更要通过实践来获得真正的成长。您觉得这条视频内容对您有帮助吗？如果有，请不要犹豫，赶紧分享给更多的朋友和群组，让这些感人的故事温暖每一个人的心。您的每一次分享，都是在让更多人了解到知识的力量和爱的深沉。让我们一起在这个温暖的秋季，传播智慧，分享爱，携手共进，创造更美好的未来！"
    # pw_srt_text = "在这个充满温情与希望的故事中，张鹏程老师用自己的行动改变了无数孩子的命运，您是否也想为这样的善举点赞呢？请您现在就点击下方黄色按钮，分享到您的家庭群、同学群和社区群，让更多人看到张老师的感人故事，感受到教育的力量。同时，别忘了点击绿色按钮，将这份感动分享给您的好友，让他们也能一起传递这份爱与责任。每一次转发，都是对张鹏程老师的支持，也是对无数孩子未来的关注。您知道吗？这样的故事值得让更多人听到，让更多人感受到教育的温暖和希望。每一个孩子都值得被关爱，他们的未来需要我们共同的努力和关心。转发这个视频，或许就能激励更多人加入到关爱贫困儿童的行列中来,赶紧分享给更多的朋友和群组"
    # pw_srt_text="每一句老话都蕴含着深深的智慧，想必大家看完这个视频后也有很多感触。请您马上点击下方黄色按钮，把这段珍贵的内容分享给您的微信群，让更多的人一起感受这些老祖宗的智慧，让我们的社区更加和谐。也别忘了点击绿色按钮，选几个亲密的朋友，单独发给他们，让大家都来思考一下这些道理。生活中，有些真理是我们需要时刻铭记的，尤其是对于我们的晚年生活来说，能帮助我们更好地面对生活的挑战。每一个分享，都是在传递这份智慧，让更多的人受益。您想想，如果大家都能明白这些道理，生活会变得多么美好。现在就动手，让这段视频在您的朋友圈传开，转发到家庭群、邻里群、老友群，让每个人都能感受到这份智慧的力量。您转发一次，就是在为大家的生活添砖加瓦，让我们共同努力，让这份智慧传播得更远。"
    # TTS.get_pw_zm(pw_srt_text, voice)
    # pw_srt = TTS.get_lone_pw_zm(pw_srt_text, voice,localpath)
    # print(pw_srt)
    # print(len(pw_srt_text))
    # {'code': 0, 'msg': 'success',
    #  'data': 'http://clipres.yishihui.com/longvideo/crawler/voice/prod/20250516/7bb5e293a0af43b38701419e28a5e3c11747364158315.mp3',
    #  'redirect': None, 'success': True}

    # mp3_id = "http://clipres.yishihui.com/longvideo/crawler/voice/prod/20250516/7bb5e293a0af43b38701419e28a5e3c11747364158315.mp3"
    # # mp3_id = "http://nls-cloud-cn-shanghai.oss-cn-shanghai.aliyuncs.com/jupiter-flow/tmp/f02e0751b96b4f4ea03d877e11fee4ae.wav?Expires=1747970050&OSSAccessKeyId=LTAI4G588hXC7P47wauY5e2K&Signature=u8Cn2WKpSv7xlmMWDy4Vzos1nV0%3D"
    # mp3_id = "http://clipres.yishihui.com/longvideo/crawler/longvoice/prod/20250516/47e9c7a769ac4a44a3e14d73ded2e49f.mp3"
    # print(TTS.getSrt(mp3_id))