Spaces:
Running
Running
File size: 12,004 Bytes
3066087 c37b36e d85921f 35b7e36 3066087 1ab9028 d85921f c37b36e d85921f c37b36e d5b84f5 3066087 d85921f 3066087 d85921f 3066087 d85921f 3066087 d85921f d5b84f5 c37b36e d85921f c37b36e 3066087 c37b36e d85921f 3066087 c37b36e 3066087 c37b36e d85921f c37b36e d842bdb c37b36e d842bdb c37b36e d842bdb bb47241 c37b36e d85921f c37b36e d85921f c37b36e 3066087 c37b36e d85921f c37b36e 3066087 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 |
# import os
# import subprocess
# import whisper
# import requests
# import tempfile
# import warnings
# import threading
# from flask import Flask, request, jsonify, send_file, render_template
# from dotenv import load_dotenv
# import requests
# warnings.filterwarnings("ignore", category=UserWarning, module="whisper")
# app = Flask(__name__)
# # Gemini API settings
# load_dotenv()
# API_KEY = os.getenv("FIRST_API_KEY")
# # Ensure the API key is loaded correctly
# if not API_KEY:
# raise ValueError("API Key not found. Make sure it is set in the .env file.")
# GEMINI_API_ENDPOINT = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent"
# GEMINI_API_KEY = API_KEY
# # Load Whisper AI model at startup
# print("Loading Whisper AI model...")
# whisper_model = whisper.load_model("base")
# print("Whisper AI model loaded successfully.")
# # Define the "/" endpoint for health check
# @app.route("/", methods=["GET"])
# def health_check():
# return jsonify({"status": "success", "message": "API is running successfully!"}), 200
# def process_video_in_background(video_file, temp_video_file_name):
# """
# This function is executed in a separate thread to handle the long-running
# video processing tasks such as transcription and querying the Gemini API.
# """
# try:
# transcription = transcribe_audio(temp_video_file_name)
# if not transcription:
# print("Audio transcription failed")
# return
# structured_data = query_gemini_api(transcription)
# # Send structured data back or store it in a database, depending on your use case
# print("Processing complete. Structured data:", structured_data)
# except Exception as e:
# print(f"Error processing video: {e}")
# finally:
# # Clean up temporary files
# if os.path.exists(temp_video_file_name):
# os.remove(temp_video_file_name)
# @app.route('/process-video', methods=['POST'])
# def process_video():
# if 'video' not in request.files:
# return jsonify({"error": "No video file provided"}), 400
# video_file = request.files['video']
# try:
# # Save video to a temporary file
# with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as temp_video_file:
# video_file.save(temp_video_file.name)
# print(f"Video file saved: {temp_video_file.name}")
# # Start the video processing in a background thread
# threading.Thread(target=process_video_in_background, args=(video_file, temp_video_file.name)).start()
# return jsonify({"message": "Video is being processed in the background."}), 202
# except Exception as e:
# return jsonify({"error": str(e)}), 500
# def transcribe_audio(video_path):
# """
# Transcribe audio directly from a video file using Whisper AI.
# """
# try:
# print(f"Transcribing video: {video_path}")
# result = whisper_model.transcribe(video_path)
# return result['text']
# except Exception as e:
# print(f"Error in transcription: {e}")
# return None
# def query_gemini_api(transcription):
# """
# Send transcription text to Gemini API and fetch structured recipe information.
# """
# try:
# # Define the structured prompt
# prompt = (
# "Analyze the provided cooking video transcription and extract the following structured information:\n"
# "1. Recipe Name: Identify the name of the dish being prepared.\n"
# "2. Ingredients List: Extract a detailed list of ingredients with their respective quantities (if mentioned).\n"
# "3. Steps for Preparation: Provide a step-by-step breakdown of the recipe's preparation process, organized and numbered sequentially.\n"
# "4. Cooking Techniques Used: Highlight the cooking techniques demonstrated in the video, such as searing, blitzing, wrapping, etc.\n"
# "5. Equipment Needed: List all tools, appliances, or utensils mentioned, e.g., blender, hot pan, cling film, etc.\n"
# "6. Nutritional Information (if inferred): Provide an approximate calorie count or nutritional breakdown based on the ingredients used.\n"
# "7. Serving size: In count of people or portion size.\n"
# "8. Special Notes or Variations: Include any specific tips, variations, or alternatives mentioned.\n"
# "9. Festive or Thematic Relevance: Note if the recipe has any special relevance to holidays, events, or seasons.\n"
# f"Text: {transcription}\n"
# )
# payload = {
# "contents": [
# {"parts": [{"text": prompt}]}
# ]
# }
# headers = {"Content-Type": "application/json"}
# # Send request to Gemini API
# response = requests.post(
# f"{GEMINI_API_ENDPOINT}?key={GEMINI_API_KEY}",
# json=payload,
# headers=headers
# )
# response.raise_for_status()
# # Extract and return the structured data
# data = response.json()
# return data.get("candidates", [{}])[0].get("content", {}).get("parts", [{}])[0].get("text", "No result found")
# except requests.exceptions.RequestException as e:
# print(f"Error querying Gemini API: {e}")
# return {"error": str(e)}
# if __name__ == '__main__':
# app.run(debug=True)
# Above code is without polling and sleep
# Below is the latest code
import os
import whisper
import requests
import tempfile
import warnings
import threading
import time
from flask import Flask, request, jsonify
from dotenv import load_dotenv
warnings.filterwarnings("ignore", category=UserWarning, module="whisper")
app = Flask(__name__)
# Gemini API settings
load_dotenv()
API_KEY = os.getenv("FIRST_API_KEY")
# Ensure the API key is loaded correctly
if not API_KEY:
raise ValueError("API Key not found. Make sure it is set in the .env file.")
GEMINI_API_ENDPOINT = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent"
GEMINI_API_KEY = API_KEY
# Load Whisper AI model at startup
print("Loading Whisper AI model...")
whisper_model = whisper.load_model("base")
print("Whisper AI model loaded successfully.")
# Define the "/" endpoint for health check
@app.route("/", methods=["GET"])
def health_check():
return jsonify({"status": "success", "message": "API is running successfully!"}), 200
def process_video_in_background(video_file, temp_video_file_name, result_container):
"""
This function is executed in a separate thread to handle the long-running
video processing tasks such as transcription and querying the Gemini API.
"""
try:
transcription = transcribe_audio(temp_video_file_name)
if not transcription:
result_container["error"] = "Audio transcription failed"
return
structured_data = query_gemini_api(transcription)
# Save structured data to the result container to return later
result_container["data"] = structured_data
except Exception as e:
result_container["error"] = f"Error processing video: {e}"
finally:
# Clean up temporary files
if os.path.exists(temp_video_file_name):
os.remove(temp_video_file_name)
@app.route('/process-video', methods=['POST'])
def process_video():
if 'video' not in request.files:
return jsonify({"error": "No video file provided"}), 400
video_file = request.files['video']
result_container = {}
try:
# Save video to a temporary file
with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as temp_video_file:
video_file.save(temp_video_file.name)
print(f"Video file saved: {temp_video_file.name}")
# Start the video processing in a background thread
threading.Thread(target=process_video_in_background, args=(video_file, temp_video_file.name, result_container)).start()
# Poll every 5 seconds to check if the result is available
while "data" not in result_container and "error" not in result_container:
print("Waiting for processing to complete...")
time.sleep(5) # Sleep for 5 seconds before checking again
# Check for the result
if "error" in result_container:
return jsonify({"error": result_container["error"]}), 500
else:
return jsonify({"message": "Processing complete", "data": result_container["data"]}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
def transcribe_audio(video_path):
"""
Transcribe audio directly from a video file using Whisper AI.
"""
try:
print(f"Transcribing video: {video_path}")
result = whisper_model.transcribe(video_path)
return result['text']
except Exception as e:
print(f"Error in transcription: {e}")
return None
def query_gemini_api(transcription):
"""
Send transcription text to Gemini API and fetch structured recipe information.
"""
try:
# Define the structured prompt
prompt = (
"Analyze the provided cooking video transcription and extract the following structured information:\n"
"1. Recipe Name: Identify the name of the dish being prepared.\n"
"2. Ingredients List: Extract a detailed list of ingredients with their respective quantities (if mentioned).\n"
"3. Steps for Preparation: Provide a step-by-step breakdown of the recipe's preparation process, organized and numbered sequentially.\n"
"4. Cooking Techniques Used: Highlight the cooking techniques demonstrated in the video, such as searing, blitzing, wrapping, etc.\n"
"5. Equipment Needed: List all tools, appliances, or utensils mentioned, e.g., blender, hot pan, cling film, etc.\n"
"6. Nutritional Information (if inferred): Provide an approximate calorie count or nutritional breakdown based on the ingredients used.\n"
"7. Serving size: In count of people or portion size.\n"
"8. Special Notes or Variations: Include any specific tips, variations, or alternatives mentioned.\n"
"9. Festive or Thematic Relevance: Note if the recipe has any special relevance to holidays, events, or seasons.\n"
f"Text: {transcription}\n"
)
payload = {
"contents": [
{"parts": [{"text": prompt}]}
]
}
headers = {"Content-Type": "application/json"}
# Send request to Gemini API
response = requests.post(
f"{GEMINI_API_ENDPOINT}?key={GEMINI_API_KEY}",
json=payload,
headers=headers
)
response.raise_for_status()
# Polling for response (in case Gemini takes time to process)
polling_wait_time = 5 # Time to wait between polling attempts
polling_max_retries = 60 # Maximum number of retries
for attempt in range(polling_max_retries):
print(f"Attempt {attempt + 1} to fetch Gemini API response...")
response_data = response.json()
# Check if the response is ready
if "candidates" in response_data and len(response_data["candidates"]) > 0:
return response_data["candidates"][0].get("content", {}).get("parts", [{}])[0].get("text", "No result found")
time.sleep(polling_wait_time) # Wait before trying again
return "Gemini API response not ready after multiple attempts."
except requests.exceptions.RequestException as e:
print(f"Error querying Gemini API: {e}")
return {"error": str(e)}
if __name__ == '__main__':
app.run(debug=True)
|