import os import time import uuid import logging import requests from io import BytesIO import yt_dlp as youtube_dl from django.views import View from django.http import JsonResponse from django.shortcuts import redirect, render from django.utils.decorators import method_decorator from django.contrib.auth.decorators import login_required from django.http import FileResponse, HttpResponseNotFound, HttpResponseForbidden from django.core.exceptions import ObjectDoesNotExist from django.core.files.storage import default_storage from apps.process.models import (File, Media, Status, UploadType, ProcessedMedia) from apps.process.utils import ( yt_down_allow, generate_name, generate_thumbnail ) from apps.process.tasks import (convert) from apps.subscriptions.utils import ( valid_extensions, file_extension_valid, retrieve_limit_of_size, retrieve_limit_of_duration, has_reached_upload_limit ) from celery.result import AsyncResult # Create a logger for this file logger = logging.getLogger(__file__) # Create your views here. def is_ajax(request): return request.META.get('HTTP_X_REQUESTED_WITH') == 'XMLHttpRequest' def normal_upload(serialize_user, media_path, file, fileName, existingPath, end, nextSlice, parts): """ Function that upload video from local """ if file=="" or fileName=="" or existingPath=="" or end=="" or nextSlice=="": # res = {'data':'Invalid Request', 'status': 'failed'} return res else: code = str(uuid.uuid4()).split("-")[0] if existingPath == 'null': # Saving File To Server DB FileFolder = File(existingPath =(media_path + generate_name(fileName)) , eof=end, name=fileName) FileFolder.save() try: media = Media.objects.get(original_file=FileFolder , user=serialize_user) except Media.DoesNotExist: media = Media(original_file=FileFolder,title=fileName , user=serialize_user, upload_source=UploadType.Local) # Path Where The File Will Be Saved path = FileFolder.existingPath # Saving File Temporarily To Local Storage temp_file_path = os.path.join("media", path) # Start Saveing The Fist Part Of The Sended File # with default_storage.open(path, 'wb+') as destination: # destination.write(file) print(f"File Will Be Setted in {temp_file_path}") with open(temp_file_path, 'wb+') as temp_file: temp_file.write(file) # TesT if The File Is Completed if int(end): # with open(temp_file_path, 'rb+') as temp_file: default_storage.save(path, temp_file) # Clean up the temporary file from local storage os.remove(temp_file_path) # Update/Create Media Object For The user How Upladed This File media.status = Status.Processing media.thumbnail.save(f"{code}_thumbnail.jpg", BytesIO(generate_thumbnail(FileFolder.existingPath)), save=True) media.size = default_storage.size(path) print(media.size) media.save() # start removing sound print(" start celery task ") convert_task = convert.apply_async(args=[FileFolder.existingPath, parts, media.id_media, serialize_user.id]) # time.sleep(5) # pause for 5 seconds task_response = AsyncResult(convert_task.id) # get final_video_id add it to dict res logger.info(f"Final Video ID:{task_response.get()}") res = {'data':'Uploaded Successfully','existingPath':FileFolder.existingPath, 'status': 'success','final_video_id':task_response.get()} else: # Update/Create Media Object For The user How Upladed This File media.status = Status.Uploading # media.size = default_storage.size(path) media.size = os.path.getsize(temp_file_path) media.save() # res = {'existingPath':FileFolder.existingPath, 'status': 'uploading'} return res else: # path = existingPath # Saving File Temporarily To Local Storage temp_file_path = os.path.join("media", path) # model_id = File.objects.get(existingPath=existingPath) # media = Media.objects.get(original_file=model_id)#, user=request.user) # if model_id.name == fileName: # if not model_id.eof: # with open(temp_file_path, 'rb') as temp_file: old_content = temp_file.read() # old_file = default_storage.open(path, 'rb') # old_content = old_file.read() # old_file.close() # file = old_content + file # print(f"File Will Be Setted in {temp_file_path} and complete uploading") with open(temp_file_path, 'wb+') as temp_file: temp_file.write(file) # with default_storage.open(path, 'wb+') as destination: # destination.write(file) # if int(end): model_id.eof = int(end) model_id.save() # with open(temp_file_path, 'rb') as temp_file: default_storage.save(path, temp_file) # Clean up the temporary file from local storage os.remove(temp_file_path) # Update/Create Media Object For The user How Upladed This File media.status = Status.Processing media.thumbnail.save(f"{code}_thumbnail.jpg", BytesIO(generate_thumbnail(existingPath)), save=True) media.size = default_storage.size(path) media.save() # start removing sound convert_task = convert.apply_async(args=[model_id.existingPath, parts, media.id_media,serialize_user.id]) # time.sleep(5) # pause for 5 seconds task_response = AsyncResult(convert_task.id) # get final_video_id add it to dict res logger.info(f"Final Video ID:{task_response.get()}") res = {'data':'Uploaded Successfully','existingPath':model_id.existingPath, 'status': 'success','task_id':convert_task.id} else: # # Update/Create Media Object For The user How Upladed This File media.status = Status.Uploading # media.size = default_storage.size(path) media.size = os.path.getsize(temp_file_path) media.save() # res = {'existingPath':model_id.existingPath, 'status': 'uploading'} return res else: # # Update/Create Media Object For The user How Upladed This File media.status = Status.Failed media.size = os.path.getsize(f"media/original_videos/{FileFolder.existingPath}") media.save() # return {'data':'EOF found. Invalid request','status': 'failed'} else: # Update/Create Media Object For The user How Upladed This File media.status = Status.Failed # media.size = default_storage.size(path) media.size = os.path.getsize(temp_file_path) media.save() return {'data':'No such file exists in the existingPath', 'status': 'failed'} def youtube_upload(serialize_user, media_path, video_id, duration, parts): """ function that upload video from youtube Args: serialize_user (_type_): get id from user to determine which user uplaod this video media_path (__str__): path of media (video) video_id (_str_): youtube video id duration (_int_): youtube video duration parts (_list_): list of all part that user selected Returns: dict: dictionary ot youtube video status """ logger.info(f'Start Downloading Video With YouTube ID {video_id} ') # try: media = Media.objects.filter(youtube_id=video_id)[:1].get() except Media.DoesNotExist: media = Media(youtube_id=video_id, user=serialize_user, upload_source=UploadType.YouTube) # if media.original_file and media.original_file != None and media.audio_path != None: # new_media = Media( youtube_id = video_id, user = serialize_user, title = media.title, author = media.author, original_file = media.original_file, thumbnail = media.thumbnail, audio_path = media.audio_path, size = media.size, duration = media.duration, status = Status.Completed, upload_source = UploadType.YouTube ) new_media.save() # ? start removing sound print(" start celery task ") convert_task = convert.apply_async(args=[new_media.original_file.existingPath, parts, new_media.id_media, serialize_user.id]) task_response = AsyncResult(convert_task.id) return {"data":"inserted Video is already downloaded and it will be proccesed", 'status': 'success'} else: # temp_dir = f"media/{media_path}" if not os.path.exists(temp_dir): os.makedirs(temp_dir) # media.user = serialize_user video_url = f"https://www.youtube.com/watch?v={video_id}" thumbnail_url = f"https://img.youtube.com/vi/{video_id}/maxresdefault.jpg" # Download the video using youtube_dl library. try: # ydl_opts = { 'outtmpl': f"{temp_dir}{video_id}.%(ext)s", 'merge_output_format': 'mp4', 'format': 'best', } # print("Start Getting The Video") with youtube_dl.YoutubeDL(ydl_opts) as ydl: info_dict = ydl.extract_info(video_url, download=True) video_title = info_dict.get('title', None) # media_path_with_file = f"{media_path}{video_id}_{video_title}.mp4" # print("Transform Video From Local To S3 Using default_storage") with default_storage.open(media_path_with_file, 'wb+') as destination: file_data = open(f"{temp_dir}{video_id}.mp4", 'rb') destination.write(file_data.read()) file_data.close() # print(f"File Saved With Path : {media_path_with_file}") media_path_without_extension = os.path.splitext(f"{media_path}{video_id}")[0] # response = requests.get(thumbnail_url) print( "Saving File In DB") # new_file = File( # existingPath=media_path_with_file, # name=video_title, # eof=True # ) # new_file.save() try: # Try to fetch an existing file with the same 'existingPath' file = File.objects.get(existingPath=media_path_with_file) # Check if the file's 'eof' attribute is False if not file.eof: # Remove the existing file since eof is False file.delete() print("Existing File Removed") file = File( existingPath=media_path_with_file, name=video_title, eof=True ) else: # Update the existing file attributes and save it file.name = video_title file.eof = True file.save() print("File Updated in DB") except File.DoesNotExist: # If the file doesn't exist, create a new 'File' object file = File( existingPath=media_path_with_file, name=video_title, eof=True ) file.save() print( "Saving Media In DB") # Save Video-Media-Info In Our DataBase media.title = video_title media.author = info_dict.get('uploader', None) media.original_file = file media.thumbnail.save(f"{video_id}_thumbnail.jpg", BytesIO(response.content), save=True) media.audio_path = f"{media_path_without_extension}.mp3" media.size = default_storage.size(media_path_with_file) media.duration = duration media.status = Status.Completed media.save() # ? start removing sound print(" Starting celery task ") convert_task = convert.apply_async(args=[media.original_file.existingPath, parts, media.id_media, serialize_user.id]) task_response = AsyncResult(convert_task.id) return {"data":"video downloaded and it will be proccesed", 'status': 'success'} except Exception as e: print(e) logger.info(e) return { 'status': 'failed', 'message': 'video not downloaded' } finally: # remove the downloaded file os.remove(f"{temp_dir}{video_id}.mp4") # remove the temporary directory if it is empty if os.path.exists(temp_dir) and not os.listdir(temp_dir): os.rmdir(temp_dir) @method_decorator(login_required(login_url="/auth/signin"), name="dispatch") class ProcessView(View): """ Process view """ def get(self, request, *args, **kwargs): # code to process a GET request: pass def post(self, request, *args, **kwargs): # code to process a POST request if is_ajax(request): # serialize_user = request.user response = {} # limit checks (nbr_videos_per_plan, file_duration, file_size, file_extention, ) # Check if current user has reached the limit of plan videos per day has_reached_limit = has_reached_upload_limit(serialize_user) # logging data_duration logger.info(has_reached_limit) if has_reached_limit["status"]: # Check if current user has reached the limit of plan video duration data_duration = retrieve_limit_of_duration(float(request.POST['file_duration']), serialize_user) # logging data_duration logger.info(data_duration) if data_duration["status"]: # Cheking if the path to upload video exist media_path = "original_videos/" duration = float(request.POST['file_duration']) if request.POST['upload_type'] == "normal": # Cheking if current video has reached the limit of plan video_size data_size = retrieve_limit_of_size(float(request.POST['file_size'])*0.001, serialize_user) # logging data_duration logger.info(data_size) # if data_size["status"]: # check file size and duration # if file_extension_valid(request): # response = {'status':'success','message':'video has a valid Extention'} # handle normal upload file = request.FILES['file'].read() fileName = request.POST['filename'] existingPath = request.POST['existingPath'] end = request.POST['end'] nextSlice = request.POST['nextSlice'] parts = request.POST['parts'] # response = normal_upload(serialize_user, media_path, file, fileName, existingPath, end, nextSlice, parts) logger.info("Parts: ", parts) # else: # response = {'status':'failed','message':'Video Exention not valid'} # else : # response = {'data':'current User has reached the limit of plan video duration.', 'status': 'failed', 'suggest':f"{data_duration['message']} \n Upgrade Your Plan Or Contact The Support For More Detailes."} if request.POST['upload_type'] == "youtube" and yt_down_allow(serialize_user.email): # check file size and duration (Api) video_id = request.POST.get('video_id') parts = request.POST.get("parts") # response = youtube_upload(serialize_user, media_path, video_id, duration, parts) else : response = {'data':'current User has reached the limit of plan video duration.', 'status': 'failed', 'suggest':f"{data_duration['message']} \n Upgrade Your Plan Or Contact The Support For More Detailes."} else : response = {'data':'User has reached the limit of videos per day.', 'status': 'failed', 'suggest':'Upgrade Your Plan Or Contact The Support For More Detailes.'} # return JsonResponse(response) # add new voice enhacement but without limits class VoiceEnhancementView(View): def get(self, request, *args, **kwargs): # code to process a GET request: if not request.user.is_authenticated: return redirect('accounts:signin') return render(request,'enhancement.html') def post(self, request, *args, **kwargs): # code to process a POST request if not request.user.is_authenticated: return redirect('accounts:signin') # code to process a POST request if is_ajax(request): logger.info("Recieve from Enhancement views") serialize_user = request.user response = {} # limit checks (nbr_videos_per_plan, file_duration, file_size, file_extention, ) # Check if current user has reached the limit of plan videos per day has_reached_limit = has_reached_upload_limit(serialize_user) # logging data_duration logger.info(has_reached_limit) # if has_reached_limit["status"]: # Check if current user has reached the limit of plan video duration # data_duration = retrieve_limit_of_duration(float(request.POST['file_duration']), serialize_user) # logging data_duration # logger.info(data_duration) # if data_duration["status"]: # Cheking if the path to upload video exist media_path = "original_videos/" if request.POST['upload_type'] == "normal": # Cheking if current video has reached the limit of plan video_size data_size = retrieve_limit_of_size(float(request.POST['file_size'])*0.001, serialize_user) # logging data_duration logger.info(data_size) # if data_size["status"]: # check file size and duration # if file_extension_valid(request): # response = {'status':'success','message':'video has a valid Extention'} # handle normal upload file = request.FILES['file'].read() fileName = request.POST['filename'] existingPath = request.POST['existingPath'] end = request.POST['end'] nextSlice = request.POST['nextSlice'] parts = request.POST['parts'] duration = float(request.POST['file_duration']) # #! TODO: try: user_plan = request.user.subscription.plan.plan_type except ObjectDoesNotExist: # check if user is admin if request.user.is_superuser: # continue with the normal flow user_plan = "default_plan_type" else: # return a response indicating that the user does not have a subscription response = { 'data': 'You do not have an active subscription to access this feature.', 'status': 'failed', 'suggest': 'Create a subscription to access this feature.' } return JsonResponse(response) response = normal_upload(serialize_user, media_path, file, fileName, existingPath, end, nextSlice, parts,) # user_plan) # else: # response = {'status':'failed','message':'Video Exention not valid'} # else : # response = {'data':'current User has reached the limit of plan video duration.', 'status': 'failed', 'suggest':f"{data_duration['message']} \n Upgrade Your Plan Or Contact The Support For More Detailes."} if request.POST['upload_type'] == "youtube" and yt_down_allow(serialize_user.email): # check file size and duration (Api) video_id = request.POST.get('video_id') parts = request.POST.get("parts") # response = youtube_upload(serialize_user, media_path, video_id, parts) # else : # response = {'data':'current User has reached the limit of plan video duration.', 'status': 'failed', 'suggest':f"{data_duration['message']} \n Upgrade Your Plan Or Contact The Support For More Detailes."} # else : # response = {'data':'User has reached the limit of videos per day.', 'status': 'failed', 'suggest':'Upgrade Your Plan Or Contact The Support For More Detailes.'} # return JsonResponse(response) class CancelMediaView(View): def post(self,request,*args,**kwargs): pass @method_decorator(login_required(login_url="/auth/signin"), name="dispatch") class DownloadMediaView(View): def get(self, request, file_id): try: file = ProcessedMedia.objects.get(id=file_id) except ProcessedMedia.DoesNotExist: return HttpResponseNotFound() # if not request.user.has_permission(file): # return HttpResponseForbidden() file_path = file.video_path file_name = (file.media.original_file.existingPath).replace('original_videos','') if default_storage.exists(file_path): try: f = default_storage.open(file_path, 'rb') response = FileResponse(f) response['Content-Disposition'] = f'attachment; filename="{file_name}"' f.close() return response except Exception as e: raise HttpResponseNotFound(e.value) # return redirect('accounts:signin') else: return HttpResponseNotFound("File not found")