# This Source Code Form is subject to the terms of the Mozilla Public

# License, v. 2.0. If a copy of the MPL was not distributed with this

# file, You can obtain one at https://mozilla.org/MPL/2.0/.

# SPDX-License-Identifier: MPL-2.0

# Copyright (c) 2026 Den Rozhnovskiy

from __future__ import annotations

import inspect

import os

from collections.abc import Mapping

from concurrent.futures import ProcessPoolExecutor, as_completed

from dataclasses import dataclass

from hashlib import sha256

from pathlib import Path

from typing import TYPE_CHECKING, Literal, cast

import orjson

from ._coerce import as_int, as_str

from .cache import (

ApiParamSpecDict,

Cache,

CacheEntry,

ClassMetricsDict,

DeadCandidateDict,

FileStat,

ModuleDepDict,

PublicSymbolDict,

SegmentReportProjection,

SourceStatsDict,

StructuralFindingGroupDict,

file_stat_signature,

)

from .contracts import ExitCode

from .domain.findings import CATEGORY_COHESION, CATEGORY_COMPLEXITY, CATEGORY_COUPLING

from .domain.quality import CONFIDENCE_HIGH, RISK_HIGH, RISK_LOW

from .extractor import extract_units_and_stats_from_source

from .golden_fixtures import (

build_suppressed_clone_groups,

split_clone_groups_for_golden_fixtures,

)

from .grouping import build_block_groups, build_groups, build_segment_groups

from .metrics import (

CoverageJoinParseError,

HealthInputs,

build_coverage_join,

build_dep_graph,

build_overloaded_modules_payload,

compute_health,

find_suppressed_unused,

find_unused,

)

from .models import (

ApiBreakingChange,

ApiParamSpec,

ApiSurfaceSnapshot,

BlockUnit,

ClassMetrics,

CoverageJoinResult,

DeadCandidate,

DeadItem,

DepGraph,

FileMetrics,

GroupItem,

GroupItemLike,

GroupMap,

MetricsDiff,

ModuleApiSurface,

ModuleDep,

ModuleDocstringCoverage,

ModuleTypingCoverage,

ProjectMetrics,

PublicSymbol,

SegmentUnit,

StructuralFindingGroup,

StructuralFindingOccurrence,

Suggestion,

SuppressedCloneGroup,

Unit,

)

from .normalize import NormalizationConfig

from .paths import is_test_filepath

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

from .suppressions import DEAD_CODE_RULE_ID, INLINE_CODECLONE_SUPPRESSION_SOURCE

if TYPE_CHECKING:

from argparse import Namespace

from collections.abc import Callable, Collection, Mapping, Sequence

MAX_FILE_SIZE = 10 * 1024 * 1024

DEFAULT_BATCH_SIZE = 100

PARALLEL_MIN_FILES_PER_WORKER = 8

PARALLEL_MIN_FILES_FLOOR = 16

DEFAULT_RUNTIME_PROCESSES = 4

_as_int = as_int

_as_str = as_str

@dataclass(frozen=True, slots=True)

class OutputPaths:

html: Path | None = None

json: Path | None = None

text: Path | None = None

md: Path | None = None

sarif: Path | None = None

@dataclass(frozen=True, slots=True)

class BootstrapResult:

root: Path

config: NormalizationConfig

args: Namespace

output_paths: OutputPaths

cache_path: Path

@dataclass(frozen=True, slots=True)

class DiscoveryResult:

files_found: int

cache_hits: int

files_skipped: int

all_file_paths: tuple[str, ...]

cached_units: tuple[GroupItem, ...]

cached_blocks: tuple[GroupItem, ...]

cached_segments: tuple[GroupItem, ...]

cached_class_metrics: tuple[ClassMetrics, ...]

cached_module_deps: tuple[ModuleDep, ...]

cached_dead_candidates: tuple[DeadCandidate, ...]

cached_referenced_names: frozenset[str]

files_to_process: tuple[str, ...]

skipped_warnings: tuple[str, ...]

cached_referenced_qualnames: frozenset[str] = frozenset()

cached_typing_modules: tuple[ModuleTypingCoverage, ...] = ()

cached_docstring_modules: tuple[ModuleDocstringCoverage, ...] = ()

