""" PPM Application — Scoring Service سرویس لایه کاربرد امتیازدهی و اولویت‌بندی. """ import logging from collections import defaultdict from datetime import datetime from decimal import Decimal from typing import Optional, List, Dict from uuid import UUID from apps.core.event_bus.events import event_bus from ...domain.entities.scoring import ( ScoringModel, ScoringCriterion, ProjectScore, ) from ...domain.events.ppm_events import ( ScoringModelCreated, ScoringModelActivated, ProjectScored, ProjectsRanked, ) from ...domain.services.scoring_engine import ScoringEngine, ProjectRanking from ...infrastructure.repositories import ( DjangoScoringModelRepository, DjangoScoringCriterionRepository, DjangoProjectScoreRepository, ) from ..dtos.ppm_dtos import ( ScoringModelCreateDTO, ScoringModelResponseDTO, ScoringCriterionCreateDTO, ScoringCriterionResponseDTO, ProjectScoreCreateDTO, ProjectScoreResponseDTO, ProjectRankingDTO, ) logger = logging.getLogger(__name__) def _publish(event) -> None: try: event_bus.publish(event) except Exception as exc: logger.warning("event_publish_failed: %s — %s", type(event).__name__, exc) class ScoringService: """سرویس مدیریت مدل امتیازدهی، معیارها، امتیازات و رتبه‌بندی.""" def __init__(self, tenant_id: UUID): self._tenant_id = tenant_id self._model_repo = DjangoScoringModelRepository() self._criterion_repo = DjangoScoringCriterionRepository() self._score_repo = DjangoProjectScoreRepository() self._engine = ScoringEngine() # ─── ScoringModel CRUD ─── def create_model(self, dto: ScoringModelCreateDTO) -> ScoringModelResponseDTO: entity = ScoringModel( tenant_id=self._tenant_id, name=dto.name, description=dto.description, portfolio_id=dto.portfolio_id, is_active=dto.is_active, metadata=dto.metadata or {}, ) saved = self._model_repo.save(entity) _publish(ScoringModelCreated( tenant_id=self._tenant_id, scoring_model_id=saved.id, name=saved.name, )) return self._to_model_response(saved) def update_model(self, model_id: UUID, **kwargs) -> Optional[ScoringModelResponseDTO]: entity = self._model_repo.get_by_id(model_id) if not entity: return None for k, v in kwargs.items(): if hasattr(entity, k): setattr(entity, k, v) saved = self._model_repo.save(entity) return self._to_model_response(saved) def get_model(self, model_id: UUID) -> Optional[ScoringModelResponseDTO]: entity = self._model_repo.get_by_id(model_id) if not entity: return None return self._to_model_response(entity) def list_models(self, **filters) -> List[ScoringModelResponseDTO]: entities = self._model_repo.find(**filters) return [self._to_model_response(e) for e in entities] def delete_model(self, model_id: UUID) -> None: self._model_repo.delete(model_id) def activate_model(self, model_id: UUID) -> Optional[ScoringModelResponseDTO]: entity = self._model_repo.get_by_id(model_id) if not entity: return None entity.activate() saved = self._model_repo.save(entity) _publish(ScoringModelActivated( tenant_id=self._tenant_id, scoring_model_id=saved.id, )) return self._to_model_response(saved) def deactivate_model(self, model_id: UUID) -> Optional[ScoringModelResponseDTO]: entity = self._model_repo.get_by_id(model_id) if not entity: return None entity.deactivate() saved = self._model_repo.save(entity) return self._to_model_response(saved) # ─── ScoringCriterion CRUD ─── def add_criterion(self, dto: ScoringCriterionCreateDTO) -> ScoringCriterionResponseDTO: entity = ScoringCriterion( tenant_id=self._tenant_id, scoring_model_id=dto.scoring_model_id, name=dto.name, description=dto.description, weight=dto.weight, scale_min=dto.scale_min, scale_max=dto.scale_max, category=dto.category, order=dto.order, ) saved = self._criterion_repo.save(entity) return self._to_criterion_response(saved) def update_criterion(self, criterion_id: UUID, **kwargs) -> Optional[ScoringCriterionResponseDTO]: entity = self._criterion_repo.get_by_id(criterion_id) if not entity: return None for k, v in kwargs.items(): if hasattr(entity, k): setattr(entity, k, v) saved = self._criterion_repo.save(entity) return self._to_criterion_response(saved) def list_criteria(self, scoring_model_id: UUID) -> List[ScoringCriterionResponseDTO]: entities = self._criterion_repo.find_by_model(scoring_model_id) return [self._to_criterion_response(e) for e in entities] def delete_criterion(self, criterion_id: UUID) -> None: self._criterion_repo.delete(criterion_id) # ─── ProjectScore CRUD ─── def score_project(self, dto: ProjectScoreCreateDTO) -> ProjectScoreResponseDTO: entity = ProjectScore( tenant_id=self._tenant_id, scoring_model_id=dto.scoring_model_id, project_id=dto.project_id, criterion_id=dto.criterion_id, score=dto.score, justification=dto.justification, scored_by_id=dto.scored_by_id, scored_at=datetime.now(), ) saved = self._score_repo.save(entity) return self._to_score_response(saved) def list_project_scores(self, project_id: UUID, scoring_model_id: UUID) -> List[ProjectScoreResponseDTO]: entities = self._score_repo.find_by_project(project_id, scoring_model_id) return [self._to_score_response(e) for e in entities] def delete_score(self, score_id: UUID) -> None: self._score_repo.delete(score_id) # ─── Ranking ─── def rank_projects(self, scoring_model_id: UUID) -> List[ProjectRankingDTO]: criteria = self._criterion_repo.find_by_model(scoring_model_id) all_score_entities = self._score_repo.find_by_model(scoring_model_id) # گروه‌بندی بر اساس project_id scores_by_project: Dict[UUID, list] = defaultdict(list) for s in all_score_entities: scores_by_project[s.project_id].append(s) rankings = self._engine.rank_projects(criteria, dict(scores_by_project)) return [ ProjectRankingDTO( project_id=r.project_id, weighted_score=r.weighted_score, rank=r.rank, scores_by_category=r.scores_by_category or {}, ) for r in rankings ] def calculate_project_score(self, project_id: UUID, scoring_model_id: UUID) -> Decimal: criteria = self._criterion_repo.find_by_model(scoring_model_id) scores = self._score_repo.find_by_project(project_id, scoring_model_id) return self._engine.calculate_weighted_score(criteria, scores) # ─── Mappers ─── def _to_model_response(self, entity: ScoringModel) -> ScoringModelResponseDTO: criteria = self._criterion_repo.find_by_model(entity.id) return ScoringModelResponseDTO( id=entity.id, name=entity.name, description=entity.description, portfolio_id=entity.portfolio_id, is_active=entity.is_active, metadata=entity.metadata or {}, criteria_count=len(criteria), ) @staticmethod def _to_criterion_response(entity: ScoringCriterion) -> ScoringCriterionResponseDTO: return ScoringCriterionResponseDTO( id=entity.id, scoring_model_id=entity.scoring_model_id, name=entity.name, description=entity.description, weight=entity.weight, scale_min=entity.scale_min, scale_max=entity.scale_max, category=entity.category, order=entity.order, ) @staticmethod def _to_score_response(entity: ProjectScore) -> ProjectScoreResponseDTO: return ProjectScoreResponseDTO( id=entity.id, scoring_model_id=entity.scoring_model_id, project_id=entity.project_id, criterion_id=entity.criterion_id, score=entity.score, justification=entity.justification, scored_by_id=entity.scored_by_id, scored_at=entity.scored_at, )