""" PM Module — Application Services (Advanced Features) سرویس‌های اپلیکیشن برای فیچرهای پیشرفته: • ActivityCodeService — مدیریت کدهای فعالیت • UDFService — مدیریت فیلدهای سفارشی (EAV) • AdvancedResourceService — مهارت، دسترسی، نرخ هزینه • CostAccountService — ساختار شکست هزینه (CBS) • FundingSourceService — منابع مالی • CashFlowService — جریان نقدی • ScheduleHealthAppService — ارزیابی سلامت زمان‌بندی • ProjectTemplateService — قالب پروژه • MonteCarloAppService — شبیه‌سازی مونت‌کارلو • ResourceLevelingAppService — تسطیح منابع """ import logging from datetime import date from decimal import Decimal from typing import Optional, List, Dict from uuid import UUID, uuid4 from apps.core.event_bus.events import event_bus from ..dtos.pm_dtos import PMBaseDTO from ...domain.entities.advanced import ( ActivityCodeType, ActivityCodeValue, TaskActivityCode, UDFDefinition, UDFValue, ResourceSkill, ResourceAvailability, ResourceCostRate, CostAccount, FundingSource, CashFlowEntry, ScheduleHealthSnapshot, ProjectTemplate, ) from ...domain.services.advanced import ( ResourceLevelingService, MonteCarloService, CCPMService, ScheduleHealthService, TaskEstimate, MonteCarloResult, LevelingResult, CCPMResult, ) from ...infrastructure.repositories.advanced_repos import ( DjangoActivityCodeTypeRepository, DjangoActivityCodeValueRepository, DjangoTaskActivityCodeRepository, DjangoUDFDefinitionRepository, DjangoUDFValueRepository, DjangoResourceSkillRepository, DjangoResourceAvailabilityRepository, DjangoResourceCostRateRepository, DjangoCostAccountRepository, DjangoFundingSourceRepository, DjangoCashFlowEntryRepository, DjangoScheduleHealthSnapshotRepository, DjangoProjectTemplateRepository, ) from ...infrastructure.persistence.models import ( TaskModel, DependencyModel, ResourceAssignmentModel, ResourceModel, ) from ...infrastructure.repositories.mappers import ( TaskMapper, ResourceAssignmentMapper, ) from ...domain.entities.task import Dependency logger = logging.getLogger(__name__) def _publish_event(event): """Publish domain event safely.""" try: event_bus.publish(event) except Exception as e: logger.warning(f"Failed to publish event: {e}") # ═══════════════════════════════════════════════════ # Activity Code Service # ═══════════════════════════════════════════════════ class ActivityCodeService: """مدیریت کدهای فعالیت (P6-style Activity Codes).""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.type_repo = DjangoActivityCodeTypeRepository(tenant_id) self.value_repo = DjangoActivityCodeValueRepository(tenant_id) self.assignment_repo = DjangoTaskActivityCodeRepository(tenant_id) # --- Code Types --- def create_code_type(self, data: dict) -> ActivityCodeType: entity = ActivityCodeType( id=uuid4(), tenant_id=self.tenant_id, name=data["name"], code=data["code"], description=data.get("description", ""), scope=data.get("scope", "global"), project_id=data.get("project_id"), is_active=True, sort_order=data.get("sort_order", 0), ) return self.type_repo.save(entity) def update_code_type(self, type_id: UUID, data: dict) -> Optional[ActivityCodeType]: entity = self.type_repo.get_by_id(type_id) if not entity: return None if "name" in data: entity.name = data["name"] if "description" in data: entity.description = data["description"] if "is_active" in data: entity.is_active = data["is_active"] if "sort_order" in data: entity.sort_order = data["sort_order"] return self.type_repo.save(entity) def get_code_type(self, type_id: UUID) -> Optional[ActivityCodeType]: return self.type_repo.get_by_id(type_id) def list_code_types(self, scope: str = None, project_id: UUID = None) -> List[ActivityCodeType]: if project_id: return self.type_repo.find_by_project(project_id) if scope: return self.type_repo.find_by_scope(scope) return self.type_repo.find() def delete_code_type(self, type_id: UUID) -> bool: return self.type_repo.delete(type_id) # --- Code Values --- def create_code_value(self, data: dict) -> ActivityCodeValue: entity = ActivityCodeValue( id=uuid4(), tenant_id=self.tenant_id, code_type_id=data["code_type_id"], parent_id=data.get("parent_id"), code=data["code"], name=data["name"], description=data.get("description", ""), color=data.get("color", ""), sort_order=data.get("sort_order", 0), ) return self.value_repo.save(entity) def update_code_value(self, value_id: UUID, data: dict) -> Optional[ActivityCodeValue]: entity = self.value_repo.get_by_id(value_id) if not entity: return None for field in ("name", "description", "color", "sort_order"): if field in data: setattr(entity, field, data[field]) return self.value_repo.save(entity) def list_code_values(self, code_type_id: UUID) -> List[ActivityCodeValue]: return self.value_repo.find_by_type(code_type_id) def delete_code_value(self, value_id: UUID) -> bool: return self.value_repo.delete(value_id) # --- Task Assignments --- def assign_code_to_task(self, task_id: UUID, code_value_id: UUID) -> TaskActivityCode: entity = TaskActivityCode( id=uuid4(), tenant_id=self.tenant_id, task_id=task_id, code_value_id=code_value_id, ) return self.assignment_repo.save(entity) def remove_code_from_task(self, assignment_id: UUID) -> bool: return self.assignment_repo.delete(assignment_id) def get_task_codes(self, task_id: UUID) -> List[TaskActivityCode]: return self.assignment_repo.find_by_task(task_id) # ═══════════════════════════════════════════════════ # UDF Service (فیلدهای سفارشی) # ═══════════════════════════════════════════════════ class UDFService: """مدیریت فیلدهای سفارشی (User Defined Fields — EAV).""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.def_repo = DjangoUDFDefinitionRepository(tenant_id) self.val_repo = DjangoUDFValueRepository(tenant_id) # --- Definitions --- def create_definition(self, data: dict) -> UDFDefinition: entity = UDFDefinition( id=uuid4(), tenant_id=self.tenant_id, name=data["name"], code=data["code"], description=data.get("description", ""), field_type=data.get("field_type", "text"), entity_type=data.get("entity_type", "task"), default_value=data.get("default_value", ""), options=data.get("options", []), is_required=data.get("is_required", False), is_active=True, sort_order=data.get("sort_order", 0), ) return self.def_repo.save(entity) def update_definition(self, def_id: UUID, data: dict) -> Optional[UDFDefinition]: entity = self.def_repo.get_by_id(def_id) if not entity: return None for field in ("name", "description", "default_value", "options", "is_required", "is_active", "sort_order"): if field in data: setattr(entity, field, data[field]) return self.def_repo.save(entity) def list_definitions(self, entity_type: str = None, active_only: bool = True) -> List[UDFDefinition]: if entity_type and active_only: return self.def_repo.find_active(entity_type) if entity_type: return self.def_repo.find_by_entity_type(entity_type) return self.def_repo.find() def delete_definition(self, def_id: UUID) -> bool: return self.def_repo.delete(def_id) # --- Values --- def set_value(self, definition_id: UUID, entity_id: UUID, entity_type: str, value: dict) -> UDFValue: existing = self.val_repo.find_value(definition_id, entity_id) if existing: for field in ("value_text", "value_number", "value_date", "value_boolean"): if field in value: setattr(existing, field, value[field]) return self.val_repo.save(existing) entity = UDFValue( id=uuid4(), tenant_id=self.tenant_id, definition_id=definition_id, entity_id=entity_id, entity_type=entity_type, value_text=value.get("value_text", ""), value_number=value.get("value_number"), value_date=value.get("value_date"), value_boolean=value.get("value_boolean"), ) return self.val_repo.save(entity) def get_entity_values(self, entity_id: UUID, entity_type: str) -> List[UDFValue]: return self.val_repo.find_by_entity(entity_id, entity_type) def delete_value(self, value_id: UUID) -> bool: return self.val_repo.delete(value_id) # ═══════════════════════════════════════════════════ # Advanced Resource Service # ═══════════════════════════════════════════════════ class AdvancedResourceService: """مدیریت مهارت‌ها، دسترسی و نرخ هزینه منابع.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.skill_repo = DjangoResourceSkillRepository(tenant_id) self.avail_repo = DjangoResourceAvailabilityRepository(tenant_id) self.rate_repo = DjangoResourceCostRateRepository(tenant_id) # --- Skills --- def add_skill(self, data: dict) -> ResourceSkill: entity = ResourceSkill( id=uuid4(), tenant_id=self.tenant_id, resource_id=data["resource_id"], skill_name=data["skill_name"], proficiency_level=data.get("proficiency_level", 1), notes=data.get("notes", ""), ) return self.skill_repo.save(entity) def get_resource_skills(self, resource_id: UUID) -> List[ResourceSkill]: return self.skill_repo.find_by_resource(resource_id) def find_by_skill(self, skill_name: str) -> List[ResourceSkill]: return self.skill_repo.find_by_skill(skill_name) def delete_skill(self, skill_id: UUID) -> bool: return self.skill_repo.delete(skill_id) # --- Availability --- def add_availability(self, data: dict) -> ResourceAvailability: entity = ResourceAvailability( id=uuid4(), tenant_id=self.tenant_id, resource_id=data["resource_id"], start_date=data["start_date"], end_date=data.get("end_date"), available_units=Decimal(str(data.get("available_units", 100))), notes=data.get("notes", ""), ) return self.avail_repo.save(entity) def get_resource_availability(self, resource_id: UUID) -> List[ResourceAvailability]: return self.avail_repo.find_by_resource(resource_id) def get_availability_in_range( self, resource_id: UUID, start_date: date, end_date: date ) -> List[ResourceAvailability]: return self.avail_repo.find_by_date_range(resource_id, start_date, end_date) def delete_availability(self, avail_id: UUID) -> bool: return self.avail_repo.delete(avail_id) # --- Cost Rates --- def add_cost_rate(self, data: dict) -> ResourceCostRate: entity = ResourceCostRate( id=uuid4(), tenant_id=self.tenant_id, resource_id=data["resource_id"], effective_date=data["effective_date"], standard_rate=Decimal(str(data.get("standard_rate", 0))), overtime_rate=Decimal(str(data.get("overtime_rate", 0))), cost_per_use=Decimal(str(data.get("cost_per_use", 0))), notes=data.get("notes", ""), ) return self.rate_repo.save(entity) def get_resource_rates(self, resource_id: UUID) -> List[ResourceCostRate]: return self.rate_repo.find_by_resource(resource_id) def get_effective_rate(self, resource_id: UUID, on_date: date) -> Optional[ResourceCostRate]: return self.rate_repo.find_effective_rate(resource_id, on_date) def delete_cost_rate(self, rate_id: UUID) -> bool: return self.rate_repo.delete(rate_id) # ═══════════════════════════════════════════════════ # Cost Account Service (CBS) # ═══════════════════════════════════════════════════ class CostAccountService: """مدیریت ساختار شکست هزینه (CBS — Cost Breakdown Structure).""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.repo = DjangoCostAccountRepository(tenant_id) def create(self, data: dict) -> CostAccount: entity = CostAccount( id=uuid4(), tenant_id=self.tenant_id, project_id=data["project_id"], parent_id=data.get("parent_id"), code=data["code"], name=data["name"], description=data.get("description", ""), budget_amount=Decimal(str(data.get("budget_amount", 0))), sort_order=data.get("sort_order", 0), ) return self.repo.save(entity) def update(self, account_id: UUID, data: dict) -> Optional[CostAccount]: entity = self.repo.get_by_id(account_id) if not entity: return None for field in ("name", "description", "budget_amount", "actual_amount", "sort_order"): if field in data: val = data[field] if field in ("budget_amount", "actual_amount"): val = Decimal(str(val)) setattr(entity, field, val) return self.repo.save(entity) def get(self, account_id: UUID) -> Optional[CostAccount]: return self.repo.get_by_id(account_id) def list_by_project(self, project_id: UUID) -> List[CostAccount]: return self.repo.find_by_project(project_id) def get_tree(self, project_id: UUID) -> List[CostAccount]: return self.repo.find_root_accounts(project_id) def delete(self, account_id: UUID) -> bool: return self.repo.delete(account_id) # ═══════════════════════════════════════════════════ # Funding Source Service # ═══════════════════════════════════════════════════ class FundingSourceService: """مدیریت منابع مالی پروژه.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.repo = DjangoFundingSourceRepository(tenant_id) def create(self, data: dict) -> FundingSource: total = Decimal(str(data.get("total_amount", 0))) entity = FundingSource( id=uuid4(), tenant_id=self.tenant_id, project_id=data["project_id"], name=data["name"], description=data.get("description", ""), total_amount=total, allocated_amount=Decimal("0"), remaining_amount=total, start_date=data.get("start_date"), end_date=data.get("end_date"), is_active=True, ) return self.repo.save(entity) def update(self, source_id: UUID, data: dict) -> Optional[FundingSource]: entity = self.repo.get_by_id(source_id) if not entity: return None for field in ("name", "description", "total_amount", "allocated_amount", "start_date", "end_date", "is_active"): if field in data: val = data[field] if field in ("total_amount", "allocated_amount"): val = Decimal(str(val)) setattr(entity, field, val) # Recalculate remaining entity.remaining_amount = entity.total_amount - entity.allocated_amount return self.repo.save(entity) def list_by_project(self, project_id: UUID) -> List[FundingSource]: return self.repo.find_by_project(project_id) def delete(self, source_id: UUID) -> bool: return self.repo.delete(source_id) # ═══════════════════════════════════════════════════ # Cash Flow Service # ═══════════════════════════════════════════════════ class CashFlowService: """مدیریت جریان نقدی دوره‌ای.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.repo = DjangoCashFlowEntryRepository(tenant_id) def create(self, data: dict) -> CashFlowEntry: entity = CashFlowEntry( id=uuid4(), tenant_id=self.tenant_id, project_id=data["project_id"], period_start=data["period_start"], period_end=data.get("period_end"), planned_income=Decimal(str(data.get("planned_income", 0))), planned_expense=Decimal(str(data.get("planned_expense", 0))), actual_income=Decimal(str(data.get("actual_income", 0))), actual_expense=Decimal(str(data.get("actual_expense", 0))), notes=data.get("notes", ""), ) return self.repo.save(entity) def update(self, entry_id: UUID, data: dict) -> Optional[CashFlowEntry]: entity = self.repo.get_by_id(entry_id) if not entity: return None for field in ("planned_income", "planned_expense", "actual_income", "actual_expense", "notes"): if field in data: val = data[field] if field != "notes": val = Decimal(str(val)) setattr(entity, field, val) return self.repo.save(entity) def list_by_project(self, project_id: UUID) -> List[CashFlowEntry]: return self.repo.find_by_project(project_id) def list_by_period( self, project_id: UUID, start_date: date, end_date: date ) -> List[CashFlowEntry]: return self.repo.find_by_period(project_id, start_date, end_date) def delete(self, entry_id: UUID) -> bool: return self.repo.delete(entry_id) # ═══════════════════════════════════════════════════ # Schedule Health Application Service # ═══════════════════════════════════════════════════ class ScheduleHealthAppService: """ارزیابی سلامت زمان‌بندی و ذخیره اسنپ‌شات.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.snapshot_repo = DjangoScheduleHealthSnapshotRepository(tenant_id) def assess_and_save( self, project_id: UUID, baseline_end: date = None ) -> ScheduleHealthSnapshot: """ارزیابی سلامت زمان‌بندی و ذخیره اسنپ‌شات.""" # Fetch data from DB task_qs = TaskModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) dep_qs = DependencyModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) assign_qs = ResourceAssignmentModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) tasks = [TaskMapper.to_entity(m) for m in task_qs] dependencies = [ Dependency( id=m.id, tenant_id=m.tenant_id, project_id=m.project_id, predecessor_id=m.predecessor_id, successor_id=m.successor_id, dependency_type=m.dependency_type, lag_days=m.lag_days, ) for m in dep_qs ] assignments = [ResourceAssignmentMapper.to_entity(m) for m in assign_qs] # Run assessment snapshot = ScheduleHealthService.assess( tasks=tasks, dependencies=dependencies, assignments=assignments, baseline_end=baseline_end, ) snapshot.project_id = project_id snapshot.tenant_id = self.tenant_id snapshot.id = uuid4() return self.snapshot_repo.save(snapshot) def get_latest(self, project_id: UUID) -> Optional[ScheduleHealthSnapshot]: return self.snapshot_repo.find_latest(project_id) def get_history( self, project_id: UUID, start_date: date = None, end_date: date = None ) -> List[ScheduleHealthSnapshot]: if start_date and end_date: return self.snapshot_repo.find_by_date_range(project_id, start_date, end_date) return self.snapshot_repo.find_by_project(project_id) # ═══════════════════════════════════════════════════ # Project Template Service # ═══════════════════════════════════════════════════ class ProjectTemplateService: """مدیریت قالب‌های پروژه.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id self.repo = DjangoProjectTemplateRepository(tenant_id) def create(self, data: dict) -> ProjectTemplate: entity = ProjectTemplate( id=uuid4(), tenant_id=self.tenant_id, name=data["name"], description=data.get("description", ""), category=data.get("category", ""), template_data=data.get("template_data", {}), is_active=True, usage_count=0, ) return self.repo.save(entity) def save_from_project(self, project_id: UUID, name: str, description: str = "") -> ProjectTemplate: """ذخیره پروژه موجود به عنوان قالب.""" # Fetch project data task_qs = TaskModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ).values( "code", "title", "description", "parent_id", "task_type", "duration", "sort_order", "is_milestone", ) dep_qs = DependencyModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ).values("predecessor_id", "successor_id", "dependency_type", "lag_days") template_data = { "tasks": list(task_qs), "dependencies": list(dep_qs), "version": "1.0", } # Serialize UUIDs to strings for item in template_data["tasks"]: for k, v in item.items(): if hasattr(v, "hex"): item[k] = str(v) for item in template_data["dependencies"]: for k, v in item.items(): if hasattr(v, "hex"): item[k] = str(v) entity = ProjectTemplate( id=uuid4(), tenant_id=self.tenant_id, name=name, description=description, category="from_project", template_data=template_data, is_active=True, ) return self.repo.save(entity) def update(self, template_id: UUID, data: dict) -> Optional[ProjectTemplate]: entity = self.repo.get_by_id(template_id) if not entity: return None for field in ("name", "description", "category", "template_data", "is_active"): if field in data: setattr(entity, field, data[field]) return self.repo.save(entity) def list_active(self) -> List[ProjectTemplate]: return self.repo.find_active() def list_by_category(self, category: str) -> List[ProjectTemplate]: return self.repo.find_by_category(category) def get(self, template_id: UUID) -> Optional[ProjectTemplate]: return self.repo.get_by_id(template_id) def delete(self, template_id: UUID) -> bool: return self.repo.delete(template_id) def increment_usage(self, template_id: UUID): entity = self.repo.get_by_id(template_id) if entity: entity.usage_count += 1 self.repo.save(entity) # ═══════════════════════════════════════════════════ # Monte Carlo Application Service # ═══════════════════════════════════════════════════ class MonteCarloAppService: """اجرای شبیه‌سازی مونت‌کارلو از داده‌های پروژه.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id def run_simulation( self, project_id: UUID, estimates: List[Dict], iterations: int = 1000, ) -> MonteCarloResult: """اجرای شبیه‌سازی مونت‌کارلو.""" task_qs = TaskModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) dep_qs = DependencyModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) tasks = [TaskMapper.to_entity(m) for m in task_qs] dependencies = [ Dependency( id=m.id, tenant_id=m.tenant_id, project_id=m.project_id, predecessor_id=m.predecessor_id, successor_id=m.successor_id, dependency_type=m.dependency_type, lag_days=m.lag_days, ) for m in dep_qs ] task_estimates = [ TaskEstimate( task_id=e["task_id"], optimistic=e["optimistic"], most_likely=e["most_likely"], pessimistic=e["pessimistic"], ) for e in estimates ] return MonteCarloService.simulate( tasks=tasks, dependencies=dependencies, estimates=task_estimates, iterations=iterations, ) # ═══════════════════════════════════════════════════ # Resource Leveling Application Service # ═══════════════════════════════════════════════════ class ResourceLevelingAppService: """اجرای تسطیح منابع از داده‌های پروژه.""" def __init__(self, tenant_id: UUID): self.tenant_id = tenant_id def run_leveling(self, project_id: UUID) -> LevelingResult: """اجرای تسطیح منابع.""" task_qs = TaskModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) dep_qs = DependencyModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id ) assign_qs = ResourceAssignmentModel.objects.filter( tenant_id=self.tenant_id, project_id=project_id, is_active=True, ) tasks = [TaskMapper.to_entity(m) for m in task_qs] dependencies = [ Dependency( id=m.id, tenant_id=m.tenant_id, project_id=m.project_id, predecessor_id=m.predecessor_id, successor_id=m.successor_id, dependency_type=m.dependency_type, lag_days=m.lag_days, ) for m in dep_qs ] assignments = [ResourceAssignmentMapper.to_entity(m) for m in assign_qs] # Fetch resource max units resource_ids = {a.resource_id for a in assignments} resource_max_units = {} for res in ResourceModel.objects.filter(id__in=resource_ids): resource_max_units[res.id] = res.max_units or Decimal("100") return ResourceLevelingService.level( tasks=tasks, dependencies=dependencies, assignments=assignments, resource_max_units=resource_max_units, ) def apply_leveling(self, project_id: UUID, result: LevelingResult) -> int: """اعمال نتیجه تسطیح — بروزرسانی planned_start تسک‌ها.""" updated = 0 for task_id, new_start in result.leveled_tasks.items(): count = TaskModel.objects.filter( id=task_id, tenant_id=self.tenant_id ).update(planned_start=new_start) updated += count return updated