oct_format.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/formatter/oct_format.py
 
"""
Option C auto-formatter implementation.
 
Purpose
-------
Detect and fix structural Option C compliance violations automatically.
Complements the linter by fixing violations while preserving code semantics.
 
Responsibilities
----------------
- Detect formatting violations using reused linter checks.
- Fix violations: headers, docstrings, imports, function patterns.
- Support report-only (dry-run) and fix modes.
- Archive originals before modification.
- Provide terminal and JSON output.
- Respect project exclusions and configuration.
 
Diagnostics
-----------
Domain: OCT.FORMATTER
Levels:
    L2 — formatter lifecycle and command routing
    L3 — file discovery and processing, fix application decisions
    L4 — detailed fixer implementations, archival operations, output formatting
 
Contracts
---------
- Files are only modified if fix_mode is True.
- Originals are archived before any modification.
- All code below headers is preserved exactly.
- Docstrings are not replaced; missing sections are appended.
- Each fixer is idempotent (running twice is same as once).
 
Dependencies
------------
- oct (package version)
- oct.linter.oct_lint (reused validation checks)
- oct.core.exclusions (directory and wildcard exclusion lists)
- oct.core.project_root (project root detection)
- oct.core.parsing (safe AST parsing helper)
- oct.core.git (git changed-file discovery)
"""
 
import sys
import json
import argparse
import re
from dataclasses import dataclass
from pathlib import Path
from datetime import datetime
import ast
from typing import Tuple, List, Dict, Any, Optional
 
from oct import __version__
from oct.linter.oct_lint import (
    validate_header_block,
    check_docstring,
    check_diagnostics_profile,
    check_dbg_import,
    check_dbg_usage,
    find_func_pattern_violations,
    _extract_module_docstring,
    find_python_files,
    LinterContext,
)
from oct.core.exclusions import LINTER_EXCLUDE_DIRS, WILDCARD_EXCLUDE_DIRS
from oct.core.project_root import find_project_root
from oct.core.parsing import safe_parse
 
 
def _git_changed_files(root: Path) -> list[Path]:
    """Return .py files modified in the git working tree (staged + unstaged).
 
    Thin shim over :func:`oct.core.git.git_changed_files` (Phase 4A).
    Collapses all ``GitError`` to empty list so the formatter never
    crashes due to git unavailability.
    """
    from oct.core import git as core_git
    try:
        return core_git.git_changed_files(root, extensions={".py"})
    except core_git.GitError:
        return []
 
# Template constants reused from generator
SHEBANG = "#!/usr/bin/env python3"
ENCODING = "# -*- coding: utf-8 -*-"
 
IMPORT_BLOCK = "from oc_diagnostics import _dbg\n"
 
DOCSTRING_SKELETON = '''\
"""
Purpose
-------
Describe the purpose of this module.
 
Responsibilities
----------------
- List the responsibilities of this module.
 
Diagnostics
-----------
Domain: <DOMAIN>
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- State the contracts and guarantees of this module.
"""
'''
 
MISSING_DOCSTRING_SECTIONS = {
    "Purpose": """
Purpose
-------
Describe the purpose of this module.""",
    "Responsibilities": """
Responsibilities
----------------
- List the responsibilities of this module.""",
    "Diagnostics": """
Diagnostics
-----------
Domain: <DOMAIN>
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing""",
    "Contracts": """
Contracts
---------
- State the contracts and guarantees of this module.""",
}
 
 
#: OI-424 — default cap on retained backups per file in
#: ``.formatter_archive/``. Projects can override via
#: ``formatter.max_backups_per_file`` in ``.octrc.json``.
_DEFAULT_MAX_BACKUPS_PER_FILE: int = 10
 
#: OI-516 — aggregate size cap per ``.formatter_archive/`` directory
#: (5 MiB). Enforced after per-file count pruning so a single giant
#: source file cannot fill the archive even within its 10-backup budget.
_DEFAULT_MAX_ARCHIVE_BYTES: int = 5 * 1024 * 1024
 
 
@dataclass
class FormatterContext:
    """Immutable per-run context for the formatter."""
    project_root: Path
    project_name: str
    diagnostics_dir: Path
    tests_dir: Path
    docs_dir: Path
    fix_mode: bool
    dry_run_mode: bool
    json_mode: bool
    verbose: bool
    max_backups_per_file: int = _DEFAULT_MAX_BACKUPS_PER_FILE
    #: OI-516 — aggregate cap across all backups in one
    #: ``.formatter_archive/`` directory.
    max_archive_bytes: int = _DEFAULT_MAX_ARCHIVE_BYTES
 
 
