""" Pydantic models for FitScore Feedback Agent API """ import re import json from pydantic import BaseModel, field_validator, ConfigDict from typing import Dict, Any, Optional, List from datetime import datetime def sanitize_text(text: str) -> str: """Sanitize text by removing invalid control characters""" if not text: return text # Remove invalid control characters (except \n, \r, \t) # Keep only printable characters and common whitespace sanitized = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', str(text)) # Normalize whitespace sanitized = re.sub(r'\s+', ' ', sanitized) # Strip leading/trailing whitespace sanitized = sanitized.strip() return sanitized class FitScoreRequest(BaseModel): """Simplified request model for FitScore calculation - only essential fields""" model_config = ConfigDict(protected_namespaces=()) job_id: str jd_text: str # Job description text resume_text: str @field_validator('job_id', 'jd_text', 'resume_text') @classmethod def sanitize_text_fields(cls, v): """Sanitize text fields to remove invalid control characters""" if v is not None: return sanitize_text(v) return v class CandidateRequest(BaseModel): """Request model for candidate evaluation""" model_config = ConfigDict(protected_namespaces=()) candidate_id: str job_id: str recruiter_id: Optional[str] = None # Made optional to match database schema name: str email: str phone: Optional[str] = None location: Optional[str] = None resume_text: str job_description: Optional[str] = None # Job description for FitScore API resume_url: Optional[str] = None linkedin_url: Optional[str] = None @field_validator('email') @classmethod def validate_email(cls, v): if v and '@' not in v: raise ValueError('Invalid email format') return v @field_validator('name', 'location', 'resume_text', 'phone', 'linkedin_url', 'resume_url', 'job_description') @classmethod def sanitize_text_fields(cls, v): """Sanitize text fields to remove invalid control characters""" if v is not None: return sanitize_text(v) return v class FeedbackRequest(BaseModel): """Request model for feedback submission""" model_config = ConfigDict(protected_namespaces=()) job_id: str company_id: str analysis_id: str feedback_type: str # hired, accepted, rejected, interviewed feedback_text: str feedback_category: str # skills, experience, location, education, other confidence_score: float email: Optional[str] = None linkedin_url: Optional[str] = None @field_validator('feedback_type') @classmethod def validate_feedback_type(cls, v): valid_types = ['hired', 'accepted', 'rejected', 'interviewed'] if v not in valid_types: raise ValueError(f'Feedback type must be one of: {", ".join(valid_types)}') return v @field_validator('feedback_category') @classmethod def validate_feedback_category(cls, v): valid_categories = ['skills', 'experience', 'location', 'education', 'other'] if v not in valid_categories: raise ValueError(f'Feedback category must be one of: {", ".join(valid_categories)}') return v @field_validator('confidence_score') @classmethod def validate_confidence_score(cls, v): if not 0.0 <= v <= 1.0: raise ValueError('Confidence score must be between 0.0 and 1.0') return v @field_validator('feedback_text', 'email', 'linkedin_url') @classmethod def sanitize_text_fields(cls, v): """Sanitize text fields to remove invalid control characters""" if v is not None: return sanitize_text(v) return v class FitScoreResponse(BaseModel): """Response model for FitScore calculation""" model_config = ConfigDict(protected_namespaces=()) success: bool evaluation_id: str fitscore: float verdict: str confidence: float category_scores: Dict[str, float] justification: str model_version: str timestamp: str class FeedbackResponse(BaseModel): """Response model for feedback submission""" model_config = ConfigDict(protected_namespaces=()) success: bool feedback_id: str learning_event_id: str message: str class RecalculateResponse(BaseModel): """Response model for FitScore recalculation""" model_config = ConfigDict(protected_namespaces=()) success: bool original_evaluation_id: str updated_evaluation_id: str original_fitscore: float updated_fitscore: float score_change: float original_verdict: str updated_verdict: str verdict_changed: bool model_version: str timestamp: str class ComparisonResponse(BaseModel): """Response model for results comparison""" model_config = ConfigDict(protected_namespaces=()) success: bool comparison: Dict[str, Any] class AnalyticsResponse(BaseModel): """Response model for analytics""" model_config = ConfigDict(protected_namespaces=()) total_feedback: int feedback_by_type: Dict[str, int] feedback_by_category: Dict[str, int] patterns_count: int class HealthResponse(BaseModel): """Response model for health check""" model_config = ConfigDict(protected_namespaces=()) status: str service: str version: str timestamp: str class RootResponse(BaseModel): """Response model for root endpoint""" model_config = ConfigDict(protected_namespaces=()) message: str version: str status: str endpoints: List[str]