import os import re import logging from decouple import config from datetime import datetime from moviepy.editor import VideoFileClip from django.contrib.auth.models import User from django.views.decorators.cache import cache_page from django.core import serializers from django.http import HttpResponse, JsonResponse , Http404 from django.contrib.auth.decorators import login_required from django.utils.decorators import method_decorator from django.shortcuts import render, redirect from django.views import View import requests import boto import boto.s3.connection from boto.s3.key import Key import aiohttp from rest_framework import status from rest_framework.views import APIView from rest_framework.response import Response from .tasks import check_transcripts from .models import File, Transcripts from .utils import generate_name, handle_uploaded_file, download_captions from .serializers import FileSerializer,TranscriptsSerializer logger = logging.getLogger(__file__) class ListTranscripts(View): @method_decorator(login_required(login_url='/auth/login')) def get(self, request): transcripts=None if request.user.is_superuser: transcripts = Transcripts.objects.all() else : transcripts = Transcripts.objects.filter(file_org__user=request.user) logger.info(transcripts) context = { "transcripts":transcripts, "page":"allTranscripts" } return render(request, 'dashboard.html', context) @method_decorator(login_required(login_url='/auth/login')) def post(self, request): if not request.user.is_authenticated: return redirect("accounts:login") @login_required(login_url='/auth/login') def upload(request): if not request.user.is_authenticated: return redirect("accounts:login") if request.method == 'POST' and ( 'filename' in request.FILES ): filename= request.FILES['filename'].name content = request.FILES['filename'] path="media/" if not os.path.exists(path): os.makedirs(path) # generate unique name generated_name = generate_name(filename) # handle upload file to local server handle_uploaded_file(content, path, generated_name) # save file info in database video_path = path+generated_name clip = VideoFileClip(video_path) filename, extension = os.path.splitext(filename) # Create File Object & Save IT new_file = File( file_title = filename, video_path = video_path, duration = clip.duration, extension = extension, size = os.path.getsize(video_path), user = request.user ) new_file.save() # Create an S3 client object using the Boto3 library conn = boto.s3.connect_to_region( region_name = config("AWS_REGION", default="us-west-2", cast=str), aws_access_key_id = config("AWS_ACCESS_KEY_ID", default="AKIAT6THLFM7NTAL4XWX", cast=str), aws_secret_access_key = config("AWS_SECRET_ACCESS_KEY", default="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", cast=str), calling_format = boto.s3.connection.OrdinaryCallingFormat(), ) # Try to get the bucket by name bucket = conn.get_bucket('input-transcribe-bucket') # Create a new Key object that can be used to interact with a new uploaded file to the bucket. k = bucket.new_key(generated_name) # start Uploading File To The Buket With The Created Key. k.set_contents_from_filename(video_path) return render( request , "upload.html", { 'id_file':new_file.id_file, 'page':'upload' } ) return render( request , "upload.html", { 'page':'upload' } ) @login_required(login_url='/auth/login') def transcript_status(request, id_file): try: transcript = Transcripts.objects.get(file_org=File.objects.get(id_file=id_file)) if transcript.path_srt_fr and transcript.path_srt_es and transcript.path_srt_en: # If all three transcript paths are present, return the transcripts as a JSON response return JsonResponse({ 'status': 2, 'detail':'Transcript Found With All Info', 'message': 'Your file Has Been processed Complety.', 'transcript': TranscriptsSerializer(transcript).data }) else: # If the transcripts are still being processed, start the task and return a message return JsonResponse({ 'status': 1, 'detail': 'Transcript Founded With Initial Creation', 'message': 'Please wait, your file is processing Right Now.', 'transcript': TranscriptsSerializer(transcript).data }) except Transcripts.DoesNotExist: return JsonResponse({ 'status': 0, 'detail': 'Transcript Not Found', 'message': 'Please wait, your file is processing Right Now.', 'transcript': '' }) class S3Input(APIView): # authentication_classes = [TokenAuthentication] # Token authentication is required for this view # permission_classes = [IsOwnerOrReadOnly] # Only the owner can modify the File object, others can only read def get_object(self, id): """ Helper method to retrieve a File object by UUID in name. """ # Returns the File object with the given uuid contains in video_path try: return File.objects.get(video_path__contains=id) except File.DoesNotExist: # Raises 404 error if no Transcripts objects are found raise Http404 def get(self, request): """ Handles GET requests to retrieve all the File objects in the database. Returns a JSON response with the serialized data. """ # Retrieves all File objects and returns them in a response queryset = File.objects.filter(user=request.user) serializer = FileSerializer(queryset, many=True) return Response(serializer.data, status.HTTP_200_OK) def put(self, request): """ Handles PUT requests to update a File object in the database. Expects the request object to have file_s3_url, language, and srt_file fields. """ # BASE_S3_URL = 'https://input-transcribe-bucket.s3.us-east-2.amazonaws.com/' # Retrieves the file_uuid using the uuid in the request data file_s3_url file_uuid = str(request.data.get('file_s3_url')).replace(BASE_S3_URL,'') # Retrieves the file language using the language attribute in the request data file_lang = str(request.data.get('language')) # get file instance based on file_name original_file = self.get_object(file_uuid) # Retrieves the corresponding File object and updates it with the request data file_serializer = FileSerializer(original_file,data=request.data) if file_serializer.is_valid(raise_exception=True): file_serializer.save() # rename file with langauage tag # file_name = f"{file_lang}_{file_uuid}.srt" # Create A New Transcripts object For The Updated File transcript = Transcripts(file_org=original_file) if 'en' in file_lang: transcript.path_srt_en = request.data.get('srt_file') if 'po' in file_lang or 'pt' in file_lang: transcript.path_srt_po = request.data.get('srt_file') if 'es' in file_lang: transcript.path_srt_es = request.data.get('srt_file') if 'fr' in file_lang: transcript.path_srt_fr = request.data.get('srt_file') transcript.save() # Run Celery Task With Transcript ID return Response(file_serializer.data, status=status.HTTP_201_CREATED) return Response(file_serializer.errors, status=status.HTTP_400_BAD_REQUEST) class S3Output(APIView): # authentication_classes = [TokenAuthentication] # Token authentication is required for this view # permission_classes = [IsOwnerOrReadOnly] # Only the owner can modify the File object, others can only read def get_objects(self): # Returns all Transcripts objects try: return Transcripts.objects.all() except Transcripts.DoesNotExist: # Raises 404 error if no Transcripts objects are found raise Http404 def get_object(self, id): # Returns the Transcripts object with the given file_org id try: return Transcripts.objects.get(file_org=id) except Transcripts.DoesNotExist: # Raises 404 error if the Transcripts object is not found raise Http404 def get(self, request): # Retrieves all Transcripts objects and returns them in a response queryset = self.get_objects() serializer = TranscriptsSerializer(queryset, many=True) return Response(data=serializer.data, status=status.HTTP_200_OK) def put(self, request): # Retrieves the file_id using the uuid in the request data file_uuid = str(request.data.get('file')).split(".")[0] # print("DATA EXAMPLE: ",request.data) # print("File UUID: ",file_uuid) file_id = File.objects.get(video_path__contains=file_uuid) # Retrieves the corresponding Transcripts object and updates it with the request data transcript = self.get_object(file_id) transcript_serializer = TranscriptsSerializer(transcript, data=request.data) if transcript_serializer.is_valid(raise_exception=True): transcript_serializer.save() # save_vtt_wget(transcript) # download_captions(transcript, request) # save_local(request) # print(transcript_serializer.data) return Response(data=transcript_serializer.data, status=status.HTTP_201_CREATED) return Response(data=transcript_serializer.errors, status=status.HTTP_400_BAD_REQUEST) class SaveSubtitleView(APIView): def post(self, request): json_data = request.data STATIC_FILES = 'static/subtitles' # Extract file name and content from the JSON payload file_name = json_data['file_name'] srt_content = json_data['srt'] vtt_content = json_data['vtt'] # Write SRT file to static directory srt_file_path = os.path.join(STATIC_FILES, "srt", file_name + '.srt') with open(srt_file_path, 'w') as srt_file: srt_file.write(srt_content) # Write VTT file to static directory vtt_file_path = os.path.join(STATIC_FILES, "vtt", file_name + '.vtt') with open(vtt_file_path, 'w') as vtt_file: vtt_file.write(vtt_content) return Response({'message': 'Files saved successfully.'}) def save_vtt_wget( instance ): STATIC_FILES = 'static/subtitles' VTT_RESPONSE = {'file': instance.file_org.file_s3_url} # INSTANCE_VTT = {key[9:]: str(path).replace('.srt', '.vtt') for key, path in vars(instance).items() if key.startswith('path_srt_') and path} INSTANCE_VTT = { 'fr': (str(instance.path_srt_fr).replace('.srt','.vtt') if instance.path_srt_fr else None), 'es': (str(instance.path_srt_es).replace('.srt','.vtt') if instance.path_srt_es else None), 'en': (str(instance.path_srt_en).replace('.srt','.vtt') if instance.path_srt_en else None), 'po': (str(instance.path_srt_po).replace('.srt','.vtt') if instance.path_srt_po else None) } for k, vtt in INSTANCE_VTT.items(): vtt_path = os.path.join(STATIC_FILES, "vtt", f"{os.path.splitext(os.path.basename(vtt))[0]}.{k}.vtt") if os.path.exists(os.path.join(STATIC_FILES,vtt_path)): VTT_RESPONSE[k] = vtt_path print("VTT Path Exists: ",vtt_path) else: # import time # print(f"Get file {vtt} with wget") # time.sleep(0.5) # os.system(f"wget {vtt} -O {vtt_path} ") # time.sleep(1.5) download_file(vtt, vtt_path) async def download_file(url, file_path): async with aiohttp.ClientSession() as session: while True: async with session.get(url) as response: if response.status == 200: content = await response.read() with open(file_path, 'wb') as file: file.write(content) break elif response.status == 403: continue else: raise Exception(f"Failed to download file from {url}: {response.status} {response.reason}") def save_vtt( instance ): STATIC_FILES = 'static/subtitles' VTT_RESPONSE = {'file': instance.file_org.file_s3_url} # INSTANCE_VTT = {key[9:]: str(path).replace('.srt', '.vtt') for key, path in vars(instance).items() if key.startswith('path_srt_') and path} INSTANCE_VTT = { 'fr': (str(instance.path_srt_fr).replace('.srt','.vtt') if instance.path_srt_fr else None), 'es': (str(instance.path_srt_es).replace('.srt','.vtt') if instance.path_srt_es else None), 'en': (str(instance.path_srt_en).replace('.srt','.vtt') if instance.path_srt_en else None), 'po': (str(instance.path_srt_po).replace('.srt','.vtt') if instance.path_srt_po else None) } for k, vtt in INSTANCE_VTT.items(): if not vtt: print(f"VTT not found for {vtt} in creating") continue vtt_path = os.path.join(STATIC_FILES, f"{os.path.splitext(os.path.basename(vtt))[0]}.{k}.vtt") if os.path.exists(os.path.join(STATIC_FILES,vtt_path)): VTT_RESPONSE[k] = vtt_path print("VTT Path Exists: ",vtt_path) else: while not os.path.exists(os.path.join(STATIC_FILES,vtt_path)): response = requests.get(vtt) print(f'{vtt} ----> {response.status_code}') if response.status_code == 200: print(f"Downloading {vtt} In Create") with open(vtt_path, 'wb') as f: f.write(response.content) VTT_RESPONSE[k] = vtt_path else: print(f"Download {vtt} Faild")