from pytubefix import YouTube from pytubefix.cli import on_progress import time import math import gradio as gr import ffmpeg from faster_whisper import WhisperModel import requests import json import arabic_reshaper # pip install arabic-reshaper from bidi.algorithm import get_display # pip install python-bidi from moviepy import VideoFileClip, TextClip, CompositeVideoClip, AudioFileClip, ColorClip import pysrt import instaloader import time import re import concurrent.futures import os api_key = "268976:66f4f58a2a905" def fetch_data(url): try: response = requests.get(url) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"An error occurred: {e}") return None def download_file(url): try: response = requests.get(url.split("#")[0], stream=True) response.raise_for_status() print(url.split("#")[1]) with open(url.split("#")[1], 'wb') as file: for chunk in response.iter_content(chunk_size=8192): if chunk: file.write(chunk) print(f"Downloaded successfully: {url.split('#')[1]}") except requests.exceptions.RequestException as e: print(f"An error occurred: {e}") def download_chunk(url, start, end, filename, index): headers = {'Range': f'bytes={start}-{end}'} response = requests.get(url, headers=headers, stream=True) response.raise_for_status() chunk_filename = f'{filename}.part{index}' with open(chunk_filename, 'wb') as file: for chunk in response.iter_content(chunk_size=8192): if chunk: file.write(chunk) return chunk_filename def merge_files(filename, num_parts): with open(filename, 'wb') as output_file: for i in range(num_parts): part_filename = f'{filename}.part{i}' with open(part_filename, 'rb') as part_file: output_file.write(part_file.read()) # Optionally, delete the part file after merging # os.remove(part_filename) def download_file_in_parallel(link, size, num_threads=4): url = link.split("#")[0] filename = link.split("#")[1] print(url+" filename: "+filename) response = requests.head(url) #file_size = int(response.headers['Content-Length']) chunk_size = size // num_threads ranges = [(i * chunk_size, (i + 1) * chunk_size - 1) for i in range(num_threads)] ranges[-1] = (ranges[-1][0], size - 1) # Adjust the last range to the end of the file with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: futures = [ executor.submit(download_chunk, url, start, end, filename, i) for i, (start, end) in enumerate(ranges) ] for future in concurrent.futures.as_completed(futures): future.result() # Ensure all threads complete merge_files(filename, num_threads) print(f'Downloaded successfully: {filename}') def one_youtube(link, api_key): # Fetch video ID video_id_url = f"https://one-api.ir/youtube/?token={api_key}&action=getvideoid&link={link}" video_data = fetch_data(video_id_url) if not video_data: return None, None video_id = video_data["result"] # Fetch video data filter_option = "" # Replace with your filter option video_data_url = f"https://youtube.one-api.ir/?token={api_key}&action=fullvideo&id={video_id}&filter={filter_option}" video_data_2 = fetch_data(video_data_url) if not video_data_2: return None, None formats_list = video_data_2["result"]["formats"] file_name = video_data_2["result"]["title"] video_name = f'{file_name}.mp4' audio_name = f'{file_name}.mp3' for f in formats_list: if f["format_note"] == "360p": download_id = f["id"] video_size = f["filesize"] for f in formats_list: if f["format_note"] == "medium": audio_id = f["id"] audio_size = f["filesize"] if not download_id or not audio_id: return None, None # Fetch video and audio links video_link_url = f"https://youtube.one-api.ir/?token={api_key}&action=download&id={download_id}" audio_link_url = f"https://youtube.one-api.ir/?token={api_key}&action=download&id={audio_id}" video_link_data = fetch_data(video_link_url) audio_link_data = fetch_data(audio_link_url) if not video_link_data or not audio_link_data: return None, None video_link = video_link_data["result"]["link"] audio_link = audio_link_data["result"]["link"] vid_str=video_link+"#"+video_name audio_str=audio_link+"#"+audio_name # Download video and audio files print(video_size , audio_size) download_file_in_parallel(vid_str, video_size) download_file_in_parallel(audio_str, audio_size) return video_name, audio_name # Define your functions here def yt_download(url): yt = YouTube(url) print(yt.title) video_path = f"{yt.title}.mp4" ys = yt.streams.get_highest_resolution() print(ys) ys.download() return video_path, yt.title def insta_oneapi(url, api_key): shortcode = url.split("/")[-1] print(shortcode) url_one="https://api.one-api.ir/instagram/v1/post/?shortcode="+shortcode request_body = [{"shortcode": shortcode},] headers = {"one-api-token": api_key, "Content-Type": "application/json"} response = requests.get(url_one, headers=headers) print(response) if response.status_code == 200: result = response.json() try: time.sleep(10) response = requests.get(result["result"]['media'][0]["url"], stream=True) response.raise_for_status() with open("video.mp4", 'wb') as file: for chunk in response.iter_content(chunk_size=8192): if chunk: file.write(chunk) print(f"Downloaded successfully") return "video.mp4" except requests.exceptions.RequestException as e: print(f"An error occurred: {e}") else: print(f"Error: {response.status_code}, {response.text}") return None def insta_download(permalink): # Create an instance of Instaloader L = instaloader.Instaloader() try: # Extract the shortcode from the permalink if "instagram.com/reel/" in permalink: shortcode = permalink.split("instagram.com/reel/")[-1].split("/")[0] elif "instagram.com/p/" in permalink: shortcode = permalink.split("instagram.com/p/")[-1].split("/")[0] else: raise ValueError("Invalid permalink format") # Load the post using the shortcode post = instaloader.Post.from_shortcode(L.context, shortcode) # Check if the post is a video if not post.is_video: raise ValueError("The provided permalink is not a video.") # Get the video URL video_url = post.video_url # Extract the filename from the URL filename = video_url.split("/")[-1] # Remove query parameters filename = filename.split("?")[0] # Download the video using requests response = requests.get(video_url, stream=True) response.raise_for_status() # Raise an error for bad responses # Save the content to a file with open(filename, 'wb') as file: for chunk in response.iter_content(chunk_size=8192): file.write(chunk) print(f"Downloaded video {filename} successfully.") return filename except Exception as e: print(f"Failed to download video from {permalink}: {e}") def extract_audio(input_video_name): # Define the input video file and output audio file mp3_file = "audio.mp3" # Load the video clip video_clip = VideoFileClip(input_video_name) # Extract the audio from the video clip audio_clip = video_clip.audio # Write the audio to a separate file audio_clip.write_audiofile(mp3_file) # Close the video and audio clips audio_clip.close() video_clip.close() print("Audio extraction successful!") return mp3_file def transcribe(audio): model = WhisperModel("tiny") segments, info = model.transcribe(audio) segments = list(segments) for segment in segments: print("[%.2fs -> %.2fs] %s" % (segment.start, segment.end, segment.text)) return segments def format_time(seconds): hours = math.floor(seconds / 3600) seconds %= 3600 minutes = math.floor(seconds / 60) seconds %= 60 milliseconds = round((seconds - math.floor(seconds)) * 1000) seconds = math.floor(seconds) formatted_time = f"{hours:02d}:{minutes:02d}:{seconds:01d},{milliseconds:03d}" return formatted_time def generate_subtitle_file(language, segments, input_video_name): subtitle_file = f"sub-{input_video_name}.{language}.srt" text = "" for index, segment in enumerate(segments): segment_start = format_time(segment.start) segment_end = format_time(segment.end) text += f"{str(index+1)} \n" text += f"{segment_start} --> {segment_end} \n" text += f"{segment.text} \n" text += "\n" f = open(subtitle_file, "w", encoding='utf8') f.write(text) f.close() return subtitle_file def read_srt_file(file_path): try: with open(file_path, 'r', encoding='utf-8') as file: srt_content = file.read() return srt_content except FileNotFoundError: print(f"The file {file_path} was not found.") except Exception as e: print(f"An error occurred: {e}") def write_srt(subtitle_text, output_file="edited_srt.srt"): with open(output_file+".srt", 'w', encoding="utf-8") as file: file.write(subtitle_text) return output_file+".srt" def generate_translated_subtitle(language, segments, input_video_name): input_video_name=input_video_name.split('/')[-1] subtitle_file = f"{input_video_name}.srt" text = "" lines = segments.split('\n') new_list = [item for item in lines if item != ''] segment_number = 1 for index, segment in enumerate(new_list): if (index+1) % 3 == 1 or (index+1)==1: text += f"{segment}\n" segment_number += 1 if (index+1) % 3 == 2 or (index+1)==2: text += segment + "\n" if (index+1) % 3 == 0: text += f"\u200F{segment}\n\n" with open(subtitle_file, "w", encoding='utf8') as f: f.write(text) return subtitle_file def clean_text(text): # Remove 'srt ' from the start of each line # Remove ''' from the start and end text = re.sub(r"^```|```$", '', text) text = re.sub(r'^srt', '', text, flags=re.MULTILINE) return text def split_srt_file(input_file, max_chars=3000): # Read the contents of the SRT file with open(input_file, 'r', encoding='utf-8') as file: content = file.read() # Split the content into individual subtitles subtitles = content.strip().split('\n\n') # Prepare to write the split files output_files = [] current_file_content = '' current_file_index = 1 for subtitle in subtitles: # Check if adding this subtitle would exceed the character limit if len(current_file_content) + len(subtitle) + 2 > max_chars: # +2 for \n\n # Write the current file output_file_name = f'split_{current_file_index}.srt' with open(output_file_name, 'w', encoding='utf-8') as output_file: output_file.write(current_file_content.strip()) output_files.append(output_file_name) # Prepare for the next file current_file_index += 1 current_file_content = subtitle + '\n\n' else: # If it fits, add the subtitle current_file_content += subtitle + '\n\n' # Write any remaining content to a new SRT file if current_file_content: output_file_name = f'split_{current_file_index}.srt' with open(output_file_name, 'w', encoding='utf-8') as output_file: output_file.write(current_file_content.strip()) output_files.append(output_file_name) return output_files def translate_text(api_key, source_lang, target_lang, text): url = "https://api.one-api.ir/translate/v1/google/" request_body = {"source": source_lang, "target": target_lang, "text": text} headers = {"one-api-token": api_key, "Content-Type": "application/json"} response = requests.post(url, headers=headers, json=request_body) if response.status_code == 200: result = response.json() enhanced_text = enhance_text(api_key, text, result['result']) return enhanced_text else: print(f"Error: {response.status_code}, {response.text}") return None def enhance_text(api_key, text): url = "https://api.one-api.ir/chatbot/v1/gpt4o/" # Prepare the request body request_body = [{ "role": "user", "content": "Translate the following English text into Persian, specifically for a voice-over. Prioritize a natural and engaging tone that resonates with a Persian-speaking audience. Make sure the translation is:Concise and Time-Conscious: It needs to be spoken fluently within approximately the same time it takes to say the original English. Avoid overly verbose phrasing.Clear and Understandable: Use vocabulary and sentence structures that are easily understood by a general Persian-speaking audience. Avoid jargon or overly technical terms unless absolutely necessary and always provide a culturally appropriate equivalent.Culturally Relevant: Adapt the translation to ensure it is culturally appropriate and resonates with Iranian sensibilities. Consider nuances of politeness, humor, and common expressions. Adapt phrases as necessary to better reflect the culture.Natural Sounding: Read the Persian translation aloud to ensure it sounds natural and flows well when spoken.Engaging: Use phrasing that keeps the listener interested and attentive. Consider using rhetorical devices or active voice where appropriate to make the delivery more impactful.[Optional: Include the intended audience demographics here, e.g., "Targeting a younger audience" or "Intended for a professional context". This helps tailor the language.][Optional: Mention the overall tone you're aiming for, e.g., "maintain a formal tone" or "use an informal and friendly tone".]Essentially, I need a translation that doesn't sound like a translation, but a message written originally in Persian that is easy to grasp and will keep the audience captivated.the text will send in next message . },] { "role": "assistant", "content": "okay" }, { "role": "user", "content": text } ] # Add the API key to the request headers = { "one-api-token": api_key, "Content-Type": "application/json" } # Make the POST request attempts = 0 max_attempts = 3 while attempts < max_attempts: response = requests.post(url, headers=headers, json=request_body) if response.status_code == 200: result = response.json() if result["status"] == 200: print("status: ", result["status"]) te = clean_text(result["result"][0]) print("result: ", te) return te else: print(f"Error: status {result['status']}, retrying in 30 seconds...") else: print(f"Error: {response.status_code}, {response.text}, retrying in 30 seconds...") attempts += 1 time.sleep(30) print("Error Max attempts reached. Could not retrieve a successful response.") return 0 def write_google(google_translate): google = "google_translate.srt" with open(google, 'a', encoding="utf-8") as f: f.write(google_translate) def time_to_seconds(time_obj): return time_obj.hours * 3600 + time_obj.minutes * 60 + time_obj.seconds + time_obj.milliseconds / 1000 def create_subtitle_clips(subtitles, videosize, fontsize, font, color, debug): subtitle_clips = [] color_clips=[] for subtitle in subtitles: start_time = time_to_seconds(subtitle.start) # Add 2 seconds offset end_time = time_to_seconds(subtitle.end) duration = end_time - start_time video_width, video_height = videosize max_width = video_width * 0.8 max_height = video_height * 0.2 text_clip = TextClip(font, subtitle.text, font_size=fontsize, size=(int(video_width * 0.8), int(video_height * 0.2)) ,text_align="center" ,color=color, method='caption').with_start(start_time).with_duration(duration) myclip = ColorClip(size=(int(video_width * 0.8), int(video_height * 0.2)) , color=(0, 0, 0)).with_opacity(0.4).with_start(start_time).with_duration(duration) subtitle_x_position = 'center' subtitle_y_position = video_height * 0.68 text_position = (subtitle_x_position, subtitle_y_position) subtitle_clips.append(text_clip.with_position(text_position)) color_clips.append(myclip.with_position(text_position)) return subtitle_clips, color_clips def process_video(video, url, type): if isinstance(video, str): input_video = video.split("/")[-1] print(input_video) input_audio = extract_audio(video) input_video_name = input_video.replace(".mp4", "") else: if type=="insta": input_video=insta_oneapi(url, api_key) input_video_name = input_video.replace(".mp4", "") input_audio = extract_audio(input_video) elif type=="youtube": #input_video, input_audio = one_youtube(url, api_key) input_video, title = yt_download(url) input_video_name = input_video.replace(".mp4", "") input_audio = extract_audio(input_video) segments = transcribe(audio=input_audio) language = "fa" subtitle_file = generate_subtitle_file(language=language, segments=segments, input_video_name=input_video_name) source_language = "en" target_language = "fa" #srt_string = read_srt_file(subtitle_file) srt_files=split_srt_file(subtitle_file) for i in srt_files: srt_string = read_srt_file(f"{i}") #google_translate = translate_text(api_key, source_language, target_language, srt_string) google_translate = enhance_text(api_key, srt_string) if google_translate == 0 : google_translate = translate_text(api_key, source_language, target_language, srt_string) write_google(google_translate) time.sleep(15) srt = read_srt_file("google_translate.srt") os.remove("google_translate.srt") return srt, video, input_audio def video_edit(srt, input_video, input_audio= 'audio.mp3'): input_video_name = input_video.replace(".mp4", "") srt_name=generate_translated_subtitle("fa", srt, input_video_name) return input_video, srt_name """ input_video_name = input_video.replace(''.mp4', '') video = VideoFileClip(input_video) audio = AudioFileClip(input_audio) video = video.with_audio(audio) print(video) output_video_file = input_video_name + '_subtitled' + '.mp4' write_srt(srt) subtitles = pysrt.open('edited_srt.srt', encoding='utf-8') subtitle_clips = create_subtitle_clips(subtitles, video.size, 32, 'arial.ttf', 'white', False) final_video = CompositeVideoClip([video] + subtitle_clips) final_video.write_videofile(output_video_file, codec='libx264', audio_codec='aac', logger=None) os.remove('google_translate.srt') print('final')""" with gr.Blocks() as demo: gr.Markdown("Start typing below and then click **Run** to see the output.") with gr.Row(): inp = gr.Textbox(placeholder="Enter URL or upload") drp = gr.Dropdown(["insta", "youtube"]) btn = gr.Button("transcribe") out = gr.Textbox(interactive=True) video_file_input = gr.Video(label="Upload Video File") video_path_output = gr.Textbox(label="Video Path", visible=False) audio_path_output = gr.Textbox(label="Video Path", visible=False) btn.click(fn=process_video, inputs=[video_file_input, inp, drp], outputs=[out, video_path_output, audio_path_output]) with gr.Row(): vid_out = gr.Video() srt_file = gr.File() btn2 = gr.Button("transcribe") gr.on( triggers=[btn2.click], fn=write_google, inputs=out, ).then(video_edit, [out, video_path_output, audio_path_output], outputs=[vid_out, srt_file]) demo.launch(debug=True)