Add github workflow CI to run flake8 linting and install dependencies.

.flake8

@@ -0,0 +1,5 @@
+[flake8]
+ignore = E226,E302,E41,C901,W504
+max-line-length = 150
+exclude = tests/*
+max-complexity = 10

.github/workflows/pytvzhen-web.yaml

@@ -0,0 +1,26 @@
+name: Pytvzhen-web application test
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+    branches: [ master ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python 3.12.3
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.12.3
+      - name: Lint with flake8
+        run: |
+          pip install flake8
+          flake8 .
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt

app.py

@@ -6,6 +6,7 @@
 from flask import Flask, request, jsonify, render_template, send_from_directory
 import zipfile
 import shutil
+import uuid
 
 from tools.audio_remove import audio_remove
 from work_space import transcribeAudioEn, srtSentanceMerge, srtFileGoogleTran, srtFileDeeplTran, srtFileGPTTran, \
@@ -179,7 +180,7 @@ def audio_serve(video_id):
     video_fn = f'{video_id}.mp4'
     audio_fn = f'{video_id}.wav'
 
-    video_path, audio_path = os.path.join(output_path, video_fn), os.path.join(output_path, audio_fn)
+    _, audio_path = os.path.join(output_path, video_fn), os.path.join(output_path, audio_fn)
 
     if os.path.exists(audio_path):
         return send_from_directory(output_path, audio_fn, as_attachment=True)
@@ -194,10 +195,10 @@ def remove_audio_bg(video_id):
     audio_fn = f'{video_id}.wav'
     audio_no_bg_fn, audio_bg_fn = f'{video_id}_no_bg.wav', f'{video_id}_bg.wav'
 
-    video_path, audio_path, audio_no_bg_path, audio_bg_fn_path = (os.path.join(output_path, video_fn),
-                                                                  os.path.join(output_path, audio_fn),
-                                                                  os.path.join(output_path, audio_no_bg_fn),
-                                                                  os.path.join(output_path, audio_bg_fn))
+    _, audio_path, audio_no_bg_path, audio_bg_fn_path = (os.path.join(output_path, video_fn),
+                                                         os.path.join(output_path, audio_fn),
+                                                         os.path.join(output_path, audio_no_bg_fn),
+                                                         os.path.join(output_path, audio_bg_fn))
 
     if os.path.exists(audio_no_bg_path) and os.path.exists(audio_bg_fn_path):
         return jsonify({"message": log_info_return_str(
@@ -318,7 +319,7 @@ def transhlate_to_zh(video_id):
     en_srt_merged_path = os.path.join(output_path, en_srt_merged_fn)
     zh_srt_merged_path = os.path.join(output_path, zh_srt_merged_fn)
 
-    if os.path.exists(en_srt_merged_path) == False:
+    if not os.path.exists(en_srt_merged_path):
         return jsonify({"message": log_warning_return_str(
             f'English SRT {en_srt_merged_fn} not found at {en_srt_merged_path}')}), 404
 
@@ -343,7 +344,7 @@ def transhlate_to_zh(video_id):
             else:
                 ret = srtFileDeeplTran(logger=app.logger, sourceFileNameAndPath=en_srt_merged_path,
                                        outputFileNameAndPath=zh_srt_merged_path, key=api_key)
-                if ret == True:
+                if ret:
                     return jsonify({"message": log_info_return_str(
                         f"using deepl translate to translate SRT from {en_srt_merged_fn} to {zh_srt_merged_fn} successfully."),
                         "video_id": video_id}), 200
@@ -357,7 +358,7 @@ def transhlate_to_zh(video_id):
                 ret = srtFileGPTTran(logger=app.logger, model=translateVendor, proxies=None,
                                      sourceFileNameAndPath=en_srt_merged_path, outputFileNameAndPath=zh_srt_merged_path,
                                      key=api_key)
-                if ret == True:
+                if ret:
                     return jsonify({"message": log_info_return_str(
                         f"using {translateVendor} translate to translate SRT from {en_srt_merged_fn} to {zh_srt_merged_fn} successfully."),
                         "video_id": video_id}), 200
@@ -395,12 +396,12 @@ def voice_connect(video_id):
     warning_log_fn = video_id + "_connect_warning.log"
     warning_log_path = os.path.join(output_path, warning_log_fn)
 
-    if os.path.exists(voiceDir) == False:
+    if not os.path.exists(voiceDir):
         return jsonify({"message": log_warning_return_str(
             f'Voice directory {voiceDir} not found at {output_path}')}), 404
 
     ret = voiceConnect(app.logger, voiceDir, voice_connect_path, warning_log_path)
-    if ret == True:
+    if ret:
         return jsonify({"message": log_info_return_str(
             f"Voice connect {voice_connect_fn} successfully."),
             "video_id": video_id}), 200
@@ -419,6 +420,7 @@ def voice_connect_log_serve(video_id):
     return jsonify({"message": log_warning_return_str(
         f'Voice connect {warning_log_path} not found at {warning_log_path}')}), 404
 
+
 @app.route('/voice_connect/<video_id>', methods=['GET'])
 def voice_connect_serve(video_id):
     voice_connect_fn = video_id + "_zh.wav"
@@ -438,37 +440,38 @@ def tts(video_id):
     video_id = data['video_id']
     srt_fn = f'{video_id}_zh_merged.srt'
     srt_path = os.path.join(output_path, srt_fn)
-    tts_dir = os.path.join(output_path, video_id+"_zh_source")
+    tts_dir = os.path.join(output_path, video_id + "_zh_source")
     charater = data['tts_character']
 
-    if os.path.exists(srt_path) == False:
+    if not os.path.exists(srt_path):
         return jsonify({"message": log_warning_return_str(
             f'Chinese SRT {srt_fn} not found at {output_path}')}), 404
 
-    if os.path.exists(tts_dir) == True:
+    if os.path.exists(tts_dir):
         # delete old tts dir
         shutil.rmtree(tts_dir)
-    
+
     try:
         ret = srtToVoiceEdge(app.logger, srt_path, tts_dir, charater)
-        if ret == True:
+        if ret:
             return jsonify({"message": log_info_return_str(
-                    f"tts success."),
-                    "video_id": video_id}), 200
+                "tts success."),
+                "video_id": video_id}), 200
         else:
             return jsonify({"message": log_warning_return_str("tts failed.")}), 404
     except Exception as e:
         print(e)
-   
+
     return jsonify({"message": log_warning_return_str("tts failed.")}), 404
 
+
 @app.route('/tts/<video_id>', methods=['GET'])
 def tts_serve(video_id):
     tts_dir = os.path.join(output_path, video_id + "_zh_source")
     tts_zip_fn = video_id + "_zh_source.zip"
     tts_zip_path = os.path.join(output_path, tts_zip_fn)
     print("tts_dir", tts_dir)
-    if os.path.exists(tts_dir) == False:
+    if not os.path.exists(tts_dir):
         return jsonify({"message": log_warning_return_str(
             f'Voice directory {tts_dir} not found at {output_path}')}), 404
 
@@ -502,17 +505,16 @@ def video_preview(video_id):
     video_out_path = os.path.join(output_path, video_out_fn)
 
     # 检查音频
-    if os.path.exists(voice_connect_path) == False or os.path.exists(audio_bg_path) == False:
+    if (not os.path.exists(voice_connect_path)) or (not os.path.exists(audio_bg_path)):
         return jsonify({"message": log_warning_return_str(
             f'Chinese Voice {voice_connect_fn} not found at {output_path}')}), 404
 
     # 检查视频
-    if os.path.exists(video_save_path) == False and os.path.exists(video_fhd_save_path) == False:
+    if (not os.path.exists(video_save_path)) and (not os.path.exists(video_fhd_save_path)):
         return jsonify({"message": log_warning_return_str(
-            f"No video found")}), 404
+            "No video found")}), 404
 
     # 选择最佳分辨率的视频
-    video_source_path = ""
     if os.path.exists(video_fhd_save_path):
         video_source_path = video_fhd_save_path
     else:
@@ -521,7 +523,7 @@ def video_preview(video_id):
     # 生成视频预览
     ret = zhVideoPreview(app.logger, video_source_path, voice_connect_path, audio_bg_path,
                          "暂时没有处理字幕文件，所以随便写", video_out_path)
-    if ret == True:
+    if ret:
         return jsonify({"message": log_info_return_str(
             f"Video preview {video_fhd} successfully."),
             "video_id": video_id}), 200

requirements.txt

@@ -60,7 +60,6 @@ llvmlite==0.42.0
 MarkupSafe==2.1.5
 matplotlib-inline==0.1.6
 mistune==3.0.2
-mkl-fft==1.3.8
 mkl-random==1.2.4
 mkl-service==2.4.0
 more-itertools==10.2.0

tools/audio_remove.py

@@ -1,5 +1,3 @@
-import os
-
 import librosa
 import numpy as np
 import soundfile as sf
@@ -9,13 +7,12 @@
 from lib import dataset
 from lib import nets
 from lib import spec_utils
-from lib import utils
-
 
 AUDIO_REMOVE_DEVICE = "gpu"
 AUDIO_REMOVE_FFT_SIZE = 2048
 AUDIO_REMOVE_HOP_SIZE = 1024
 
+
 class Separator(object):
 
     def __init__(self, model, device=None, batchsize=1, cropsize=256, postprocess=False):
@@ -112,9 +109,9 @@ def audio_remove(audioFileNameAndPath, voiceFileNameAndPath, instrumentFileNameA
         device = device = torch.device('cuda:0')
     else:
         raise ValueError("Invalid device: {}".format(AUDIO_REMOVE_DEVICE))
-    
+
     print("Loading model " + AUDIO_REMOVE_DEVICE)
-    model = nets.CascadedNet(AUDIO_REMOVE_FFT_SIZE, AUDIO_REMOVE_HOP_SIZE, 32, 128)#模型参数
+    model = nets.CascadedNet(AUDIO_REMOVE_FFT_SIZE, AUDIO_REMOVE_HOP_SIZE, 32, 128)  # 模型参数
     model.load_state_dict(torch.load(modelNameAndPath, map_location='cpu'))
     model.to(device)
     print("Model loaded")
@@ -151,8 +148,9 @@ def audio_remove(audioFileNameAndPath, voiceFileNameAndPath, instrumentFileNameA
     wave = spec_utils.spectrogram_to_wave(v_spec, hop_length=AUDIO_REMOVE_HOP_SIZE)
     print('done')
     sf.write(voiceFileNameAndPath, wave.T, sr)
-
 
 
 if __name__ == '__main__':
-    audio_remove("d:\\document\\AI_Work\\whisper\\videos\\proxy\\RXXRguaHZs0.wav", "d:\\document\\AI_Work\\whisper\\videos\\proxy\\RXXRguaHZs0_voice.wav", "d:\\document\\AI_Work\\whisper\\videos\\proxy\\RXXRguaHZs0_instrument.wav")
+    audio_remove("d:\\document\\AI_Work\\whisper\\videos\\proxy\\RXXRguaHZs0.wav",
+                 "d:\\document\\AI_Work\\whisper\\videos\\proxy\\RXXRguaHZs0_voice.wav",
+                 "d:\\document\\AI_Work\\whisper\\videos\\proxy\\RXXRguaHZs0_instrument.wav")

tools/merge_subtitle.py

@@ -1,13 +1,14 @@
 import pysubs2
 
+
 class SubtitleMerger:
-    def __init__(self, 
-                 chinese_sub_path, 
-                 english_sub_path, 
+    def __init__(self,
+                 chinese_sub_path,
+                 english_sub_path,
                  output_path,
-                 chinese_font_name="Arial", 
+                 chinese_font_name="Arial",
                  chinese_font_size=10,
-                 english_font_name="Arial", 
+                 english_font_name="Arial",
                  english_font_size=6):
         self.chinese_sub_path = chinese_sub_path
         self.english_sub_path = english_sub_path
@@ -26,7 +27,7 @@ def merge_subtitles(self):
         # Define subtitle styles
         chinese_style = 'ChineseStyle'
         merged_subs.styles[chinese_style] = pysubs2.SSAStyle(
-            fontname=self.chinese_font_name, 
+            fontname=self.chinese_font_name,
             fontsize=self.chinese_font_size,
             primarycolor=pysubs2.Color(255, 255, 255),
             outlinecolor=pysubs2.Color(0, 0, 0),
@@ -57,15 +58,15 @@ def merge_subtitles(self):
 
 def main():
     # Example usage
-    merger = SubtitleMerger(chinese_sub_path=r"C:\Users\le\Videos\pytvzhen\conver\eMlx5fFNoYc\eMlx5fFNoYc_zh_merge.srt", 
+    merger = SubtitleMerger(chinese_sub_path=r"C:\Users\le\Videos\pytvzhen\conver\eMlx5fFNoYc\eMlx5fFNoYc_zh_merge.srt",
                             english_sub_path=r"C:\Users\le\Videos\pytvzhen\conver\eMlx5fFNoYc\eMlx5fFNoYc_en_merge.srt",
                             output_path=r"C:\Users\le\Videos\pytvzhen\conver\eMlx5fFNoYc\merged_subtitles.ass",
-                            chinese_font_name="Arial", 
+                            chinese_font_name="Arial",
                             chinese_font_size=10,
-                            english_font_name="Arial", 
+                            english_font_name="Arial",
                             english_font_size=6)
     merger.merge_subtitles()
 
 
-if __name__=="__main__":
-    main()
+if __name__ == "__main__":
+    main()

tools/merge_video_srt.py

@@ -1,13 +1,13 @@
 import subprocess
 
-def add_subtitles_and_mix_audio(input_video, 
-                                input_subtitle, 
-                                input_audio1, 
-                                input_audio2, 
+
+def add_subtitles_and_mix_audio(input_video,
+                                input_subtitle,
+                                input_audio1,
+                                input_audio2,
                                 output_video):
     """
     使用 FFmpeg 将字幕和两个音频文件合成到视频文件中。
-    
     参数:
       input_video (str): 输入视频文件的路径。
       input_subtitle (str): 字幕文件（ASS格式）的路径。
@@ -45,12 +45,12 @@ def add_subtitles_and_mix_audio(input_video,
 
 def main():
     # 使用示例
-    add_subtitles_and_mix_audio(input_video='eMlx5fFNoYc_fhd.mp4', 
-                                input_subtitle='merged_subtitles.ass', 
-                                input_audio1='eMlx5fFNoYc_insturment.wav', 
-                                input_audio2='eMlx5fFNoYc_zh.wav', 
+    add_subtitles_and_mix_audio(input_video='eMlx5fFNoYc_fhd.mp4',
+                                input_subtitle='merged_subtitles.ass',
+                                input_audio1='eMlx5fFNoYc_insturment.wav',
+                                input_audio2='eMlx5fFNoYc_zh.wav',
                                 output_video='eMlx5fFNoYc_fhd_merge.mp4')
 
 
 if __name__ == "__main__":
-    main()
+    main()