media_utils.py

import gradio as gr
import subprocess
import os
from tqdm import tqdm
import time
from PIL import Image
from concurrent.futures import ThreadPoolExecutor
import torch

def get_device():
    if torch.cuda.is_available():
        return torch.device('cuda')
    else:
        return torch.device('cpu')

device = get_device()
print(f"Using device: {device}")

def get_unique_filename(base_name, extension):
    counter = 1
    unique_name = f"{base_name}{extension}"
    while os.path.exists(unique_name):
        unique_name = f"{base_name}_{counter}{extension}"
        counter += 1
    return unique_name

def combine_videos(video_files):
    if not video_files:
        return "No video files selected.", None
    output_file = get_unique_filename("combined_video", ".mp4")
    filelist_path = os.path.abspath("filelist.txt")
    with open(filelist_path, "w") as filelist:
        for video in video_files:
            video_path = os.path.abspath(video).replace('\\', '/')
            filelist.write(f"file '{video_path}'\n")
    
    command = [
        "ffmpeg", "-f", "concat", "-safe", "0", "-i", filelist_path,
        "-c", "copy", output_file
    ]
    
    with ThreadPoolExecutor() as executor:
        future = executor.submit(subprocess.run, command, text=True, capture_output=True)
        result = future.result()  # Waits for the command to complete and returns the result

    if result.returncode == 0:
        return f"Videos combined successfully into {output_file}", output_file
    else:
        return f"Error combining videos: {result.stderr}", None

def combine_audios(audio_files):
    if not audio_files:
        return "No audio files selected.", None
    output_file = get_unique_filename("combined_audio", ".mp3")
    filelist_path = os.path.abspath("filelist.txt")
    with open(filelist_path, "w") as filelist:
        for audio in audio_files:
            filelist.write(f"file '{os.path.abspath(audio).replace('\\', '/')}'\n")
    
    command = [
        "ffmpeg", "-f", "concat", "-safe", "0", "-i", filelist_path,
        "-c", "copy", output_file
    ]
    
    with ThreadPoolExecutor() as executor:
        future = executor.submit(subprocess.run, command, text=True, capture_output=True)
        result = future.result()  # Waits for the command to complete and returns the result

    if result.returncode == 0:
        return f"Audios combined successfully into {output_file}", output_file
    else:
        return f"Error combining audios: {result.stderr}", None

def combine_images(image_files):
    if not image_files:
        return "No image files selected.", None
    output_file = get_unique_filename("combined_image", ".mp4")
    command = ["convert"] + image_files + [output_file]
    
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    total_time = 0
    with tqdm(total=100, desc="Combining Images") as pbar:
        while True:
            output = process.stderr.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                total_time += 1
                pbar.update(1)
                time.sleep(0.1)

    return f"Images combined successfully into {output_file}", output_file

def adjust_speed(media_file, speed):
    if not media_file:
        return "No media file selected.", None
    output_file = get_unique_filename(f"adjusted_speed_{os.path.basename(media_file)}", ".mp4")
    command = [
        "ffmpeg", "-i", media_file, "-filter:v", f"setpts={1/speed}*PTS",
        "-filter:a", f"atempo={speed}", output_file
    ]
    
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    total_time = 0
    with tqdm(total=100, desc="Adjusting Speed") as pbar:
        while True:
            output = process.stderr.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                total_time += 1
                pbar.update(1)
                time.sleep(0.1)

    return f"Speed adjusted successfully to {speed}x in {output_file}", output_file

def adjust_speed_by_length(media_file, desired_length_hhmmss):
    if not media_file:
        return "No media file selected.", None
    original_length = get_media_length(media_file)
    desired_length = hhmmss_to_seconds(desired_length_hhmmss)
    speed = original_length / desired_length
    return adjust_speed(media_file, speed)

def adjust_speed_combined(media_file, speed, hours, minutes, seconds, compress):
    if hours or minutes or seconds:
        desired_length = f"{int(hours):02}:{int(minutes):02}:{int(seconds):02}"
        output_message, output_file = adjust_speed_by_length(media_file, desired_length)
    else:
        output_message, output_file = adjust_speed(media_file, speed)
    
    if compress and output_file:
        compressed_output_file = get_unique_filename(f"compressed_{os.path.basename(output_file)}", ".mp4")
        compress_video(output_file, compressed_output_file)
        output_message = f"{output_message} and compressed to {compressed_output_file}"
        output_file = compressed_output_file
    
    return output_message, output_file

