import os, datetime, cv2, json, logging from django.conf import settings from apps.accounts.models import User from apps.subscriptions.models import Subscription from apps.process.models import Media, ProcessedMedia, Status from apps.process.serializers import ProcessedMediaSerializer from django.core.serializers.json import DjangoJSONEncoder # Create a logger for this file logger = logging.getLogger(__file__) # has_download_youtube_permission def yt_down_allow(value): user = User.objects.get(email=value) if user.is_superuser or user.is_staff : return True else: subscription = Subscription.objects.get(user=user) plan_type = subscription.plan.download_youtube_perm return plan_type # has_upload_youtube_permission def yt_up_allow(value): user = User.objects.get(email=value) if user.is_superuser or user.is_staff : return True else: subscription = Subscription.objects.get(user=user) plan_type = subscription.plan.upload_youtube_perm return plan_type def get_client_ip(request): x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = request.META.get('REMOTE_ADDR') return ip def generate_name(filename): filechar, ext = os.path.splitext(filename) # rewrite file name and remove any unwanted character filechar = filechar.replace(".", "").replace(" ", "_").replace("(", "_").replace(")", "_").replace("&", "AND") # add date and time to video name to avoide conflect by name now = (str(datetime.datetime.now()).split(" ")[1].replace(":", "_").split(".")[0]) # return new filename return filechar + "_" + now + ext def generate_thumbnail_old(video_path): # Generate thumbnail if not os.path.exists("media/thumbnails/"): os.makedirs("media/thumbnails/") img_output_path = f"media/thumbnails/{os.path.splitext(video_path)[0]}_thumbnail.png" command = f"ffmpeg -ss 00:00:01.005 -i {f'media/original_videos/{video_path}'} -frames:v 1 -q:v 2 {img_output_path}" os.system(command) return img_output_path def generate_thumbnail(video_path): # command = f"ffmpeg -ss 00:00:01.005 -i {f'media/original_videos/{video_path}'} -frames:v 1 -q:v 2 {img_output_path}" input_path = os.path.join(settings.MEDIA_URL, video_path) vidcap = cv2.VideoCapture(input_path) # print("reading Video ...") success, image = vidcap.read() print("reading video success ...") count = 0 while success: success, image = vidcap.read() if count == 90: # cv2.imwrite(img_output_path,image) im_resize = cv2.resize(image, (image.shape[1], image.shape[0])) _, im_buf_arr = cv2.imencode(".jpg", im_resize) return im_buf_arr # break count += 1 # os.system(command) # print(command) # print("Finich generate thumbnails !") # return img_output_path return def get_sec(time_str): """Get seconds from time.""" h, m, s = time_str.split(":") seconds = int(h) * 3600 + int(m) * 60 + int(s) return seconds * 1000 def get_audio_parts(audio_duration, parts_list): starts = [] ends = [] non_selected_list = [] for part in parts_list: starts.append(part[0]) ends.append(part[1]) if max(ends) != audio_duration: non_selected_list.append([ends[ends.index(max(ends))], audio_duration]) if min(starts) != 0: non_selected_list.append([0, starts[starts.index(min(starts))]]) for index in range(len(starts) - 1): non_selected_list.append([ends[index], starts[index + 1]]) return sorted(non_selected_list + parts_list) def getlastmediauploaded(request): # return Media.objects.filter(user=request.user).latest('-uploaded_at') try: all_videos = ProcessedMedia.objects.filter(user=request.user) if len(all_videos) > 0: return all_videos.order_by('-uploaded_at')[0] else: return ProcessedMedia.objects.all()[:1].get() except ProcessedMedia.DoesNotExist: return ProcessedMedia.objects.all()[:1].get() # def get_media(user, nbr=None): # if ProcessedMedia.objects.all().count() == 0: # return None # if user.is_superuser: # return ProcessedMedia.objects.all().exclude(status=Status.Deleted)[:nbr] # else : # default_video = ProcessedMedia.objects.filter(user__is_superuser=True).exclude(status=Status.Deleted).first() # if ProcessedMedia.objects.filter(user=user).count() == 0 : # return default_video # elif ProcessedMedia.objects.filter(user=user).count() > 0 : # user_videos = ProcessedMedia.objects.filter(user=user).exclude(status=Status.Deleted)[:nbr] # if default_video: # user_videos = [default_video] + list(user_videos) # return user_videos def get_media(user, nbr=None): if ProcessedMedia.objects.all().count() == 0: return None if user.is_superuser: return list(ProcessedMedia.objects.all().exclude(status=Status.Deleted)[:nbr]) else: default_video = ProcessedMedia.objects.filter(user__is_superuser=True,media__id_media="c2a8dacc-1249-40c2-a5fd-40931c003435").exclude(status=Status.Deleted).first() if ProcessedMedia.objects.filter(user=user).count() == 0: return [default_video] if default_video else None elif ProcessedMedia.objects.filter(user=user).count() > 0: user_videos = list(ProcessedMedia.objects.filter(user=user).exclude(status=Status.Deleted)[:nbr]) if default_video: user_videos = user_videos + [default_video] return user_videos def last_uploaded_pm(user): """ get all media based on last converting time input: user:obj """ try: if user.is_superuser: last_uploaded_pm = ProcessedMedia.objects.latest('converted_time') else: last_uploaded_pm = ProcessedMedia.objects.filter(user=user).latest('converted_time') serialized_data = ProcessedMediaSerializer(last_uploaded_pm).data return json.dumps(serialized_data, cls=DjangoJSONEncoder) except ProcessedMedia.DoesNotExist: default_video = ProcessedMedia.objects.filter(user__is_superuser=True, media__id_media="c2a8dacc-1249-40c2-a5fd-40931c003435").exclude(status=Status.Deleted).first() serialized_data = ProcessedMediaSerializer(default_video).data return json.dumps(serialized_data, cls=DjangoJSONEncoder)