import gc import os import json from fastapi.responses import JSONResponse import httpx import torch import requests import ffmpeg from typing import List, Dict, Any, Optional from fastapi import FastAPI, HTTPException, Body, status from pydantic import BaseModel, Field import whisperx from groq import Groq import uvicorn from dotenv import load_dotenv from fastapi.middleware.cors import CORSMiddleware from fastapi import Request # Initialize FastAPI app = FastAPI() # Initialize Groq client load_dotenv() API_KEY = os.getenv("GROQ_API_KEY") client = Groq(api_key=API_KEY) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Set device and compute type for Whisper device = "cpu" batch_size = 4 compute_type = "int8" # Define Pydantic models class VideoData(BaseModel): question: str videoUrl: str class RequestModel(BaseModel): candidateId: int jd: str apiUrl: str data: List[VideoData] class IndividualEvaluation(BaseModel): question: str answer: str rating: int = Field(ge=0, le=5) explanation: Optional[str] = "No explanation provided." class Evaluation(BaseModel): candidateId: int individualEvaluations: List[IndividualEvaluation] overallRating: int = Field(ge=0, le=5) performanceSummary: str hiringRecommendation: str class Config: extra = "forbid" # Function to download video and extract audio def get_video_audio(video_url: str, video_path: str, audio_path: str): try: print("Downloading Video....") response = requests.get(video_url, stream=True) response.raise_for_status() with open(video_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) print("Download Complete !!!") print("Extracting Audio...") ( ffmpeg .input(video_path) .output(audio_path, format="wav", acodec="libmp3lame") .run(overwrite_output=True, quiet=True) ) print(f"Audio extracted and saved as {audio_path}") os.remove(video_path) except requests.exceptions.RequestException as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Failed to download video: {str(e)}" ) except ffmpeg.Error as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to extract audio: {str(e)}" ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An unexpected error occurred: {str(e)}" ) # Function to transcribe audio using WhisperX def transcribe_audio(audio_file_path: str): try: whisperx_model = whisperx.load_model("large-v2", device, compute_type=compute_type) audio = whisperx.load_audio(audio_file_path) result = whisperx_model.transcribe(audio, batch_size=batch_size) gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() del whisperx_model return result except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to transcribe audio: {str(e)}" ) # Function to evaluate interview using Groq API def evaluate_interview(job_description: str, qa_dict: Dict[str, str]): print(job_description) print(qa_dict) try: prompt = f""" You are an AI interviewer evaluating a candidate based on the following job description: {job_description} Evaluate the candidate's responses and provide: - A rating (0-5) for each response based on relevance, clarity, and correctness (must be whole numbers only). - A detailed explanation for each rating. After evaluating all responses, provide: - An **overall rating** (0-5) for the candidate's performance using whole numbers only. - A **summary** of strengths and weaknesses. - A **final hiring recommendation** (e.g., "Strongly Recommend", "Consider with Reservations", "Not Recommended") based on the job description. **Return the response in valid JSON format.** Do not include Markdown formatting (`json`, `\`\`\``, etc.). JSON Output Example: {{ "individualEvaluations": [ {{ "question": "", "answer": "", "rating": <0-5>, "explanation": "" }}, ... ], "overallRating": <0-5>, "performanceSummary": "", "hiringRecommendation": "" }} Candidate Responses: """ for question, answer in qa_dict.items(): prompt += f"\nQuestion: {question}\nAnswer: {answer}\n" # Call Groq API completion = client.chat.completions.create( model="llama-3.3-70b-versatile", messages=[ {"role": "system", "content": "You are an expert interviewer providing structured interview evaluations."}, {"role": "user", "content": prompt} ], temperature=0.7, max_completion_tokens=1500, top_p=1, stream=False, ) response_text = completion.choices[0].message.content try: evaluation_response = json.loads(response_text) # Ensure the actual questions are used in the evaluation response for eval in evaluation_response.get("individualEvaluations", []): eval["question"] = qa_dict.get(eval["question"], eval["question"]) return evaluation_response except json.JSONDecodeError: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="AI response is not valid JSON" ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to evaluate interview: {str(e)}" ) # Function to standardize individual evaluations def standardize_individual_evaluations(raw_evaluations: List[dict], qa_dict: Dict[str, str]) -> List[IndividualEvaluation]: standardized_evaluations = [] for eval in raw_evaluations: try: # Ensure the actual question is used actual_question = qa_dict.get(eval["question"], eval["question"]) standardized_eval = IndividualEvaluation( question=actual_question, answer=eval.get("answer", "No Answer Provided"), rating=eval.get("rating", 0), explanation=eval.get("explanation", "No explanation provided.") ) standardized_evaluations.append(standardized_eval) except Exception as e: print(f"Skipping invalid evaluation due to error: {e}") return standardized_evaluations # Function to standardize the entire evaluation response def standardize_evaluation_response(model_response: Dict[str, Any], candidate_id: int, qa_dict: Dict[str, str]) -> Evaluation: try: raw_individual_evaluations = model_response.get("individualEvaluations", []) standardized_individual_evaluations = standardize_individual_evaluations(raw_individual_evaluations, qa_dict) overallRating = model_response.get("overallRating", 0) performanceSummary = model_response.get("performanceSummary", "No summary provided.") hiringRecommendation = model_response.get("hiringRecommendation", "Not Recommended") evaluation = Evaluation( candidateId=candidate_id, individualEvaluations=standardized_individual_evaluations, overallRating=overallRating, performanceSummary=performanceSummary, hiringRecommendation=hiringRecommendation, ) return evaluation except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to standardize evaluation response: {str(e)}" ) # API Endpoints @app.get("/") async def root(): return {"message": "Hello"} @app.post("/get-report") async def report_process(interview_data: RequestModel = Body(...)): try: job_description = interview_data.jd qa_dict = {} apiUrl = interview_data.apiUrl for idx, item in enumerate(interview_data.data, start=1): video_url = item.videoUrl video_path = f"videos/video_ques_{idx}.mp4" audio_path = f"audios/audio_ques_{idx}.wav" get_video_audio(video_url, video_path, audio_path) transcribe_result = transcribe_audio(audio_path) detected_lang = transcribe_result.get('language', '').lower() answer = " ".join(seg["text"] for seg in transcribe_result["segments"]) if detected_lang != 'en' or len(answer.strip()) < 3: answer = "No answer given by candidate." else: answer = answer qa_dict[item.question.strip()] = answer.strip() os.remove(audio_path) evaluation_response = evaluate_interview(job_description, qa_dict) if evaluation_response: if "apiUrl" in evaluation_response: del evaluation_response["apiUrl"] standardized_evaluation = standardize_evaluation_response(evaluation_response, interview_data.candidateId, qa_dict) print("Found response, will send to PUT request") put_payload = standardized_evaluation.dict() put_request_payload = { "success": True, "statusCode": 200, "message": "Interview Report Generated", "data": put_payload } async with httpx.AsyncClient() as client: print("Going to send response to PUT Request") # Create the new format for the PUT request input put_request_payload = { "success": True, "statusCode": 200, "message": "Interview Report Generated", "data": put_payload } # Debug: Print API URL and payload print(f"API URL: {apiUrl}") print(f"Payload being sent: {json.dumps(put_request_payload, indent=4)}") print(put_request_payload) try: put_response = await client.put( apiUrl, json=put_request_payload ) print("Response sent to PUT successfully!!") print(f"Status Code: {put_response.status_code}") print(f"Headers: {put_response.headers}") put_response.raise_for_status() try: response_json = put_response.json() print("Response JSON:", json.dumps(response_json, indent=4)) except Exception: print("Response Text:", put_response.text) return { "success": True, "statusCode": 200, "message": "Interview Report Generated", "data": standardized_evaluation.dict(exclude={"apiUrl"}) } except httpx.HTTPStatusError as e: print(f"HTTP Error: {e.response.status_code} - {e.response.text}") return { "success": False, "statusCode": e.response.status_code, "message": f"Error sending to {apiUrl}: {e.response.text}" } except httpx.RequestError as e: print(f"Request Error: {str(e)}") return { "success": False, "statusCode": 404, "message": f"Request Error: {str(e)}" } except Exception as e: print(f"Unexpected Error: {str(e)}") return { "success": False, "statusCode": 500, "message": f"Unexpected Error: {str(e)}" } # If PUT request fails or there was no valid response return { "success": False, "statusCode": 404, "message": "Failed to generate interview report" } except httpx.HTTPStatusError as e: return { "success": False, "statusCode": e.response.status_code, "message": f"Error sending to {apiUrl}: {e.response.text}" } except Exception as e: return { "success": False, "statusCode": 500, "message": f"An unexpected error occurred: {str(e)}" } if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=6263, reload=True)