def hhmmss_to_seconds(hhmmss):
    h, m, s = map(int, hhmmss.split(':'))
    return h * 3600 + m * 60 + s

def get_media_length(media_file):
    result = subprocess.run(
        ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", media_file],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT
    )
    return float(result.stdout)

def auto_color_correct(video_file):
    if not video_file:
        return "No video file selected.", None
    output_file = get_unique_filename(f"color_corrected_{os.path.basename(video_file)}", "")
    command = [
        "ffmpeg", "-i", video_file, "-vf", "eq=brightness=0.06:saturation=1.5", output_file
    ]
    subprocess.run(command, check=True)
    return f"Auto color correction applied successfully to {output_file}", output_file

def extract_audio(video_file):
    if not video_file:
        return "No video file selected.", None
    output_file = get_unique_filename(f"extracted_audio_{os.path.basename(video_file)}", ".mka")
    command = [
        "ffmpeg", "-i", video_file, "-vn", "-acodec", "copy", output_file
    ]
    try:
        subprocess.run(command, check=True)
    except subprocess.CalledProcessError as e:
        return f"Error extracting audio: {e}. Command: {' '.join(command)}", None
    return f"Audio extracted successfully into {output_file}", output_file

def compress_video(input_file, output_file):
    command = [
        "ffmpeg", "-i", input_file, "-vcodec", "libx265", "-crf", "28", output_file
    ]
    subprocess.run(command, check=True)
    return output_file

def add_watermark(video_file, watermark_type, watermark_text, watermark_image, opacity, position_x, position_y, font_size, font_color, resize_width, resize_height):
    if not video_file:
        return "No video file selected.", None

    output_file = get_unique_filename(f"watermarked_{os.path.basename(video_file)}", ".mp4")
    drawtext = f"drawtext=text='{watermark_text}':x={position_x}:y={position_y}:fontsize={font_size}:fontcolor={font_color}@{opacity}" if watermark_type == "text" else ""
def add_watermark(video_file, watermark_type, watermark_text, watermark_image, opacity, position, font_size, font_color, resize):
    if not video_file:
        return "No video file selected.", None

    output_file = get_unique_filename(f"watermarked_{os.path.basename(video_file)}", ".mp4")
    drawtext = f"drawtext=text='{watermark_text}':x={position[0]}:y={position[1]}:fontsize={font_size}:fontcolor={font_color}@{opacity}" if watermark_type == "text" else ""
    overlay = f"overlay={position[0]}:{position[1]}" if watermark_type == "image" else ""
    resize_cmd = f"scale={resize[0]}:{resize[1]}" if resize else "scale=iw:ih"

    command = [
        "ffmpeg", "-i", video_file,
        "-vf", f"{resize_cmd},{drawtext if watermark_type == 'text' else ''}{overlay if watermark_type == 'image' else ''}",
        output_file
    ]

    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    total_time = 0
    with tqdm(total=100, desc="Adding Watermark") as pbar:
        while True:
            output = process.stderr.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                total_time += 1
                pbar.update(1)
                time.sleep(0.1)

    return f"Watermark added successfully to {output_file}", output_file

def compress_image(input_image):
    if not input_image:
        return "No image file selected.", None
    output_file = get_unique_filename(f"compressed_{os.path.basename(input_image)}", ".jpg")
    
    with Image.open(input_image) as img:
        img.save(output_file, "JPEG", quality=95)  # Adjust quality as needed
    
    return f"Image compressed successfully into {output_file}", output_file

def compress_image_lossless(input_image):
    if not input_image:
        return "No image file selected.", None
    output_file = get_unique_filename(f"compressed_lossless_{os.path.basename(input_image)}", ".png")
    
    with Image.open(input_image) as img:
        img.save(output_file, "PNG", optimize=True)  # PNG is lossless
    
    return f"Image compressed losslessly into {output_file}", output_file