cached_api_modules: tuple[ModuleApiSurface, ...] = ()

cached_structural_findings: tuple[StructuralFindingGroup, ...] = ()

cached_segment_report_projection: SegmentReportProjection | None = None

cached_lines: int = 0

cached_functions: int = 0

cached_methods: int = 0

cached_classes: int = 0

cached_source_stats_by_file: tuple[tuple[str, int, int, int, int], ...] = ()

@dataclass(frozen=True, slots=True)

class FileProcessResult:

filepath: str

success: bool

error: str | None = None

units: list[Unit] | None = None

blocks: list[BlockUnit] | None = None

segments: list[SegmentUnit] | None = None

lines: int = 0

functions: int = 0

methods: int = 0

classes: int = 0

stat: FileStat | None = None

error_kind: str | None = None

file_metrics: FileMetrics | None = None

structural_findings: list[StructuralFindingGroup] | None = None

@dataclass(frozen=True, slots=True)

class ProcessingResult:

units: tuple[GroupItem, ...]

blocks: tuple[GroupItem, ...]

segments: tuple[GroupItem, ...]

class_metrics: tuple[ClassMetrics, ...]

module_deps: tuple[ModuleDep, ...]

dead_candidates: tuple[DeadCandidate, ...]

referenced_names: frozenset[str]

files_analyzed: int

files_skipped: int

analyzed_lines: int

analyzed_functions: int

analyzed_methods: int

analyzed_classes: int

failed_files: tuple[str, ...]

source_read_failures: tuple[str, ...]

referenced_qualnames: frozenset[str] = frozenset()

typing_modules: tuple[ModuleTypingCoverage, ...] = ()

docstring_modules: tuple[ModuleDocstringCoverage, ...] = ()

api_modules: tuple[ModuleApiSurface, ...] = ()

structural_findings: tuple[StructuralFindingGroup, ...] = ()

source_stats_by_file: tuple[tuple[str, int, int, int, int], ...] = ()

@dataclass(frozen=True, slots=True)

class AnalysisResult:

func_groups: GroupMap

block_groups: GroupMap

block_groups_report: GroupMap

segment_groups: GroupMap

suppressed_segment_groups: int

block_group_facts: dict[str, dict[str, str]]

func_clones_count: int

block_clones_count: int

segment_clones_count: int

files_analyzed_or_cached: int

project_metrics: ProjectMetrics | None

metrics_payload: dict[str, object] | None

suggestions: tuple[Suggestion, ...]

segment_groups_raw_digest: str

suppressed_clone_groups: tuple[SuppressedCloneGroup, ...] = ()

coverage_join: CoverageJoinResult | None = None

suppressed_dead_code_items: int = 0

structural_findings: tuple[StructuralFindingGroup, ...] = ()

@dataclass(frozen=True, slots=True)

class GatingResult:

exit_code: int

reasons: tuple[str, ...]

@dataclass(frozen=True, slots=True)

class ReportArtifacts:

html: str | None = None

json: str | None = None

text: str | None = None

md: str | None = None

sarif: str | None = None

report_document: dict[str, object] | None = None

@dataclass(frozen=True, slots=True)

class MetricGateConfig:

fail_complexity: int

fail_coupling: int

fail_cohesion: int

fail_cycles: bool

fail_dead_code: bool

fail_health: int

fail_on_new_metrics: bool

fail_on_typing_regression: bool = False

fail_on_docstring_regression: bool = False

fail_on_api_break: bool = False

fail_on_untested_hotspots: bool = False

min_typing_coverage: int = -1

min_docstring_coverage: int = -1

coverage_min: int = 50

def _as_sorted_str_tuple(value: object) -> tuple[str, ...]:

if not isinstance(value, list):

return ()

return tuple(sorted({item for item in value if isinstance(item, str) and item}))

def _group_item_sort_key(item: GroupItemLike) -> tuple[str, int, int, str]:

return (

_as_str(item.get("filepath")),

_as_int(item.get("start_line")),

_as_int(item.get("end_line")),

_as_str(item.get("qualname")),

)

def _segment_projection_item_sort_key(item: GroupItemLike) -> tuple[str, str, int, int]:

