""" PPM Domain Service — Dependency Analyzer DependencyAnalyzer — تحلیل وابستگی‌های بین پروژه‌ای. """ from collections import defaultdict from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Set from uuid import UUID @dataclass class CircularDependency: """یک حلقه وابستگی تشخیص‌داده‌شده.""" cycle: List[str] = field(default_factory=list) # List of project_id strings in the cycle description: str = "" @dataclass class CriticalChainLink: """یک حلقه در زنجیره بحرانی.""" project_id: str = "" predecessor_id: Optional[str] = None dependency_type: str = "FS" lag_days: int = 0 total_slack: int = 0 @dataclass class ImpactResult: """نتیجه تحلیل اثر تأخیر.""" delayed_project_id: str = "" delay_days: int = 0 affected_projects: List[Dict[str, Any]] = field(default_factory=list) total_cascade_days: int = 0 class DependencyAnalyzer: """ سرویس دامنه — تحلیل وابستگی‌های بین پروژه‌ای (Stateless). متدها: - detect_circular_dependencies: تشخیص وابستگی‌های حلقوی - calculate_critical_chain: محاسبه زنجیره بحرانی - impact_analysis: تحلیل اثر تأخیر یک پروژه """ def detect_circular_dependencies( self, dependencies: List[Any], ) -> List[CircularDependency]: """ تشخیص وابستگی‌های حلقوی (Circular Dependencies). با الگوریتم DFS در گراف جهت‌دار، حلقه‌ها را پیدا می‌کند. """ # Build adjacency list graph: Dict[str, Set[str]] = defaultdict(set) for dep in dependencies: pred = str(dep.predecessor_project_id) if dep.predecessor_project_id else None succ = str(dep.successor_project_id) if dep.successor_project_id else None if pred and succ and dep.status == "active": graph[pred].add(succ) # DFS cycle detection visited: Set[str] = set() rec_stack: Set[str] = set() cycles: List[CircularDependency] = [] path: List[str] = [] def _dfs(node: str) -> None: visited.add(node) rec_stack.add(node) path.append(node) for neighbor in graph.get(node, set()): if neighbor not in visited: _dfs(neighbor) elif neighbor in rec_stack: # Found a cycle cycle_start = path.index(neighbor) cycle = path[cycle_start:] + [neighbor] cycles.append(CircularDependency( cycle=cycle, description=f"Circular dependency detected: {' → '.join(cycle)}", )) path.pop() rec_stack.discard(node) all_nodes = set(graph.keys()) for targets in graph.values(): all_nodes.update(targets) for node in all_nodes: if node not in visited: _dfs(node) return cycles def calculate_critical_chain( self, dependencies: List[Any], ) -> List[CriticalChainLink]: """ محاسبه زنجیره بحرانی. بر اساس وابستگی‌های فعال، طولانی‌ترین مسیر (Critical Chain) را تشخیص می‌دهد. """ # Build adjacency list with lag graph: Dict[str, List[Dict[str, Any]]] = defaultdict(list) all_nodes: Set[str] = set() for dep in dependencies: pred = str(dep.predecessor_project_id) if dep.predecessor_project_id else None succ = str(dep.successor_project_id) if dep.successor_project_id else None if pred and succ and dep.status == "active": graph[pred].append({ "successor": succ, "dependency_type": dep.dependency_type, "lag_days": dep.lag_days, }) all_nodes.add(pred) all_nodes.add(succ) if not all_nodes: return [] # Topological sort + longest path in_degree: Dict[str, int] = defaultdict(int) for node in all_nodes: in_degree.setdefault(node, 0) for edges in graph.values(): for edge in edges: in_degree[edge["successor"]] += 1 # Find starting nodes (in_degree == 0) queue = [n for n in all_nodes if in_degree[n] == 0] dist: Dict[str, int] = {n: 0 for n in all_nodes} predecessor_map: Dict[str, Optional[Dict[str, Any]]] = {n: None for n in all_nodes} # BFS-based topological traversal topo_order: List[str] = [] q = list(queue) while q: node = q.pop(0) topo_order.append(node) for edge in graph.get(node, []): succ = edge["successor"] new_dist = dist[node] + edge["lag_days"] if new_dist > dist[succ]: dist[succ] = new_dist predecessor_map[succ] = {"predecessor": node, **edge} in_degree[succ] -= 1 if in_degree[succ] == 0: q.append(succ) # Reconstruct critical chain from the node with max dist if not dist: return [] end_node = max(dist, key=dist.get) chain: List[CriticalChainLink] = [] current = end_node while current is not None: pred_info = predecessor_map.get(current) chain.append(CriticalChainLink( project_id=current, predecessor_id=pred_info["predecessor"] if pred_info else None, dependency_type=pred_info["dependency_type"] if pred_info else "FS", lag_days=pred_info["lag_days"] if pred_info else 0, total_slack=0, )) current = pred_info["predecessor"] if pred_info else None chain.reverse() return chain def impact_analysis( self, dependencies: List[Any], delayed_project_id: str, delay_days: int, ) -> ImpactResult: """ تحلیل اثر تأخیر یک پروژه بر پروژه‌های وابسته. به صورت BFS از پروژه تأخیردار، اثر آبشاری را محاسبه می‌کند. """ # Build successor graph graph: Dict[str, List[Dict[str, Any]]] = defaultdict(list) for dep in dependencies: pred = str(dep.predecessor_project_id) if dep.predecessor_project_id else None succ = str(dep.successor_project_id) if dep.successor_project_id else None if pred and succ and dep.status == "active": graph[pred].append({ "successor": succ, "dependency_type": dep.dependency_type, "lag_days": dep.lag_days, }) affected: List[Dict[str, Any]] = [] visited: Set[str] = set() queue = [(delayed_project_id, delay_days)] max_cascade = 0 while queue: current_id, current_delay = queue.pop(0) if current_id in visited: continue visited.add(current_id) for edge in graph.get(current_id, []): succ = edge["successor"] # For FS dependencies, the cascade delay includes lag cascade_delay = max(0, current_delay - edge["lag_days"]) if edge["dependency_type"] == "FS": cascade_delay = current_delay elif edge["dependency_type"] == "SS": cascade_delay = current_delay elif edge["dependency_type"] == "FF": cascade_delay = current_delay elif edge["dependency_type"] == "SF": cascade_delay = max(0, current_delay - edge["lag_days"]) if cascade_delay > 0 and succ not in visited: affected.append({ "project_id": succ, "cascade_delay_days": cascade_delay, "dependency_type": edge["dependency_type"], "from_project": current_id, }) max_cascade = max(max_cascade, cascade_delay) queue.append((succ, cascade_delay)) return ImpactResult( delayed_project_id=delayed_project_id, delay_days=delay_days, affected_projects=affected, total_cascade_days=max_cascade, )