import fnmatch import json import logging import os from datetime import datetime, timezone from typing import Dict, List, Optional, Set import networkx as nx from models.dead_code import ( DeadCodeResult, DeadFile, OrphanModule, DeadDependencyChain, ) from services.github_service import GitHubService from services.graph_service import GraphService from services.architecture_service import ArchitectureService from core.repository_context import RepositoryContext logger = logging.getLogger(__name__) class DeadCodeService: """Orchestrates Dead Code Intelligence (PH2-005A).""" def __init__( self, github_service: Optional[GitHubService] = None, graph_service: Optional[GraphService] = None, architecture_service: Optional[ArchitectureService] = None, scores_file_path: Optional[str] = None, ) -> None: self.github_service = github_service or GitHubService() self.graph_service = graph_service or GraphService() self.architecture_service = architecture_service or ArchitectureService() if scores_file_path is None: project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) self.scores_file_path = os.path.join( project_root, "data", "dead_code_scores.json" ) else: self.scores_file_path = scores_file_path self.default_ignores = [ "migrations/", "examples/", "playground/", "templates/", "legacy/", "**pycache__/", "dist/", "build/", ] def is_ignored(self, file_path: str, patterns: List[str]) -> bool: """Determines if a file path matches any of the ignore patterns (case-insensitive).""" norm_path = file_path.replace("\\", "/").lower() for pat in patterns: pat_clean = pat.replace("\\", "/").lower() # Directory pattern (e.g. migrations/) if pat_clean.endswith("/"): dir_name = pat_clean[:-1] parts = norm_path.split("/") if dir_name in parts: return True # Standard glob matching if fnmatch.fnmatch(norm_path, pat_clean) or fnmatch.fnmatch( os.path.basename(norm_path), pat_clean ): return True return False def load_ignore_patterns(self) -> List[str]: """Loads default patterns and appends customization from data/dead_code_ignore.json.""" patterns = list(self.default_ignores) project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ignore_file = os.path.join(project_root, "data", "dead_code_ignore.json") if os.path.exists(ignore_file): try: with open(ignore_file, "r", encoding="utf-8") as fh: data = json.load(fh) if isinstance(data, list): patterns.extend(data) except Exception as exc: logger.warning("Failed to load dead_code_ignore.json: %s", exc) return patterns def _find_last_reachable_parent( self, graph: nx.DiGraph, orphan: str, reachable_nodes: Set[str] ) -> Optional[str]: """Finds the closest reachable node connected to the orphan in the undirected graph.""" if not reachable_nodes: return None ug = graph.to_undirected() if orphan not in ug: return None best_node = None min_dist = float("inf") for r_node in reachable_nodes: if r_node in ug: try: if nx.has_path(ug, orphan, r_node): dist = nx.shortest_path_length(ug, orphan, r_node) if dist < min_dist: min_dist = dist best_node = r_node except Exception: pass return best_node def _find_dead_chains( self, unreachable_subgraph: nx.DiGraph, centrality: Dict[str, float] ) -> List[DeadDependencyChain]: """Finds weakly connected components in the unreachable graph and extracts path chains.""" chains = [] H = unreachable_subgraph components = list(nx.weakly_connected_components(H)) for comp in components: comp_sub = H.subgraph(comp) comp_nodes = list(comp) if len(comp_nodes) < 2: continue roots = [n for n in comp_nodes if comp_sub.in_degree(n) == 0] if not roots: roots = [comp_nodes[0]] all_paths = [] visited = set() def dfs_paths(node, current_path): visited.add(node) current_path.append(node) successors = [s for s in comp_sub.successors(node) if s not in visited] if not successors: if len(current_path) >= 2: all_paths.append(list(current_path)) else: for s in successors: dfs_paths(s, list(current_path)) visited.remove(node) for r in roots: dfs_paths(r, []) if not all_paths: continue # Sort paths by length descending all_paths.sort(key=len, reverse=True) # De-duplicate sub-paths filtered_paths = [] for p in all_paths: is_subpath = False for fp in filtered_paths: for i in range(len(fp) - len(p) + 1): if fp[i : i + len(p)] == p: is_subpath = True break if is_subpath: break if not is_subpath: filtered_paths.append(p) for path in filtered_paths: length = len(path) - 1 total_nodes = len(path) max_cent = max(centrality.get(n, 0.0) for n in path) chain_rep = " -> ".join(os.path.basename(n) for n in path) recommendation = f"Dependency chain [{chain_rep}] appears unreachable and may be removable as a unit." chains.append( DeadDependencyChain( chain=path, confidence=0.95, risk_level="SAFE", recommendation=recommendation, length=length, total_nodes=total_nodes, max_centrality=round(max_cent, 4), ) ) return chains def _load_previous_score(self, repo_fullName: str) -> Optional[int]: if os.path.exists(self.scores_file_path): try: with open(self.scores_file_path, "r", encoding="utf-8") as fh: data = json.load(fh) if isinstance(data, dict) and repo_fullName in data: return data[repo_fullName].get("score") except Exception as exc: logger.warning("Failed to load previous score: %s", exc) return None def _save_new_score(self, repo_fullName: str, score: int) -> None: data = {} os.makedirs(os.path.dirname(self.scores_file_path), exist_ok=True) if os.path.exists(self.scores_file_path): try: with open(self.scores_file_path, "r", encoding="utf-8") as fh: data = json.load(fh) except Exception: pass data[repo_fullName] = { "score": score, "updated_at": datetime.now(timezone.utc).isoformat(), } try: with open(self.scores_file_path, "w", encoding="utf-8") as fh: json.dump(data, fh, indent=2) except Exception as exc: logger.warning("Failed to save score to %s: %s", self.scores_file_path, exc) def analyze(self, repo_name: str) -> DeadCodeResult: """Wrapper method for compatibility with ReportComposer and test mocks.""" owner, repo = repo_name.split("/", 1) return self.analyze_dead_code(owner, repo) def analyze_dead_code( self, owner: str, repo: str, context: Optional[RepositoryContext] = None, ) -> DeadCodeResult: """Orchestrates reachability dead code calculations.""" repo_fullName = f"{owner}/{repo}" if context is None: context = RepositoryContext(repo_fullName) if not self.graph_service.graph_exists(repo_fullName): raise ValueError( f"No dependency graph found for '{repo_fullName}'. " "Run repository architecture build first." ) summary = self.architecture_service.get_summary(repo_fullName) if not summary: raise ValueError( f"No architecture summary found for '{repo_fullName}'. " "Run repository architecture build first." ) G = context.dependency_graph if G is None: G = self.graph_service.load_graph(repo_fullName) if G is None: G = nx.DiGraph() ignore_patterns = self.load_ignore_patterns() # Find entry points entry_points = set(summary.entry_points or []) for node in G.nodes: basename = os.path.basename(node).lower() if basename in { "main.py", "__main__.py", "app.py", "application.py", "server.py", "index.js", "server.js", "app.js", "main.tsx", "app.tsx", "api.py", "run.py", "manage.py", "wsgi.py", "asgi.py", }: entry_points.add(node) # Reachability calculations reachable_nodes = set() for ep in entry_points: if ep in G: reachable_nodes.add(ep) reachable_nodes.update(nx.descendants(G, ep)) all_unreachable = set(G.nodes) - reachable_nodes unreachable_filtered = { node for node in all_unreachable if not self.is_ignored(node, ignore_patterns) } reachable_filtered = { node for node in reachable_nodes if not self.is_ignored(node, ignore_patterns) } H = G.subgraph(unreachable_filtered) centrality = {} if G.number_of_nodes() > 1: centrality = nx.degree_centrality(G) else: centrality = {n: 0.0 for n in G.nodes} unused_files_list = [] orphan_modules_list = [] total_deductions = 0.0 for v in unreachable_filtered: cent_val = centrality.get(v, 0.0) out_deg = G.out_degree(v) subtree_size = len(nx.descendants(H, v)) # Graph-weighted score: factors in centrality, out degree, and subtree size node_weight = 1.0 + 10.0 * cent_val + 0.5 * out_deg + 1.0 * subtree_size in_deg = G.in_degree(v) if in_deg == 0: base_factor = 6.0 total_deductions += base_factor * node_weight unused_files_list.append( DeadFile( file_path=v, confidence=0.95, risk_level="SAFE", recommendation=f"Consider removing unused file {v}", ) ) else: base_factor = 4.0 total_deductions += base_factor * node_weight last_parent = self._find_last_reachable_parent(G, v, reachable_filtered) rec = ( f"Review orphaned module {v}; no active execution path reaches it." ) orphan_modules_list.append( OrphanModule( file_path=v, confidence=0.90, risk_level="REVIEW", recommendation=rec, last_reachable_parent=last_parent, ) ) chains = self._find_dead_chains(H, centrality) for c in chains: total_deductions += 3.0 * len(c.chain) cleanup_score = max(0, min(100, 100 - int(total_deductions))) total_findings_count = ( len(unused_files_list) + len(orphan_modules_list) + len(chains) ) if total_findings_count <= 2: effort = "LOW" elif total_findings_count <= 10: effort = "MEDIUM" else: effort = "HIGH" # Generate recommendations sorted by impact recommendations_with_weights = [] for f in unused_files_list: cent = centrality.get(f.file_path, 0.0) od = G.out_degree(f.file_path) st = len(nx.descendants(H, f.file_path)) w = 6.0 * (1.0 + 10.0 * cent + 0.5 * od + 1.0 * st) if st > 0: msg = f"Remove unused subtree starting at {f.file_path} (cascades to {st} dead modules)." else: msg = f"Remove unused file {f.file_path} (no active imports)." recommendations_with_weights.append((msg, w)) for o in orphan_modules_list: cent = centrality.get(o.file_path, 0.0) od = G.out_degree(o.file_path) st = len(nx.descendants(H, o.file_path)) w = 4.0 * (1.0 + 10.0 * cent + 0.5 * od + 1.0 * st) if o.last_reachable_parent: msg = f"Review orphaned module {o.file_path} (previously connected via {o.last_reachable_parent})." else: msg = f"Review orphaned module {o.file_path} (completely isolated from active graph)." recommendations_with_weights.append((msg, w)) for c in chains: w = 3.0 * len(c.chain) chain_rep = " -> ".join(os.path.basename(n) for n in c.chain) msg = f"Delete unreachable dependency chain: {chain_rep} (removes {len(c.chain)} dead files)." recommendations_with_weights.append((msg, w)) recommendations_with_weights.sort(key=lambda x: x[1], reverse=True) cleanup_recommendations = [r[0] for r in recommendations_with_weights[:10]] prev_score = self._load_previous_score(repo_fullName) self._save_new_score(repo_fullName, cleanup_score) return DeadCodeResult( repo=repo_fullName, cleanup_score=cleanup_score, previous_cleanup_score=prev_score, estimated_cleanup_effort=effort, unused_files=unused_files_list, orphan_modules=orphan_modules_list, dead_dependency_chains=chains, cleanup_recommendations=cleanup_recommendations, analyzed_at=datetime.now(timezone.utc).isoformat(), )