Update app.py

app.py

@@ -7,8 +7,8 @@ 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
+import arabic_reshaper  # pip install arabic-reshaper
+from bidi.algorithm import get_display  # pip install python-bidi
 from moviepy import VideoFileClip, TextClip, CompositeVideoClip, AudioFileClip, ImageClip
 import pysrt
 import instaloader
@@ -17,10 +17,10 @@ import concurrent.futures
 import re
 from io import BytesIO
 from PIL import Image
-api_key = "268976:66f4f58a2a905"
-
-
+import yt_dlp
+import os
 
+api_key = "268976:66f4f58a2a905"
 
 def fetch_data(url):
     try:
@@ -70,9 +70,9 @@ def merge_files(filename, num_parts):
 def download_file_in_parallel(link, size, num_threads=4):
     url = link.split("#")[0]
     filename = link.split("#")[1]
-    print(url+" filename: "+filename)
+    print(url + " filename: " + filename)
     response = requests.head(url)
-    #file_size = int(response.headers['Content-Length'])
+    # 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)]
@@ -89,88 +89,15 @@ def download_file_in_parallel(link, size, num_threads=4):
     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 download_image(url, save_path='downloaded_image.jpg'): 
-    response = requests.get(url) 
-    image = Image.open(BytesIO(response.content)) 
-    image.save(save_path) 
-    return save_path
-    
 def insta_oneapi(url, api_key):
     shortcode = url.split("/")[-2]
     print(shortcode)
-    url_one="https://api.one-api.ir/instagram/v1/post/?shortcode="+shortcode
-    request_body = [{"shortcode": 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)
@@ -182,7 +109,7 @@ def insta_oneapi(url, api_key):
                         file.write(chunk)
             file.close()
             print(f"Downloaded successfully")
-            image_url = result["result"]['media'][0]["cover"] 
+            image_url = result["result"]['media'][0]["cover"]
             image_file_path = download_image(image_url)
             return "video.mp4", image_file_path
         except requests.exceptions.RequestException as e:
@@ -191,48 +118,11 @@ def insta_oneapi(url, api_key):
         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 download_image(url, save_path='downloaded_image.jpg'):
+    response = requests.get(url)
+    image = Image.open(BytesIO(response.content))
+    image.save(save_path)
+    return save_path
 
 def extract_audio(input_video_name):
     # Define the input video file and output audio file
@@ -278,7 +168,7 @@ def generate_subtitle_file(language, segments, input_video_name):
     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"{str(index + 1)} \n"
         text += f"{segment_start} --> {segment_end} \n"
         text += f"{segment.text} \n"
         text += "\n"
@@ -312,7 +202,7 @@ def enhance_text(api_key, text, google):
     request_body = [{
         "role": "user",
         "content": f"{text} ROLE: Expert translator for Persian motivational anime subtitles. TASK: Translate input English SRT (`[X.XXs -> Y.YYs] English text`) to Persian SRT. REQUIREMENTS: 1. STRICT FORMAT: Output MUST be `[X.XXs -> Y.YYs] Persian Text` per line. 2. TIMESTAMPS: CRITICAL - NEVER change timestamps; copy EXACTLY. 3. TRANSLATION: Create motivational, touching, natural, understandable Persian; capture spirit, avoid literalism. 4. LENGTH: Ensure Persian text is concise & readable in time. AVOID: Format errors, timestamp changes, unnatural translation, overly long text.  in respose dont add any thing exept for the srt formated translation.Now translate:"
-    },]
+    }, ]
 
     # Add the API key to the request
     headers = {
@@ -360,46 +250,73 @@ def time_to_seconds(time_obj):
 def create_subtitle_clips(subtitles, videosize, fontsize, font, color, debug):
     subtitle_clips = []
     for subtitle in subtitles:
-        start_time = time_to_seconds(subtitle.start) # Add 2 seconds offset
+        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_width = video_width * 0.8
         max_height = video_height * 0.2
-        #reshaped_text = arabic_reshaper.reshape(subtitle.text)
-        #bidi_text = get_display(reshaped_text)
-        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)
+        # reshaped_text = arabic_reshaper.reshape(subtitle.text)
+        # bidi_text = get_display(reshaped_text)
+        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)
         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))
     return subtitle_clips
 