return (

_as_str(item.get("filepath")),

_as_str(item.get("qualname")),

_as_int(item.get("start_line")),

_as_int(item.get("end_line")),

)

def _segment_groups_digest(segment_groups: GroupMap) -> str:

normalized_rows: list[

tuple[str, tuple[tuple[str, str, int, int, int, str, str], ...]]

] = []

for group_key in sorted(segment_groups):

items = sorted(segment_groups[group_key], key=_segment_projection_item_sort_key)

normalized_items: list[tuple[str, str, int, int, int, str, str]] = [

(

_as_str(item.get("filepath")),

_as_str(item.get("qualname")),

_as_int(item.get("start_line")),

_as_int(item.get("end_line")),

_as_int(item.get("size")),

_as_str(item.get("segment_hash")),

_as_str(item.get("segment_sig")),

)

for item in items

]

normalized_rows.append((group_key, tuple(normalized_items)))

payload = orjson.dumps(tuple(normalized_rows), option=orjson.OPT_SORT_KEYS)

return sha256(payload).hexdigest()

def _coerce_segment_report_projection(

value: object,

) -> SegmentReportProjection | None:

if not isinstance(value, dict):

return None

digest = value.get("digest")

suppressed = value.get("suppressed")

groups = value.get("groups")

if (

not isinstance(digest, str)

or not isinstance(suppressed, int)

or not isinstance(groups, dict)

):

return None

if not all(

isinstance(group_key, str) and isinstance(items, list)

for group_key, items in groups.items()

):

return None

return cast("SegmentReportProjection", value)

def _module_dep_sort_key(dep: ModuleDep) -> tuple[str, str, str, int]:

return dep.source, dep.target, dep.import_type, dep.line

def _class_metric_sort_key(metric: ClassMetrics) -> tuple[str, int, int, str]:

return metric.filepath, metric.start_line, metric.end_line, metric.qualname

def _dead_candidate_sort_key(item: DeadCandidate) -> tuple[str, int, int, str]:

return item.filepath, item.start_line, item.end_line, item.qualname

def _unit_to_group_item(unit: Unit) -> GroupItem:

return {

"qualname": unit.qualname,

"filepath": unit.filepath,

"start_line": unit.start_line,

"end_line": unit.end_line,

"loc": unit.loc,

"stmt_count": unit.stmt_count,

"fingerprint": unit.fingerprint,

"loc_bucket": unit.loc_bucket,

"cyclomatic_complexity": unit.cyclomatic_complexity,

"nesting_depth": unit.nesting_depth,

"risk": unit.risk,

"raw_hash": unit.raw_hash,

"entry_guard_count": unit.entry_guard_count,

"entry_guard_terminal_profile": unit.entry_guard_terminal_profile,

"entry_guard_has_side_effect_before": unit.entry_guard_has_side_effect_before,

"terminal_kind": unit.terminal_kind,

"try_finally_profile": unit.try_finally_profile,

"side_effect_order_profile": unit.side_effect_order_profile,

}

def _block_to_group_item(block: BlockUnit) -> GroupItem:

return {

"block_hash": block.block_hash,

"filepath": block.filepath,

"qualname": block.qualname,

"start_line": block.start_line,

"end_line": block.end_line,

"size": block.size,

}

def _segment_to_group_item(segment: SegmentUnit) -> GroupItem:

return {

"segment_hash": segment.segment_hash,

"segment_sig": segment.segment_sig,

"filepath": segment.filepath,

"qualname": segment.qualname,

"start_line": segment.start_line,

"end_line": segment.end_line,

"size": segment.size,

}

def _parallel_min_files(processes: int) -> int:

return max(PARALLEL_MIN_FILES_FLOOR, processes * PARALLEL_MIN_FILES_PER_WORKER)

def _resolve_process_count(processes: object) -> int:

if processes is None:

return DEFAULT_RUNTIME_PROCESSES

return max(1, _as_int(processes, DEFAULT_RUNTIME_PROCESSES))

def _should_collect_structural_findings(output_paths: OutputPaths) -> bool:

return any(

path is not None

for path in (

output_paths.html,

output_paths.json,

output_paths.md,

output_paths.sarif,

output_paths.text,

)

)

