Spaces:
Running
Running
File size: 5,529 Bytes
4cc0ea8 |
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 |
import os
import tempfile
import subprocess
from typing import Optional, Tuple, List
import pytube
from src.video_processing import extract_audio_from_video
from src.quiz_processing import analyze_document
import docx
import PyPDF2
import re
def parse_quiz_content(quiz_text):
questions = []
lines = quiz_text.split('\n')
current_question = None
for line in lines:
line = line.strip()
if not line:
continue
q_match = re.match(r'^(?:\d+\.|\[?Q\d+\]?\.?)\s+(.*)', line, re.IGNORECASE)
if q_match:
if current_question:
questions.append(current_question)
current_question = {"question": q_match.group(1), "answer": ""}
elif current_question and line.lower().startswith(("answer:", "a:", "ans:")):
answer_text = re.sub(r'^(?:answer:|a:|ans:)\s*', '', line, flags=re.IGNORECASE)
current_question["answer"] = answer_text.strip()
if current_question:
questions.append(current_question)
return {"questions": questions}
def transcribe_audio(audio_path, elevenlabs_api_key, model_id):
import requests
import json
try:
with open(audio_path, 'rb') as audio_file:
response = requests.post(
'https://api.elevenlabs.io/v1/transcribe',
headers={'xi-api-key': elevenlabs_api_key},
files={'audio': audio_file},
data={'model_id': model_id}
)
if response.status_code == 200:
transcription = response.json().get('transcription', '')
transcript_path = tempfile.mktemp(suffix='.txt')
with open(transcript_path, 'w', encoding='utf-8') as f:
f.write(transcription)
return transcription, transcript_path, "Transcription completed successfully"
else:
return None, None, f"Transcription failed: {response.text}"
except Exception as e:
return None, None, f"Transcription error: {str(e)}"
def process_video_file(video_path, audio_format, elevenlabs_api_key, model_id, gemini_api_key, language, content_type):
try:
audio_path = extract_audio_from_video(video_path, audio_format)
transcription, transcript_path, transcription_status = transcribe_audio(
audio_path,
elevenlabs_api_key,
model_id
)
if not transcription:
return audio_path, "Audio extracted, but transcription failed", None, transcription_status, None, None, None
formatted_output, json_path, txt_path = analyze_document(
transcription,
gemini_api_key,
language,
content_type
)
return audio_path, "Processing completed successfully", transcript_path, transcription_status, formatted_output, txt_path, json_path
except Exception as e:
error_message = f"Error processing video: {str(e)}"
return None, error_message, None, error_message, error_message, None, None
def process_youtube_video(youtube_url, audio_format, elevenlabs_api_key, model_id, gemini_api_key, language, content_type):
try:
yt = pytube.YouTube(youtube_url)
stream = yt.streams.filter(progressive=True, file_extension='mp4').order_by('resolution').desc().first()
if not stream:
raise Exception("No suitable video stream found")
# Download to temporary file
video_path = tempfile.mktemp(suffix='.mp4')
stream.download(filename=video_path)
audio_path = extract_audio_from_video(video_path, audio_format)
transcription, transcript_path, transcription_status = transcribe_audio(
audio_path,
elevenlabs_api_key,
model_id
)
if not transcription:
return audio_path, "Audio extracted, but transcription failed", None, transcription_status, None, None, None
formatted_output, json_path, txt_path = analyze_document(
transcription,
gemini_api_key,
language,
content_type
)
return audio_path, "Processing completed successfully", transcript_path, transcription_status, formatted_output, txt_path, json_path
except Exception as e:
error_message = f"Error processing YouTube video: {str(e)}"
return None, error_message, None, error_message, error_message, None, None
def process_audio_document(audio_path, elevenlabs_api_key, model_id, gemini_api_key, language, content_type):
"""Process an audio file - transcribe and generate summary or quiz."""
try:
transcription, transcript_path, transcription_status = transcribe_audio(
audio_path,
elevenlabs_api_key,
model_id
)
if not transcription:
return "Transcription failed", None, None, None, None
formatted_output, json_path, txt_path = analyze_document(
transcription,
gemini_api_key,
language,
content_type
)
return "Processing completed successfully", transcript_path, formatted_output, txt_path, json_path
except Exception as e:
error_message = f"Error processing audio: {str(e)}"
return error_message, None, error_message, None, None
|