def mp3_to_video(mp3_file, image_file):
    if not mp3_file or not image_file:
        return "MP3 file or image file not selected.", None
    output_file = get_unique_filename(f"{os.path.splitext(os.path.basename(mp3_file))[0]}", ".mp4")
    command = [
        "ffmpeg", "-loop", "1", "-i", image_file, "-i", mp3_file, "-c:v", "libx264", "-c:a", "aac", "-b:a", "192k", "-shortest", output_file
    ]
    subprocess.run(command, check=True)
    return f"MP3 converted to video successfully into {output_file}", output_file

def video_to_mp3(video_file):
    if not video_file:
        return "No video file selected.", None
    output_file = get_unique_filename(f"{os.path.splitext(os.path.basename(video_file))[0]}", ".mp3")
    command = [
        "ffmpeg", "-i", video_file, "-q:a", "0", "-map", "a", output_file
    ]
    subprocess.run(command, check=True)
    return f"Video converted to MP3 successfully into {output_file}", output_file

def convert_image_format(input_image, output_format):
    if not input_image:
        return "No image file selected.", None
    output_file = get_unique_filename(f"{os.path.splitext(os.path.basename(input_image))[0]}", f".{output_format}")
    
    with Image.open(input_image) as img:
        img.save(output_file, output_format.upper())
    
    return f"Image converted to {output_format} format successfully into {output_file}", output_file

