| """ |
| 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 |
|
|
| |
| |
| |
| client = Groq(api_key=os.getenv("GROQ_API_KEY")) |
|
|
| TRANSCRIPTION_MODEL = "whisper-large-v3" |
| REWRITE_MODEL = "openai/gpt-oss-120b" |
| CHUNK_SIZE = 300 |
| REWRITE_CHUNK_SIZE = 4000 |
| MAX_RETRIES = 5 |
|
|
| |
| |
| |
| 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 = [] |
|
|
| |
| 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}\"") |
|
|
| |
| 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) |
|
|
| |
| 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)}") |
|
|
| |
| |
| |
| 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 |
| 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() |
|
|
| |
| |
| |
| 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 / گرامر اور پنکچویشن درست کی گئی" |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| 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 |
|
|
| |
| |
| |
| def create_ui(): |
| with gr.Blocks(title="AI Graduate Job Matcher") as demo: |
| gr.HTML("<h1>🎓 AI Graduate Job-Matching Platform</h1><p>Developer: Najaf Ali Sharqi</p>") |
| |
| 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) |
| |
| |
| transcribe_btn = gr.Button("🚀 Generate Transcript / ٹرانسکرپٹ بنائیں") |
| rewrite_btn = gr.Button("✏️ Correct Grammar & Punctuation / گرامر درست کریں") |
| download_btn = gr.Button("💾 Download Transcript / ٹرانسکرپٹ ڈاؤن لوڈ کریں") |
| |
| |
| transcribe_btn.click( |
| fn=generate_transcript, |
| inputs=[video_input, language_dropdown], |
| outputs=[transcript_output, status_output] |
| ) |
| |
| |
| rewrite_btn.click( |
| fn=correct_transcript, |
| inputs=[transcript_output, language_dropdown], |
| outputs=[transcript_output, status_output] |
| ) |
| |
| |
| download_btn.click( |
| fn=download_transcript, |
| inputs=[transcript_output, language_dropdown], |
| outputs=gr.File(label="Download your transcript") |
| ) |
| |
| return demo |
|
|
| |
| |
| |
| 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) |
|
|
