Coder19/interview_system
0
1from interview.app.services.openai import InterviewAI2from interview.app.services.monitor import InterviewMonitor3import json4from fastapi import WebSocket5 6class InterviewService:7 def __init__(self):8 try:9 self.openai = InterviewAI()10 self.monitor = InterviewMonitor()11 self.websocket = None12 self.current_context = []13 self.is_interview_ended = False14 except ValueError as e:15 print(f"Error initializing OpenAI service: {e}")16 self.openai = None17 18 async def start_interview(self, websocket: WebSocket) -> None:19 try:20 self.websocket = websocket21 self.is_interview_ended = False22 self.monitor.reset()23 24 if not self.openai:25 await self.send_message("Error: OpenAI service is not properly configured. Please check your API key.")26 return27 28 # Start the interview with a greeting29 greeting = "Hello! I'm your AI interviewer today. I'll be asking you questions and we'll have a conversation. When you're ready to end the interview, just type 'end interview' and I'll provide feedback. Are you ready to begin?"30 await self.send_message(greeting)31 32 except Exception as e:33 print(f"Error in interview: {e}")34 35 async def handle_user_response(self, response: str) -> None:36 if not self.openai:37 await self.send_message("Error: OpenAI service is not properly configured. Please check your API key.")38 return39 40 try:41 # Check if user wants to end the interview42 if response.lower().strip() == "end interview":43 await self.end_interview()44 return45 46 # Monitor response for suspicious activity47 suspicious = self.monitor.record_response_time(response)48 if suspicious and suspicious['severity'] in ['high', 'medium']:49 print(f"Suspicious activity detected: {suspicious['details']}")50 51 # Add user's response to context52 self.current_context.append({"role": "user", "content": response})53 54 # Generate next question or response using OpenAI55 ai_response = await self.openai.generate_response(self.current_context)56 57 # Add AI's response to context58 self.current_context.append({"role": "assistant", "content": ai_response})59 60 # Send the response back to the client61 await self.send_message(ai_response)62 63 except Exception as e:64 print(f"Error handling user response: {e}")65 error_message = "I apologize, but I encountered an error. Could you please try again?"66 await self.send_message(error_message)67 68 async def detect_cheating(self, video_frame: dict):69 """Process video frame with YOLOv8 and return detection results"""70 try:71 # Extract image data from video frame72 image_data = video_frame.get('image_data')73 if not image_data:74 return []75 76 # Run YOLOv8 detection77 detections = await self.cheating_detector.detect_objects(image_data)78 79 # Filter detections for suspicious objects80 suspicious_detections = []81 for detection in detections:82 if detection['confidence'] > 0.5 and detection['class'] != 'person':83 suspicious_detections.append({84 'class': detection['class'],85 'confidence': detection['confidence'],86 'bbox': detection['bbox']87 })88 89 return suspicious_detections90 91 except Exception as e:92 print(f"Error in YOLOv8 detection: {e}")93 return []94 95 async def handle_audio_data(self, audio_data: dict) -> None:96 """Handle incoming audio data for monitoring"""97 suspicious = self.monitor.analyze_audio(audio_data)98 if suspicious and suspicious['severity'] in ['high', 'medium']:99 print(f"Suspicious audio activity: {suspicious['details']}")100 101 async def handle_typing_data(self, keystrokes: list) -> None:102 """Handle typing pattern data for monitoring"""103 suspicious = self.monitor.analyze_typing_pattern(keystrokes)104 if suspicious and suspicious['severity'] in ['high', 'medium']:105 print(f"Suspicious typing activity: {suspicious['details']}")106 107 async def end_interview(self) -> None:108 try:109 await self.send_message("Thank you for participating in the interview. I'll now provide you with feedback...")110 111 # Get cheating detection report112 cheating_report = self.monitor.get_cheating_report()113 114 # Generate feedback using the conversation history and cheating report115 feedback_prompt = {116 "role": "system",117 "content": f"""Please provide constructive feedback for this interview. Consider:118 1. Technical knowledge demonstrated119 2. Communication clarity120 3. Problem-solving approach121 4. Areas of strength122 5. Areas for improvement123 124 Additionally, the following suspicious activities were detected:125 - Total suspicious activities: {cheating_report['total_suspicious_activities']}126 - High severity issues: {cheating_report['severity_breakdown']['high']}127 - Medium severity issues: {cheating_report['severity_breakdown']['medium']}128 - Low severity issues: {cheating_report['severity_breakdown']['low']}129 130 Recommendation: {cheating_report['recommendation']}131 132 Please incorporate this information into your feedback professionally and constructively.133 Keep the feedback professional, specific, and actionable."""134 }135 136 # Add feedback prompt to conversation137 feedback_context = self.current_context + [feedback_prompt]138 feedback = await self.openai.generate_response(feedback_context)139 140 # Send feedback141 await self.send_message("\nInterview Feedback\n" + feedback)142 143 # Mark interview as ended144 self.is_interview_ended = True145 146 except Exception as e:147 print(f"Error generating feedback: {e}")148 await self.send_message("I apologize, but I encountered an error generating feedback.")149 150 async def send_message(self, text: str) -> None:151 if self.websocket:152 try:153 await self.websocket.send_json({154 "type": "bot_response",155 "text": text156 })157 except Exception as e:158 print(f"Error sending message: {e}")159 160 async def cleanup(self) -> None:161 self.websocket = None162 self.current_context = []163 self.is_interview_ended = False164 self.monitor.reset()