service.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/complexity/service.py
 
"""
Purpose
-------
Provide shared analysis, ordering, and rendering for command adapters.
 
Responsibilities
----------------
- Discover Python files while excluding environment and build trees.
- Record every parse failure instead of silently omitting files.
- Produce stable worst-first CC or CRAP ordering.
- Render full JSON data and compact ASCII human output.
- Expose the shared relative-path display that gate output and baseline
  identities are built from.
 
Diagnostics
-----------
Domain: OCT-COMPLEXITY
Levels:
    L2 - analysis lifecycle
    L3 - file and result details
    L4 - deep tracing
 
Contracts
---------
- JSON results are never truncated by ``--top``.
- CC-only rows explicitly identify coverage as not measured.
- The service measures and never gates.
 
Dependencies
------------
- oct.complexity.coverage (optional coverage mapping)
- oct.complexity.engine (complexity records)
- oct.complexity.metrics (CRAP formula)
"""
 
# OCT-LINT: disable-file=dbg_usage reason="The measurement service emits only its explicit report and must not add diagnostics noise to JSON or ASCII output"
# OCT-LINT: disable-file=dbg_import reason="Diagnostics would contaminate explicit JSON and ASCII report output"
 
from __future__ import annotations
 
import importlib.util
import os
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable, Sequence
 
from oct.complexity.coverage import CoveredFunction, map_coverage
from oct.complexity.engine import FunctionComplexity, analyze_file
from oct.complexity.metrics import calculate_crap
 
 
_EXCLUDED_DIRECTORY_NAMES = frozenset(
    {
        ".git",
        ".hg",
        ".mypy_cache",
        ".nox",
        ".pytest_cache",
        ".ruff_cache",
        ".svn",
        ".tox",
        ".venv",
        "__pycache__",
        "build",
        "dist",
        "node_modules",
        "tests",
        "venv",
    }
)
 
 
@dataclass(frozen=True, slots=True)
class ParseFailure:
    """One source file that could not be parsed or read."""
 
    file: str
    error: str
    lineno: int | None
    offset: int | None
 
    def to_dict(self) -> dict[str, str | int | None]:
        """Return a JSON-ready failure record."""
        return asdict(self)
 
 
@dataclass(frozen=True, slots=True)
class FunctionResult:
    """One complete result row, with nullable P2 fields in CC-only mode."""
 
    file: str
    qualname: str
    lineno: int
    end_lineno: int
    complexity: int
    coverage: float | None
    coverage_status: str
    covered_statements: int | None
    statement_count: int | None
    crap: float | None
 
    def to_dict(self) -> dict[str, str | int | float | None]:
        """Return every stable field for JSON output."""
        return asdict(self)
 
 
@dataclass(frozen=True, slots=True)
class AnalysisReport:
    """All measurements and all parse failures for one command invocation."""
 
    functions: tuple[FunctionResult, ...]
    parse_errors: tuple[ParseFailure, ...]
    coverage_message: str | None
 
    def to_dict(self) -> dict[str, object]:
        """Return full, untruncated machine-readable output."""
        return {
            "functions": [item.to_dict() for item in self.functions],
            "parse_errors": [item.to_dict() for item in self.parse_errors],
            "coverage_message": self.coverage_message,
        }
 
 
def analyze_paths(
    paths: Sequence[str | Path],
    coverage_data: str | Path | None = None,
) -> AnalysisReport:
    """Analyze Python paths and optionally enrich every record with coverage."""
    records: list[FunctionComplexity] = []
    parse_errors: list[ParseFailure] = []
    for filename in discover_python_files(paths):
        try:
            records.extend(analyze_file(filename))
        except SyntaxError as exc:
            parse_errors.append(
                ParseFailure(
                    file=str(filename),
                    error=str(exc),
                    lineno=exc.lineno,
                    offset=exc.offset,
                )
            )
        except (OSError, UnicodeError) as exc:
            parse_errors.append(
                ParseFailure(
                    file=str(filename),
                    error=str(exc),
                    lineno=None,
                    offset=None,
                )
            )
    if coverage_data is None:
        results = [_without_coverage(record) for record in records]
        coverage_message = _coverage_absence_message()
        results.sort(key=_complexity_sort_key)
    else:
        results = [
            _with_coverage(item)
            for item in map_coverage(records, coverage_data)
        ]
        coverage_message = None
        results.sort(key=_crap_sort_key)
    parse_errors.sort(key=lambda item: (item.file.casefold(), item.lineno or 0))
    return AnalysisReport(
        functions=tuple(results),
        parse_errors=tuple(parse_errors),
        coverage_message=coverage_message,
    )
 
 
def discover_python_files(paths: Sequence[str | Path]) -> list[Path]:
    """Return unique Python files under paths with build/cache trees excluded."""
    discovered: dict[str, Path] = {}
    for raw_path in paths:
        path = Path(raw_path).resolve()
        if not path.exists():
            raise FileNotFoundError(f"path not found: {path}")
        candidates: Iterable[Path]
        if path.is_file():
            candidates = (path,) if path.suffix.casefold() == ".py" else ()
        else:
            candidates = path.rglob("*.py")
        for candidate in candidates:
            resolved = candidate.resolve()
            if is_excluded_path(resolved, path if path.is_dir() else path.parent):
                continue
            discovered[_normalized_path(resolved)] = resolved
    return sorted(discovered.values(), key=lambda item: str(item).casefold())
 
 