def _should_use_parallel(files_count: int, processes: int) -> bool:

if processes <= 1:

return False

return files_count >= _parallel_min_files(processes)

def _new_discovery_buffers() -> tuple[

list[GroupItem],

list[GroupItem],

list[GroupItem],

list[ClassMetrics],

list[ModuleDep],

list[DeadCandidate],

set[str],

set[str],

list[ModuleTypingCoverage],

list[ModuleDocstringCoverage],

list[ModuleApiSurface],

list[str],

list[str],

]:

return [], [], [], [], [], [], set(), set(), [], [], [], [], []

def _decode_cached_structural_finding_group(

group_dict: StructuralFindingGroupDict,

filepath: str,

) -> StructuralFindingGroup:

"""Convert a StructuralFindingGroupDict (from cache) to a StructuralFindingGroup."""

finding_kind = group_dict["finding_kind"]

finding_key = group_dict["finding_key"]

signature = group_dict["signature"]

items = tuple(

StructuralFindingOccurrence(

finding_kind=finding_kind,

finding_key=finding_key,

file_path=filepath,

qualname=item["qualname"],

start=item["start"],

end=item["end"],

signature=signature,

)

for item in group_dict["items"]

)

return StructuralFindingGroup(

finding_kind=finding_kind,

finding_key=finding_key,

signature=signature,

items=items,

)

def bootstrap(

*,

args: Namespace,

root: Path,

output_paths: OutputPaths,

cache_path: Path,

) -> BootstrapResult:

return BootstrapResult(

root=root,

config=NormalizationConfig(),

args=args,

output_paths=output_paths,

cache_path=cache_path,

)

def _resolve_optional_runtime_path(value: object, *, root: Path) -> Path | None:

text = str(value).strip() if value is not None else ""

if not text:

return None

candidate = Path(text).expanduser()

resolved = candidate if candidate.is_absolute() else root / candidate

try:

return resolved.resolve()

except OSError:

return resolved.absolute()

def _cache_entry_has_metrics(entry: CacheEntry) -> bool:

metric_keys = (

"class_metrics",

"module_deps",

"dead_candidates",

"referenced_names",

"referenced_qualnames",

"import_names",

"class_names",

)

return all(key in entry and isinstance(entry.get(key), list) for key in metric_keys)

def _cache_entry_has_structural_findings(entry: CacheEntry) -> bool:

return "structural_findings" in entry

def _cache_entry_source_stats(entry: CacheEntry) -> tuple[int, int, int, int] | None:

stats_obj = entry.get("source_stats")

if not isinstance(stats_obj, dict):

return None

lines = stats_obj.get("lines")

functions = stats_obj.get("functions")

methods = stats_obj.get("methods")

classes = stats_obj.get("classes")

if not (

isinstance(lines, int)

and isinstance(functions, int)

and isinstance(methods, int)

and isinstance(classes, int)

and lines >= 0

and functions >= 0

and methods >= 0

and classes >= 0

):

return None

return lines, functions, methods, classes

def _usable_cached_source_stats(

entry: CacheEntry,

*,

skip_metrics: bool,

collect_structural_findings: bool,

) -> tuple[int, int, int, int] | None:

if not skip_metrics and not _cache_entry_has_metrics(entry):

return None

if collect_structural_findings and not _cache_entry_has_structural_findings(entry):

return None

return _cache_entry_source_stats(entry)

def _cache_dict_module_fields(

value: object,

) -> tuple[Mapping[str, object], str, str] | None:

if not isinstance(value, dict):

return None

row = cast("Mapping[str, object]", value)

module = row.get("module")

filepath = row.get("filepath")

if not isinstance(module, str) or not isinstance(filepath, str):

return None

return row, module, filepath

def _cache_dict_int_fields(

row: Mapping[str, object],

*keys: str,

) -> tuple[int, ...] | None:

values: list[int] = []

for key in keys:

value = row.get(key)

if not isinstance(value, int):

return None

values.append(value)

return tuple(values)

def _typing_coverage_from_cache_dict(

value: object,

) -> ModuleTypingCoverage | None:

row_info = _cache_dict_module_fields(value)

if row_info is None:

return None

row, module, filepath = row_info

