Spaces:
Runtime error
Runtime error
| from uuid import UUID | |
| import uuid | |
| from sqlalchemy import select | |
| from src.repositories import BaseRepository | |
| from src.models import ( | |
| AnalysisType, | |
| Analysis, | |
| Proposal, | |
| ComparativeWeights, | |
| Weights, | |
| Evaluations, | |
| EvaluationType, | |
| AnalysisStatus, | |
| ) | |
| class ScoreClient: | |
| def __init__(self): | |
| self.repository = BaseRepository | |
| self.weighted_scores = {"NONE": 0, "LOW": 1, "MEDIUM": 3, "HIGH": 7} | |
| async def __aenter__(self): | |
| return self | |
| async def __aexit__(self, exc_type, exc_value, traceback): | |
| pass | |
| async def update_ai_score( | |
| self, evaluation_id: str, analysis_score_type: str = None | |
| ): | |
| async with self.repository(model=Evaluations).get_session() as session: | |
| query = select(Evaluations).where(Evaluations.id == evaluation_id) | |
| result = await session.execute(query) | |
| evaluation = result.scalars().first() | |
| proposal_id = evaluation.proposal_id | |
| async with self.repository(model=Proposal).get_session() as session: | |
| query = select(Proposal).where(Proposal.id == proposal_id) | |
| result = await session.execute(query) | |
| proposal = result.scalars().first() | |
| rfp_id = proposal.rfp_id | |
| async with self.repository(model=Analysis).get_session() as session: | |
| query = select(Analysis).where(Analysis.evaluation_id == evaluation_id) | |
| result = await session.execute(query) | |
| results = result.scalars().all() | |
| deficiency = [ | |
| result.insights | |
| for result in results | |
| if result.analysis_type == AnalysisType.DEFICIENCIES | |
| ] | |
| weaknesses = [ | |
| result.insights | |
| for result in results | |
| if result.analysis_type == AnalysisType.WEAKNESSES | |
| ] | |
| strengths = [ | |
| result.insights | |
| for result in results | |
| if result.analysis_type == AnalysisType.STRENGTHS | |
| ] | |
| if analysis_score_type != "AI": | |
| deficiency = [ | |
| result.insights | |
| for result in results | |
| if result.analysis_type == AnalysisType.DEFICIENCIES | |
| and result.status != AnalysisStatus.REJECTED | |
| ] | |
| weaknesses = [ | |
| result.insights | |
| for result in results | |
| if result.analysis_type == AnalysisType.WEAKNESSES | |
| and result.status != AnalysisStatus.REJECTED | |
| ] | |
| strengths = [ | |
| result.insights | |
| for result in results | |
| if result.analysis_type == AnalysisType.STRENGTHS | |
| and result.status != AnalysisStatus.REJECTED | |
| ] | |
| start_score = 0 if deficiency else 100 | |
| async with self.repository(model=ComparativeWeights).get_session() as session: | |
| query = select(ComparativeWeights).where( | |
| ComparativeWeights.rfp_id == rfp_id | |
| ) | |
| result = await session.execute(query) | |
| comparative_weights = result.scalars().first() | |
| start_score += len(strengths) * comparative_weights.strengths_weight | |
| start_score -= len(weaknesses) * comparative_weights.weaknesses_weight | |
| async with self.repository(model=Evaluations).get_session() as session: | |
| query = select(Evaluations).where(Evaluations.id == evaluation_id) | |
| result = await session.execute(query) | |
| evaluation = result.scalars().first() | |
| if analysis_score_type == "AI": | |
| evaluation.ai_score = start_score | |
| evaluation.adjusted_score = start_score | |
| else: | |
| evaluation.adjusted_score = start_score | |
| await session.commit() | |
| await session.refresh(evaluation) | |
| query = select(Evaluations).where(Evaluations.proposal_id == proposal_id) | |
| result = await session.execute(query) | |
| evaluations = result.scalars().all() | |
| technical_ai_score = [ | |
| evaluation.ai_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.TECHNICAL | |
| and evaluation.ai_score is not None | |
| ] | |
| management_ai_score = [ | |
| evaluation.ai_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.MANAGEMENT | |
| and evaluation.ai_score is not None | |
| ] | |
| past_performance_ai_score = [ | |
| evaluation.ai_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.PAST_PERFORMANCE | |
| and evaluation.ai_score is not None | |
| ] | |
| price_ai_score = [ | |
| evaluation.ai_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.PRICE | |
| and evaluation.ai_score is not None | |
| ] | |
| if analysis_score_type != "AI": | |
| technical_ai_score = [ | |
| evaluation.adjusted_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.TECHNICAL | |
| and evaluation.adjusted_score is not None | |
| ] | |
| management_ai_score = [ | |
| evaluation.adjusted_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.MANAGEMENT | |
| and evaluation.adjusted_score is not None | |
| ] | |
| past_performance_ai_score = [ | |
| evaluation.adjusted_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.PAST_PERFORMANCE | |
| and evaluation.adjusted_score is not None | |
| ] | |
| price_ai_score = [ | |
| evaluation.adjusted_score | |
| for evaluation in evaluations | |
| if evaluation.evaluation_type == EvaluationType.PRICE | |
| and evaluation.adjusted_score is not None | |
| ] | |
| technical_ai_score = technical_ai_score[0] if technical_ai_score else 0 | |
| management_ai_score = management_ai_score[0] if management_ai_score else 0 | |
| past_performance_ai_score = ( | |
| past_performance_ai_score[0] if past_performance_ai_score else 0 | |
| ) | |
| price_ai_score = price_ai_score[0] if price_ai_score else 0 | |
| technical_weight = self.weighted_scores[ | |
| comparative_weights.technical_weight.name | |
| ] | |
| management_weight = self.weighted_scores[ | |
| comparative_weights.management_weight.name | |
| ] | |
| past_performance_weight = self.weighted_scores[ | |
| comparative_weights.past_performance_weight.name | |
| ] | |
| price_weight = self.weighted_scores[comparative_weights.price_weight.name] | |
| ai_score = ( | |
| technical_ai_score * technical_weight | |
| + management_ai_score * management_weight | |
| + past_performance_ai_score * past_performance_weight | |
| + price_ai_score * price_weight | |
| ) | |
| ai_score /= ( | |
| technical_weight | |
| + management_weight | |
| + past_performance_weight | |
| + price_weight | |
| ) | |
| async with self.repository(model=Proposal).get_session() as session: | |
| query = select(Proposal).where(Proposal.id == proposal_id) | |
| result = await session.execute(query) | |
| evaluation = result.scalars().first() | |
| if analysis_score_type == "AI": | |
| evaluation.ai_score = ai_score | |
| evaluation.final_score = ai_score | |
| else: | |
| evaluation.final_score = ai_score | |
| await session.commit() | |
| await session.refresh(evaluation) | |