def is_excluded_path(path: Path, scan_root: Path) -> bool:
    """Return whether a discovered path crosses an excluded directory."""
    try:
        relative_parts = path.relative_to(scan_root).parts[:-1]
    except ValueError:
        relative_parts = path.parts[:-1]
    return any(part in _EXCLUDED_DIRECTORY_NAMES for part in relative_parts)
 
 
def _without_coverage(record: FunctionComplexity) -> FunctionResult:
    """Convert an engine record into an explicit CC-only result."""
    return FunctionResult(
        file=record.file,
        qualname=record.qualname,
        lineno=record.lineno,
        end_lineno=record.end_lineno,
        complexity=record.complexity,
        coverage=None,
        coverage_status="not-measured",
        covered_statements=None,
        statement_count=None,
        crap=None,
    )
 
 
def _with_coverage(item: CoveredFunction) -> FunctionResult:
    """Convert a covered record and calculate its CRAP score."""
    return FunctionResult(
        file=item.file,
        qualname=item.qualname,
        lineno=item.lineno,
        end_lineno=item.end_lineno,
        complexity=item.complexity,
        coverage=item.coverage,
        coverage_status=item.coverage_status,
        covered_statements=item.covered_statements,
        statement_count=item.statement_count,
        crap=calculate_crap(item.complexity, item.coverage),
    )
 
 
def _complexity_sort_key(item: FunctionResult) -> tuple[object, ...]:
    """Sort CC-only rows worst-first with deterministic tie breaking."""
    return (
        -item.complexity,
        item.file.casefold(),
        item.lineno,
        item.qualname,
    )
 
 
def _crap_sort_key(item: FunctionResult) -> tuple[object, ...]:
    """Sort covered rows by CRAP worst-first, then complexity."""
    return (
        -(item.crap or 0.0),
        -item.complexity,
        item.file.casefold(),
        item.lineno,
        item.qualname,
    )
 
 
def _coverage_absence_message() -> str:
    """Explain exactly why CRAP columns are absent in CC-only mode."""
    if importlib.util.find_spec("coverage") is None:
        return (
            "Coverage unavailable: coverage.py is not installed; "
            "CRAP columns are absent."
        )
    return (
        "Coverage unavailable: no --coverage-data was provided; "
        "CRAP columns are absent."
    )
 
 
def render_human(report: AnalysisReport, top: int | None = None) -> str:
    """Render a compact worst-first ASCII report for terminal users."""
    shown = report.functions[:top] if top is not None else report.functions
    lines: list[str] = []
    if report.coverage_message:
        lines.append(report.coverage_message)
    if shown and shown[0].coverage is not None:
        lines.append(" CC  Coverage       CRAP  Status          Function  Span")
        for item in shown:
            lines.append(
                f"{item.complexity:>3}  {item.coverage * 100:>7.1f}%  "
                f"{item.crap:>9.2f}  {item.coverage_status:<14}  "
                f"{item.qualname}  {_span(item)}"
            )
    else:
        lines.append(" CC  Function  Span")
        for item in shown:
            lines.append(
                f"{item.complexity:>3}  {item.qualname}  {_span(item)}"
            )
    lines.append(f"Showing {len(shown)} of {len(report.functions)} functions.")
    if report.parse_errors:
        lines.append("Parse errors:")
        for error in report.parse_errors:
            location = f":{error.lineno}" if error.lineno is not None else ""
            lines.append(f"  {error.file}{location}: {error.error}")
    return "\n".join(lines)
 
 
def _span(item: FunctionResult) -> str:
    """Return one source span suitable for terminal output."""
    return f"{item.file}:{item.lineno}-{item.end_lineno}"
 
 
def _normalized_path(path: Path) -> str:
    """Return a deterministic filesystem identity key."""
    return os.path.normcase(str(path))
 
 
def relative_display_path(file: str | Path, root: str | Path) -> str:
    """Return *file* relative to *root* with forward separators.
 
    Gate output and baseline identities have to survive being read on another
    machine, so the separator is normalised and the path is made relative
    whenever it lies under *root*. A file outside *root* (a different drive,
    a temporary fixture) keeps its absolute POSIX spelling rather than
    growing a misleading chain of parent segments.
    """
    resolved = Path(file).resolve()
    try:
        return resolved.relative_to(Path(root).resolve()).as_posix()
    except ValueError:
        return resolved.as_posix()
 
 
def function_identity(file: str | Path, qualname: str, root: str | Path) -> str:
    """Return the stable ``relpath::qualname`` identity of one function.
 
    A renamed or moved function is honestly a new function under this
    identity, which is the point: the ratchet grandfathers recorded debt,
    not the name it happened to carry.
    """
    return f"{relative_display_path(file, root)}::{qualname}"