adding score to DB

backend/alembic/versions/290ee4ce077e_add_score_and_question_scores_columns_.py

@@ -0,0 +1,34 @@
+"""Add score and question_scores columns to applications table
+
+Revision ID: 290ee4ce077e
+Revises: f9f1aa7380ab
+Create Date: 2026-02-09 11:29:38.655897
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '290ee4ce077e'
+down_revision: Union[str, Sequence[str], None] = 'f9f1aa7380ab'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('applications', sa.Column('score', sa.Float(), nullable=True))
+    op.add_column('applications', sa.Column('question_scores', sa.Text(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('applications', 'question_scores')
+    op.drop_column('applications', 'score')
+    # ### end Alembic commands ###

backend/api/application_routes.py

@@ -53,6 +53,15 @@ def get_applications_list(jid: str, aid: str, page: int = 1, limit: int = 10, db
         from services.user_service import get_user
         user = get_user(db, application.user_id)
 
+        # Parse question scores from JSON string
+        question_scores = []
+        if application.question_scores:
+            try:
+                question_scores = json.loads(application.question_scores)
+            except json.JSONDecodeError:
+                logger.warning(f"Failed to parse question scores for application ID: {application.id}")
+                question_scores = []
+
         # Create response object that matches technical requirements exactly
         application_response = {
             'id': application.id,
@@ -62,6 +71,7 @@ def get_applications_list(jid: str, aid: str, page: int = 1, limit: int = 10, db
             'answers': [],  # Not including answers in the list view for performance
             'score': score,
             'passing_score': assessment.passing_score,
+            'question_scores': question_scores,  # Include individual question scores
             'assessment_details': {
                 'id': assessment.id,
                 'title': assessment.title,
@@ -202,6 +212,15 @@ def get_application_detail(jid: str, aid: str, id: str, db: Session = Depends(ge
             logger.error(f"Error creating assessment details: {str(e)}")
             assessment_details_obj = None
 
+    # Parse question scores from JSON string
+    question_scores = []
+    if application.question_scores:
+        try:
+            question_scores = json.loads(application.question_scores)
+        except json.JSONDecodeError:
+            logger.warning(f"Failed to parse question scores for application ID: {application.id}")
+            question_scores = []
+
     application_detail = ApplicationDetailedResponse(
         id=application.id,
         job_id=application.job_id,
@@ -210,6 +229,7 @@ def get_application_detail(jid: str, aid: str, id: str, db: Session = Depends(ge
         answers=enriched_answers,
         score=score,
         passing_score=assessment.passing_score,
+        question_scores=question_scores,  # Include individual question scores
         assessment_details=assessment_details_obj,
         user={
             'id': user.id if user else None,
@@ -416,6 +436,15 @@ def get_my_application(id: str, db: Session = Depends(get_db), current_user: Use
             logger.error(f"Error creating assessment details: {str(e)}")
             assessment_details_obj = None
 
+    # Parse question scores from JSON string
+    question_scores = []
+    if application.question_scores:
+        try:
+            question_scores = json.loads(application.question_scores)
+        except json.JSONDecodeError:
+            logger.warning(f"Failed to parse question scores for application ID: {application.id}")
+            question_scores = []
+
     application_detail = ApplicationDetailedResponse(
         id=application.id,
         job_id=application.job_id,
@@ -424,6 +453,7 @@ def get_my_application(id: str, db: Session = Depends(get_db), current_user: Use
         answers=enriched_answers,
         score=score,
         passing_score=assessment.passing_score,
+        question_scores=question_scores,  # Include individual question scores
         assessment_details=assessment_details_obj,
         user={
             'id': user.id if user else None,

backend/models/application.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, String, Text, ForeignKey, DateTime
+from sqlalchemy import Column, String, Text, ForeignKey, DateTime, Float
 from sqlalchemy.sql import func
 from .base import Base
 import uuid
@@ -11,5 +11,7 @@ class Application(Base):
     assessment_id = Column(String, ForeignKey("assessments.id"), nullable=False)
     user_id = Column(String, ForeignKey("users.id"), nullable=False)
     answers = Column(Text)  # Stored as JSON string
+    score = Column(Float)  # Overall application score
+    question_scores = Column(Text)  # Individual question scores stored as JSON string
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now())

backend/schemas/application.py

@@ -23,6 +23,11 @@ class ApplicationQuestion(BaseModel):
     options: Optional[List[dict]] = []  # Using dict for simplicity
     correct_options: Optional[List[str]] = []
 
+class ApplicationAnswerScore(BaseModel):
+    question_id: str
+    score: float  # Score between 0 and 1
+    rationale: str
+
 class ApplicationAnswerWithQuestion(ApplicationAnswer):
     question_text: str = Field(..., min_length=1, max_length=1000)
     weight: int = Field(..., ge=1, le=5)  # range 1-5
@@ -31,6 +36,7 @@ class ApplicationAnswerWithQuestion(ApplicationAnswer):
     question_options: Optional[List[dict]] = []  # Options for the question
     correct_options: Optional[List[str]] = []
     rationale: str = Field(..., min_length=1, max_length=1000)
+    score: Optional[float] = None  # Score for this specific answer
 
 class ApplicationBase(BaseSchema):
     job_id: str = Field(..., min_length=1)
@@ -52,9 +58,10 @@ class ApplicationAssessment(BaseModel):
 
 class ApplicationResponse(ApplicationBase):
     id: str
-    score: Optional[float] = None
+    score: Optional[float] = None  # Overall application score
     passing_score: Optional[float] = None
     assessment_details: Optional[ApplicationAssessment] = None
+    question_scores: Optional[List[ApplicationAnswerScore]] = None  # Individual question scores
 
     class Config:
         from_attributes = True

backend/services/application_service.py

@@ -51,19 +51,25 @@ def get_applications_by_user(db: Session, user_id: str, skip: int = 0, limit: in
     return applications
 
 def create_application(db: Session, application: ApplicationCreate) -> Application:
-    """Create a new application"""
+    """Create a new application and calculate scores"""
     logger.info(f"Creating new application for job ID: {application.job_id}, assessment ID: {application.assessment_id}, user ID: {application.user_id}")
+    
+    # Calculate scores for the application
+    score, question_scores = calculate_detailed_application_score(db, application)
+    
     db_application = Application(
         id=str(uuid.uuid4()),
         job_id=application.job_id,
         assessment_id=application.assessment_id,
         user_id=application.user_id,
-        answers=json.dumps([ans.dict() for ans in application.answers])  # Store as JSON string
+        answers=json.dumps([ans.dict() for ans in application.answers]),  # Store as JSON string
+        score=score,  # Store the overall application score
+        question_scores=json.dumps([qs.dict() for qs in question_scores])  # Store individual question scores as JSON
     )
     db.add(db_application)
     db.commit()
     db.refresh(db_application)
-    logger.info(f"Successfully created application with ID: {db_application.id}")
+    logger.info(f"Successfully created application with ID: {db_application.id} and overall score: {score}")
     return db_application
 
 def update_application(db: Session, application_id: str, **kwargs) -> Optional[Application]:
@@ -95,6 +101,106 @@ def delete_application(db: Session, application_id: str) -> bool:
     logger.warning(f"Failed to delete application - application not found: {application_id}")
     return False
 
+def calculate_detailed_application_score(db: Session, application_create: ApplicationCreate):
+    """Calculate detailed scores for an application including individual question scores"""
+    from models.assessment import Assessment
+    from schemas.application import ApplicationAnswerScore
+    
+    logger.debug(f"Calculating detailed scores for application - job ID: {application_create.job_id}, assessment ID: {application_create.assessment_id}")
+    
+    # Get the associated assessment to compare answers with correct answers
+    assessment = db.query(Assessment).filter(Assessment.id == application_create.assessment_id).first()
+    if not assessment:
+        logger.warning(f"Assessment not found for ID: {application_create.assessment_id}")
+        return 0.0, []
+
+    # Parse the questions
+    import json
+    try:
+        questions = json.loads(assessment.questions) if assessment.questions else []
+    except json.JSONDecodeError:
+        logger.error(f"Failed to parse questions for assessment ID: {application_create.assessment_id}")
+        return 0.0, []
+
+    # Create a mapping of question_id to question for easy lookup
+    question_map = {q['id']: q for q in questions}
+
+    # Calculate the scores
+    total_points = 0
+    earned_points = 0
+    question_scores = []
+
+    for answer in application_create.answers:
+        question_id = answer.question_id
+        if not question_id or question_id not in question_map:
+            continue
+
+        question_data = question_map[question_id]
+
+        # Calculate weighted score
+        question_weight = question_data.get('weight', 1)  # Default weight is 1
+        total_points += question_weight
+
+        # Initialize question score object
+        question_score_obj = ApplicationAnswerScore(
+            question_id=question_id,
+            score=0.0,
+            rationale="No rationale available"
+        )
+
+        # For multiple choice questions, score directly without AI
+        if question_data['type'] in ['choose_one', 'choose_many']:
+            correct_options = set(question_data.get('correct_options', []))
+            selected_options = set(answer.options or [])
+
+            # Check if the selected options match the correct options exactly
+            if selected_options == correct_options:
+                earned_points += question_weight  # Full points for correct answer
+                question_score_obj.score = 1.0  # Perfect score
+                question_score_obj.rationale = "Correct answer"
+            else:
+                question_score_obj.score = 0.0  # No points for incorrect answer
+                question_score_obj.rationale = f"Incorrect. Correct options: {list(correct_options)}, Selected: {list(selected_options)}"
+
+        # For text-based questions, use AI to evaluate the answer
+        elif question_data['type'] == 'text_based':
+            # Convert the question data to an AssessmentQuestion object
+            from schemas.assessment import AssessmentQuestion, AssessmentQuestionOption
+            from schemas.enums import QuestionType
+            question_obj = 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', [])
+            )
+
+            # Use AI service to score the text-based answer
+            from services.ai_service import score_answer
+            score_result = score_answer(
+                question=question_obj,
+                answer_text=answer.text or '',
+                selected_options=answer.options or []
+            )
+
+            earned_points += score_result['score'] * question_weight
+            question_score_obj.score = score_result['score']
+            question_score_obj.rationale = score_result['rationale']
+
+        question_scores.append(question_score_obj)
+
+    # Calculate percentage score
+    if total_points > 0:
+        overall_score = (earned_points / total_points) * 100
+    else:
+        overall_score = 0.0
+
+    logger.debug(f"Calculated detailed scores: overall {overall_score}% ({earned_points}/{total_points} points), {len(question_scores)} questions scored")
+    return round(overall_score, 2), question_scores
+
+
 def calculate_application_score(db: Session, application_id: str) -> float:
     """Calculate the score for an application"""
     logger.debug(f"Calculating score for application ID: {application_id}")