| 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:
|
| filelist.write(f"file '{os.path.abspath(video).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()
|
|
|
| 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()
|
|
|
| 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)
|
|
|
| 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)
|
|
|
| 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:
|
| 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)
|
|
|
