Application to make short gifs for your videos

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} 

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)