int_fields = _cache_dict_int_fields(

row,

"callable_count",

"params_total",

"params_annotated",

"returns_total",

"returns_annotated",

"any_annotation_count",

)

if int_fields is None:

return None

(

callable_count,

params_total,

params_annotated,

returns_total,

returns_annotated,

any_annotation_count,

) = int_fields

return ModuleTypingCoverage(

module=module,

filepath=filepath,

callable_count=callable_count,

params_total=params_total,

params_annotated=params_annotated,

returns_total=returns_total,

returns_annotated=returns_annotated,

any_annotation_count=any_annotation_count,

)

def _docstring_coverage_from_cache_dict(

value: object,

) -> ModuleDocstringCoverage | None:

row_info = _cache_dict_module_fields(value)

if row_info is None:

return None

row, module, filepath = row_info

totals = _cache_dict_int_fields(

row,

"public_symbol_total",

"public_symbol_documented",

)

if totals is None:

return None

public_symbol_total, public_symbol_documented = totals

return ModuleDocstringCoverage(

module=module,

filepath=filepath,

public_symbol_total=public_symbol_total,

public_symbol_documented=public_symbol_documented,

)

def _api_param_spec_from_cache_dict(value: ApiParamSpecDict) -> ApiParamSpec | None:

name = value.get("name")

kind = value.get("kind")

has_default = value.get("has_default")

annotation_hash = value.get("annotation_hash", "")

if (

not isinstance(name, str)

or not isinstance(kind, str)

or not isinstance(has_default, bool)

or not isinstance(annotation_hash, str)

):

return None

return ApiParamSpec(

name=name,

kind=cast(

"Literal['pos_only', 'pos_or_kw', 'vararg', 'kw_only', 'kwarg']",

kind,

),

has_default=has_default,

annotation_hash=annotation_hash,

)

def _public_symbol_from_cache_dict(

value: PublicSymbolDict,

) -> PublicSymbol | None:

qualname = value.get("qualname")

kind = value.get("kind")

start_line = value.get("start_line")

end_line = value.get("end_line")

exported_via = value.get("exported_via", "name")

returns_hash = value.get("returns_hash", "")

params_raw = value.get("params", [])

if (

not isinstance(qualname, str)

or not isinstance(kind, str)

or not isinstance(start_line, int)

or not isinstance(end_line, int)

or not isinstance(exported_via, str)

or not isinstance(returns_hash, str)

or not isinstance(params_raw, list)

):

return None

params = []

for param in params_raw:

if not isinstance(param, dict):

return None

parsed = _api_param_spec_from_cache_dict(param)

if parsed is None:

return None

params.append(parsed)

return PublicSymbol(

qualname=qualname,

kind=cast("Literal['function', 'class', 'method', 'constant']", kind),

start_line=start_line,

end_line=end_line,

params=tuple(params),

returns_hash=returns_hash,

exported_via=cast("Literal['all', 'name']", exported_via),

)

def _api_surface_from_cache_dict(value: object) -> ModuleApiSurface | None:

row_info = _cache_dict_module_fields(value)

if row_info is None:

return None

row, module, filepath = row_info

all_declared_raw = row.get("all_declared", [])

symbols_raw = row.get("symbols", [])

if (

not isinstance(all_declared_raw, list)

or not isinstance(symbols_raw, list)

or not all(isinstance(item, str) for item in all_declared_raw)

):

return None

symbols: list[PublicSymbol] = []

for item in symbols_raw:

if not isinstance(item, dict):

return None

parsed = _public_symbol_from_cache_dict(cast("PublicSymbolDict", item))

if parsed is None:

return None

symbols.append(parsed)

return ModuleApiSurface(

module=module,

filepath=filepath,

all_declared=tuple(sorted(set(all_declared_raw))) or None,

symbols=tuple(sorted(symbols, key=lambda item: item.qualname)),

)

def _load_cached_metrics_extended(

entry: CacheEntry,

*,

filepath: str,

) -> tuple[

tuple[ClassMetrics, ...],

tuple[ModuleDep, ...],

tuple[DeadCandidate, ...],

frozenset[str],

frozenset[str],

ModuleTypingCoverage | None,

ModuleDocstringCoverage | None,

ModuleApiSurface | None,

]:

