fix errors and adding mistral ai

backend/.env.example

@@ -21,7 +21,10 @@ APP_NAME=AI-Powered Hiring Assessment Platform
 APP_VERSION=0.1.0
 APP_DESCRIPTION=MVP for managing hiring assessments using AI
 
+
+
 # AI Provider Configuration (for future use)
+MISTRAL_API_KEY=your_mistral_api_key
 OPENAI_API_KEY=
 ANTHROPIC_API_KEY=
 GOOGLE_AI_API_KEY=

backend/api/application_routes.py

@@ -16,7 +16,7 @@ logger = get_logger(__name__)
 
 router = APIRouter(prefix="/applications", tags=["applications"])
 
-@router.get("/jobs/{jid}/assessments/{aid}", response_model=ApplicationDetailedListResponse)
+@router.get("/jobs/{jid}/assessments/{aid}")
 def get_applications_list(jid: str, aid: str, page: int = 1, limit: int = 10, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
     """Get list of applications for an assessment"""
     logger.info(f"Retrieving applications list for job ID: {jid}, assessment ID: {aid}, page: {page}, limit: {limit} by user: {current_user.id}")
@@ -33,28 +33,170 @@ def get_applications_list(jid: str, aid: str, page: int = 1, limit: int = 10, db
     # Calculate total count
     total = len(get_applications_by_job_and_assessment(db, jid, aid, skip=0, limit=1000))  # Simplified for demo
 
-    # Convert answers from JSON string to list and calculate scores
+    # Get the assessment to retrieve the passing score
+    assessment = get_assessment(db, aid)
+    if not assessment:
+        logger.error(f"Assessment not found for ID: {aid}")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Assessment not found"
+        )
+
+    # Calculate scores and create responses
     application_responses = []
     for application in applications:
-        application_dict = application.__dict__
-        if application.answers:
-            application_dict['answers'] = json.loads(application.answers)
-        else:
-            application_dict['answers'] = []
+        # Calculate score
+        score = calculate_application_score(db, application.id)
 
-        # Calculate score (placeholder)
-        application_dict['score'] = calculate_application_score(db, application.id)
-        application_dict['passing_score'] = 0.0  # Placeholder
+        # Get user information
+        from services.user_service import get_user
+        user = get_user(db, application.user_id)
 
-        application_responses.append(ApplicationResponse(**application_dict))
+        # Create response object that matches technical requirements exactly
+        application_response = {
+            'id': application.id,
+            'score': score,
+            'passing_score': assessment.passing_score,
+            'user': {
+                'id': user.id if user else None,
+                'first_name': user.first_name if user else None,
+                'last_name': user.last_name if user else None,
+                'email': user.email if user else None
+            } if user else None
+        }
+
+        application_responses.append(application_response)
 
     logger.info(f"Successfully retrieved {len(applications)} applications out of total {total} for job ID: {jid}, assessment ID: {aid}")
-    return ApplicationDetailedListResponse(
-        count=len(applications),
-        total=total,
-        data=application_responses
+    return {
+        'count': len(applications),
+        'total': total,
+        'data': application_responses
+    }
+
+@router.get("/jobs/{jid}/assessment_id/{aid}/applications/{id}", response_model=ApplicationDetailedResponse)
+def get_application_detail(jid: str, aid: str, id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
+    """Get detailed application information including answers"""
+    logger.info(f"Retrieving application detail for job ID: {jid}, assessment ID: {aid}, application ID: {id} by user: {current_user.id}")
+    # Only HR users can view application details
+    if current_user.role != "hr":
+        logger.warning(f"Unauthorized attempt to view application detail by user: {current_user.id} with role: {current_user.role}")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only HR users can view application details"
+        )
+
+    # Get the application
+    application = get_application(db, id)
+    if not application or application.job_id != jid or application.assessment_id != aid:
+        logger.warning(f"Application not found for job ID: {jid}, assessment ID: {aid}, application ID: {id}")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Application not found for this job and assessment"
+        )
+
+    # Get the assessment to retrieve the passing score
+    assessment = get_assessment(db, aid)
+    if not assessment:
+        logger.error(f"Assessment not found for ID: {aid}")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Assessment not found"
+        )
+
+    # Calculate score
+    score = calculate_application_score(db, application.id)
+
+    # Get user information
+    from services.user_service import get_user
+    user = get_user(db, application.user_id)
+
+    # Parse answers from JSON string
+    import json
+    answers = json.loads(application.answers) if application.answers else []
+
+    # Get the assessment questions to enrich the answers with question details
+    assessment_questions = json.loads(assessment.questions) if assessment.questions else []
+    question_map = {q['id']: q for q in assessment_questions}
+
+    # Enrich answers with question details and rationales
+    enriched_answers = []
+    for answer in answers:
+        question_id = answer.get('question_id')
+        question_data = question_map.get(question_id, {})
+
+        # For text-based questions, we might want to add rationale from AI scoring
+        rationale = 'No rationale available'
+        if question_data.get('type') == 'text_based':
+            # Use AI service to get rationale for text-based answers
+            from schemas.assessment import AssessmentQuestion, AssessmentQuestionOption
+            from schemas.enums import QuestionType
+
+            # Create a temporary question object for AI scoring
+            temp_question = AssessmentQuestion(
+                id=question_data['id'],
+                text=question_data['text'],
+                weight=question_data['weight'],
+                skill_categories=question_data['skill_categories'],
+                type=QuestionType(question_data['type']),
+                options=[AssessmentQuestionOption(text=opt['text'], value=opt['value']) for opt in question_data.get('options', [])],
+                correct_options=question_data.get('correct_options', [])
+            )
+
+            from services.ai_service import score_answer
+            try:
+                score_result = score_answer(
+                    question=temp_question,
+                    answer_text=answer.get('text', ''),
+                    selected_options=answer.get('options', [])
+                )
+                rationale = score_result.get('rationale', 'No rationale provided') or 'No rationale provided'
+            except Exception:
+                rationale = 'Unable to generate rationale'
+
+        # Create an ApplicationAnswerWithQuestion object with proper field assignments
+        # The 'options' field in the parent class refers to selected options (List[str])
+        # The 'question_options' field in the child class refers to question options (List[dict])
+        from schemas.application import ApplicationAnswerWithQuestion
+        from schemas.enums import QuestionType
+        enriched_answer = ApplicationAnswerWithQuestion(
+            question_id=answer.get('question_id'),
+            text=answer.get('text'),
+            options=answer.get('options', []),  # Selected options from the applicant (List[str])
+            question_text=question_data.get('text', ''),
+            weight=question_data.get('weight', 1),
+            skill_categories=question_data.get('skill_categories', []),
+            type=QuestionType(question_data.get('type', 'text_based')),  # Convert to enum
+            question_options=question_data.get('options', []),  # Question's possible options (List[dict])
+            correct_options=question_data.get('correct_options', []),
+            rationale=rationale
+        )
+
+        # Add the selected options as an additional attribute if needed
+        # But for now, we'll rely on the schema as defined
+        enriched_answers.append(enriched_answer)
+
+    # Create the detailed response
+    application_detail = ApplicationDetailedResponse(
+        id=application.id,
+        job_id=application.job_id,
+        assessment_id=application.assessment_id,
+        user_id=application.user_id,
+        answers=enriched_answers,
+        score=score,
+        passing_score=assessment.passing_score,
+        user={
+            'id': user.id if user else None,
+            'first_name': user.first_name if user else None,
+            'last_name': user.last_name if user else None,
+            'email': user.email if user else None
+        } if user else None
     )
 
