Add script to create videos for speaking questions.

helper/file_helper.py

@@ -9,8 +9,9 @@ def delete_files_older_than_one_day(directory):
     for entry in os.scandir(directory):
         if entry.is_file():
             file_path = Path(entry)
+            file_name = file_path.name
             file_modified_time = datetime.datetime.fromtimestamp(file_path.stat().st_mtime)
             time_difference = current_time - file_modified_time
-            if time_difference.days > 1:
+            if time_difference.days > 1  and "placeholder" not in file_name:
                 file_path.unlink()
                 print(f"Deleted file: {file_path}")

helper/firebase_helper.py

@@ -22,5 +22,24 @@ def upload_file_firebase(bucket_name, destination_blob_name, source_file_name):
         print("Error uploading file to Google Cloud Storage:", e)
         return False
 
+def upload_file_firebase_get_url(bucket_name, destination_blob_name, source_file_name):
+    # Uploads a file to Firebase Storage.
+    storage_client = storage.Client()
+    bucket = storage_client.bucket(bucket_name)
+    try:
+        blob = bucket.blob(destination_blob_name)
+        blob.upload_from_filename(source_file_name)
+        print(f"File uploaded to {destination_blob_name}")
+
+        # Make the file public
+        blob.make_public()
+
+        # Get the public URL
+        url = blob.public_url
+        return url
+    except Exception as e:
+        print("Error uploading file to Google Cloud Storage:", e)
+        return None
+
 
 

helper/heygen_api.py

@@ -0,0 +1,87 @@
+import os
+import requests
+import time
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Get HeyGen token
+TOKEN = os.getenv("HEY_GEN_TOKEN")
+
+# POST TO CREATE VIDEO
+CREATE_VIDEO_URL = 'https://api.heygen.com/v1/video.generate'
+GET_VIDEO_URL = 'https://api.heygen.com/v1/video_status.get'
+POST_HEADER = {
+    'X-Api-Key': TOKEN,
+    'Content-Type': 'application/json'
+}
+GET_HEADER = {
+    'X-Api-Key': TOKEN
+}
+
+def create_video(text):
+    # POST TO CREATE VIDEO
+    data = {
+        "background": "#ffffff",
+        "clips": [
+            {
+                "avatar_id": "Mido-lite-20221128",
+                "avatar_style": "normal",
+                "input_text": text,
+                "offset": {
+                    "x": 0,
+                    "y": 0
+                },
+                "scale": 1,
+                "voice_id": "ccb30e87c6b34ca8941f88352c71612d"
+            }
+        ],
+        "ratio": "16:9",
+        "test": True,
+        "version": "v1alpha"
+    }
+    response = requests.post(CREATE_VIDEO_URL, headers=POST_HEADER, json=data)
+    print(response.status_code)
+    print(response.json())
+
+    # GET TO CHECK STATUS AND GET VIDEO WHEN READY
+    video_id = response.json()["data"]["video_id"]
+    params = {
+        'video_id': response.json()["data"]["video_id"]
+    }
+    response = {}
+    status = "processing"
+    error = None
+
+    while status != "completed" and error is None:
+        response = requests.get(GET_VIDEO_URL, headers=GET_HEADER, params=params)
+        response_data = response.json()
+
+        status = response_data["data"]["status"]
+        error = response_data["data"]["error"]
+
+        if status != "completed" and error is None:
+            print(f"Status: {status}")
+            time.sleep(5)  # Wait for 5 second before the next request
+
+    print(response.status_code)
+    print(response.json())
+
+    # DOWNLOAD VIDEO
+    download_url = response.json()['data']['video_url']
+    output_directory = 'download-video/'
+    output_filename = video_id + '.mp4'
+
+    response = requests.get(download_url)
+
+    if response.status_code == 200:
+        os.makedirs(output_directory, exist_ok=True)  # Create the directory if it doesn't exist
+        output_path = os.path.join(output_directory, output_filename)
+        with open(output_path, 'wb') as f:
+            f.write(response.content)
+        print(f"File '{output_filename}' downloaded successfully.")
+        return output_filename
+    else:
+        print(f"Failed to download file. Status code: {response.status_code}")
+        return None