# ============================================================================
# Helper Functions
# ============================================================================
 
 
def _get_relative_path(path: Path, root: Path) -> str:
    """Get relative path from root to path."""
    try:
        return str(path.relative_to(root))
    except ValueError:
        return str(path)
 
 
def _prune_archive(
    archive_dir: Path, original_name: str, max_backups: int,
) -> int:
    """Keep only the ``max_backups`` most recent backups of ``original_name``.
 
    Matches files shaped like ``{original_name}.*.bak`` and sorts them
    by mtime (newest first). Older entries beyond the cap are removed.
    Best-effort: any OS error is swallowed so a failed prune can never
    block ``oct format`` (OI-424).
 
    Returns the number of files removed. ``max_backups <= 0`` disables
    pruning entirely (useful when a project deliberately wants unbounded
    history).
    """
    func = "_prune_archive"
    if max_backups <= 0:
        return 0
    try:
        candidates = sorted(
            archive_dir.glob(f"{original_name}.*.bak"),
            key=lambda p: p.stat().st_mtime,
            reverse=True,
        )
    except OSError:
        return 0
    removed = 0
    for stale in candidates[max_backups:]:
        try:
            stale.unlink()
            removed += 1
        except OSError:
            continue
    return removed
 
 
def _prune_archive_by_size(
    archive_dir: Path, max_total_bytes: int,
) -> int:
    """OI-516: evict oldest ``.bak`` files until total size fits under ``max_total_bytes``.
 
    Runs *after* per-file count pruning so the archive floor is always
    correct under the two coexisting policies. Any OS error on size or
    unlink is swallowed — a rotation failure must never block formatting.
    Returns the number of files removed.
    """
    func = "_prune_archive_by_size"
    if max_total_bytes <= 0:
        return 0
    try:
        entries = [(p, p.stat().st_mtime, p.stat().st_size)
                   for p in archive_dir.glob("*.bak") if p.is_file()]
    except OSError:
        return 0
    total = sum(size for _, _, size in entries)
    if total <= max_total_bytes:
        return 0
    # Oldest first — evict until we fit.
    entries.sort(key=lambda row: row[1])
    removed = 0
    for path, _, size in entries:
        if total <= max_total_bytes:
            break
        try:
            path.unlink()
            total -= size
            removed += 1
        except OSError:
            continue
    return removed
 
 
def _archive_file(
    path: Path,
    max_backups: int = _DEFAULT_MAX_BACKUPS_PER_FILE,
    max_archive_bytes: int = _DEFAULT_MAX_ARCHIVE_BYTES,
) -> None:
    """Archive original file to ``.formatter_archive/`` with timestamp.
 
    After writing the new backup, prune in two passes:
 
    1. Per-file count cap (OI-424): keep only the N newest backups per
       source file name.
    2. Per-directory size cap (OI-516): enforce aggregate ``max_archive_bytes``
       across *all* backups in the archive, evicting oldest first.
    """
    func = "archive_file"
    archive_dir = path.parent / ".formatter_archive"
    archive_dir.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_path = archive_dir / f"{path.name}.{timestamp}.bak"
    backup_path.write_bytes(path.read_bytes())
    _prune_archive(archive_dir, path.name, max_backups)
    _prune_archive_by_size(archive_dir, max_archive_bytes)
 
 
def _detect_indent_style(lines: List[str]) -> str:
    """Return the dominant body indentation unit used in the file.
 
    Returns ``"\\t"`` if the file uses tabs for body indentation more
    often than spaces, otherwise ``"    "`` (four spaces). Only
    considers lines with a non-whitespace character so blank lines do
    not skew the result (OI-425).
    """
    func = "_detect_indent_style"
    tabs = 0
    spaces = 0
    for line in lines:
        if not line or not line.strip():
            continue
        if line.startswith("\t"):
            tabs += 1
        elif line.startswith("    "):
            spaces += 1
    return "\t" if tabs > spaces else "    "
 
 
