make api stateless with tempfile.

Dockerfile

@@ -1,11 +1,15 @@
 # Use an official Python runtime as a parent image
 FROM python:3.11.7-slim-bullseye
 
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
 # Set the working directory in the container to /app
 WORKDIR /app
 
 # Copy the current directory contents into the container at /app
-COPY . /app
+COPY --chown=user . /app
 
 #Install ImageMagick
 RUN apt-get update && apt-get install -y imagemagick && sed -i '91d' /etc/ImageMagick-6/policy.xml

main.py

@@ -1,31 +1,21 @@
-from fastapi import FastAPI, UploadFile, HTTPException, Form, Depends
-from fastapi.responses import FileResponse, HTMLResponse
-from fastapi.security import HTTPBasic, HTTPBasicCredentials
+import shutil, os, logging, uvicorn, tempfile
 from typing import Optional
-from pydantic import BaseModel, field_validator
+
 from utils.process_video import process_video
 from utils.zip_response import zip_response
 from utils.api_configs import api_configs
 from utils.read_html import read_html
-from utils.archiver import archiver
 from utils.logger import setup_logger
-import shutil, os, logging, uvicorn, secrets
+
+from fastapi import FastAPI, UploadFile, HTTPException, Form, Depends
+from fastapi.responses import HTMLResponse, Response
+from fastapi.security import HTTPBasic
+from pydantic import BaseModel, field_validator
 
 app = FastAPI()
 security = HTTPBasic()
 api_configs_file = os.path.abspath("api_config.yml")
 
-async def get_current_user(credentials: HTTPBasicCredentials = Depends(security)):
-    correct_username = secrets.compare_digest(credentials.username, api_configs(api_configs_file)["secrets"]["username"])
-    correct_password = secrets.compare_digest(credentials.password, api_configs(api_configs_file)["secrets"]["password"])
-    if not (correct_username and correct_password):
-        raise HTTPException(
-            status_code=401,
-            detail="Incorrect email or password",
-            headers={"WWW-Authenticate": "Basic"},
-        )
-    return credentials.username
-
 class MP4Video(BaseModel):
     video_file: UploadFile
     
@@ -61,7 +51,6 @@ class SRTFile(BaseModel):
             raise HTTPException(status_code=422, detail='Invalid subtitle file type. Please upload an SRT file.')
         return v
 
-
 @app.get("/")
 async def root():
     html_content = f"""
@@ -77,6 +66,13 @@ async def get_form():
     """
     return HTMLResponse(content=html_content)
 