+def download_youtube(url, save_path):
+    """
+    Downloads a YouTube video to the specified save path using yt_dlp.
 
+    Args:
+        url (str): The URL of the YouTube video to download.
+        save_path (str): The directory where the video will be saved.
 
-def process_video(url, type):
-
+    Returns:
+        str: The full path to the downloaded video file.
+    """
+    ydl_opts = {
+        'format': 'best',  # Downloads the best quality format with both video and audio
+        'outtmpl': os.path.join(save_path, '%(title)s.%(ext)s'),  # Saves with title and extension
+    }
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            # Get video info without downloading to determine the file name
+            info = ydl.extract_info(url, download=False)
+            file_path = ydl.prepare_filename(info)
+            # Download the video
+            ydl.download([url])
+        print(f'Video downloaded successfully to {file_path}')
+        return file_path
+    except Exception as e:
+        print(f"Error downloading video: {e}")
+        return None
 
-    if type=="insta":
-        input_video, image_path=insta_oneapi(url, api_key)
+def process_video(url, type):
+    if type == "insta":
+        input_video, image_path = insta_oneapi(url, api_key)
         input_video_name = input_video.replace(".mp4", "")
         video = VideoFileClip(input_video)
         image_clip = ImageClip(image_path).with_duration(1)
-        # Set the position and size of the image (optional) 
         image_clip = image_clip.with_position(("center", "center")).resized(height=video.size[1])
         first_video = CompositeVideoClip([video.with_start(1), image_clip])
-        input_video = input_video_name+"_cover.mp4"
-        input_video_name = input_video.replace(".mp4", "")
+        input_video = input_video_name + "_cover.mp4"
         first_video.write_videofile(input_video, codec="libx264", audio_codec="aac", logger=None)
         input_audio = extract_audio(input_video)
-    elif type=="youtube":
-        input_video, input_audio = one_youtube(url, api_key)
-        input_video_name = input_video.replace(".mp4", "")
+    elif type == "youtube":
+        input_video = download_youtube(url, ".")
+        if input_video is None:
+            raise ValueError("Failed to download YouTube video")
+        input_audio = extract_audio(input_video)
+        input_video_name = os.path.splitext(input_video)[0]
+
     # Get the current local time
     t = time.localtime()
-    # Format the time as a string
     current_time = time.strftime("%H:%M:%S", t)
     print("Current Time =", current_time)
+
     segments = transcribe(audio=input_audio)
     language = "fa"
     subtitle_file = generate_subtitle_file(language=language, segments=segments, input_video_name=input_video_name)
@@ -408,31 +325,24 @@ def process_video(url, type):
     srt_string = read_srt_file(subtitle_file)
     google_translate = translate_text(api_key, source_language, target_language, srt_string)
     write_google(google_translate)
+
+    # Load the video, which already has audio
     video = VideoFileClip(input_video)
-    audio = AudioFileClip(input_audio)
-    video = video.with_audio(audio)
-    print(video)
     subtitles = pysrt.open("google_translate.srt", encoding="utf-8")
     output_video_file = input_video_name + '_subtitled' + ".mp4"
     subtitle_clips = create_subtitle_clips(subtitles, video.size, 24, 'arial.ttf', 'yellow', False)
     final_video = CompositeVideoClip([video] + subtitle_clips)
     final_video.write_videofile(output_video_file, codec="libx264", audio_codec="aac", logger=None)
     video.close()
-    audio.close()
     print('final')
+
     # Get the current local time
     t = time.localtime()
-    
-    # Format the time as a string
     current_time = time.strftime("%H:%M:%S", t)
     print("Current Time =", current_time)
-    
-    # Generate the URL for the file
-    return output_video_file
 
-def download_file(file_path):
-    return gr.File.update(file_path)
+    return output_video_file
 
-iface = gr.Interface(fn=process_video, inputs=[gr.Text(),gr.Dropdown(["insta","youtube"])], outputs="file")
+iface = gr.Interface(fn=process_video, inputs=[gr.Text(), gr.Dropdown(["insta", "youtube"])], outputs="file")
 
 iface.launch(debug=True)