""" AI Graduate Job-Matching Platform Production-Safe Version with Retry & Chunk Handling Developer: Najaf Ali Sharqi MoviePy-free (uses ffmpeg + tempfile) """ import os import gradio as gr import tempfile import subprocess import math import time from groq import Groq # ============================================================================ # CONFIGURATION # ============================================================================ client = Groq(api_key=os.getenv("GROQ_API_KEY")) TRANSCRIPTION_MODEL = "whisper-large-v3" REWRITE_MODEL = "openai/gpt-oss-120b" CHUNK_SIZE = 300 # 5 min chunks for transcription to avoid rate limit REWRITE_CHUNK_SIZE = 4000 # words per rewrite chunk MAX_RETRIES = 5 # retries for 429 errors # ============================================================================ # UTILITY FUNCTIONS # ============================================================================ def get_video_duration(video_path): """Get video duration using ffprobe""" try: result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", video_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) duration = float(result.stdout.strip()) return duration except Exception as e: raise RuntimeError(f"Failed to get video duration: {str(e)}") def extract_audio_chunks(video_path, chunk_size=CHUNK_SIZE): """Extract audio from video and split into chunks""" try: duration = get_video_duration(video_path) num_chunks = math.ceil(duration / chunk_size) audio_chunks = [] # extract full audio first temp_audio_full = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) audio_full_path = temp_audio_full.name temp_audio_full.close() os.system(f"ffmpeg -y -i \"{video_path}\" -q:a 0 -map a \"{audio_full_path}\"") # split audio into chunks for i in range(num_chunks): start = i * chunk_size end = min((i + 1) * chunk_size, duration) temp_chunk = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) chunk_path = temp_chunk.name temp_chunk.close() os.system(f"ffmpeg -y -i \"{audio_full_path}\" -ss {start} -to {end} -c copy \"{chunk_path}\"") audio_chunks.append(chunk_path) # remove full audio file if os.path.exists(audio_full_path): os.unlink(audio_full_path) return audio_chunks except Exception as e: raise RuntimeError(f"Audio extraction failed: {str(e)}") # ============================================================================ # TRANSCRIPTION WITH RETRY # ============================================================================ def transcribe_audio_chunks(audio_chunks, language="English"): transcript_text = "" for chunk in audio_chunks: for attempt in range(MAX_RETRIES): try: with open(chunk, "rb") as audio_file: transcription = client.audio.transcriptions.create( file=(os.path.basename(chunk), audio_file.read()), model=TRANSCRIPTION_MODEL, temperature=0, response_format="verbose_json", language="en" if language=="English" else "ur" ) transcript_text += transcription.text + "\n\n" break # success except Exception as e: err_str = str(e) if "Rate limit reached" in err_str or "429" in err_str: wait_time = 5 + attempt * 5 print(f"Rate limit hit. Waiting {wait_time}s before retrying...") time.sleep(wait_time) else: transcript_text += f"[Error in chunk transcription: {err_str}]\n\n" break if os.path.exists(chunk): os.unlink(chunk) return transcript_text.strip() # ============================================================================ # GRAMMAR & PUNCTUATION CORRECTION (Chunked) # ============================================================================ def rewrite_transcript_chunked(text, language="English", chunk_size=REWRITE_CHUNK_SIZE): """Rewrite transcript in smaller chunks to avoid token limits""" try: words = text.split() rewritten_chunks = [] for i in range(0, len(words), chunk_size): chunk_text = " ".join(words[i:i+chunk_size]) prompt = f"Correct grammar, punctuation, and make it readable. Language: {language}\n\n{chunk_text}" for attempt in range(MAX_RETRIES): try: response = client.chat.completions.create( model=REWRITE_MODEL, messages=[{"role": "user", "content": prompt}], temperature=0 ) rewritten_chunks.append(response.choices[0].message.content) break except Exception as e: err_str = str(e) if "Rate limit reached" in err_str or "429" in err_str: wait_time = 5 + attempt*5 print(f"Rewrite rate limit hit. Waiting {wait_time}s before retrying...") time.sleep(wait_time) else: rewritten_chunks.append(f"[Error rewriting chunk: {err_str}]") break return "\n\n".join(rewritten_chunks) except Exception as e: return f"[Error rewriting transcript: {str(e)}]" def correct_transcript(transcript_text, language): lang = "English" if "English" in language else "Urdu" if not transcript_text: return "", "Please generate transcript first." if lang=="English" else "پہلے ٹرانسکرپٹ بنائیں۔" rewritten_text = rewrite_transcript_chunked(transcript_text, language=lang) return rewritten_text, "✅ Grammar & punctuation corrected / گرامر اور پنکچویشن درست کی گئی" # ============================================================================ # PROCESS VIDEO # ============================================================================ def generate_transcript(video_file, language): lang = "English" if "English" in language else "Urdu" if not video_file: return "", "Please upload a video file." if lang=="English" else "براہ کرم ایک ویڈیو فائل اپ لوڈ کریں۔" try: audio_chunks = extract_audio_chunks(video_file.name) transcript = transcribe_audio_chunks(audio_chunks, language=lang) return transcript, "✅ Transcription completed / ٹرانسکرپشن مکمل" except Exception as e: error_msg = str(e) return "", error_msg if lang=="English" else "خرابی: " + error_msg # ============================================================================ # DOWNLOAD FUNCTION # ============================================================================ def download_transcript(text, language): lang_suffix = "en" if "English" in language else "ur" temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=f"_{lang_suffix}.txt") temp_file.write(text.encode("utf-8")) temp_file.close() return temp_file.name # ============================================================================ # GRADIO UI # ============================================================================ def create_ui(): with gr.Blocks(title="AI Graduate Job Matcher") as demo: gr.HTML("

🎓 AI Graduate Job-Matching Platform

Developer: Najaf Ali Sharqi

") video_input = gr.File(label="Upload Video / ویڈیو اپ لوڈ کریں", file_types=[".mp4", ".mov", ".avi"]) language_dropdown = gr.Dropdown(choices=["English", "Urdu / اردو"], value="English", label="Output Language / آؤٹ پٹ زبان") transcript_output = gr.Textbox(label="Transcript / ٹرانسکرپٹ", lines=20) status_output = gr.Textbox(label="Status / حیثیت", lines=1) # Buttons transcribe_btn = gr.Button("🚀 Generate Transcript / ٹرانسکرپٹ بنائیں") rewrite_btn = gr.Button("✏️ Correct Grammar & Punctuation / گرامر درست کریں") download_btn = gr.Button("💾 Download Transcript / ٹرانسکرپٹ ڈاؤن لوڈ کریں") # Transcription click transcribe_btn.click( fn=generate_transcript, inputs=[video_input, language_dropdown], outputs=[transcript_output, status_output] ) # Grammar correction click rewrite_btn.click( fn=correct_transcript, inputs=[transcript_output, language_dropdown], outputs=[transcript_output, status_output] ) # Download click download_btn.click( fn=download_transcript, inputs=[transcript_output, language_dropdown], outputs=gr.File(label="Download your transcript") ) return demo # ============================================================================ # MAIN # ============================================================================ if __name__ == "__main__": if not os.getenv("GROQ_API_KEY"): print("⚠️ WARNING: GROQ_API_KEY not found!") demo = create_ui() demo.launch(server_name="0.0.0.0", server_port=7860)