fix: reduce cognitive complexity across diffctx pipeline (SonarCloud S3776)

Author: nikolay-e

benchmarks/summarize_results.py

@@ -7,26 +7,25 @@
 from common import load_results
 
 
+def _print_txt_section(txt: Path, prefix: str, title: str, markers: tuple[str, ...]) -> None:
+    mode = txt.stem.replace(prefix, "")
+    print(f"### {title} ({mode})\n```")
+    for line in txt.read_text().splitlines():
+        if line.startswith(markers):
+            print(line)
+    print("```\n")
+
+
 def main() -> None:
     results_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("results")
 
     print("## Benchmark Results\n")
 
     for txt in sorted(results_dir.glob("cb_*.txt")):
-        mode = txt.stem.replace("cb_", "")
-        print(f"### ContextBench ({mode})\n```")
-        for line in txt.read_text().splitlines():
-            if line.startswith(("Avg ", "Total:")):
-                print(line)
-        print("```\n")
+        _print_txt_section(txt, "cb_", "ContextBench", ("Avg ", "Total:"))
 
     for txt in sorted(results_dir.glob("loo_*.txt")):
-        mode = txt.stem.replace("loo_", "")
-        print(f"### LOO ({mode})\n```")
-        for line in txt.read_text().splitlines():
-            if line.startswith(("Total LOO", "Found")):
-                print(line)
-        print("```\n")
+        _print_txt_section(txt, "loo_", "LOO", ("Total LOO", "Found"))
 
     for jf in sorted(results_dir.glob("loo_*.json")):
         mode = jf.stem.replace("loo_", "")

src/treemapper/cli.py

@@ -63,34 +63,42 @@ def _resolve_root_dir(directory: str) -> Path:
         _exit_error(f"Cannot access '{directory}': {e}")
 
 
-def _expand_paths(raw_paths: list[str]) -> tuple[list[Path], list[Path]]:
+def _resolve_glob_pattern(pattern: str) -> list[str]:
     import glob as globmod
 
+    matches = sorted(globmod.glob(pattern, recursive=True))
+    if matches:
+        return matches
+    try:
+        p = Path(pattern).resolve(strict=True)
+    except FileNotFoundError:
+        _exit_error(f"No matches for '{pattern}'")
+    except OSError as e:
+        _exit_error(f"Cannot access '{pattern}': {e}")
+    return [str(p)]
+
+
+def _classify_resolved(resolved: Path, dirs: list[Path], files: list[Path]) -> None:
+    if resolved.is_dir():
+        dirs.append(resolved)
+    elif resolved.is_file():
+        files.append(resolved)
+
+
+def _expand_paths(raw_paths: list[str]) -> tuple[list[Path], list[Path]]:
     dirs: list[Path] = []
     files: list[Path] = []
     seen: set[Path] = set()
     for pattern in raw_paths:
-        matches = sorted(globmod.glob(pattern, recursive=True))
-        if not matches:
-            try:
-                p = Path(pattern).resolve(strict=True)
-            except FileNotFoundError:
-                _exit_error(f"No matches for '{pattern}'")
-            except OSError as e:
-                _exit_error(f"Cannot access '{pattern}': {e}")
-            matches = [str(p)]
-        for m in matches:
+        for m in _resolve_glob_pattern(pattern):
             try:
                 resolved = Path(m).resolve()
             except OSError as e:
                 _exit_error(f"Cannot access '{m}': {e}")
             if resolved in seen:
                 continue
             seen.add(resolved)
-            if resolved.is_dir():
-                dirs.append(resolved)
-            elif resolved.is_file():
-                files.append(resolved)
+            _classify_resolved(resolved, dirs, files)
     return dirs, files
 
 

src/treemapper/diffctx/edges/semantic/javascript.py

@@ -472,6 +472,21 @@ def build(self, fragments: list[Fragment], repo_root: Path | None = None) -> Edg
     _IMPORT_WEIGHT = 0.55
     _REEXPORT_WEIGHT_FACTOR = 0.8
 