def _get_indent_at_line(
    lines: List[str], line_num: int, fallback: str = "    ",
) -> str:
    """Return the indentation string of a given line.
 
    If the requested line is out of range or whitespace-only, return
    ``fallback``. The default fallback preserves the historical four-
    space behaviour, so existing callers remain unchanged (OI-425).
    """
    if line_num >= len(lines):
        return fallback
    line = lines[line_num]
    stripped = line.lstrip()
    if stripped:
        return line[:len(line) - len(stripped)]
    return fallback
 
 
def _detect_missing_docstring_sections(text: str) -> List[str]:
    """Detect which docstring sections are missing."""
    func = "_detect_missing_docstring_sections"
    docstring = _extract_module_docstring(text)
    if not docstring:
        return ["Purpose", "Responsibilities", "Diagnostics", "Contracts"]
 
    missing = []
    lines = docstring.split('\n')
    for section in ["Purpose", "Responsibilities", "Diagnostics", "Contracts"]:
        found = any(line.strip().startswith(section) for line in lines)
        if not found:
            missing.append(section)
    return missing
 
 
# ============================================================================
# Fix Functions
# ============================================================================
 
 
def fix_header_block(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure file has correct 4-line header block.
 
    Returns (changed, new_text, messages). Caller is responsible for
    archiving and writing to disk.
    """
    func = "fix_header_block"
 
    # Validate using linter
    header_lines = text.split('\n')[:4] if len(text.split('\n')) >= 4 else text.split('\n')
    ok, errors = validate_header_block(header_lines, path, ctx)
 
    if ok:
        return False, text, []
 
    # Reconstruct correct header
    relative_path = _get_relative_path(path, ctx.project_root)
    new_shebang = f"{SHEBANG}"
    new_encoding = f"{ENCODING}"
    new_identity = f"# {relative_path}"
    new_blank = ""
 
    new_header = new_shebang + "\n" + new_encoding + "\n" + new_identity + "\n" + new_blank + "\n"
 
    # Find where the actual file content starts (after header block).
    # OI-509: delegate to shared oct.core.parsing.header_boundary so the
    # formatter and linter use identical scanning rules.
    from oct.core.parsing import header_boundary
    lines = text.split('\n')
    relative_identity = f"# {relative_path}"
    content_start = header_boundary(
        lines, path, expected_identity=relative_identity,
    )
 
    # Take content after the old header
    if content_start < len(lines):
        remaining_content = '\n'.join(lines[content_start:])
    else:
        remaining_content = ""
 
    new_text = new_header + remaining_content
 
    messages = ["Fixed header block (shebang, encoding, file identity, blank line)"]
    return True, new_text, messages
 
 
def fix_docstring(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure module has complete docstring with all required sections.
 
    If no docstring: insert skeleton.
    If incomplete: add missing sections.
    If complete: no change.
    """
    func = "fix_docstring"
 
    docstring = _extract_module_docstring(text)
 
    if not docstring:
        # Insert complete skeleton after header + blank line
        lines = text.split('\n')
        insert_pos = 4  # After header block (4 lines)
 
        new_text = '\n'.join(lines[:insert_pos]) + '\n' + DOCSTRING_SKELETON + '\n'.join(lines[insert_pos:])
 
        return True, new_text, ["Added complete module docstring"]
 
    # Check for missing sections
    missing = _detect_missing_docstring_sections(text)
    if not missing:
        return False, text, []
 
    # Add missing sections to existing docstring
    sections_to_add = "\n".join(MISSING_DOCSTRING_SECTIONS[section] for section in missing)
 
    # Find docstring closing """ and insert before it
    docstring_start = text.find('"""')
    docstring_end = text.find('"""', docstring_start + 3)
 
    if docstring_end == -1:
        return False, text, []
 
    new_text = (text[:docstring_end] + "\n" + sections_to_add + '\n' + text[docstring_end:])
 
    messages = [f"Added missing docstring sections: {', '.join(missing)}"]
    return True, new_text, messages
 
 
def fix_dbg_import(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure canonical _dbg import exists at module level.
 
    Skips trivial files (< 20 lines).
    """
    func = "fix_dbg_import"
 
    # Skip trivial files
    if len(text.splitlines()) < 20:
        return False, text, []
 
    # Check if canonical import exists
    ok, msg = check_dbg_import(text)
    if ok:
        return False, text, []
 
    # Parse AST to find import insertion point
    tree, _ = safe_parse(text)
    if tree is None:
        return False, text, []
 
    # Find position after header + blank line
    lines = text.split('\n')
 
    # Find any existing _dbg imports and remove them
    new_lines = []
    removed_count = 0
 
    for i, line in enumerate(lines):
        # Skip incorrect _dbg imports (aliased or from wrong module)
        # Only remove if it's clearly a _dbg import line
        if ("from oc_diagnostics import" in line and "_dbg" in line and "import _dbg" not in line) or \
           ("import _dbg" in line and "from oc_diagnostics" not in line):
            removed_count += 1
            continue
 
        new_lines.append(line)
 
    # Insert canonical import after header + blank line
    insert_pos = min(5, len(new_lines))  # After header block (lines 0-4)
    new_lines.insert(insert_pos, IMPORT_BLOCK.rstrip())
    new_text = '\n'.join(new_lines)
 
    msg = "Added canonical _dbg import"
    if removed_count > 0:
        msg += f" (removed {removed_count} incorrect import(s))"
 
    return True, new_text, [msg]
 
 
def fix_func_pattern(path: Path, text: str, ctx: FormatterContext) -> Tuple[bool, str, List[str]]:
    """
    Ensure every function with _dbg() calls has func = "function_name" as first statement.
 
    Delegates violation detection to :func:`oct.linter.oct_lint.find_func_pattern_violations`
    so the linter remains the single source of truth for what constitutes a
    func-pattern violation.  This function only handles the fix (inserting
    the ``func = "..."`` line).
 
    Skips trivial files (< 20 lines).
    """
    func = "fix_func_pattern"
 
    functions_to_fix = find_func_pattern_violations(text)
    if not functions_to_fix:
        return False, text, []
 
    lines = text.split('\n')
 
    # OI-425: detect the file's dominant indentation unit once so the
    # fallback used when insert_idx points at a blank or out-of-range
    # line matches the surrounding file (tab-indented files must not
    # receive space-indented ``func = "..."`` injections).
    file_indent_unit = _detect_indent_style(lines)
 
    # Insert func = "..." for each function (in reverse order to maintain line numbers)
    for func_name, func_lineno in sorted(functions_to_fix, key=lambda x: x[1], reverse=True):
        # func_lineno is 1-indexed, we need 0-indexed
        func_line_idx = func_lineno - 1
 
        # Validate that AST lineno points to actual def line
        if func_line_idx >= len(lines) or "def " not in lines[func_line_idx]:
            continue
 
        # Find the first statement after the function def and docstring
        insert_idx = func_line_idx + 1
        while insert_idx < len(lines):
            line_content = lines[insert_idx].strip()
            # Skip docstrings, comments, and empty lines
            if not line_content or line_content.startswith("#"):
                insert_idx += 1
            elif line_content.startswith('"""') or line_content.startswith("'''"):
                # Skip entire docstring
                quote = '"""' if '"""' in line_content else "'''"
                if line_content.count(quote) == 1:
                    # Docstring continues on next lines
                    insert_idx += 1
                    while insert_idx < len(lines) and quote not in lines[insert_idx]:
                        insert_idx += 1
                insert_idx += 1
            else:
                break
 
        # OI-425: derive a tab-safe fallback from the ``def`` line itself.
        # ``def_indent`` is the leading whitespace on the ``def`` line
        # (empty for module-level functions, one unit deeper for nested
        # defs); adding ``file_indent_unit`` yields the expected body
        # indentation in whatever style the rest of the file uses.
        def_indent = _get_indent_at_line(
            lines, func_line_idx, fallback="",
        )
        body_fallback = def_indent + file_indent_unit
        indent = _get_indent_at_line(
            lines, insert_idx, fallback=body_fallback,
        )
 
        pattern = f'{indent}func = "{func_name}"'
        lines.insert(insert_idx, pattern)
 
    new_text = '\n'.join(lines)
 
    fixed_names = [name for name, _ in functions_to_fix]
    messages = [f"Added func pattern to: {', '.join(fixed_names)}"]
    return True, new_text, messages
 
 
# ============================================================================
# File Processing
# ============================================================================
 
 
def format_file(path: Path, ctx: FormatterContext) -> Dict[str, Any]:
    """Process a single file through all formatters."""
    func = "format_file"
 
    try:
        text = path.read_text(encoding="utf-8")
    except Exception as e:
        return {
            "path": _get_relative_path(path, ctx.project_root),
            "changed": False,
            "fixes": {
                "header": {"changed": False, "messages": []},
                "docstring": {"changed": False, "messages": []},
                "dbg_import": {"changed": False, "messages": []},
                "func_pattern": {"changed": False, "messages": []},
            },
            "errors": [f"Could not read file: {str(e)}"],
        }
 
    result = {
        "path": _get_relative_path(path, ctx.project_root),
        "changed": False,
        "fixes": {},
        "errors": [],
    }
 
    # Apply fixers in sequence
    fixers = [
        ("header", fix_header_block),
        ("docstring", fix_docstring),
        ("dbg_import", fix_dbg_import),
        ("func_pattern", fix_func_pattern),
    ]
 
    any_changed = False
 
    for fix_name, fixer in fixers:
        try:
            changed, text, messages = fixer(path, text, ctx)
            result["fixes"][fix_name] = {"changed": changed, "messages": messages}
            if changed:
                any_changed = True
        except Exception as e:
            result["errors"].append(f"{fix_name}: {str(e)}")
 
    # Single archive + write at end
    if any_changed and ctx.fix_mode and not ctx.dry_run_mode:
        _archive_file(
            path,
            max_backups=ctx.max_backups_per_file,
            max_archive_bytes=ctx.max_archive_bytes,
        )
        path.write_text(text, encoding="utf-8")
 
    result["changed"] = any_changed
    return result
 
 
def compute_summary(file_results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate results into summary."""
    func = "compute_summary"
 
    summary = {
        "total_files": len(file_results),
        "total_changed": sum(1 for r in file_results if r.get("changed")),
        "fixes": {
            "header": 0,
            "docstring": 0,
            "dbg_import": 0,
            "func_pattern": 0,
        },
    }
 
    for result in file_results:
        for fix_type in summary["fixes"]:
            if result["fixes"].get(fix_type, {}).get("changed"):
                summary["fixes"][fix_type] += 1
 
    return summary
 
 
# ============================================================================
# Output Formatting
# ============================================================================
 
 
def _print_terminal_report(summary: Dict[str, Any], file_results: List[Dict[str, Any]],
                           verbose: bool, dry_run: bool) -> None:
    """Print human-readable terminal report."""
    func = "_print_terminal_report"
 
    mode = "DRY-RUN" if dry_run else "FIXED"
 
    print(f"\n{'=' * 70}")
    print(f"OCT FORMATTER REPORT — {mode}")
    print(f"{'=' * 70}\n")
 
    print(f"Files processed:      {summary['total_files']}")
    print(f"Files with changes:   {summary['total_changed']}")
    print()
    print("Fixes applied:")
    print(f"  Header blocks:      {summary['fixes']['header']}")
    print(f"  Docstrings:         {summary['fixes']['docstring']}")
    print(f"  _dbg imports:       {summary['fixes']['dbg_import']}")
    print(f"  func patterns:      {summary['fixes']['func_pattern']}")
 
    if verbose:
        print(f"\n{'-' * 70}")
        print("Detailed results:")
        print(f"{'-' * 70}\n")
 
        for result in file_results:
            if result["changed"] or result["errors"]:
                print(f"{result['path']}")
 
                for fix_type, fix_result in result["fixes"].items():
                    if fix_result["changed"]:
                        print(f"  [OK] {fix_type}: {', '.join(fix_result['messages'])}")
 
                for error in result["errors"]:
                    print(f"  [ERROR] {error}")
 
                print()
 
    print(f"{'=' * 70}\n")
 
 
def _print_json_report(summary: Dict[str, Any], file_results: List[Dict[str, Any]],
                       ctx: FormatterContext) -> None:
    """Print JSON output."""
    func = "_print_json_report"
 
    output = {
        "project": ctx.project_name,
        "oct_version": __version__,
        "timestamp": datetime.now().isoformat(),
        "dry_run": ctx.dry_run_mode,
        "summary": summary,
        "files": file_results,
    }
 
    print(json.dumps(output, indent=2))
 
 
# ============================================================================
# Main Entry Point
# ============================================================================
 
 
def run_formatter(project_root: Path, argv: Optional[List[str]] = None) -> None:
    """
    Run the Option C formatter on the given project root.
 
    Parameters
    ----------
    project_root : Path
        The detected Option C project root directory.
    argv : list[str] | None
        Optional list of command-line arguments.
        If None, argparse will use sys.argv[1:].
    """
    func = "run_formatter"
 
    # Parse arguments
    parser = argparse.ArgumentParser(
        description="Auto-fix Option C compliance violations",
        prog="oct format",
    )
    parser.add_argument(
        "--fix",
        action="store_true",
        help="Apply formatting changes. Without this, only reports what would change.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Report what would change without modifying files (default behavior).",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output machine-readable JSON instead of colorized terminal report.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Show per-file details in addition to summary.",
    )
    parser.add_argument(
        "--changed",
        action="store_true",
        help="Format only git-modified Python files.",
    )
 
    args = parser.parse_args(argv)
 
    # Create context
    diagnostics_dir = project_root / "oc_diagnostics"
    tests_dir = project_root / "tests"
    docs_dir = project_root / "docs"
 
    # Find and process all Python files
    exclude_dirs = set(LINTER_EXCLUDE_DIRS)
    wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
 
    # Load per-project overrides from .octrc.json (same as linter).
    # OI-519: routed through central oct.core.octrc loader.
    max_backups_per_file = _DEFAULT_MAX_BACKUPS_PER_FILE
    from oct.core.octrc import load_octrc_safe
    octrc = load_octrc_safe(project_root)
    linter_cfg = octrc.get("linter", {})
    if isinstance(linter_cfg, dict):
        for d in linter_cfg.get("exclude_dirs_add", []):
            exclude_dirs.add(d)
        for d in linter_cfg.get("exclude_dirs_remove", []):
            exclude_dirs.discard(d)
        for d in linter_cfg.get("wildcard_exclude_add", []):
            if d not in wildcard_exclude:
                wildcard_exclude.append(d)
    # OI-424: optional formatter.max_backups_per_file override.
    # OI-516: optional formatter.max_archive_mib override — written as MiB
    # in .octrc.json (readable) and converted to bytes on load.
    max_archive_bytes = _DEFAULT_MAX_ARCHIVE_BYTES
    formatter_cfg = octrc.get("formatter", {})
    if isinstance(formatter_cfg, dict):
        raw = formatter_cfg.get("max_backups_per_file")
        if isinstance(raw, int) and raw >= 0:
            max_backups_per_file = raw
        raw_mib = formatter_cfg.get("max_archive_mib")
        if isinstance(raw_mib, (int, float)) and raw_mib >= 0:
            max_archive_bytes = int(raw_mib * 1024 * 1024)
 
    ctx = FormatterContext(
        project_root=project_root,
        project_name=project_root.name,
        diagnostics_dir=diagnostics_dir,
        tests_dir=tests_dir,
        docs_dir=docs_dir,
        fix_mode=args.fix,
        dry_run_mode=not args.fix,
        json_mode=args.json,
        verbose=args.verbose,
        max_backups_per_file=max_backups_per_file,
        max_archive_bytes=max_archive_bytes,
    )
 
    if args.changed:
        from oct.linter.oct_lint import is_excluded_dir
        python_files = _git_changed_files(project_root)
        python_files = [
            p for p in python_files
            if p.is_file() and not any(
                is_excluded_dir(Path(part), exclude_dirs, wildcard_exclude)
                for part in p.relative_to(project_root).parts
            )
        ]
        if not python_files:
            if not args.json:
                print("No modified Python files found.", file=sys.stderr)
            return
    else:
        python_files = find_python_files(project_root, exclude_dirs, wildcard_exclude)
 
    file_results = []
    for py_file in python_files:
        result = format_file(py_file, ctx)
        file_results.append(result)
 
    # Compute summary
    summary = compute_summary(file_results)
 
    # Output report
    if ctx.json_mode:
        _print_json_report(summary, file_results, ctx)
    else:
        _print_terminal_report(summary, file_results, ctx.verbose, ctx.dry_run_mode)
 
 
if __name__ == "__main__":
    project_root = find_project_root(Path.cwd())
    run_formatter(project_root)