|
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() |
|
|
|
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) |