def interface():
    with gr.Blocks(theme="small_and_pretty") as demo:  # Applying the "small_and_pretty" theme
        gr.Markdown("### Media Combiner Tool")
        
        with gr.Tab("Combine Videos"):
            video_files = gr.File(label="Select Video Files", type="filepath", file_count="multiple")
            video_submit = gr.Button("Combine Videos")
            video_output = gr.Textbox(label="Output")
            video_download = gr.File(label="Download Combined Video")
            video_submit.click(combine_videos, inputs=[video_files], outputs=[video_output, video_download])

        with gr.Tab("Combine Audios"):
            audio_files = gr.File(label="Select Audio Files", type="filepath", file_count="multiple")
            audio_submit = gr.Button("Combine Audios")
            audio_output = gr.Textbox(label="Output")
            audio_download = gr.File(label="Download Combined Audio")
            audio_submit.click(combine_audios, inputs=[audio_files], outputs=[audio_output, audio_download])

        with gr.Tab("Combine Images"):
            image_files = gr.File(label="Select Image Files", type="filepath", file_count="multiple")
            image_submit = gr.Button("Combine Images")
            image_output = gr.Textbox(label="Output")
            image_download = gr.File(label="Download Combined Image")
            image_submit.click(combine_images, inputs=[image_files], outputs=[image_output, image_download])

        with gr.Tab("Adjust Speed"):
            media_file = gr.File(label="Select Media File", type="filepath")
            speed = gr.Slider(label="Speed", minimum=0.5, maximum=2.0, step=0.1, value=1.0)
            with gr.Row():
                hours = gr.Number(label="Hours", value=0, precision=0)
                minutes = gr.Number(label="Minutes", value=0, precision=0)
                seconds = gr.Number(label="Seconds", value=0, precision=0)
            compress = gr.Checkbox(label="Compress Video", value=False)
            speed_submit = gr.Button("Adjust Speed")
            speed_output = gr.Textbox(label="Output")
            speed_download = gr.File(label="Download Adjusted Media")
            
            speed_submit.click(adjust_speed_combined, inputs=[media_file, speed, hours, minutes, seconds, compress], outputs=[speed_output, speed_download])

        with gr.Tab("Auto Color Correction"):
            video_file = gr.File(label="Select Video File", type="filepath")
            color_submit = gr.Button("Apply Color Correction")
            color_output = gr.Textbox(label="Output")
            color_download = gr.File(label="Download Color Corrected Video")
            color_submit.click(auto_color_correct, inputs=[video_file], outputs=[color_output, color_download])

        with gr.Tab("Extract Audio"):
            video_file = gr.File(label="Select Video File", type="filepath")
            extract_submit = gr.Button("Extract Audio")
            extract_output = gr.Textbox(label="Output")
            extract_download = gr.File(label="Download Extracted Audio")
            extract_submit.click(extract_audio, inputs=[video_file], outputs=[extract_output, extract_download])

        with gr.Tab("Add Watermark"):
            video_file = gr.File(label="Select Video File", type="filepath")
            watermark_type = gr.Radio(label="Watermark Type", choices=["text", "image"], value="text")
            watermark_text = gr.Textbox(label="Watermark Text", visible=True)
            watermark_image = gr.File(label="Watermark Image", type="filepath", visible=False)
            opacity = gr.Slider(label="Opacity", minimum=0.0, maximum=1.0, step=0.1, value=1.0)
            position_x = gr.Slider(label="Position X", minimum=0, maximum=1920, step=1, value=0)
            position_y = gr.Slider(label="Position Y", minimum=0, maximum=1080, step=1, value=0)
            font_size = gr.Slider(label="Font Size", minimum=10, maximum=100, step=1, value=24, visible=True)
            font_color = gr.ColorPicker(label="Font Color", value="#FFFFFF", visible=True)
            resize_width = gr.Number(label="Resize Width", visible=False)
            resize_height = gr.Number(label="Resize Height", visible=False)
            watermark_submit = gr.Button("Add Watermark")
            watermark_output = gr.Textbox(label="Output")
            watermark_download = gr.File(label="Download Watermarked Video")

            def update_visibility(watermark_type):
                return {
                    watermark_text: gr.update(visible=watermark_type == "text"),
                    watermark_image: gr.update(visible=watermark_type == "image"),
                    font_size: gr.update(visible=watermark_type == "text"),
                    font_color: gr.update(visible=watermark_type == "text"),
                    resize_width: gr.update(visible=watermark_type == "image"),
                    resize_height: gr.update(visible=watermark_type == "image")
                }

            watermark_type.change(update_visibility, inputs=[watermark_type], outputs=[watermark_text, watermark_image, font_size, font_color, resize_width, resize_height])
            watermark_submit.click(add_watermark, inputs=[video_file, watermark_type, watermark_text, watermark_image, opacity, position_x, position_y, font_size, font_color, resize_width, resize_height], outputs=[watermark_output, watermark_download])


        with gr.Tab("Compress Image Losslessly"):
            image_file = gr.File(label="Select Image File", type="filepath")
            compress_submit = gr.Button("Compress Image Losslessly")
            compress_output = gr.Textbox(label="Output")
            compress_download = gr.File(label="Download Compressed Image")
            compress_submit.click(compress_image_lossless, inputs=[image_file], outputs=[compress_output, compress_download])

        with gr.Tab("Convert MP3 to Video"):
            mp3_file = gr.File(label="Select MP3 File", type="filepath")
            image_file = gr.File(label="Select Image File", type="filepath")
            convert_submit = gr.Button("Convert MP3 to Video")
            convert_output = gr.Textbox(label="Output")
            convert_download = gr.File(label="Download Converted Video")
            convert_submit.click(mp3_to_video, inputs=[mp3_file, image_file], outputs=[convert_output, convert_download])

        with gr.Tab("Convert Video to MP3"):
            video_file = gr.File(label="Select Video File", type="filepath")
            convert_submit = gr.Button("Convert Video to MP3")
            convert_output = gr.Textbox(label="Output")
            convert_download = gr.File(label="Download Converted MP3")
            convert_submit.click(video_to_mp3, inputs=[video_file], outputs=[convert_output, convert_download])

        with gr.Tab("Convert Image Format"):
            input_image = gr.File(label="Select Image File", type="filepath")
            output_format = gr.Dropdown(label="Output Format", choices=["png", "jpg", "bmp", "gif"], value="png")
            convert_submit = gr.Button("Convert Image Format")
            convert_output = gr.Textbox(label="Output")
            convert_download = gr.File(label="Download Converted Image")
            convert_submit.click(convert_image_format, inputs=[input_image, output_format], outputs=[convert_output, convert_download])

    return demo

if __name__ == "__main__":
    demo = interface()
    demo.launch(share=True)