Merge pull request #15 from PerfectThymeTech/marvinbuss/update_video_…

…tagging

Update Video Processing

code/durablefunction/models/videoextraction.py

@@ -21,6 +21,7 @@ def from_json(data: str):
 
 
 class VideoTimestamp(BaseModel):
+    id: int
     start_time: time
     end_time: time
 
@@ -66,6 +67,7 @@ def from_json(data: str):
 
 class ExtractVideoClipRequest(BaseModel):
     video_file_path: str
+    id: int
     start_time: time
     end_time: time
     instance_id: str
@@ -81,6 +83,7 @@ def from_json(data: str):
 
 class ExtractVideoClipResponse(BaseModel):
     video_clip_file_path: str
+    id: int
     start_time: time
     end_time: time
 
@@ -95,6 +98,7 @@ def from_json(data: str):
 
 class UploadVideoRequest(BaseModel):
     video_file_path: str
+    id: int
     start_time: time
     end_time: time
     instance_id: str
@@ -110,6 +114,7 @@ def from_json(data: str):
 
 class UploadVideoResponse(BaseModel):
     content_url_videoclip: str
+    id: int
     start_time: time
     end_time: time
 

code/durablefunction/shared/utils.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import shutil
+from urllib.parse import unquote
 
 import azure.durable_functions as df
 from azure.identity.aio import DefaultAzureCredential
