perf(core): add lazy suppression/report fast-paths and streamline extractor binding

Author: orenlab

codeclone/_cli_reports.py

@@ -22,38 +22,14 @@ class _QuietArgs(Protocol):
     quiet: bool
 
 
-class _OutputPaths(Protocol):
-    @property
-    def html(self) -> Path | None: ...
+def _path_attr(obj: object, name: str) -> Path | None:
+    value = getattr(obj, name, None)
+    return value if isinstance(value, Path) else None
 
-    @property
-    def json(self) -> Path | None: ...
 
-    @property
-    def md(self) -> Path | None: ...
-
-    @property
-    def sarif(self) -> Path | None: ...
-
-    @property
-    def text(self) -> Path | None: ...
-
-
-class _ReportArtifacts(Protocol):
-    @property
-    def html(self) -> str | None: ...
-
-    @property
-    def json(self) -> str | None: ...
-
-    @property
-    def md(self) -> str | None: ...
-
-    @property
-    def sarif(self) -> str | None: ...
-
-    @property
-    def text(self) -> str | None: ...
+def _text_attr(obj: object, name: str) -> str | None:
+    value = getattr(obj, name, None)
+    return value if isinstance(value, str) else None
 
 
 def _write_report_output(
@@ -83,60 +59,70 @@ def _open_html_report_in_browser(*, path: Path) -> None:
 def write_report_outputs(
     *,
     args: _QuietArgs,
-    output_paths: _OutputPaths,
-    report_artifacts: _ReportArtifacts,
+    output_paths: object,
+    report_artifacts: object,
     console: _PrinterLike,
     open_html_report: bool = False,
 ) -> str | None:
     html_report_path: str | None = None
     saved_reports: list[tuple[str, Path]] = []
-
-    if output_paths.html and report_artifacts.html is not None:
-        out = output_paths.html
+    html_path = _path_attr(output_paths, "html")
+    json_path = _path_attr(output_paths, "json")
+    md_path = _path_attr(output_paths, "md")
+    sarif_path = _path_attr(output_paths, "sarif")
+    text_path = _path_attr(output_paths, "text")
+    html_report = _text_attr(report_artifacts, "html")
+    json_report = _text_attr(report_artifacts, "json")
+    md_report = _text_attr(report_artifacts, "md")
+    sarif_report = _text_attr(report_artifacts, "sarif")
+    text_report = _text_attr(report_artifacts, "text")
+
+    if html_path and html_report is not None:
+        out = html_path
         _write_report_output(
             out=out,
-            content=report_artifacts.html,
+            content=html_report,
             label="HTML",
             console=console,
         )
         html_report_path = str(out)
         saved_reports.append(("HTML", out))
 
-    if output_paths.json and report_artifacts.json is not None:
-        out = output_paths.json
+    if json_path and json_report is not None:
+        out = json_path
         _write_report_output(
             out=out,
-            content=report_artifacts.json,
+            content=json_report,
             label="JSON",
             console=console,
         )
         saved_reports.append(("JSON", out))
 
-    if output_paths.md and report_artifacts.md is not None:
-        out = output_paths.md
+    if md_path and md_report is not None:
+        out = md_path
         _write_report_output(
             out=out,
-            content=report_artifacts.md,
+            content=md_report,
             label="Markdown",
             console=console,
         )
         saved_reports.append(("Markdown", out))
 
-    if output_paths.sarif and report_artifacts.sarif is not None:
-        out = output_paths.sarif
+    if sarif_path and sarif_report is not None:
+        out = sarif_path
         _write_report_output(
             out=out,
-            content=report_artifacts.sarif,
+            content=sarif_report,
             label="SARIF",
             console=console,
         )
         saved_reports.append(("SARIF", out))
 
-    if output_paths.text and report_artifacts.text is not None:
-        out = output_paths.text
+    if text_path and text_report is not None:
+        out = text_path
         _write_report_output(
             out=out,
-            content=report_artifacts.text,
+            content=text_report,
             label="text",
             console=console,
         )
@@ -152,12 +138,10 @@ def write_report_outputs(
                 display = path
             console.print(f"  [bold]{label} report saved:[/bold] [dim]{display}[/dim]")
 
-    if open_html_report and output_paths.html is not None:
+    if open_html_report and html_path is not None:
         try:
-            _open_html_report_in_browser(path=output_paths.html)
+            _open_html_report_in_browser(path=html_path)
         except Exception as exc:
-            console.print(
-                ui.fmt_html_report_open_failed(path=output_paths.html, error=exc)
-            )
+            console.print(ui.fmt_html_report_open_failed(path=html_path, error=exc))
 
     return html_report_path

codeclone/extractor.py

@@ -77,6 +77,7 @@ class _ParseTimeoutError(Exception):
 
 FunctionNode = ast.FunctionDef | ast.AsyncFunctionDef
 _NamedDeclarationNode = FunctionNode | ast.ClassDef
+_DeclarationTokenIndexKey = tuple[int, int, str]
 
 
 def _consumed_cpu_seconds(resource_module: object) -> float:
@@ -177,7 +178,10 @@ def _declaration_token_index(
     start_line: int,
     start_col: int,
     declaration_token: str,
+    source_token_index: Mapping[_DeclarationTokenIndexKey, int] | None = None,
 ) -> int | None:
+    if source_token_index is not None:
+        return source_token_index.get((start_line, start_col, declaration_token))
     for idx, token in enumerate(source_tokens):
         if token.start != (start_line, start_col):
             continue
@@ -186,6 +190,19 @@ def _declaration_token_index(
     return None
 
 
+def _build_declaration_token_index(
+    source_tokens: tuple[tokenize.TokenInfo, ...],
+) -> Mapping[_DeclarationTokenIndexKey, int]:
+    indexed: dict[_DeclarationTokenIndexKey, int] = {}
+    for idx, token in enumerate(source_tokens):
+        if token.type != tokenize.NAME:
+            continue
+        if token.string not in {"def", "async", "class"}:
+            continue
+        indexed[(token.start[0], token.start[1], token.string)] = idx
+    return indexed
+
+
 def _scan_declaration_colon_line(
     *,
     source_tokens: tuple[tokenize.TokenInfo, ...],
@@ -223,6 +240,7 @@ def _declaration_end_line(
     node: ast.AST,
     *,
     source_tokens: tuple[tokenize.TokenInfo, ...],
+    source_token_index: Mapping[_DeclarationTokenIndexKey, int] | None = None,
 ) -> int:
     start_line = int(getattr(node, "lineno", 0))
     start_col = int(getattr(node, "col_offset", 0))
@@ -235,6 +253,7 @@ def _declaration_end_line(
         start_line=start_line,
         start_col=start_col,
         declaration_token=declaration_token,
+        source_token_index=source_token_index,
     )
     if start_index is None:
         return _fallback_declaration_end_line(node, start_line=start_line)
@@ -788,7 +807,9 @@ def _collect_declaration_targets(
     filepath: str,
     module_name: str,
     collector: _QualnameCollector,
-    source_tokens: tuple[tokenize.TokenInfo, ...],
+    source_tokens: tuple[tokenize.TokenInfo, ...] = (),
+    source_token_index: Mapping[_DeclarationTokenIndexKey, int] | None = None,
+    include_inline_lines: bool = False,
 ) -> tuple[DeclarationTarget, ...]:
     declarations: list[DeclarationTarget] = []
 
@@ -797,9 +818,14 @@ def _collect_declaration_targets(
         end = int(getattr(node, "end_lineno", 0))
         if start <= 0 or end <= 0:
             continue
-        declaration_end_line = _declaration_end_line(
-            node,
-            source_tokens=source_tokens,
+        declaration_end_line = (
+            _declaration_end_line(
+                node,
+                source_tokens=source_tokens,
+                source_token_index=source_token_index,
+            )
+            if include_inline_lines
+            else None
         )
         kind: Literal["function", "method"] = (
             "method" if "." in local_name else "function"
@@ -820,9 +846,14 @@ def _collect_declaration_targets(
         end = int(getattr(class_node, "end_lineno", 0))
         if start <= 0 or end <= 0:
             continue
-        declaration_end_line = _declaration_end_line(
-            class_node,
-            source_tokens=source_tokens,
+        declaration_end_line = (
+            _declaration_end_line(
+                class_node,
+                source_tokens=source_tokens,
+                source_token_index=source_token_index,
+            )
+            if include_inline_lines
+            else None
         )
         declarations.append(
             DeclarationTarget(
@@ -849,6 +880,42 @@ def _collect_declaration_targets(
     )
 
 
+def _build_suppression_index_for_source(
+    *,
+    source: str,
+    filepath: str,
+    module_name: str,
+    collector: _QualnameCollector,
+) -> Mapping[SuppressionTargetKey, tuple[str, ...]]:
+    suppression_directives = extract_suppression_directives(source)
+    if not suppression_directives:
+        return {}
+
+    needs_inline_binding = any(
+        directive.binding == "inline" for directive in suppression_directives
+    )
+    source_tokens: tuple[tokenize.TokenInfo, ...] = ()
+    source_token_index: Mapping[_DeclarationTokenIndexKey, int] | None = None
+    if needs_inline_binding:
+        source_tokens = _source_tokens(source)
+        if source_tokens:
+            source_token_index = _build_declaration_token_index(source_tokens)
+
+    declaration_targets = _collect_declaration_targets(
+        filepath=filepath,
+        module_name=module_name,
+        collector=collector,
+        source_tokens=source_tokens,
+        source_token_index=source_token_index,
+        include_inline_lines=needs_inline_binding,
+    )
+    suppression_bindings = bind_suppressions_to_declarations(
+        directives=suppression_directives,
+        declarations=declaration_targets,
+    )
+    return build_suppression_index(suppression_bindings)
+
+
 # =========================
 # Public API
 # =========================
@@ -883,7 +950,6 @@ def extract_units_and_stats_from_source(
     collector = _QualnameCollector()
     collector.visit(tree)
     source_lines = source.splitlines()
-    source_tokens = _source_tokens(source)
     source_line_count = len(source_lines)
 
     is_test_file = is_test_filepath(filepath)
@@ -902,18 +968,12 @@ def extract_units_and_stats_from_source(
     protocol_symbol_aliases = _walk.protocol_symbol_aliases
     protocol_module_aliases = _walk.protocol_module_aliases
 
-    suppression_directives = extract_suppression_directives(source)
-    declaration_targets = _collect_declaration_targets(
+    suppression_index = _build_suppression_index_for_source(
+        source=source,
         filepath=filepath,
         module_name=module_name,
         collector=collector,
-        source_tokens=source_tokens,
-    )
-    suppression_bindings = bind_suppressions_to_declarations(
-        directives=suppression_directives,
-        declarations=declaration_targets,
     )
-    suppression_index = build_suppression_index(suppression_bindings)
     class_names = frozenset(class_node.name for _, class_node in collector.class_nodes)
     module_import_names = set(import_names)
     module_class_names = set(class_names)

codeclone/pipeline.py

@@ -56,16 +56,11 @@
 )
 from .normalize import NormalizationConfig
 from .paths import is_test_filepath
-from .report import (
-    build_block_group_facts,
-    prepare_block_report_groups,
-    prepare_segment_report_groups,
-    render_json_report_document,
-    render_text_report_document,
-    to_markdown_report,
-    to_sarif_report,
-)
+from .report.blocks import prepare_block_report_groups
+from .report.explain import build_block_group_facts
 from .report.json_contract import build_report_document
+from .report.segments import prepare_segment_report_groups
+from .report.serialize import render_json_report_document, render_text_report_document
 from .report.suggestions import generate_suggestions
 from .scanner import iter_py_files, module_name_from_path
 from .structural_findings import build_clone_cohort_structural_findings
@@ -1531,6 +1526,8 @@ def report(
         contents["json"] = render_json_report_document(report_document)
 
     if boot.output_paths.md and report_document is not None:
+        from .report.markdown import to_markdown_report
+
         contents["md"] = to_markdown_report(
             report_document=report_document,
             meta=report_meta,
@@ -1548,6 +1545,8 @@ def report(
         )
 
     if boot.output_paths.sarif and report_document is not None:
+        from .report.sarif import to_sarif_report
+
         contents["sarif"] = to_sarif_report(
             report_document=report_document,
             meta=report_meta,