+async def get_temp_dir():
+    dir = tempfile.TemporaryDirectory()
+    try:
+        yield dir.name
+    finally:
+        del dir
+
 @app.post("/process_video/")
 async def process_video_api(video_file: MP4Video = Depends(),
                             srt_file: SRTFile = Depends(),
@@ -87,40 +83,36 @@ async def process_video_api(video_file: MP4Video = Depends(),
                             bg_color: Optional[str] = Form("#070a13b3"),
                             text_color: Optional[str] = Form("white"),
                             caption_mode: Optional[str] = Form("desktop"),
-                            username: str = Depends(get_current_user)
+                            temp_dir: str = Depends(get_temp_dir)
                             ):
     try:
         logging.info("Creating temporary directories")
-        temp_dir = os.path.join(os.getcwd(),"temp")
-        os.makedirs(temp_dir, exist_ok=True)
-        temp_vid_dir = os.path.join(temp_dir,video_file.filename.split('.')[0])
-        os.makedirs(temp_vid_dir, exist_ok=True)
-        temp_input_path = os.path.join(temp_vid_dir, video_file.filename)
-        logging.info("Copying video UploadFile to the temp_input_path")
-        with open(temp_input_path, 'wb') as buffer:
+        with open(os.path.join(temp_dir, video_file.filename), 'w+b') as temp_file:
+            logging.info("Copying video UploadFile to the temporary directory")
             try:
-                shutil.copyfileobj(video_file.file, buffer)
+                shutil.copyfileobj(video_file.file, temp_file)
             finally:
                 video_file.file.close()
-        logging.info("Copying SRT UploadFile to the temp_input_path")
-        if srt_file.size > 0:
-            SRT_PATH = os.path.abspath(f"{temp_input_path.split('.')[0]}.srt")
-            with open(SRT_PATH, 'wb') as buffer:
-                try:
-                    shutil.copyfileobj(srt_file.file, buffer)
-                finally:
-                    srt_file.file.close()
-            logging.info("Processing the video...")
-            output_path, _ = process_video(temp_input_path, SRT_PATH, task, max_words_per_line, fontsize, font, bg_color, text_color, caption_mode)
-            logging.info("Zipping response...")
-            zip_path = zip_response(os.path.join(temp_vid_dir,"archive.zip"), [output_path, SRT_PATH])
-            return FileResponse(zip_path, media_type='application/zip', filename=f"result_{video_file.filename.split('.')[0]}.zip")
-        logging.info("Processing the video...")
-        output_path, srt_path = process_video(temp_input_path, None, task, max_words_per_line, fontsize, font, bg_color, text_color, caption_mode, api_configs_file)
-        logging.info("Zipping response...")
-        zip_path = zip_response(os.path.join(temp_vid_dir,"archive.zip"), [output_path, srt_path])
-        return  FileResponse(zip_path, media_type='application/zip', filename=f"result_{video_file.filename.split('.')[0]}.zip")
-                
+            logging.info("Copying SRT UploadFile to the temp_input_path")
+            if srt_file.size > 0:
+                with open(os.path.join(temp_dir, f"{video_file.filename.split('.')[0]}.srt"), 'w+b') as temp_srt_file:
+                    try:
+                        shutil.copyfileobj(srt_file.file, temp_srt_file)
+                    finally:
+                        srt_file.file.close()
+                logging.info("Processing the video...")
+                output_path, _ = process_video(temp_file.name, temp_srt_file.name, task, max_words_per_line, fontsize, font, bg_color, text_color, caption_mode)
+                logging.info("Zipping response...")
+                with open(os.path.join(temp_dir, f"{video_file.filename.split('.')[0]}.zip"), 'w+b') as temp_zip_file:
+                    zip_file = zip_response(temp_zip_file.name, [output_path, srt_path])
+                return Response(content = zip_file)
+            with open(os.path.join(temp_dir, f"{video_file.filename.split('.')[0]}.srt"), 'w+b') as temp_srt_file:
+                logging.info("Processing the video...")
+                output_path, srt_path = process_video(temp_file.name, None, task, max_words_per_line, fontsize, font, bg_color, text_color, caption_mode, api_configs_file)
+                logging.info("Zipping response...")
+                with open(os.path.join(temp_dir, f"{video_file.filename.split('.')[0]}.zip"), 'w+b') as temp_zip_file:
+                    zip_file = zip_response(temp_zip_file.name, [output_path, srt_path])
+                return Response(content = zip_file)  
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
     

utils/archiver.py

@@ -1,17 +0,0 @@
-import shutil, os
-from datetime import datetime
-
-def archiver(timestamp:datetime=datetime.now()):
-    TIME = f"{timestamp.year:4d}-{timestamp.month:02d}-{timestamp.day:02d}_{timestamp.hour:02d}-{timestamp.minute:02d}"
-    ARCHIVE = os.path.abspath(f"archive/{TIME}")
-    TEMP_DIR = os.path.abspath("temp/")
-    LOG_FILE = os.path.abspath("main.log")
-    if os.path.exists(TEMP_DIR):
-        shutil.make_archive(os.path.join(ARCHIVE, "files"), 'zip', TEMP_DIR)
-        shutil.rmtree(TEMP_DIR)
-    if os.path.exists(LOG_FILE):
-        shutil.copy(LOG_FILE, os.path.join(ARCHIVE, f"{TIME}.log"))
-        os.remove(LOG_FILE)
-
-if __name__ == '__main__':
-    archiver(timestamp=datetime.now())

utils/process_video.py

@@ -1,17 +1,9 @@
+import logging, os
 from utils.transcriber import transcriber
 from utils.subtitler import subtitler
-import logging, os
-
-# Set up logging
-logging.basicConfig(filename='main.log',
-                    encoding='utf-8',
-                    level=logging.DEBUG,
-                    format='%(asctime)s %(levelname)s %(message)s',
-                    datefmt='%m/%d/%Y %I:%M:%S %p')
 
-# API Function
-def process_video(invideo_filename:str,
-                  srt_path: str,
+def process_video(invideo_file: str,
+                  srt_file: str | None,
                   task: str,
                   max_words_per_line:int,
                   fontsize:str,
@@ -21,17 +13,15 @@ def process_video(invideo_filename:str,
                   caption_mode:str,
                   config_file:str
                   ):
-    invideo_filename = os.path.normpath(invideo_filename)
-    invideo_path_parts = invideo_filename.split(os.path.sep)
-    VIDEO_NAME = invideo_path_parts[-1]
-    OUTVIDEO_PATH = os.path.join(invideo_path_parts[-3], invideo_path_parts[-2], f"result_{VIDEO_NAME}")
-    if srt_path:
-        subtitler(invideo_filename, srt_path, OUTVIDEO_PATH, fontsize, font, bg_color, text_color, caption_mode)
-        return OUTVIDEO_PATH, srt_path
-    SRT_PATH = os.path.abspath(f"{invideo_filename.split('.')[0]}.srt") 
-    logging.info("Transcribing...")
-    if not os.path.exists(SRT_PATH):
-        transcriber(invideo_filename, SRT_PATH, max_words_per_line, task, config_file)
-    logging.info("Subtitling...")
-    subtitler(invideo_filename, SRT_PATH, OUTVIDEO_PATH, fontsize, font, bg_color, text_color, caption_mode)
-    return OUTVIDEO_PATH, SRT_PATH
+    invideo_path_parts = os.path.normpath(invideo_file).split(os.path.sep)
+    VIDEO_NAME = os.path.basename(invideo_file)
+    OUTVIDEO_PATH = os.path.join(os.path.normpath('/'.join(invideo_path_parts[:-1])), f"result_{VIDEO_NAME}")
+    if srt_file:
+        logging.info("Subtitling...")
+        subtitler(invideo_file, srt_file, OUTVIDEO_PATH, fontsize, font, bg_color, text_color, caption_mode)
+    else:
+        srt_file = os.path.normpath(f"{invideo_file.split('.')[0]}.srt")
+        transcriber(invideo_file, srt_file, max_words_per_line, task, config_file)
+        logging.info("Subtitling...")
+        subtitler(invideo_file, srt_file, OUTVIDEO_PATH, fontsize, font, bg_color, text_color, caption_mode)
+    return OUTVIDEO_PATH, srt_file

utils/transcriber.py

@@ -1,19 +1,19 @@
 from gradio_client import Client, handle_file
 from utils.api_configs import api_configs
+import tempfile
 
-def transcriber(invideo_path:str, srt_path:str,
+def transcriber(invideo_file:str, srt_file:str,
         max_words_per_line:int, task:str,
         config_file:str):
-        
         HF_TOKEN = api_configs(config_file)["secrets"]["hf-token"]
         HF_SPACE = api_configs(config_file)["secrets"]["hf-space"]
         client = Client(HF_SPACE, hf_token=HF_TOKEN)
         result = client.predict(
-                video_input=handle_file(invideo_path),
+                video_input=handle_file(invideo_file),
                 max_words_per_line=max_words_per_line,
                 task=task,
                 api_name="/predict"
         )
-        with open(srt_path, "w", encoding='utf-8') as file:
+        with open(srt_file, "w", encoding='utf-8') as file:
                 file.write(result[0])
-        return srt_path
+        return srt_file

utils/zip_response.py

@@ -1,9 +1,9 @@
 import zipfile, os
 
-def zip_response(archive_name: str, files: list):
-    # Create a new zip file
-    with zipfile.ZipFile(archive_name, 'w') as zipf:
-    # Add specific files to the zip file
+def zip_response(temp_zip_file: str, files: list):
+    with zipfile.ZipFile(temp_zip_file, 'w') as zipf:
         for file in files:
             zipf.write(file, arcname=os.path.basename(file))
-    return archive_name
+    with open(temp_zip_file, 'rb') as zip_file:
+        zip_bytes = zip_file.read()
+    return zip_bytes