@@ -23,6 +24,12 @@ async def download_blob(
     """
     logging.info(f"Start downloading file from blob storage to '{file_path}'.")
 
+    # Preprocess storage blob name
+    storage_blob_name_cleansed = unquote(storage_blob_name)
+    logging.debug(
+        f"Storage Details: Domain='{storage_domain_name}', Container='{storage_container_name}', Blob='{storage_blob_name_cleansed}'."
+    )
+
     # Create credentials
     credential = DefaultAzureCredential()
 
@@ -31,7 +38,7 @@ async def download_blob(
         f"https://{storage_domain_name}", credential=credential
     )
     blob_client = blob_service_client.get_blob_client(
-        container=storage_container_name, blob=storage_blob_name
+        container=storage_container_name, blob=storage_blob_name_cleansed
     )
 
     # Download blob
@@ -62,6 +69,12 @@ async def upload_blob(
     """
     logging.info(f"Start uploading file '{file_path}' to blob storage.")
 
+    # Preprocess storage blob name
+    storage_blob_name_cleansed = unquote(storage_blob_name)
+    logging.debug(
+        f"Storage Details: Domain='{storage_domain_name}', Container='{storage_container_name}', Blob='{storage_blob_name_cleansed}'."
+    )
+
     # Create credentials
     credential = DefaultAzureCredential()
 
@@ -76,7 +89,7 @@ async def upload_blob(
     # Upload blob
     with open(file=file_path, mode="rb") as data:
         blob_client = await container_client.upload_blob(
-            name=storage_blob_name, data=data, overwrite=True
+            name=storage_blob_name_cleansed, data=data, overwrite=True
         )
 
     logging.info(f"Finished uploading file '{file_path}' to blob storage.")
@@ -100,6 +113,12 @@ async def load_blob(
     """
     logging.info(f"Start downloading file from blob storage.")
 
+    # Preprocess storage blob name
+    storage_blob_name_cleansed = unquote(storage_blob_name)
+    logging.debug(
+        f"Storage Details: Domain='{storage_domain_name}', Container='{storage_container_name}', Blob='{storage_blob_name_cleansed}'."
+    )
+
     # Create credentials
     credential = DefaultAzureCredential()
 
@@ -108,7 +127,7 @@ async def load_blob(
         f"https://{storage_domain_name}", credential=credential
     )
     blob_client = blob_service_client.get_blob_client(
-        container=storage_container_name, blob=storage_blob_name
+        container=storage_container_name, blob=storage_blob_name_cleansed
     )
 
     # Download blob

code/durablefunction/videoextraction/orchestration.py

@@ -97,41 +97,43 @@ def video_extraction_orchestrator(context: df.DurableOrchestrationContext):
         )
     )
 
-    # Extract video clips
-    logging.info("Extract video clips")
+    # Extract video clips and upload video clips
+    logging.info("Extract video clips and upload video clips")
     utils.set_custom_status(
-        context=context, completion_percentage=25.0, status="Extracting Video Clips"
+        context=context,
+        completion_percentage=25.0,
+        status="Extracting Video Clips and Uploading Video Clips",
     )
-    tasks_extract_video_clip = []
+    results_extract_video_clip: List[ExtractVideoClipResponse] = []
+    tasks_upload_video_clips = []
     for video_timestamp in result_load_openai_content.video_timestamps:
+        logging.info(
+            f"Extract video clip: {video_timestamp.start_time}-{video_timestamp.end_time}"
+        )
         input_extract_video_clip: ExtractVideoClipRequest = ExtractVideoClipRequest(
             video_file_path=result_load_video_content.video_file_path,
+            id=video_timestamp.id,
             start_time=video_timestamp.start_time,
             end_time=video_timestamp.end_time,
             instance_id=context.instance_id,
         )
-        tasks_extract_video_clip.append(
-            context.call_activity_with_retry(
+        result_extract_video_clip: ExtractVideoClipResponse = (
+            yield context.call_activity_with_retry(
                 name="extract_video_clip",
                 retry_options=retry_options,
                 input_=input_extract_video_clip,
             )
         )
-    results_extract_video_clip: List[ExtractVideoClipResponse] = yield context.task_all(
-        tasks_extract_video_clip
-    )
+        results_extract_video_clip.append(result_extract_video_clip)
 
-    # Upload video clip
-    logging.info("Upload video clips")
-    utils.set_custom_status(
-        context=context, completion_percentage=70.0, status="Uploading Video Clips"
-    )
-    tasks_upload_video_clips = []
-    for video_clip in results_extract_video_clip:
+        logging.info(
+            f"Uploading video clip: {video_timestamp.start_time}-{video_timestamp.end_time}"
+        )
         input_upload_video: UploadVideoRequest = UploadVideoRequest(
-            video_file_path=video_clip.video_clip_file_path,
-            start_time=video_clip.start_time,
-            end_time=video_clip.end_time,
+            video_file_path=result_extract_video_clip.video_clip_file_path,
+            id=result_extract_video_clip.id,
+            start_time=result_extract_video_clip.start_time,
+            end_time=result_extract_video_clip.end_time,
             instance_id=context.instance_id,
         )
         tasks_upload_video_clips.append(
@@ -141,6 +143,11 @@ def video_extraction_orchestrator(context: df.DurableOrchestrationContext):
                 input_=input_upload_video,
             )
         )
+    utils.set_custom_status(
+        context=context,
+        completion_percentage=80.0,
+        status="Waiting for Upload of Video Clips to Complete",
+    )
     results_upload_video: List[UploadVideoResponse] = yield context.task_all(
         tasks_upload_video_clips
     )
@@ -198,6 +205,7 @@ async def load_openai_content(
     response: LoadOpenaiContentResponse = LoadOpenaiContentResponse(video_timestamps=[])
     for scene in data_obj.scenes:
         video_timestamp = VideoTimestamp(
+            id=scene.id,
             start_time=scene.start_time,
             end_time=scene.end_time,
         )
@@ -279,6 +287,7 @@ def extract_video_clip(inputData: ExtractVideoClipRequest) -> ExtractVideoClipRe
     # Generate response
     response = ExtractVideoClipResponse(
         video_clip_file_path=video_clip_file_path,
+        id=inputData.id,
         start_time=inputData.start_time,
         end_time=inputData.end_time,
     )
@@ -305,6 +314,7 @@ async def upload_video(inputData: UploadVideoRequest) -> UploadVideoResponse:
     # Generate response
     response = UploadVideoResponse(
         content_url_videoclip=result_upload_file,
+        id=inputData.id,
         start_time=inputData.start_time,
         end_time=inputData.end_time,
     )