💡 Introduction
This article is a related to my previous article, in this I will discuss and share the code to make a gif maker of your own. If you want to know the idea behind this project do read my previous article.
🤔 Why is it useful or what's the importance of this project?
Have you tried converting a video to a gif using an online tool? even a 5 sec gif can go upto 10 of MBs, and gifs can improve the overall look of the application by replacing the static images with a moving short glimpse of the video or reels.
🏛️ Architecture
Our current architecture for gif maker is, we have a consumer(RabbitMQ) written in python and when a video is uploaded and processes successfully, an event is fired with params "mp4Url" and "s3Destination", then the script downloads the video from the given url convert it to a compressed gif and upload it to the S3 destination.
🥸 Code and Explanation
from moviepy.editor import VideoFileClip
import os
import urllib.request
import uuid
import logging
import boto3
from pathlib import Path
from dotenv import dotenv_values
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
config = dotenv_values(".env")
logging.basicConfig(
filename=str(Path(__file__).parents[2]) + "/logs/app.log",
format="%(name)s - %(levelname)s - %(message)s",
)
mp4UrlPrefix = config["CLOUDFRONT_VIDEO_PREFIX"]
def getGifUrlFromMp4Driver(params):
logging.info("Started getGifUrlFromMp4Driver")
if "mp4Url" not in params:
logging.error("mp4Url not in params")
return {"gifUrl": None}
if "s3Destination" not in params:
logging.error("s3Destination not in params")
return {"gifUrl": None}
# get full mp4 url
mp4Url = mp4UrlPrefix + "/" + params["mp4Url"]
s3Destination = params["s3Destination"]
# if tweaking params are not passed, then use the default
# value
fps = params.get("fps", 20)
fuzz = params.get("fuzz", 1)
start = params.get("start", 5)
duration = params.get("duration", 5)
videoWidth = params.get("videoWidth", 144)
videoHeight = params.get("videoHeight", 256)
try:
# initializing s3 session
session = boto3.Session(
aws_access_key_id=config["AWS_KEY"],
aws_secret_access_key=config["AWS_SECRET"],
)
s3 = session.resource("s3")
mp4_folder = ROOT_DIR + "/videos/"
gif_folder = ROOT_DIR + "/gifs/"
# creating a unique name for mp4 video download path
# and gif
name = str(uuid.uuid4())
# download mp4
downloadedVideoPath = f"{mp4_folder}{name}.mp4"
urllib.request.urlretrieve(mp4Url, downloadedVideoPath)
# to reduce size of gif as well as to not take a long
# time we will try 3 times with reduced frame rates
# and increased fuzz to reduce size of gif
counter = 0
convertedGifPath = f"{gif_folder}{name}.gif"
while True:
counter += 1
videoClip = VideoFileClip(downloadedVideoPath)
# take a clip of video from x to y
videoClip = videoClip.subclip(start, start + duration)
# resizing video dimensions to desired width and
# height, this also reduces gif size to choose it
# wisely
videoClip = videoClip.resize((videoWidth, videoHeight))
# setting video fps, this also reduces gif size
videoClip = videoClip.set_fps(fps)
videoClip.write_gif(
filename=convertedGifPath,
program="ImageMagick",
opt="optimizeplus",
tempfiles=True,
verbose=False,
fuzz=fuzz,
logger=None,
)
# get size of converted gif
file_size = os.path.getsize(convertedGifPath)
# greater than 500Kb then reduce fps
if file_size > 500000 and counter <= 3:
if counter == 1:
fps = 15
elif counter == 2:
fps = 10
elif counter == 3:
fps = 5
continue
break
# remove downloaded video from disk
os.remove(downloadedVideoPath)
destBucketName = config["AWS_BUCKET_IMAGE_NAME"]
if s3Destination[-1] != "/":
s3Destination += "/"
gifPath = "gif" + ".gif"
# upload gif to s3 bucket
s3.Bucket(destBucketName).upload_file(convertedGifPath,
s3Destination + gifPath)
gifUrl = f"{s3Destination}{gifPath}"
os.remove(convertedGifPath)
# return back the uploaded gif url
return {"gifUrl": gifUrl}
except Exception as e:
logging.error(f"Error in getGifUrlFromMp4Driver: {e}")
os.remove(downloadedVideoPath)
return {"gifUrl": None}
The above code is executed when the consumer listening to the event is triggered, here's an example of consumer
import json
from dotenv import dotenv_values
import logging
from getGifUrlFromMp4 import getGifUrlFromMp4Driver
import requests
config = dotenv_values(".env")
def gifConsumer(ch, method, properties, body):
try:
params = json.loads(body)
print(f" Received: {params}")
res = getGifUrlFromMp4Driver(params)
print(f" Gif created, url: {res}")
if res["gifUrl"] == None:
print("Gif url not found")
ch.basic_ack(delivery_tag=method.delivery_tag)
return
res["id"] = res["gifUrl"].split("/")[1]
mediaResponse = requests.post(
<url>, data=res
)
print(f"Response from media endpoint status code: {mediaResponse.status_code}")
print(f"Response from endpoint: {mediaResponse.json()}")
ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e:
print("Error in gifConsumer: ", e)
logging.error(f"Error in gifConsumer: {e}")
ch.basic_ack(delivery_tag=method.delivery_tag)
This consumer calls getGifUrlFromMp4Driver with mp4Url and s3Destination, the code is well explained in the comment, so do read them. After successfull or unsuccessfull conversion of gif controls comes back to the consumer function, now in case there's a gif url we give a api call to our endpoint informing it a successfull conversion of the gif and to save the gif url with the reel.
Here's the gist code in case formatting is messed up here: gist
Top comments (1)
great buddy