app.py

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, 'w', encoding="utf-8") as file:
        file.write(subtitle_text)
        
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": f"{text} Translate the above text into Persian, converting the English terms used in it into common Persian terms. in respose dont add any thing exept for the srt formated translation."
    },]

    # 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 = []
    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
        #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_opacity(0.8).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 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'):

    return input_video, 'edited_srt.srt'
    """    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)