+    logger.info(f"Successfully retrieved application detail for job ID: {jid}, assessment ID: {aid}, application ID: {id}")
+    return application_detail
+
+
 @router.post("/jobs/{jid}/assessments/{aid}", response_model=dict)  # Returns just id as per requirements
 def create_new_application(jid: str, aid: str, application: ApplicationCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
     """Create a new application for an assessment"""

backend/api/assessment_routes.py

@@ -85,7 +85,7 @@ def create_new_assessment(id: str, assessment: AssessmentCreate, db: Session = D
     return {"id": db_assessment.id}
 
 @router.patch("/jobs/{jid}/{aid}/regenerate")
-def regenerate_assessment(jid: str, aid: str, regenerate_data: AssessmentRegenerate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
+def regenerate_assessment_route(jid: str, aid: str, regenerate_data: AssessmentRegenerate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
     """Regenerate an assessment"""
     logger.info(f"Regenerating assessment for job ID: {jid}, assessment ID: {aid} by user: {current_user.id}")
     # Only HR users can regenerate assessments
@@ -95,7 +95,11 @@ def regenerate_assessment(jid: str, aid: str, regenerate_data: AssessmentRegener
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Only HR users can regenerate assessments"
         )
-    updated_assessment = regenerate_assessment(db, aid, **regenerate_data.model_dump(exclude_unset=True))
+    # Extract parameters from the request data using dict() to maintain consistency with other routes
+    regenerate_params = regenerate_data.dict(exclude_unset=True)
+
+    # Call the service function with the extracted parameters
+    updated_assessment = regenerate_assessment(db, aid, **regenerate_params)
     if not updated_assessment:
         logger.warning(f"Assessment not found for regeneration with job ID: {jid}, assessment ID: {aid}")
         raise HTTPException(

backend/config.py

@@ -25,6 +25,9 @@ class Settings(BaseSettings):
     algorithm: str = "HS256"
     access_token_expire_minutes: int = 30
 
+    # AI Provider Configuration
+    mistral_api_key: Optional[str] = None
+
     # Application Configuration
     app_name: str = "AI-Powered Hiring Assessment Platform"
     app_version: str = "0.1.0"

backend/integrations/ai_integration/ai_factory.py

@@ -5,6 +5,7 @@ from integrations.ai_integration.mock_ai_generator import MockAIGenerator
 from integrations.ai_integration.openai_generator import OpenAIGenerator
 from integrations.ai_integration.anthropic_generator import AnthropicGenerator
 from integrations.ai_integration.google_ai_generator import GoogleAIGenerator
+from integrations.ai_integration.mistral_generator import MistralGenerator
 
 
 class AIProvider(Enum):
@@ -15,6 +16,7 @@ class AIProvider(Enum):
     OPENAI = "openai"
     ANTHROPIC = "anthropic"
     GOOGLE = "google"
+    MISTRAL = "mistral"
 
 
 class AIGeneratorFactory:
@@ -72,7 +74,8 @@ AIGeneratorFactory.register_provider(AIProvider.MOCK, MockAIGenerator)
 AIGeneratorFactory.register_provider(AIProvider.OPENAI, OpenAIGenerator)
 AIGeneratorFactory.register_provider(AIProvider.ANTHROPIC, AnthropicGenerator)
 AIGeneratorFactory.register_provider(AIProvider.GOOGLE, GoogleAIGenerator)
+AIGeneratorFactory.register_provider(AIProvider.MISTRAL, MistralGenerator)
 
 
 # Optional: Create a default provider
-DEFAULT_PROVIDER = AIProvider.MOCK
+DEFAULT_PROVIDER = AIProvider.MISTRAL

backend/integrations/ai_integration/mistral_generator.py

@@ -0,0 +1,280 @@
+import json
+import os
+from typing import List, Dict, Any
+from mistralai import Mistral
+from schemas.assessment import AssessmentQuestion, AssessmentQuestionOption
+from schemas.enums import QuestionType
+from integrations.ai_integration.ai_generator_interface import AIGeneratorInterface
+from config import settings
+
+
+class MistralGenerator(AIGeneratorInterface):
+    """
+    Mistral Generator implementation for generating assessment questions using Mistral AI API.
+    """
+
+    def __init__(self):
+        """
+        Initialize the MistralGenerator with API key from settings.
+        """
+        api_key = os.getenv("MISTRAL_API_KEY") or getattr(settings, 'mistral_api_key', None)
+
+        if not api_key:
+            raise ValueError("MISTRAL_API_KEY environment variable is not set")
+
+        self.client = Mistral(api_key=api_key)
+
+    def generate_questions(
+        self,
+        title: str,
+        questions_types: List[str],
+        additional_note: str = None,
+        job_info: Dict[str, Any] = None
+    ) -> List[AssessmentQuestion]:
+        """
+        Generate questions using Mistral AI API based on the assessment title, job information, and specified question types.
+        """
+        # Prepare the prompt for Mistral AI
+        prompt = self._create_prompt(title, questions_types, additional_note, job_info)
+
+        messages = [
+            {"role": "system", "content": "You generate technical assessment questions."},
+            {"role": "user", "content": prompt},
+        ]
+
+        response = self.client.chat.complete(
+            model="mistral-small-latest",
+            messages=messages,
+            temperature=0.2,
+        )
+
+        content = response.choices[0].message.content
+        passed = 0
+        while passed < 5:
+            try:
+                try:
+                    # Parse the JSON response from Mistral
+                    questions_data = json.loads(content)
+                    passed = 10
+                except json.JSONDecodeError:
+                    content = content[7:-3].strip()
+                    questions_data = json.loads(content)
+                    passed = 10
+            except json.JSONDecodeError:
+                raise ValueError("Mistral returned invalid JSON")
+
+        # Convert the response to AssessmentQuestion objects
+        return self._convert_to_assessment_questions(questions_data)
+
+    def score_answer(
+        self,
+        question: AssessmentQuestion,
+        answer_text: str,
+        selected_options: List[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Score an answer using Mistral AI API based on the question and the provided answer.
+        """
+        # Create a prompt for scoring the answer
+        if question.type == QuestionType.text_based:
+            prompt = f"""
+            Evaluate the following answer to a text-based question:
+
+            Question: {question.text}
+            Answer: {answer_text}
+
+            Please provide a score between 0 and 1, where 1 means completely correct and 0 means completely incorrect.
+            Also provide a brief rationale for the score.
+
+            Respond in the following JSON format:
+            {{
+                "score": float,
+                "rationale": str,
+                "correct": bool
+            }}
+            """
+        else:
+            # For multiple choice questions
+            selected_str = ", ".join(selected_options) if selected_options else "No options selected"
+            correct_str = ", ".join(question.correct_options) if question.correct_options else "Unknown"
+
+            prompt = f"""
+            Evaluate the following answer to a multiple-choice question:
+
+            Question: {question.text}
+            Selected Options: {selected_str}
+            Correct Options: {correct_str}
+
+            Please provide a score between 0 and 1, where 1 means completely correct and 0 means completely incorrect.
+            Also provide a brief rationale for the score.
+
+            Respond in the following JSON format:
+            {{
+                "score": float,
+                "rationale": str,
+                "correct": bool
+            }}
+            """
+
+        messages = [
+            {"role": "system", "content": "You are an expert at evaluating assessment answers."},
+            {"role": "user", "content": prompt},
+        ]
+
+        response = self.client.chat.complete(
+            model="mistral-small-latest",
+            messages=messages,
+            temperature=0.2,
+        )
+
+        content = response.choices[0].message.content
+        passed = 0
+        while passed < 5:
+            try:
+                try:
+                    result = json.loads(content)
+                    passed = 10  # Exit the loop successfully
+                except json.JSONDecodeError:
+                    # Try to strip markdown code block markers
+                    content = content[7:-3].strip()
+                    result = json.loads(content)
+                    passed = 10  # Exit the loop successfully
+            except json.JSONDecodeError:
+                raise ValueError("Mistral returned invalid JSON for answer scoring")
+
+        return {
+            'score': result.get('score', 0.0),
+            'rationale': result.get('rationale', ''),
+            'correct': result.get('correct', False)
+        }
+
+    def _create_prompt(
+        self,
+        title: str,
+        questions_types: List[str],
+        additional_note: str = None,
+        job_info: Dict[str, Any] = None
+    ) -> str:
+        """
+        Create a prompt for Mistral AI based on the assessment requirements.
+        """
+        # Map question types to the expected format for Mistral
+        type_mapping = {
+            QuestionType.choose_one.value: "MCQ",
+            QuestionType.choose_many.value: "MCQ",  # Multiple choice with multiple correct answers
+            QuestionType.text_based.value: "TEXT"
+        }
+
+        # Count the number of each type of question needed
+        mcq_count = questions_types.count(QuestionType.choose_one.value) + \
+                    questions_types.count(QuestionType.choose_many.value)
+        text_count = questions_types.count(QuestionType.text_based.value)
+
+        # Build the job information section of the prompt
+        job_details = ""
+        if job_info:
+            job_title = job_info.get('title', '')
+            job_skills = job_info.get('skill_categories', [])
+            job_seniority = job_info.get('seniority', '')
+
+            job_details = f"""
+Job Information:
+- Title: {job_title}
+- Skills: {', '.join(job_skills)}
+- Seniority: {job_seniority}
+"""
+        else:
+            job_details = f"""
+Job Information:
+- Title: {title}
+"""
+
+        # Add additional note if provided
+        if additional_note:
+            job_details += f"- Additional Note: {additional_note}\n"
+
+        prompt = f"""
+You are an assessment generator.
+
+Generate EXACTLY {len(questions_types)} questions for the following job.
+
+{job_details}
+
+MANDATORY RULES:
+1. Output MUST be a JSON ARRAY with EXACTLY {len(questions_types)} objects.
+2. The list MUST contain:
+   - {mcq_count} MCQ questions (multiple choice)
+   - {text_count} TEXT questions (text-based)
+3. Do NOT include explanations or markdown.
+4. Follow the schema EXACTLY.
+
+Schema for each question:
+
+{{
+  "type": "MCQ | TEXT",
+  "prompt": "string",
+  "choices": ["string"],  // For MCQ questions only
+  "correct_answer": "string | null",  // For MCQ questions, string for correct choice; for TEXT questions, null
+  "difficulty": "easy | medium | hard",
+  "skill": "string"
+}}
+
+Rules per type:
+- MCQ → 4 choices + correct_answer as the text of the correct choice
+- TEXT → correct_answer = null
+
+Return ONLY the JSON array.
+"""
+
+        return prompt
+
+    def _convert_to_assessment_questions(self, questions_data: List[Dict]) -> List[AssessmentQuestion]:
+        """
+        Convert the JSON response from Mistral to AssessmentQuestion objects.
+        """
+        assessment_questions = []
+
+        for i, q_data in enumerate(questions_data):
+            # Generate a unique ID for the question
+            question_id = f"mistral_{i}"
+
+            # Determine the question type based on the response
+            if q_data.get("type") == "MCQ":
+                # For multiple choice questions
+                question_type = QuestionType.choose_one  # Default to choose_one
+
+                # Create options
+                options = []
+                for choice in q_data.get("choices", []):
+                    option = AssessmentQuestionOption(text=choice, value=choice)
+                    options.append(option)
+
+                # Find the correct option
+                correct_options = []
+                correct_answer = q_data.get("correct_answer")
+                if correct_answer:
+                    # Find the option that matches the correct answer
+                    for opt in options:
+                        if opt.text == correct_answer:
+                            correct_options.append(opt.value)
+                            break
+            else:
+                # For text-based questions
+                question_type = QuestionType.text_based
+                options = []
+                correct_options = []
+
+            # Create the AssessmentQuestion object
+            question = AssessmentQuestion(
+                id=question_id,
+                text=q_data.get("prompt", ""),
+                weight=3,  # Default weight
+                skill_categories=[q_data.get("skill", "General")],  # Default to General if no skill specified
+                type=question_type,
+                options=options,
+                correct_options=correct_options
+            )
+
+            assessment_questions.append(question)
+
+        return assessment_questions

backend/schemas/application.py

@@ -28,7 +28,7 @@ class ApplicationAnswerWithQuestion(ApplicationAnswer):
     weight: int = Field(..., ge=1, le=5)  # range 1-5
     skill_categories: List[str] = Field(..., min_items=1)
     type: QuestionType
-    options: Optional[List[dict]] = []
+    question_options: Optional[List[dict]] = []  # Options for the question
     correct_options: Optional[List[str]] = []
     rationale: str = Field(..., min_length=1, max_length=1000)
 

backend/services/application_service.py

@@ -102,7 +102,14 @@ def calculate_application_score(db: Session, application_id: str) -> float:
     # Parse the answers and questions
     import json
     try:
-        answers = json.loads(application.answers) if application.answers else []
+        # Check if answers is already a list (parsed) or a string (needs parsing)
+        if isinstance(application.answers, str):
+            answers = json.loads(application.answers) if application.answers else []
+        else:
+            # Assume it's already a list object
+            answers = application.answers if application.answers else []
+
+        # Questions should always be a JSON string from the database
         questions = json.loads(assessment.questions) if assessment.questions else []
     except json.JSONDecodeError:
         logger.error(f"Failed to parse answers or questions for application ID: {application_id}")