Spaces:
Sleeping
Sleeping
| import openai | |
| from pytube import YouTube | |
| import argparse | |
| import os | |
| from pathlib import Path | |
| from tqdm import tqdm | |
| from src.srt_util.srt import SrtScript | |
| from src.Pigeon import Pigeon | |
| import stable_whisper | |
| import whisper | |
| from src.srt_util import srt2ass | |
| import logging | |
| from datetime import datetime | |
| import torch | |
| import subprocess | |
| import time | |
| def parse_args(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument("--link", help="youtube video link here", default=None, type=str, required=False) | |
| parser.add_argument("--video_file", help="local video path here", default=None, type=str, required=False) | |
| parser.add_argument("--audio_file", help="local audio path here", default=None, type=str, required=False) | |
| parser.add_argument("--srt_file", help="srt file input path here", default=None, type=str, | |
| required=False) # New argument | |
| parser.add_argument("--download", help="download path", default='./downloads', type=str, required=False) | |
| parser.add_argument("--output_dir", help="translate result path", default='./results', type=str, required=False) | |
| parser.add_argument("--video_name", | |
| help="video name, if use video link as input, the name will auto-filled by youtube video name", | |
| default='placeholder', type=str, required=False) | |
| parser.add_argument("--model_name", help="model name only support gpt-4 and gpt-3.5-turbo", type=str, | |
| required=False, default="gpt-4") # default change to gpt-4 | |
| parser.add_argument("--log_dir", help="log path", default='./logs', type=str, required=False) | |
| parser.add_argument("-only_srt", help="set script output to only .srt file", action='store_true') | |
| parser.add_argument("-v", help="auto encode script with video", action='store_true') | |
| args = parser.parse_args() | |
| return args | |
| def get_sources(args, download_path, result_path, video_name): | |
| # get source audio | |
| audio_path = None | |
| audio_file = None | |
| video_path = None | |
| if args.link is not None and args.video_file is None: | |
| # Download audio from YouTube | |
| video_link = args.link | |
| video = None | |
| audio = None | |
| try: | |
| yt = YouTube(video_link,use_oauth=True, allow_oauth_cache=True) | |
| video = yt.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc().first() | |
| if video: | |
| video.download(f'{download_path}/video') | |
| print('Video download completed!') | |
| else: | |
| print("Error: Video stream not found") | |
| audio = yt.streams.filter(only_audio=True, file_extension='mp4').first() | |
| if audio: | |
| audio.download(f'{download_path}/audio') | |
| print('Audio download completed!') | |
| else: | |
| print("Error: Audio stream not found") | |
| except Exception as e: | |
| print("Connection Error") | |
| print(e) | |
| exit() | |
| video_path = f'{download_path}/video/{video.default_filename}' | |
| audio_path = '{}/audio/{}'.format(download_path, audio.default_filename) | |
| audio_file = open(audio_path, "rb") | |
| if video_name == 'placeholder': | |
| video_name = audio.default_filename.split('.')[0] | |
| elif args.video_file is not None: | |
| # Read from local | |
| video_path = args.video_file | |
| if args.audio_file is not None: | |
| audio_file = open(args.audio_file, "rb") | |
| audio_path = args.audio_file | |
| else: | |
| output_audio_path = f'{download_path}/audio/{video_name}.mp3' | |
| subprocess.run(['ffmpeg', '-i', video_path, '-f', 'mp3', '-ab', '192000', '-vn', output_audio_path]) | |
| audio_file = open(output_audio_path, "rb") | |
| audio_path = output_audio_path | |
| if not os.path.exists(f'{result_path}/{video_name}'): | |
| os.mkdir(f'{result_path}/{video_name}') | |
| if args.audio_file is not None: | |
| audio_file = open(args.audio_file, "rb") | |
| audio_path = args.audio_file | |
| pass | |
| return audio_path, audio_file, video_path, video_name | |
| def get_srt_class(srt_file_en, result_path, video_name, audio_path, audio_file=None, whisper_model='large', | |
| method="stable"): | |
| # Instead of using the script_en variable directly, we'll use script_input | |
| if srt_file_en is not None: | |
| srt = SrtScript.parse_from_srt_file(srt_file_en) | |
| else: | |
| # using whisper to perform speech-to-text and save it in <video name>_en.txt under RESULT PATH. | |
| srt_file_en = "{}/{}/{}_en.srt".format(result_path, video_name, video_name) | |
| if not os.path.exists(srt_file_en): | |
| devices = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") | |
| # use OpenAI API for transcribe | |
| if method == "api": | |
| transcript = openai.Audio.transcribe("whisper-1", audio_file) | |
| # use local whisper model | |
| elif method == "basic": | |
| model = whisper.load_model(whisper_model, | |
| device=devices) # using base model in local machine (may use large model on our server) | |
| transcript = model.transcribe(audio_path) | |
| srt = SRT_script(transcript['segments']) # read segments to SRT class | |
| # use stable-whisper | |
| elif method == "stable": | |
| # use cuda if available | |
| model = stable_whisper.load_model(whisper_model, device=devices) | |
| transcript = model.transcribe(audio_path, regroup=False, | |
| initial_prompt="Hello, welcome to my lecture. Are you good my friend?") | |
| ( | |
| transcript | |
| .split_by_punctuation(['.', '。', '?']) | |
| .merge_by_gap(.15, max_words=3) | |
| .merge_by_punctuation([' ']) | |
| .split_by_punctuation(['.', '。', '?']) | |
| ) | |
| transcript = transcript.to_dict() | |
| srt = SRT_script(transcript['segments']) # read segments to SRT class | |
| else: | |
| raise ValueError("invalid speech to text method") | |
| srt = SrtScript(transcript['segments']) # read segments to SRT class | |
| else: | |
| srt = SrtScript.parse_from_srt_file(srt_file_en) | |
| return srt_file_en, srt | |
| # Split the video script by sentences and create chunks within the token limit | |
| def script_split(script_in, chunk_size=1000): | |
| script_split = script_in.split('\n\n') | |
| script_arr = [] | |
| range_arr = [] | |
| start = 1 | |
| end = 0 | |
| script = "" | |
| for sentence in script_split: | |
| if len(script) + len(sentence) + 1 <= chunk_size: | |
| script += sentence + '\n\n' | |
| end += 1 | |
| else: | |
| range_arr.append((start, end)) | |
| start = end + 1 | |
| end += 1 | |
| script_arr.append(script.strip()) | |
| script = sentence + '\n\n' | |
| if script.strip(): | |
| script_arr.append(script.strip()) | |
| range_arr.append((start, len(script_split) - 1)) | |
| assert len(script_arr) == len(range_arr) | |
| return script_arr, range_arr | |
| def check_translation(sentence, translation): | |
| """ | |
| check merge sentence issue from openai translation | |
| """ | |
| sentence_count = sentence.count('\n\n') + 1 | |
| translation_count = translation.count('\n\n') + 1 | |
| if sentence_count != translation_count: | |
| # print("sentence length: ", len(sentence), sentence_count) | |
| # print("translation length: ", len(translation), translation_count) | |
| return False | |
| else: | |
| return True | |
| def get_response(model_name, sentence): | |
| """ | |
| Generates a translated response for a given sentence using a specified OpenAI model. | |
| Args: | |
| model_name (str): The name of the OpenAI model to be used for translation, either "gpt-3.5-turbo" or "gpt-4". | |
| sentence (str): The English sentence related to StarCraft 2 videos that needs to be translated into Chinese. | |
| Returns: | |
| str: The translated Chinese sentence, maintaining the original format, meaning, and number of lines. | |
| """ | |
| if model_name == "gpt-3.5-turbo" or model_name == "gpt-4": | |
| response = openai.ChatCompletion.create( | |
| model=model_name, | |
| messages=[ | |
| # {"role": "system", "content": "You are a helpful assistant that translates English to Chinese and have decent background in starcraft2."}, | |
| # {"role": "system", "content": "Your translation has to keep the orginal format and be as accurate as possible."}, | |
| # {"role": "system", "content": "Your translation needs to be consistent with the number of sentences in the original."}, | |
| # {"role": "system", "content": "There is no need for you to add any comments or notes."}, | |
| # {"role": "user", "content": 'Translate the following English text to Chinese: "{}"'.format(sentence)} | |
| {"role": "system", | |
| "content": "你是一个翻译助理，你的任务是翻译星际争霸视频，你会被提供一个按行分割的英文段落，你需要在保证句意和行数的情况下输出翻译后的文本。"}, | |
| {"role": "user", "content": sentence} | |
| ], | |
| temperature=0.15 | |
| ) | |
| return response['choices'][0]['message']['content'].strip() | |
| # Translate and save | |
| def translate(srt, script_arr, range_arr, model_name, video_name, video_link, attempts_count=5): | |
| """ | |
| Translates the given script array into another language using the chatgpt and writes to the SRT file. | |
| This function takes a script array, a range array, a model name, a video name, and a video link as input. It iterates | |
| through sentences and range in the script and range arrays. If the translation check fails for five times, the function | |
| will attempt to resolve merge sentence issues and split the sentence into smaller tokens for a better translation. | |
| Args: | |
| srt (Subtitle): An instance of the Subtitle class representing the SRT file. | |
| script_arr (list): A list of strings representing the original script sentences to be translated. | |
| range_arr (list): A list of tuples representing the start and end positions of sentences in the script. | |
| model_name (str): The name of the translation model to be used. | |
| video_name (str): The name of the video. | |
| video_link (str): The link to the video. | |
| attempts_count (int): Number of attemps of failures for unmatched sentences. | |
| """ | |
| logging.info("start translating...") | |
| previous_length = 0 | |
| for sentence, range in tqdm(zip(script_arr, range_arr)): | |
| # update the range based on previous length | |
| range = (range[0] + previous_length, range[1] + previous_length) | |
| # using chatgpt model | |
| print(f"now translating sentences {range}") | |
| logging.info(f"now translating sentences {range}, time: {datetime.now()}") | |
| flag = True | |
| while flag: | |
| flag = False | |
| try: | |
| translate = get_response(model_name, sentence) | |
| # detect merge sentence issue and try to solve for five times: | |
| while not check_translation(sentence, translate) and attempts_count > 0: | |
| translate = get_response(model_name, sentence) | |
| attempts_count -= 1 | |
| # if failure still happen, split into smaller tokens | |
| if attempts_count == 0: | |
| single_sentences = sentence.split("\n\n") | |
| logging.info("merge sentence issue found for range", range) | |
| translate = "" | |
| for i, single_sentence in enumerate(single_sentences): | |
| if i == len(single_sentences) - 1: | |
| translate += get_response(model_name, single_sentence) | |
| else: | |
| translate += get_response(model_name, single_sentence) + "\n\n" | |
| # print(single_sentence, translate.split("\n\n")[-2]) | |
| logging.info("solved by individually translation!") | |
| except Exception as e: | |
| logging.debug("An error has occurred during translation:", e) | |
| print("An error has occurred during translation:", e) | |
| print("Retrying... the script will continue after 30 seconds.") | |
| time.sleep(30) | |
| flag = True | |
| srt.set_translation(translate, range, model_name, video_name, video_link) | |
| def main_old(): | |
| args = parse_args() | |
| # input check: input should be either video file or youtube video link. | |
| if args.link is None and args.video_file is None and args.srt_file is None and args.audio_file is None: | |
| raise TypeError("need video source or srt file") | |
| # set up | |
| start_time = time.time() | |
| openai.api_key = os.getenv("OPENAI_API_KEY") | |
| DOWNLOAD_PATH = Path(args.download) | |
| if not DOWNLOAD_PATH.exists(): | |
| DOWNLOAD_PATH.mkdir(parents=False, exist_ok=False) | |
| DOWNLOAD_PATH.joinpath('audio').mkdir(parents=False, exist_ok=False) | |
| DOWNLOAD_PATH.joinpath('video').mkdir(parents=False, exist_ok=False) | |
| RESULT_PATH = Path(args.output_dir) | |
| if not RESULT_PATH.exists(): | |
| RESULT_PATH.mkdir(parents=False, exist_ok=False) | |
| # set video name as the input file name if not specified | |
| if args.video_name == 'placeholder': | |
| # set video name to upload file name | |
| if args.video_file is not None: | |
| VIDEO_NAME = args.video_file.split('/')[-1].split('.')[0] | |
| elif args.audio_file is not None: | |
| VIDEO_NAME = args.audio_file.split('/')[-1].split('.')[0] | |
| elif args.srt_file is not None: | |
| VIDEO_NAME = args.srt_file.split('/')[-1].split('.')[0].split("_")[0] | |
| else: | |
| VIDEO_NAME = args.video_name | |
| else: | |
| VIDEO_NAME = args.video_name | |
| audio_path, audio_file, video_path, VIDEO_NAME = get_sources(args, DOWNLOAD_PATH, RESULT_PATH, VIDEO_NAME) | |
| if not os.path.exists(args.log_dir): | |
| os.makedirs(args.log_dir) | |
| logging.basicConfig(level=logging.INFO, handlers=[ | |
| logging.FileHandler("{}/{}_{}.log".format(args.log_dir, VIDEO_NAME, datetime.now().strftime("%m%d%Y_%H%M%S")), | |
| 'w', encoding='utf-8')]) | |
| logging.info("---------------------Video Info---------------------") | |
| logging.info("Video name: {}, translation model: {}, video link: {}".format(VIDEO_NAME, args.model_name, args.link)) | |
| srt_file_en, srt = get_srt_class(args.srt_file, RESULT_PATH, VIDEO_NAME, audio_path, audio_file, method="api") | |
| # SRT class preprocess | |
| logging.info("---------------------Start Preprocessing SRT class---------------------") | |
| srt.write_srt_file_src(srt_file_en) | |
| srt.form_whole_sentence() | |
| # srt.spell_check_term() | |
| # srt.correct_with_force_term() | |
| processed_srt_file_en = srt_file_en.split('.srt')[0] + '_processed.srt' | |
| srt.write_srt_file_src(processed_srt_file_en) | |
| script_input = srt.get_source_only() | |
| # write ass | |
| if not args.only_srt: | |
| logging.info("write English .srt file to .ass") | |
| assSub_en = srt2ass(processed_srt_file_en, "default", "No", "Modest") | |
| logging.info('ASS subtitle saved as: ' + assSub_en) | |
| script_arr, range_arr = script_split(script_input) | |
| logging.info("---------------------Start Translation--------------------") | |
| translate(srt, script_arr, range_arr, args.model_name, VIDEO_NAME, args.link) | |
| # SRT post-processing | |
| logging.info("---------------------Start Post-processing SRT class---------------------") | |
| srt.check_len_and_split() | |
| srt.remove_trans_punctuation() | |
| srt.write_srt_file_translate(f"{RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}_zh.srt") | |
| srt.write_srt_file_bilingual(f"{RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}_bi.srt") | |
| # write ass | |
| if not args.only_srt: | |
| logging.info("write Chinese .srt file to .ass") | |
| assSub_zh = srt2ass(f"{RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}_zh.srt", "default", "No", "Modest") | |
| logging.info('ASS subtitle saved as: ' + assSub_zh) | |
| # encode to .mp4 video file | |
| if args.v: | |
| logging.info("encoding video file") | |
| if args.only_srt: | |
| os.system( | |
| f'ffmpeg -i {video_path} -vf "subtitles={RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}_zh.srt" {RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}.mp4') | |
| else: | |
| os.system( | |
| f'ffmpeg -i {video_path} -vf "subtitles={RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}_zh.ass" {RESULT_PATH}/{VIDEO_NAME}/{VIDEO_NAME}.mp4') | |
| end_time = time.time() | |
| logging.info( | |
| "Pipeline finished, time duration:{}".format(time.strftime("%H:%M:%S", time.gmtime(end_time - start_time)))) | |
| def main(): | |
| pigeon = Pigeon() | |
| pigeon.run() | |
| if __name__ == "__main__": | |
| main() | |