+    def _link_resolved_import(
+        self,
+        src_path: Path,
+        resolved: Path,
+        file_to_frags: dict[Path, list[FragmentId]],
+        fragment_paths: set[Path],
+        edges: EdgeDict,
+    ) -> None:
+        if resolved == src_path:
+            return
+        target_ids = file_to_frags.get(resolved, [])
+        if target_ids:
+            self._link_import_pairs(file_to_frags[src_path], target_ids, edges)
+            self._follow_reexports(resolved, file_to_frags[src_path], file_to_frags, fragment_paths, edges)
+
     def _add_import_edges(
         self,
         js_frags: list[Fragment],
@@ -484,43 +499,17 @@ def _add_import_edges(
             file_to_frags[f.path].append(f.id)
 
         fragment_paths = set(file_to_frags.keys())
-        file_imports, alias_resolved = self._collect_imports(
-            js_frags,
-            info_cache,
-            tsconfig_resolver,
-            fragment_paths,
-        )
+        file_imports, alias_resolved = self._collect_imports(js_frags, info_cache, tsconfig_resolver, fragment_paths)
 
         for src_path, import_sources in file_imports.items():
             for import_source in import_sources:
                 resolved = _resolve_relative_import(src_path, import_source, fragment_paths)
-                if resolved is None or resolved == src_path:
-                    continue
-                target_ids = file_to_frags.get(resolved, [])
-                if target_ids:
-                    self._link_import_pairs(file_to_frags[src_path], target_ids, edges)
-                    self._follow_reexports(
-                        resolved,
-                        file_to_frags[src_path],
-                        file_to_frags,
-                        fragment_paths,
-                        edges,
-                    )
+                if resolved is not None:
+                    self._link_resolved_import(src_path, resolved, file_to_frags, fragment_paths, edges)
 
         for src_path, resolved_targets in alias_resolved.items():
             for resolved in resolved_targets:
-                if resolved == src_path:
-                    continue
-                target_ids = file_to_frags.get(resolved, [])
-                if target_ids:
-                    self._link_import_pairs(file_to_frags[src_path], target_ids, edges)
-                    self._follow_reexports(
-                        resolved,
-                        file_to_frags[src_path],
-                        file_to_frags,
-                        fragment_paths,
-                        edges,
-                    )
+                self._link_resolved_import(src_path, resolved, file_to_frags, fragment_paths, edges)
 
     @staticmethod
     def _collect_imports(

src/treemapper/diffctx/edges/semantic/jvm.py

@@ -295,56 +295,51 @@ def _compute_import_dirs(repo_root: Path | None, import_packages: set[str]) -> s
                     import_dirs.add(repo_root / src_prefix / Path(*pkg.split(".")))
         return import_dirs
 
-    def _discover_single_hop(
-        self,
-        source_files: list[Path],
-        candidates: list[Path],
-        repo_root: Path | None,
-    ) -> list[Path]:
-        type_refs, import_packages = self._collect_source_refs(source_files)
-        source_dirs = {f.parent for f in source_files}
-        eligible_dirs = source_dirs | self._compute_import_dirs(repo_root, import_packages)
-        source_set = set(source_files)
-
+    @staticmethod
+    def _collect_frontier_classes(source_files: list[Path]) -> set[str]:
         frontier_classes: set[str] = set()
-        frontier_packages: set[str] = set()
         for f in source_files:
             try:
                 content = f.read_text(encoding="utf-8")
                 frontier_classes.update(_extract_classes(content, f))
-                pkg = _extract_package(content)
-                if pkg:
-                    frontier_packages.add(pkg)
             except (OSError, UnicodeDecodeError):
                 pass
+        return frontier_classes
 
-        discovered: list[Path] = []
-        for candidate in candidates:
-            if candidate in source_set:
-                continue
-            try:
-                content = candidate.read_text(encoding="utf-8")
-                cand_classes = _extract_classes(content, candidate)
-
-                if candidate.parent in eligible_dirs and cand_classes & type_refs:
-                    discovered.append(candidate)
-                    continue
-
-                cand_type_refs = _extract_type_refs(content)
-                if cand_type_refs & frontier_classes:
-                    discovered.append(candidate)
-                    continue
-
-                cand_imports = _extract_imports(content, candidate)
-                for imp in cand_imports:
-                    imp_class = imp.rsplit(".", 1)[-1]
-                    if imp_class in frontier_classes:
-                        discovered.append(candidate)
-                        break
-            except (OSError, UnicodeDecodeError):
-                pass
+    @staticmethod
+    def _candidate_matches_frontier(
+        candidate: Path,
+        eligible_dirs: set[Path],
+        type_refs: set[str],
+        frontier_classes: set[str],
+    ) -> bool:
+        try:
+            content = candidate.read_text(encoding="utf-8")
+            cand_classes = _extract_classes(content, candidate)
+            if candidate.parent in eligible_dirs and cand_classes & type_refs:
+                return True
+            if _extract_type_refs(content) & frontier_classes:
+                return True
+            return any(imp.rsplit(".", 1)[-1] in frontier_classes for imp in _extract_imports(content, candidate))
+        except (OSError, UnicodeDecodeError):
+            return False
+
+    def _discover_single_hop(
+        self,
+        source_files: list[Path],
+        candidates: list[Path],
+        repo_root: Path | None,
+    ) -> list[Path]:
+        type_refs, import_packages = self._collect_source_refs(source_files)
+        eligible_dirs = {f.parent for f in source_files} | self._compute_import_dirs(repo_root, import_packages)
+        source_set = set(source_files)
+        frontier_classes = self._collect_frontier_classes(source_files)
 
-        return discovered
+        return [
+            c
+            for c in candidates
+            if c not in source_set and self._candidate_matches_frontier(c, eligible_dirs, type_refs, frontier_classes)
+        ]
 
     def build(self, fragments: list[Fragment], repo_root: Path | None = None) -> EdgeDict:
         jvm_frags = [f for f in fragments if _is_jvm_file(f.path)]