Upload 3 files

app.py

@@ -0,0 +1,154 @@
+import gradio as gr
+import os
+
+# update modelscope
+os.system("pip install -U modelscope -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.html -i https://mirror.sjtu.edu.cn/pypi/web/simple")
+    
+import datetime
+from modelscope.pipelines import pipeline
+from modelscope.utils.constant import Tasks
+from subtitle_utils import generate_srt
+
+#获取当前北京时间
+utc_dt = datetime.datetime.utcnow()
+beijing_dt = utc_dt.astimezone(datetime.timezone(datetime.timedelta(hours=16)))
+formatted = beijing_dt.strftime("%Y-%m-%d_%H")
+print(f"北京时间: {beijing_dt.year}年{beijing_dt.month}月{beijing_dt.day}日 "
+      f"{beijing_dt.hour}时{beijing_dt.minute}分{beijing_dt.second}秒")
+#创建作品存放目录
+works_path = '../works_audio_video_recognize/' + formatted
+if not os.path.exists(works_path):
+  os.makedirs(works_path)
+print('作品目录：' + works_path)
+
+inference_pipeline = pipeline(
+    task=Tasks.auto_speech_recognition,
+    model='damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch')
+    
+def transcript(audiofile, text_file, srt_file):
+    rec_result = inference_pipeline(audio_in=audiofile)
+    
+    text_output = rec_result['text']
+    with open(text_file, "w") as f:
+        f.write(text_output)
+        
+    srt_output = generate_srt(rec_result['sentences'])
+    with open(srt_file, "w") as f:
+        f.write(srt_output)
+        
+    return text_output, srt_output
+
+def audio_recog(audiofile):
+    utc_dt = datetime.datetime.utcnow()
+    beijing_dt = utc_dt.astimezone(datetime.timezone(datetime.timedelta(hours=16)))
+    formatted = beijing_dt.strftime("%Y-%m-%d_%H-%M-%S")
+    print(f"开始时间: {beijing_dt.year}年{beijing_dt.month}月{beijing_dt.day}日 "
+      f"{beijing_dt.hour}时{beijing_dt.minute}分{beijing_dt.second}秒")
+      
+    print("音频文件：" + audiofile)
+    
+    filename = os.path.splitext(os.path.basename(audiofile))[0]
+    text_file = works_path + '/' + filename + '.txt'
+    srt_file = works_path + '/' + filename + '.srt'
+    text_output, srt_output = transcript(audiofile, text_file, srt_file)
+    
+    utc_dt = datetime.datetime.utcnow()
+    beijing_dt = utc_dt.astimezone(datetime.timezone(datetime.timedelta(hours=16)))
+    formatted = beijing_dt.strftime("%Y-%m-%d_%H-%M-%S")
+    print(f"结束时间: {beijing_dt.year}年{beijing_dt.month}月{beijing_dt.day}日 "
+      f"{beijing_dt.hour}时{beijing_dt.minute}分{beijing_dt.second}秒")
+    
+    return text_output, text_file, srt_output, srt_file
+
+def video_recog(filepath):
+    filename = os.path.splitext(os.path.basename(filepath))[0]
+    worksfile = works_path + '/works_' + filename + '.mp4'
+    print("视频文件：" + filepath)
+    
+    utc_dt = datetime.datetime.utcnow()
+    beijing_dt = utc_dt.astimezone(datetime.timezone(datetime.timedelta(hours=16)))
+    formatted = beijing_dt.strftime("%Y-%m-%d_%H-%M-%S.%f")
+    
+    # 提取音频为mp3
+    audiofile = works_path + '/' + formatted + '.mp3'
+    os.system(f"ffmpeg -i {filepath} -vn -c:a libmp3lame -q:a 4 {audiofile}")
+    
+    #识别音频文件
+    text_output, text_file, srt_output, srt_file = audio_recog(audiofile)
+    
+#    # 给视频添加字幕
+#    os.system(f"ffmpeg -i {filepath} -i {srt_file} -c:s mov_text -c:v copy -c:a copy {worksfile}")
+#    print("作品：" + worksfile)
+    
+    return text_output, text_file, srt_output, srt_file
+
+css_style = "#fixed_size_img {height: 240px;} " \
+            "#overview {margin: auto;max-width: 400px; max-height: 400px;}"
+
+title = "音视频识别 by宁侠"
+description = "您只需要上传一段音频或视频文件，我们的服务会快速对其进行语音识别，然后生成相应的文字和字幕。这样，您就可以轻松地记录下重要的语音内容，或者为视频添加精准的字幕。现在就来试试我们的音视频识别服务吧，让您的生活和工作更加便捷！"
+
+examples_path = 'examples/'
+examples = [[examples_path + 'demo_shejipuhui.mp4']]
+
+# gradio interface
+with gr.Blocks(title=title, css=css_style) as demo:
+    gr.HTML('''
+      <div style="text-align: center; max-width: 720px; margin: 0 auto;">
+                  <div
+                    style="
+                      display: inline-flex;
+                      align-items: center;
+                      gap: 0.8rem;
+                      font-size: 1.75rem;
+                    "
+                  >
+                    <h1 style="font-family:  PingFangSC; font-weight: 500; font-size: 36px; margin-bottom: 7px;">
+                      音视频识别
+                    </h1>
+                    <h1 style="font-family: PingFangSC; font-weight: 500; line-height: 1.5em; font-size: 16px; margin-bottom: 7px;">
+                      by宁侠
+                    </h1>
+      ''')
+    gr.Markdown(description)
+    
+    with gr.Tab("🔊音频识别 Audio Transcribe"):
+        with gr.Row():
+            with gr.Column():
+                audio_input = gr.Audio(label="🔊音频输入 Audio Input", type="filepath")
+                gr.Examples(['examples/paddlespeech.asr-zh.wav', 'examples/demo_shejipuhui.mp3'], [audio_input])
+                audio_recog_button = gr.Button("👂音频识别 Recognize")
+            with gr.Column():
+                audio_text_output = gr.Textbox(label="✏️识别结果 Recognition Result", max_lines=5)
+                audio_text_file = gr.File(label="✏️识别结果文件 Recognition Result File")
+                audio_srt_output = gr.Textbox(label="📖SRT字幕内容 SRT Subtitles", max_lines=10)
+                audio_srt_file = gr.File(label="📖SRT字幕文件 SRT File")
+                audio_subtitles_button = gr.Button("添加字幕\nGenerate Subtitles", visible=False)
+                audio_output = gr.Audio(label="🔊音频 Audio", visible=False)
+    
+    audio_recog_button.click(audio_recog, inputs=[audio_input], outputs=[audio_text_output, audio_text_file, audio_srt_output, audio_srt_file])
+#    audio_subtitles_button.click(audio_subtitles, inputs=[audio_text_input], outputs=[audio_output])
+
+    with gr.Tab("🎥视频识别 Video Transcribe"):
+        with gr.Row():
+            with gr.Column():
+                video_input = gr.Video(label="🎥视频输入 Video Input")
+                gr.Examples(['examples/demo_shejipuhui.mp4'], [video_input], label='语音识别示例 ASR Demo')
+                video_recog_button = gr.Button("👂视频识别 Recognize")
+                video_output = gr.Video(label="🎥视频 Video", visible=False)
+            with gr.Column():
+                video_text_output = gr.Textbox(label="✏️识别结果 Recognition Result", max_lines=5)
+                video_text_file = gr.File(label="✏️识别结果文件 Recognition Result File")
+                video_srt_output = gr.Textbox(label="📖SRT字幕内容 SRT Subtitles", max_lines=10)
+                video_srt_file = gr.File(label="📖SRT字幕文件 SRT File")
+                with gr.Row(visible=False):
+                    font_size = gr.Slider(minimum=10, maximum=100, value=32, step=2, label="🔠字幕字体大小 Subtitle Font Size")
+                    font_color = gr.Radio(["black", "white", "green", "red"], label="🌈字幕颜色 Subtitle Color", value='white')
+                video_subtitles_button = gr.Button("添加字幕\nGenerate Subtitles", visible=False)
+                
+    
+    video_recog_button.click(video_recog, inputs=[video_input], outputs=[video_text_output, video_text_file, video_srt_output, video_srt_file])
+#    video_subtitles_button.click(video_subtitles, inputs=[video_text_input], outputs=[video_output])
+
+# start gradio service in local
+demo.queue(api_open=False).launch(debug=True)

