App/Worker.py

from celery import Celery
import os
import time
from App import celery_config
import yt_dlp
import tempfile
from App.Transcription.Utils.audio_transcription import transcribe_file

celery = Celery()
celery.config_from_object(celery_config)


@celery.task(name="transcription", bind=True)
def transcription_task(self, file_path, model_size="tiny"):
    return transcribe_file(state=self, file_path=file_path, model_size=model_size)


@celery.task(name="download", bind=True)
def downloadfile(self, url, ydl_opts, model_size="tiny"):
    # updated
    self.update_state(state="Downloading File..", meta={})

    ####
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        ydl.download([url])

    # updated
    self.update_state(state="Downloading complete", meta={})
    audio_file = ydl_opts["outtmpl"]
    print(audio_file["default"])
    return transcribe_file(
        state=self, file_path=audio_file["default"], model_size=model_size
    )