""" PM Application Services — Project & Task Use Cases. سرویس‌های لایه Application برای پروژه و تسک. """ import logging from datetime import date, datetime from typing import Optional, List from uuid import UUID, uuid4 from apps.core.event_bus.events import event_bus from ...domain.entities.project import Project from ...domain.entities.task import Task, Dependency from ...domain.events.pm_events import ( ProjectCreated, ProjectStatusChanged, ProjectCompleted, TaskCreated, TaskStatusChanged, TaskCompleted, TaskAssigned, TaskProgressUpdated, MilestoneReached, ) from ...domain.exceptions.pm_exceptions import ( ProjectNotFoundError, TaskNotFoundError, InvalidStatusTransitionError, CircularDependencyError, DuplicateCodeError, ) from ...domain.services.scheduling import CriticalPathService, WBSService from ...infrastructure.repositories import ( DjangoProjectRepository, DjangoTaskRepository, DjangoDependencyRepository, DjangoActivityLogRepository, ) from ..dtos.pm_dtos import ( CreateProjectDTO, UpdateProjectDTO, ProjectResponseDTO, CreateTaskDTO, UpdateTaskDTO, TaskResponseDTO, MoveTaskDTO, CreateDependencyDTO, DependencyResponseDTO, ) logger = logging.getLogger(__name__) def _publish_event(event) -> None: """Publish a domain event via the platform event bus (fire-and-forget).""" try: event_bus.publish(event) except Exception as exc: logger.warning("event_publish_failed: %s — %s", type(event).__name__, exc) class ProjectService: """سرویس مدیریت پروژه.""" def __init__(self, tenant_id: UUID): self._tenant_id = tenant_id self._repo = DjangoProjectRepository(tenant_id) self._task_repo = DjangoTaskRepository(tenant_id) self._activity_repo = DjangoActivityLogRepository(tenant_id) def create_project(self, dto: CreateProjectDTO) -> ProjectResponseDTO: # بررسی یکتایی کد existing = self._repo.find_by_code(dto.code) if existing: raise DuplicateCodeError("project", dto.code) project = Project( id=uuid4(), tenant_id=self._tenant_id, code=dto.code, title=dto.title, description=dto.description, project_type=dto.project_type, parent_id=dto.parent_id, status="draft", priority=dto.priority, planned_start=dto.planned_start, planned_end=dto.planned_end, total_budget=dto.total_budget, estimated_cost=dto.estimated_cost, company_id=dto.company_id, org_unit_id=dto.org_unit_id, currency_id=dto.currency_id, manager_id=dto.manager_id, calendar_id=dto.calendar_id, color=dto.color, tags=dto.tags, metadata=dto.metadata, created_by=dto.created_by, ) saved = self._repo.save(project) # Publish event _publish_event(ProjectCreated( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.id, code=saved.code, title=saved.title, project_type=saved.project_type, company_id=saved.company_id, manager_id=saved.manager_id, )) return self._to_response(saved) def update_project(self, dto: UpdateProjectDTO) -> ProjectResponseDTO: project = self._repo.get_by_id(dto.id) if not project: raise ProjectNotFoundError(dto.id) if dto.title is not None: project.title = dto.title if dto.description is not None: project.description = dto.description if dto.project_type is not None: project.project_type = dto.project_type if dto.parent_id is not None: project.parent_id = dto.parent_id if dto.priority is not None: project.priority = dto.priority if dto.planned_start is not None: project.planned_start = dto.planned_start if dto.planned_end is not None: project.planned_end = dto.planned_end if dto.total_budget is not None: project.total_budget = dto.total_budget if dto.estimated_cost is not None: project.estimated_cost = dto.estimated_cost if dto.company_id is not None: project.company_id = dto.company_id if dto.org_unit_id is not None: project.org_unit_id = dto.org_unit_id if dto.currency_id is not None: project.currency_id = dto.currency_id if dto.manager_id is not None: project.manager_id = dto.manager_id if dto.calendar_id is not None: project.calendar_id = dto.calendar_id if dto.color is not None: project.color = dto.color if dto.tags is not None: project.tags = dto.tags if dto.metadata is not None: project.metadata = dto.metadata if dto.updated_by is not None: project.updated_by = dto.updated_by saved = self._repo.save(project) return self._to_response(saved) def get_project(self, project_id: UUID) -> ProjectResponseDTO: project = self._repo.get_by_id(project_id) if not project: raise ProjectNotFoundError(project_id) return self._to_response(project) def list_projects(self, status: Optional[str] = None, manager_id: Optional[UUID] = None, company_id: Optional[UUID] = None) -> List[ProjectResponseDTO]: if status: projects = self._repo.find_by_status(status) elif manager_id: projects = self._repo.find_by_manager(manager_id) elif company_id: projects = self._repo.find_by_company(company_id) else: projects = self._repo.find() return [self._to_response(p) for p in projects] def change_status(self, project_id: UUID, new_status: str, user_id: Optional[UUID] = None) -> ProjectResponseDTO: project = self._repo.get_by_id(project_id) if not project: raise ProjectNotFoundError(project_id) valid_transitions = { "draft": ["planning", "cancelled"], "planning": ["active", "cancelled"], "active": ["on_hold", "completed", "cancelled"], "on_hold": ["active", "cancelled"], "completed": ["archived"], "cancelled": ["archived"], } allowed = valid_transitions.get(project.status, []) if new_status not in allowed: raise InvalidStatusTransitionError("project", project.status, new_status) old_status = project.status project.status = new_status if new_status == "active" and not project.actual_start: project.actual_start = date.today() elif new_status == "completed": project.actual_end = date.today() project.progress = 100 if user_id: project.updated_by = user_id saved = self._repo.save(project) # Publish status change event _publish_event(ProjectStatusChanged( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.id, old_status=old_status, new_status=new_status, )) # Publish completion event if new_status == "completed": _publish_event(ProjectCompleted( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.id, actual_end=saved.actual_end, final_cost=saved.actual_cost, )) return self._to_response(saved) def delete_project(self, project_id: UUID) -> bool: project = self._repo.get_by_id(project_id) if not project: raise ProjectNotFoundError(project_id) if project.status not in ("draft", "cancelled"): raise InvalidStatusTransitionError( "project", project.status, "deleted", ) return self._repo.delete(project_id) def get_dashboard(self, project_id: UUID) -> dict: """داده‌های داشبورد پروژه.""" project = self._repo.get_by_id(project_id) if not project: raise ProjectNotFoundError(project_id) tasks = self._task_repo.find_by_project(project_id) total = len(tasks) completed = sum(1 for t in tasks if t.status == "completed") overdue = sum( 1 for t in tasks if t.planned_end and t.planned_end < date.today() and t.status not in ("completed", "cancelled") ) milestones = [t for t in tasks if t.task_type == "milestone"] return { "project": self._to_response(project).__dict__, "task_summary": { "total": total, "completed": completed, "in_progress": sum(1 for t in tasks if t.status == "in_progress"), "overdue": overdue, }, "milestones": [ {"id": str(m.id), "title": m.title, "date": str(m.planned_end), "status": m.status} for m in milestones ], } def _to_response(self, project: Project) -> ProjectResponseDTO: return ProjectResponseDTO( id=project.id, code=project.code, title=project.title, description=project.description, project_type=project.project_type, parent_id=project.parent_id, status=project.status, priority=project.priority, planned_start=project.planned_start, planned_end=project.planned_end, actual_start=project.actual_start, actual_end=project.actual_end, progress=project.progress, total_budget=project.total_budget, actual_cost=project.actual_cost, estimated_cost=project.estimated_cost, company_id=project.company_id, org_unit_id=project.org_unit_id, currency_id=project.currency_id, manager_id=project.manager_id, calendar_id=project.calendar_id, color=project.color, tags=project.tags, metadata=project.metadata, created_at=project.created_at, updated_at=project.updated_at, created_by=project.created_by, updated_by=project.updated_by, ) class TaskService: """سرویس مدیریت تسک.""" def __init__(self, tenant_id: UUID): self._tenant_id = tenant_id self._repo = DjangoTaskRepository(tenant_id) self._dep_repo = DjangoDependencyRepository(tenant_id) self._project_repo = DjangoProjectRepository(tenant_id) def create_task(self, dto: CreateTaskDTO) -> TaskResponseDTO: project = self._project_repo.get_by_id(dto.project_id) if not project: raise ProjectNotFoundError(dto.project_id) # محاسبه WBS code code = dto.code if not code: wbs = WBSService() existing_tasks = self._repo.find_by_project(dto.project_id) if dto.parent_id: parent = self._repo.get_by_id(dto.parent_id) parent_code = parent.code if parent else "" sibling_count = sum(1 for t in existing_tasks if t.parent_id == dto.parent_id) code = wbs.generate_code(parent_code, sibling_count + 1) else: root_count = sum(1 for t in existing_tasks if t.parent_id is None) code = wbs.generate_code("", root_count + 1) # محاسبه sort_order sort_order = dto.sort_order if sort_order is None: sort_order = self._repo.get_max_sort_order(dto.project_id, dto.parent_id) + 1 is_milestone = dto.task_type == "milestone" task = Task( id=uuid4(), tenant_id=self._tenant_id, project_id=dto.project_id, code=code, title=dto.title, description=dto.description, parent_id=dto.parent_id, sort_order=sort_order, task_type=dto.task_type, status="not_started", planned_start=dto.planned_start, planned_end=dto.planned_end, duration=dto.duration if not is_milestone else 0, planned_cost=dto.planned_cost, is_milestone=is_milestone, assignee_id=dto.assignee_id, color=dto.color, tags=dto.tags, notes=dto.notes, ) saved = self._repo.save(task) # Publish event _publish_event(TaskCreated( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, title=saved.title, parent_id=saved.parent_id, )) return self._to_response(saved) def update_task(self, dto: UpdateTaskDTO) -> TaskResponseDTO: task = self._repo.get_by_id(dto.id) if not task: raise TaskNotFoundError(dto.id) old_progress = task.progress old_assignee = task.assignee_id if dto.title is not None: task.title = dto.title if dto.description is not None: task.description = dto.description if dto.parent_id is not None: task.parent_id = dto.parent_id if dto.task_type is not None: task.task_type = dto.task_type task.is_milestone = dto.task_type == "milestone" if dto.planned_start is not None: task.planned_start = dto.planned_start if dto.planned_end is not None: task.planned_end = dto.planned_end if dto.duration is not None: task.duration = dto.duration if dto.progress is not None: task.progress = dto.progress if dto.progress > 0 and not task.actual_start: task.actual_start = date.today() task.status = "in_progress" if dto.progress >= 100: task.status = "completed" task.actual_end = date.today() if dto.planned_cost is not None: task.planned_cost = dto.planned_cost if dto.actual_cost is not None: task.actual_cost = dto.actual_cost if dto.assignee_id is not None: task.assignee_id = dto.assignee_id if dto.color is not None: task.color = dto.color if dto.tags is not None: task.tags = dto.tags if dto.notes is not None: task.notes = dto.notes if dto.sort_order is not None: task.sort_order = dto.sort_order saved = self._repo.save(task) # Publish progress event if dto.progress is not None and dto.progress != old_progress: _publish_event(TaskProgressUpdated( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, old_progress=old_progress, new_progress=dto.progress, )) # Publish assignee change event if dto.assignee_id is not None and dto.assignee_id != old_assignee: _publish_event(TaskAssigned( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, assignee_id=saved.assignee_id, )) # Publish completion events if saved.status == "completed" and saved.is_milestone: _publish_event(MilestoneReached( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, title=saved.title, reached_date=saved.actual_end, )) return self._to_response(saved) def change_status(self, task_id: UUID, new_status: str) -> TaskResponseDTO: task = self._repo.get_by_id(task_id) if not task: raise TaskNotFoundError(task_id) valid_transitions = { "not_started": ["in_progress", "cancelled"], "in_progress": ["completed", "on_hold", "cancelled"], "on_hold": ["in_progress", "cancelled"], } allowed = valid_transitions.get(task.status, []) if new_status not in allowed: raise InvalidStatusTransitionError("task", task.status, new_status) old_status = task.status task.status = new_status if new_status == "in_progress" and not task.actual_start: task.actual_start = date.today() elif new_status == "completed": task.actual_end = date.today() task.progress = 100 saved = self._repo.save(task) # Publish status change event _publish_event(TaskStatusChanged( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, old_status=old_status, new_status=new_status, )) # Publish completion / milestone events if new_status == "completed": _publish_event(TaskCompleted( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, actual_end=saved.actual_end, )) if saved.is_milestone: _publish_event(MilestoneReached( event_id=uuid4(), tenant_id=self._tenant_id, occurred_at=datetime.utcnow(), project_id=saved.project_id, task_id=saved.id, title=saved.title, reached_date=saved.actual_end, )) return self._to_response(saved) def get_task(self, task_id: UUID) -> TaskResponseDTO: task = self._repo.get_by_id(task_id) if not task: raise TaskNotFoundError(task_id) return self._to_response(task) def list_tasks(self, project_id: UUID) -> List[TaskResponseDTO]: tasks = self._repo.find_by_project(project_id) return [self._to_response(t) for t in tasks] def get_task_tree(self, project_id: UUID) -> List[dict]: """دریافت ساختار درختی تسک‌ها.""" all_tasks = self._repo.find_by_project(project_id) task_map = {t.id: t for t in all_tasks} def build_tree(parent_id=None): children = sorted( [t for t in all_tasks if t.parent_id == parent_id], key=lambda t: t.sort_order, ) result = [] for t in children: node = self._to_response(t).__dict__ node["children"] = build_tree(t.id) node["children_count"] = len(node["children"]) result.append(node) return result return build_tree() def move_task(self, dto: MoveTaskDTO) -> TaskResponseDTO: task = self._repo.get_by_id(dto.id) if not task: raise TaskNotFoundError(dto.id) if dto.parent_id is not None: task.parent_id = dto.parent_id if dto.sort_order is not None: task.sort_order = dto.sort_order saved = self._repo.save(task) return self._to_response(saved) def delete_task(self, task_id: UUID) -> bool: task = self._repo.get_by_id(task_id) if not task: raise TaskNotFoundError(task_id) self._dep_repo.delete_by_task(task_id) return self._repo.delete(task_id) def calculate_critical_path(self, project_id: UUID) -> List[TaskResponseDTO]: """محاسبه مسیر بحرانی و بروزرسانی تسک‌ها.""" tasks = self._repo.find_by_project(project_id) deps = self._dep_repo.find_by_project(project_id) cpm = CriticalPathService() updated_tasks = cpm.calculate(tasks, deps) self._repo.bulk_update_cpm(updated_tasks) critical = [t for t in updated_tasks if t.is_critical] return [self._to_response(t) for t in critical] def _to_response(self, task: Task) -> TaskResponseDTO: return TaskResponseDTO( id=task.id, project_id=task.project_id, code=task.code, title=task.title, description=task.description, parent_id=task.parent_id, sort_order=task.sort_order, task_type=task.task_type, status=task.status, planned_start=task.planned_start, planned_end=task.planned_end, actual_start=task.actual_start, actual_end=task.actual_end, duration=task.duration, progress=task.progress, planned_cost=task.planned_cost, actual_cost=task.actual_cost, is_critical=task.is_critical, is_milestone=task.is_milestone, early_start=task.early_start, early_finish=task.early_finish, late_start=task.late_start, late_finish=task.late_finish, total_float=task.total_float, free_float=task.free_float, assignee_id=task.assignee_id, color=task.color, tags=task.tags, notes=task.notes, metadata=task.metadata, created_at=task.created_at, updated_at=task.updated_at, ) class DependencyService: """سرویس مدیریت وابستگی.""" def __init__(self, tenant_id: UUID): self._tenant_id = tenant_id self._repo = DjangoDependencyRepository(tenant_id) self._task_repo = DjangoTaskRepository(tenant_id) def create_dependency(self, dto: CreateDependencyDTO) -> DependencyResponseDTO: # بررسی وجود تسک‌ها pred = self._task_repo.get_by_id(dto.predecessor_id) if not pred: raise TaskNotFoundError(dto.predecessor_id) succ = self._task_repo.get_by_id(dto.successor_id) if not succ: raise TaskNotFoundError(dto.successor_id) # بررسی وابستگی تکراری if self._repo.exists(dto.predecessor_id, dto.successor_id): raise DuplicateCodeError("dependency", f"{dto.predecessor_id}->{dto.successor_id}") # بررسی حلقه if self._would_create_cycle(dto.predecessor_id, dto.successor_id, dto.project_id): raise CircularDependencyError(dto.predecessor_id, dto.successor_id) dep = Dependency( id=uuid4(), tenant_id=self._tenant_id, project_id=dto.project_id, predecessor_id=dto.predecessor_id, successor_id=dto.successor_id, dependency_type=dto.dependency_type, lag_days=dto.lag_days, ) saved = self._repo.save(dep) return self._to_response(saved) def delete_dependency(self, dep_id: UUID) -> bool: return self._repo.delete(dep_id) def list_by_project(self, project_id: UUID) -> List[DependencyResponseDTO]: deps = self._repo.find_by_project(project_id) return [self._to_response(d) for d in deps] def _would_create_cycle(self, pred_id: UUID, succ_id: UUID, project_id: UUID) -> bool: """بررسی ایجاد حلقه با DFS.""" if pred_id == succ_id: return True deps = self._repo.find_by_project(project_id) graph = {} for d in deps: if d.predecessor_id not in graph: graph[d.predecessor_id] = [] graph[d.predecessor_id].append(d.successor_id) # اضافه‌کردن وابستگی جدید if pred_id not in graph: graph[pred_id] = [] graph[pred_id].append(succ_id) # DFS از succ_id visited = set() stack = [succ_id] while stack: node = stack.pop() if node == pred_id: return True if node in visited: continue visited.add(node) stack.extend(graph.get(node, [])) return False def _to_response(self, dep: Dependency) -> DependencyResponseDTO: return DependencyResponseDTO( id=dep.id, project_id=dep.project_id, predecessor_id=dep.predecessor_id, successor_id=dep.successor_id, dependency_type=dep.dependency_type, lag_days=dep.lag_days, created_at=dep.created_at, )