subtitle_utils.py

@@ -0,0 +1,130 @@
+def time_convert(ms):
+    ms = int(ms)
+    tail = ms % 1000
+    s = ms // 1000
+    mi = s // 60
+    s = s % 60
+    h = mi // 60
+    mi = mi % 60
+    h = "00" if h == 0 else str(h)
+    mi = "00" if mi == 0 else str(mi)
+    s = "00" if s == 0 else str(s)
+    tail = str(tail)
+    if len(h) == 1: h = '0' + h
+    if len(mi) == 1: mi = '0' + mi
+    if len(s) == 1: s = '0' + s
+    return "{}:{}:{},{}".format(h, mi, s, tail)
+
+
+class Text2SRT():
+    def __init__(self, text_seg, ts_list, offset=0):
+        self.token_list = [i for i in text_seg.split() if len(i)]
+        self.ts_list = ts_list
+        start, end = ts_list[0][0] - offset, ts_list[-1][1] - offset
+        self.start_sec, self.end_sec = start, end
+        self.start_time = time_convert(start)
+        self.end_time = time_convert(end)
+    def text(self):
+        res = ""
+        for word in self.token_list:
+            if '\u4e00' <= word <= '\u9fff':
+                res += word
+            else:
+                res += " " + word
+        return res
+    def len(self):
+        return len(self.token_list)
+    def srt(self, acc_ost=0.0):
+        return "{} --> {}\n{}\n".format(
+            time_convert(self.start_sec+acc_ost*1000),
+            time_convert(self.end_sec+acc_ost*1000), 
+            self.text())
+    def time(self, acc_ost=0.0):
+        return (self.start_sec/1000+acc_ost, self.end_sec/1000+acc_ost)
+
+def distribute_spk(sentence_list, sd_time_list):
+    sd_sentence_list = []
+    for d in sentence_list:
+        sentence_start = d['ts_list'][0][0]
+        sentence_end = d['ts_list'][-1][1]
+        sentence_spk = 0
+        max_overlap = 0
+        for sd_time in sd_time_list:
+            spk_st, spk_ed, spk = sd_time
+            spk_st = spk_st*1000
+            spk_ed = spk_ed*1000
+            overlap = max(
+                min(sentence_end, spk_ed) - max(sentence_start, spk_st), 0)
+            if overlap > max_overlap:
+                max_overlap = overlap
+                sentence_spk = spk
+        d['spk'] = sentence_spk
+        sd_sentence_list.append(d)
+    return sd_sentence_list
+
+def generate_srt(sentence_list):
+    srt_total = ''
+    for i, d in enumerate(sentence_list):
+        t2s = Text2SRT(d['text_seg'], d['ts_list'])
+        if 'spk' in d:
+            srt_total += "{}  spk{}\n{}".format(i, d['spk'], t2s.srt())
+        else:
+            srt_total += "{}\n{}".format(i, t2s.srt())
+    return srt_total
+
+def generate_srt_clip(sentence_list, start, end, begin_index=0, time_acc_ost=0.0):
+    start, end = int(start * 1000), int(end * 1000)
+    srt_total = ''
+    cc = 1 + begin_index
+    subs = []
+    for i, d in enumerate(sentence_list):
+        if d['ts_list'][-1][1] <= start:
+            continue
+        if d['ts_list'][0][0] >= end:
+            break
+        # parts in between
+        if (d['ts_list'][-1][1] <= end and d['ts_list'][0][0] > start) or (d['ts_list'][-1][1] == end and d['ts_list'][0][0] == start):
+            t2s = Text2SRT(d['text_seg'], d['ts_list'], offset=start)
+            srt_total += "{}\n{}".format(cc, t2s.srt(time_acc_ost))
+            subs.append((t2s.time(time_acc_ost), t2s.text()))
+            cc += 1
+            continue
+        if d['ts_list'][0][0] <= start:
+            if not d['ts_list'][-1][1] > end:
+                for j, ts in enumerate(d['ts_list']):
+                    if ts[1] > start:
+                        break
+                _text = " ".join(d['text_seg'].split()[j:])
+                _ts = d['ts_list'][j:]
+            else:
+                for j, ts in enumerate(d['ts_list']):
+                    if ts[1] > start:
+                        _start = j
+                        break
+                for j, ts in enumerate(d['ts_list']):
+                    if ts[1] > end:
+                        _end = j
+                        break
+                _text = " ".join(d['text_seg'].split()[_start:_end])
+                _ts = d['ts_list'][_start:_end]
+            if len(ts):
+                t2s = Text2SRT(_text, _ts, offset=start)
+                srt_total += "{}\n{}".format(cc, t2s.srt(time_acc_ost))
+                subs.append((t2s.time(time_acc_ost), t2s.text()))
+                cc += 1
+            continue
+        if d['ts_list'][-1][1] > end:
+            for j, ts in enumerate(d['ts_list']):
+                if ts[1] > end:
+                    break
+            _text = " ".join(d['text_seg'].split()[:j])
+            _ts = d['ts_list'][:j]
+            if len(_ts):
+                t2s = Text2SRT(_text, _ts, offset=start)
+                srt_total += "{}\n{}".format(cc, t2s.srt(time_acc_ost))
+                subs.append(
+                    (t2s.time(time_acc_ost), t2s.text())
+                    )
+                cc += 1
+            continue
+    return srt_total, subs, cc