class_metrics_rows: list[ClassMetricsDict] = entry.get("class_metrics", [])

class_metrics = tuple(

ClassMetrics(

qualname=row["qualname"],

filepath=row["filepath"],

start_line=row["start_line"],

end_line=row["end_line"],

cbo=row["cbo"],

lcom4=row["lcom4"],

method_count=row["method_count"],

instance_var_count=row["instance_var_count"],

risk_coupling=cast(

"Literal['low', 'medium', 'high']",

row["risk_coupling"],

),

risk_cohesion=cast(

"Literal['low', 'medium', 'high']",

row["risk_cohesion"],

),

coupled_classes=_as_sorted_str_tuple(row.get("coupled_classes", [])),

)

for row in class_metrics_rows

if row.get("qualname") and row.get("filepath")

)

module_dep_rows: list[ModuleDepDict] = entry.get("module_deps", [])

module_deps = tuple(

ModuleDep(

source=row["source"],

target=row["target"],

import_type=cast("Literal['import', 'from_import']", row["import_type"]),

line=row["line"],

)

for row in module_dep_rows

if row.get("source") and row.get("target")

)

dead_rows: list[DeadCandidateDict] = entry.get("dead_candidates", [])

dead_candidates = tuple(

DeadCandidate(

qualname=row["qualname"],

local_name=row["local_name"],

filepath=row["filepath"],

start_line=row["start_line"],

end_line=row["end_line"],

kind=cast(

"Literal['function', 'class', 'method', 'import']",

row["kind"],

),

suppressed_rules=tuple(sorted(set(row.get("suppressed_rules", [])))),

)

for row in dead_rows

if row.get("qualname") and row.get("local_name") and row.get("filepath")

)

referenced_names = (

frozenset()

if is_test_filepath(filepath)

else frozenset(entry.get("referenced_names", []))

)

referenced_qualnames = (

frozenset()

if is_test_filepath(filepath)

else frozenset(entry.get("referenced_qualnames", []))

)

typing_coverage = _typing_coverage_from_cache_dict(entry.get("typing_coverage"))

docstring_coverage = _docstring_coverage_from_cache_dict(

entry.get("docstring_coverage")

)

api_surface = _api_surface_from_cache_dict(entry.get("api_surface"))

return (

class_metrics,

module_deps,

dead_candidates,

referenced_names,

referenced_qualnames,

typing_coverage,

docstring_coverage,

api_surface,

)

def discover(*, boot: BootstrapResult, cache: Cache) -> DiscoveryResult:

files_found = 0

cache_hits = 0

files_skipped = 0

collect_structural_findings = _should_collect_structural_findings(boot.output_paths)

cached_segment_projection = _coerce_segment_report_projection(

getattr(cache, "segment_report_projection", None)

)

(

cached_units,

cached_blocks,

cached_segments,

cached_class_metrics,

cached_module_deps,

cached_dead_candidates,

cached_referenced_names,

cached_referenced_qualnames,

cached_typing_modules,

cached_docstring_modules,

cached_api_modules,

files_to_process,

skipped_warnings,

) = _new_discovery_buffers()

cached_sf: list[StructuralFindingGroup] = []

cached_source_stats_by_file: list[tuple[str, int, int, int, int]] = []

cached_lines = 0

cached_functions = 0

cached_methods = 0

cached_classes = 0

all_file_paths: list[str] = []

for filepath in iter_py_files(str(boot.root)):

files_found += 1

all_file_paths.append(filepath)

try:

stat = file_stat_signature(filepath)

except OSError as exc:

files_skipped += 1

skipped_warnings.append(f"{filepath}: {exc}")

continue

cached = cache.get_file_entry(filepath)

if cached and cached.get("stat") == stat:

cached_source_stats = _usable_cached_source_stats(

cached,

skip_metrics=boot.args.skip_metrics,

collect_structural_findings=collect_structural_findings,

)

if cached_source_stats is None:

files_to_process.append(filepath)

continue

cache_hits += 1

lines, functions, methods, classes = cached_source_stats

cached_lines += lines

cached_functions += functions

cached_methods += methods

cached_classes += classes

cached_source_stats_by_file.append(

(filepath, lines, functions, methods, classes)

)

