Milestone5API.py

import os
import re
import sentencepiece
from pytube import YouTube
from youtube_transcript_api import YouTubeTranscriptApi
from transformers import MarianMTModel, MarianTokenizer
import torch
from TTS.api import TTS
from moviepy.editor import VideoFileClip
from pydub import AudioSegment
import subprocess
from moviepy.editor import *

def add_text_to_video(input_video_path, output_video_path, text):
    video = VideoFileClip(input_video_path)
    txt_clip = TextClip(text, fontsize=24, color='white').set_position('center').set_duration(video.duration)
    result = CompositeVideoClip([video, txt_clip])
    result.write_videofile(output_video_path, codec='libx264', audio_codec='aac', remove_temp=True)


def replace_video_audio(video_file_path, new_audio_file_path, output_video_file_path, translated_text,new_video_path):
    print("Loading video...")
    video = VideoFileClip(video_file_path)
    print("Video loaded successfully.")

    print("Loading new audio...")
    new_audio = AudioFileClip(new_audio_file_path)
    print("New audio loaded successfully.")

    print("Trimming audio...")
    new_audio = new_audio.subclip(0, video.duration)
    print("Audio trimmed successfully.")

    print("Setting audio to video...")
    video = video.set_audio(new_audio)
    print("Audio set to video successfully.")

    try:
        video.write_videofile(output_video_file_path, codec="libx264", audio_codec="aac", remove_temp=True)
        print("Video file written successfully.")
    except Exception as e:
        print("Error writing video file:", e)



def text_to_speech(translated_text_file_path, video_path):
    device = "cuda" if torch.cuda.is_available() else "cpu"
    tts = TTS("tts_models/fr/mai/tacotron2-DDC").to(device)
    output_audio_dir = 'CS370_Milestone5/audio'
    os.makedirs(output_audio_dir, exist_ok=True)

    file_name_only = os.path.splitext(os.path.basename(translated_text_file_path))[0]
    output_file_path = os.path.join(output_audio_dir, f"{file_name_only}.wav")

    with open(translated_text_file_path, 'r', encoding='utf-8') as file:
        translated_text = file.read()

    tts.tts_to_file(text=translated_text, file_path=output_file_path)

    output_video_path = f"CS370_Milestone5/videos/{file_name_only}_new.mp4"
    new_video_path = f"CS370_Milestone5/videos/{file_name_only}_new2.mp4"
  
    replace_video_audio(video_path, output_file_path, output_video_path,translated_text,new_video_path)
    return output_video_path

def translate_text_file(input_text_path, output_text_path, source_lang='en', target_lang='fr', model_name="Helsinki-NLP/opus-mt-en-fr", batch_size=8):
    model = MarianMTModel.from_pretrained(model_name)
    tokenizer = MarianTokenizer.from_pretrained(model_name)
    
    def translate_batch(model, tokenizer, sentences):
        sentences = [f"{source_lang}: {sentence}" for sentence in sentences]
        input_ids = tokenizer(sentences, return_tensors="pt", padding=True, truncation=True)["input_ids"]
        translation_ids = model.generate(input_ids)
        translated_texts = tokenizer.batch_decode(translation_ids, skip_special_tokens=True)
        return translated_texts

    def read_text_from_file(file_path):
        with open(file_path, 'r', encoding='utf-8') as file:
            text = file.readlines()
        return text

    def write_text_to_file(file_path, translated_texts):
        with open(file_path, 'w', encoding='utf-8') as file:
            file.writelines([f"{line}\n" for line in translated_texts])

    translated_lines = []

    input_lines = read_text_from_file(input_text_path)

    for i in range(0, len(input_lines), batch_size):
        batch = input_lines[i:i + batch_size]
        translated_batch = translate_batch(model, tokenizer, batch)
        translated_lines.extend(translated_batch)

    write_text_to_file(output_text_path, translated_lines)
    return translated_lines

def download_video_transcript(video_url, target_lang='fr'):
    try:
        video_id = re.search(r"(?<=v=)[\w-]+", video_url)
        if video_id:
            video_id = video_id.group()
            yt = YouTube(video_url)
            stream = yt.streams.get_highest_resolution()

            print(f'Downloading {yt.title}...')
            modified_title = yt.title.replace(" ", "_")
            download_path = 'CS370_Milestone5/videos'  # Updated path to CS370_Milestone5 repository
            captions_path = 'CS370_Milestone5/captions'  # Updated path to CS370_Milestone5 repository

            # Create directories if they don't exist
            os.makedirs(download_path, exist_ok=True)
            os.makedirs(captions_path, exist_ok=True)

            video_file = f'{download_path}/{modified_title}.mp4'
            stream.download(output_path=download_path, filename=modified_title + '.mp4')
            print('Download completed!')

            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
            print("Available Transcripts:")
            for transcript in transcript_list:
                print(f"Language: {transcript.language}")

            transcript = None

            for source_lang in ['en', 'auto']:
                try:
                    transcript = transcript_list.find_generated_transcript([source_lang]).fetch()
                    break  # Stop if found
                except Exception as e:
                    continue 

            original_captions = ""
            for i, line in enumerate(transcript):
                start_time = line['start']
                formatted_time = f"{int(start_time // 60):02d}:{int(start_time % 60):02d}"
                original_captions += f"{formatted_time} {line['text']}\n"

            original_filename = f'{captions_path}/{modified_title}_original.txt'
            with open(original_filename, 'w', encoding='utf-8') as file:
                file.write(original_captions)
            
            # Translation part
            translated_captions = translate_text_file(original_filename,
                                                      f'{captions_path}/{modified_title}_translated.txt',
                                                      source_lang=source_lang,
                                                      target_lang=target_lang)

            translated_text_filename = f'{captions_path}/{modified_title}_translated.txt'
            new_video_path = text_to_speech(translated_text_filename, video_file)
            return original_captions, translated_captions, original_filename, translated_text_filename, new_video_path
        else:
            print("Video ID not found in URL.")
            return None, None, None, None, None
    except Exception as e:
        print("Error:", e)
        return None, None, None, None, None