videoclipper.py

@@ -0,0 +1,172 @@
+import sys
+import copy
+import librosa
+import logging
+import argparse
+import numpy as np
+import soundfile as sf
+import moviepy.editor as mpy
+# from modelscope.pipelines import pipeline
+# from modelscope.utils.constant import Tasks
+from subtitle_utils import generate_srt, generate_srt_clip, distribute_spk
+from trans_utils import pre_proc, proc, write_state, load_state, proc_spk, generate_vad_data
+# from argparse_tools import ArgumentParser, get_commandline_args
+
+from moviepy.editor import *
+from moviepy.video.tools.subtitles import SubtitlesClip
+
+
+class VideoClipper():
+    def __init__(self, asr_pipeline, sd_pipeline=None):
+        logging.warning("Initializing VideoClipper.")
+        self.asr_pipeline = asr_pipeline
+        self.sd_pipeline = sd_pipeline
+
+    def recog(self, audio_input, sd_switch='no', state=None):
+        if state is None:
+            state = {}
+        sr, data = audio_input
+        assert sr == 16000, "16kHz sample rate required, {} given.".format(sr)
+        if len(data.shape) == 2:  # multi-channel wav input
+            logging.warning("Input wav shape: {}, only first channel reserved.").format(data.shape)
+            data = data[:,0]
+        state['audio_input'] = (sr, data)
+        data = data.astype(np.float64)
+        rec_result = self.asr_pipeline(audio_in=data)
+        if sd_switch == 'yes':
+            vad_data = generate_vad_data(data.astype(np.float32), rec_result['sentences'], sr)
+            sd_result = self.sd_pipeline(audio=vad_data, batch_size=1)
+            rec_result['sd_sentences'] = distribute_spk(rec_result['sentences'], sd_result['text'])
+            res_srt = generate_srt(rec_result['sd_sentences'])
+            state['sd_sentences'] = rec_result['sd_sentences']
+        else:
+            res_srt = generate_srt(rec_result['sentences'])
+        state['recog_res_raw'] = rec_result['text_postprocessed']
+        state['timestamp'] = rec_result['time_stamp']
+        state['sentences'] = rec_result['sentences']
+        res_text = rec_result['text']
+        return res_text, res_srt, state
+
+    def clip(self, dest_text, start_ost, end_ost, state, dest_spk=None):
+        # get from state
+        audio_input = state['audio_input']
+        recog_res_raw = state['recog_res_raw']
+        timestamp = state['timestamp']
+        sentences = state['sentences']
+        sr, data = audio_input
+        data = data.astype(np.float64)
+
+        all_ts = []
+        if dest_spk is None or dest_spk == '' or 'sd_sentences' not in state:
+            for _dest_text in dest_text.split('#'):
+                _dest_text = pre_proc(_dest_text)
+                ts = proc(recog_res_raw, timestamp, _dest_text)
+                for _ts in ts: all_ts.append(_ts)
+        else:
+            for _dest_spk in dest_spk.split('#'):
+                ts = proc_spk(_dest_spk, state['sd_sentences'])
+                for _ts in ts: all_ts.append(_ts)
+        ts = all_ts
+        ts.sort()
+        srt_index = 0
+        clip_srt = ""
+        if len(ts):
+            start, end = ts[0]
+            start = min(max(0, start+start_ost*16), len(data))
+            end = min(max(0, end+end_ost*16), len(data))
+            res_audio = data[start:end]
+            start_end_info = "from {} to {}".format(start/16000, end/16000)
+            srt_clip, _, srt_index = generate_srt_clip(sentences, start/16000.0, end/16000.0, begin_index=srt_index)
+            clip_srt += srt_clip
+            for _ts in ts[1:]:  # multiple sentence input or multiple output matched
+                start, end = _ts
+                start = min(max(0, start+start_ost*16), len(data))
+                end = min(max(0, end+end_ost*16), len(data))
+                start_end_info += ", from {} to {}".format(start, end)
+                res_audio = np.concatenate([res_audio, data[start+start_ost*16:end+end_ost*16]], -1)
+                srt_clip, _, srt_index = generate_srt_clip(sentences, start/16000.0, end/16000.0, begin_index=srt_index-1)
+                clip_srt += srt_clip
+        if len(ts):
+            message = "{} periods found in the speech: ".format(len(ts)) + start_end_info
+        else:
+            message = "No period found in the speech, return raw speech. You may check the recognition result and try other destination text."
+            res_audio = data
+        return (sr, res_audio), message, clip_srt
+
+    def video_recog(self, vedio_filename, sd_switch='no'):
+        vedio_filename = vedio_filename
+        clip_video_file = vedio_filename[:-4] + '_clip.mp4'
+        video = mpy.VideoFileClip(vedio_filename)
+        audio_file = vedio_filename[:-3] + 'wav'
+        video.audio.write_audiofile(audio_file)
+        wav = librosa.load(audio_file, sr=16000)[0]
+        state = {
+            'vedio_filename': vedio_filename,
+            'clip_video_file': clip_video_file,
+            'video': video,
+        }
+        # res_text, res_srt = self.recog((16000, wav), state)
+        return self.recog((16000, wav), sd_switch, state)
+
+    def video_clip(self, dest_text, start_ost, end_ost, state, font_size=32, font_color='white', add_sub=False, dest_spk=None):
+        # get from state
+        recog_res_raw = state['recog_res_raw']
+        timestamp = state['timestamp']
+        sentences = state['sentences']
+        video = state['video']
+        clip_video_file = state['clip_video_file']
+        vedio_filename = state['vedio_filename']
+        
+        all_ts = []
+        srt_index = 0
+        if dest_spk is None or dest_spk == '' or 'sd_sentences' not in state:
+            for _dest_text in dest_text.split('#'):
+                _dest_text = pre_proc(_dest_text)
+                ts = proc(recog_res_raw, timestamp, _dest_text)
+                for _ts in ts: all_ts.append(_ts)
+        else:
+            for _dest_spk in dest_spk.split('#'):
+                ts = proc_spk(_dest_spk, state['sd_sentences'])
+                for _ts in ts: all_ts.append(_ts)
+        time_acc_ost = 0.0
+        ts = all_ts
+        ts.sort()
+        clip_srt = ""
+        if len(ts):
+            start, end = ts[0][0] / 16000, ts[0][1] / 16000
+            srt_clip, subs, srt_index = generate_srt_clip(sentences, start, end, begin_index=srt_index, time_acc_ost=time_acc_ost)
+            start, end = start+start_ost/1000.0, end+end_ost/1000.0
+            video_clip = video.subclip(start, end)
+            start_end_info = "from {} to {}".format(start, end)
+            clip_srt += srt_clip
+            if add_sub:
+                generator = lambda txt: TextClip(txt, font='./font/STHeitiMedium.ttc', fontsize=font_size, color=font_color)
+                subtitles = SubtitlesClip(subs, generator)
+                video_clip = CompositeVideoClip([video_clip, subtitles.set_pos(('center','bottom'))])
+            concate_clip = [video_clip]
+            time_acc_ost += end+end_ost/1000.0 - (start+start_ost/1000.0)
+            for _ts in ts[1:]:
+                start, end = _ts[0] / 16000, _ts[1] / 16000
+                srt_clip, subs, srt_index = generate_srt_clip(sentences, start, end, begin_index=srt_index-1, time_acc_ost=time_acc_ost)
+                start, end = start+start_ost/1000.0, end+end_ost/1000.0
+                _video_clip = video.subclip(start, end)
+                start_end_info += ", from {} to {}".format(start, end)
+                clip_srt += srt_clip
+                if add_sub:
+                    generator = lambda txt: TextClip(txt, font='./font/STHeitiMedium.ttc', fontsize=font_size, color=font_color)
+                    subtitles = SubtitlesClip(subs, generator)
+                    _video_clip = CompositeVideoClip([_video_clip, subtitles.set_pos(('center','bottom'))])
+                concate_clip.append(copy.copy(_video_clip))
+                time_acc_ost += end+end_ost/1000.0 - (start+start_ost/1000.0)
+            message = "{} periods found in the audio: ".format(len(ts)) + start_end_info
+            logging.warning("Concating...")
+            if len(concate_clip) > 1:
+                video_clip = concatenate_videoclips(concate_clip)
+            video_clip.write_videofile(clip_video_file, audio_codec="aac")
+        else:
+            clip_video_file = vedio_filename
+            message = "No period found in the audio, return raw speech. You may check the recognition result and try other destination text."
+            srt_clip = ''
+        return clip_video_file, message, clip_srt
+
+