cached_units.extend(cast("list[GroupItem]", cast(object, cached["units"])))

cached_blocks.extend(

cast("list[GroupItem]", cast(object, cached["blocks"]))

)

cached_segments.extend(

cast("list[GroupItem]", cast(object, cached["segments"]))

)

if not boot.args.skip_metrics:

(

class_metrics,

module_deps,

dead_candidates,

referenced_names,

referenced_qualnames,

typing_coverage,

docstring_coverage,

api_surface,

) = _load_cached_metrics_extended(cached, filepath=filepath)

cached_class_metrics.extend(class_metrics)

cached_module_deps.extend(module_deps)

cached_dead_candidates.extend(dead_candidates)

cached_referenced_names.update(referenced_names)

cached_referenced_qualnames.update(referenced_qualnames)

if typing_coverage is not None:

cached_typing_modules.append(typing_coverage)

if docstring_coverage is not None:

cached_docstring_modules.append(docstring_coverage)

if api_surface is not None:

cached_api_modules.append(api_surface)

if collect_structural_findings:

cached_sf.extend(

_decode_cached_structural_finding_group(group_dict, filepath)

for group_dict in cached.get("structural_findings") or []

)

continue

files_to_process.append(filepath)

return DiscoveryResult(

files_found=files_found,

cache_hits=cache_hits,

files_skipped=files_skipped,

all_file_paths=tuple(all_file_paths),

cached_units=tuple(sorted(cached_units, key=_group_item_sort_key)),

cached_blocks=tuple(sorted(cached_blocks, key=_group_item_sort_key)),

cached_segments=tuple(sorted(cached_segments, key=_group_item_sort_key)),

cached_class_metrics=tuple(

sorted(cached_class_metrics, key=_class_metric_sort_key)

),

cached_module_deps=tuple(sorted(cached_module_deps, key=_module_dep_sort_key)),

cached_dead_candidates=tuple(

sorted(cached_dead_candidates, key=_dead_candidate_sort_key)

),

cached_referenced_names=frozenset(cached_referenced_names),

cached_referenced_qualnames=frozenset(cached_referenced_qualnames),

cached_typing_modules=tuple(

sorted(cached_typing_modules, key=lambda item: (item.filepath, item.module))

),

cached_docstring_modules=tuple(

sorted(

cached_docstring_modules,

key=lambda item: (item.filepath, item.module),

)

),

cached_api_modules=tuple(

sorted(cached_api_modules, key=lambda item: (item.filepath, item.module))

),

files_to_process=tuple(files_to_process),

skipped_warnings=tuple(sorted(skipped_warnings)),

cached_structural_findings=tuple(cached_sf),

cached_segment_report_projection=cached_segment_projection,

cached_lines=cached_lines,

cached_functions=cached_functions,

cached_methods=cached_methods,

cached_classes=cached_classes,

cached_source_stats_by_file=tuple(

sorted(cached_source_stats_by_file, key=lambda row: row[0])

),

)

def process_file(

filepath: str,

root: str,

cfg: NormalizationConfig,

min_loc: int,

min_stmt: int,

collect_structural_findings: bool = True,

collect_api_surface: bool = False,

api_include_private_modules: bool = False,

block_min_loc: int = 20,

block_min_stmt: int = 8,

segment_min_loc: int = 20,

segment_min_stmt: int = 10,

) -> FileProcessResult:

try:

try:

stat_result = os.stat(filepath)

if stat_result.st_size > MAX_FILE_SIZE:

return FileProcessResult(

filepath=filepath,

success=False,

error=(

f"File too large: {stat_result.st_size} bytes "

f"(max {MAX_FILE_SIZE})"

),

error_kind="file_too_large",

)

except OSError as exc:

return FileProcessResult(

filepath=filepath,

success=False,

error=f"Cannot stat file: {exc}",

error_kind="stat_error",

)

stat: FileStat = {

"mtime_ns": stat_result.st_mtime_ns,

"size": stat_result.st_size,

}

try:

source = Path(filepath).read_text("utf-8")

except UnicodeDecodeError as exc:

return FileProcessResult(

filepath=filepath,

success=False,

error=f"Encoding error: {exc}",

error_kind="source_read_error",