scripts.ast_report

Reference › scripts › ast_report

Produces an advisory AST health report: one Markdown digest of the codebase built from the AST toolkit's analyses. A human or a CI step runs it, and main always returns 0 because the report is advisory.

Invocation: --source takes a comma-separated list of directories or files (a default is built in), and --output writes the Markdown to a file instead of standard output. build_report loads the toolkit with _load_tool, runs its checks over the sources and assembles sections with _section and _item_line, each giving a count line followed by a limited list of rows. It reads source only and writes only the optional output file.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/ast_report.py

"""Advisory AST health report — a single markdown digest of the codebase.

Aggregates the read-only analyses from ``scripts/serialize_codebase_ast.py``
(cyclomatic complexity, nesting depth, long parameter lists, unused symbols,
the logging/exception audit, and untraced public defs) into one markdown
document.

It is purely advisory: it never fails. The process always exits 0 regardless
of how many findings surface, so it can run as a non-blocking CI step, a final
pipeline stage, or a manual pre-commit hook without ever gating a commit.

Run:
    uv run python scripts/ast_report.py                       # report to stdout
    uv run python scripts/ast_report.py --output ast-health.md
    uv run python scripts/ast_report.py --source src/pkg,tests
"""

from __future__ import annotations

import argparse
import importlib.util
import sys
from collections import Counter
from pathlib import Path

_TOOL_PATH = Path(__file__).resolve().parent / "serialize_codebase_ast.py"

_DEFAULT_SOURCE = "src/orchestrata_ultimate"

# Per-section thresholds and how many top items to surface.
_COMPLEXITY_THRESHOLD = 10
_NESTING_THRESHOLD = 4
_LONG_PARAM_THRESHOLD = 5
_TOP_N = 15


def _load_tool():
    """Load the AST toolkit module by path (same pattern as ``ast_api.py``)."""
    spec = importlib.util.spec_from_file_location("serialize_codebase_ast", _TOOL_PATH)
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load AST toolkit from {_TOOL_PATH}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _sources(source: str) -> list[str]:
    """Split a comma-separated ``--source`` value into a clean list."""
    return [s.strip() for s in source.split(",") if s.strip()]


def _item_line(row: dict) -> str:
    """Render one finding as ``- file:line  name`` (name optional)."""
    location = f"{row.get('file', '?')}:{row.get('line', '?')}"
    name = row.get("name")
    return f"- {location}  {name}" if name else f"- {location}"


def _section(heading: str, count_line: str, rows: list[dict], limit: int) -> list[str]:
    """Build a ``## heading`` section: a one-line count plus the top items."""
    lines = [f"## {heading}", "", count_line]
    if rows:
        lines.append("")
        lines.extend(_item_line(r) for r in rows[:limit])
    lines.append("")
    return lines


def build_report(sources: list[str]) -> str:
    """Build the full markdown advisory health report for *sources*.

    *sources* is a list of directories/files (e.g. ``["src/orchestrata_ultimate"]``).
    Returns the report as a markdown string; never raises for "findings", only
    for genuinely unreadable inputs.
    """
    tool = _load_tool()

    complexity = tool.measure_complexity(sources, threshold=_COMPLEXITY_THRESHOLD)
    nesting = tool.measure_nesting(sources, threshold=_NESTING_THRESHOLD)
    long_params = tool.find_long_param_lists(sources, threshold=_LONG_PARAM_THRESHOLD)
    unused = tool.find_unused(sources)
    audit = tool.audit_error_handling(sources)
    untraced = tool.find_untraced(sources)

    out: list[str] = [
        "# AST Health Report (advisory)",
        "",
        f"Sources: {', '.join(sources)}",
        "",
        "Advisory only — findings never block a commit, merge, or pipeline.",
        "",
    ]

    out += _section(
        "Complexity",
        f"{len(complexity)} function(s) with cyclomatic complexity "
        f">= {_COMPLEXITY_THRESHOLD} (top {_TOP_N}).",
        [{**r, "name": f"{r['name']} (cc={r['complexity']})"} for r in complexity],
        _TOP_N,
    )

    out += _section(
        "Nesting",
        f"{len(nesting)} function(s) with nesting depth "
        f">= {_NESTING_THRESHOLD} (top {_TOP_N}).",
        [{**r, "name": f"{r['name']} (depth={r['depth']})"} for r in nesting],
        _TOP_N,
    )

    out += _section(
        "Long-params",
        f"{len(long_params)} function(s) with more than "
        f"{_LONG_PARAM_THRESHOLD} parameters (top {_TOP_N}).",
        [{**r, "name": f"{r['name']} ({r['params']} params)"} for r in long_params],
        _TOP_N,
    )

    out += _section(
        "Unused",
        f"{len(unused)} module-level def(s) never referenced "
        "(precision-filtered — review before deleting).",
        unused,
        len(unused),
    )

    severities = Counter(f.get("severity", "info") for f in audit)
    audit_rows = [
        {**f, "name": f"[{f.get('rule', '?')}] {f.get('detail', '')}"} for f in audit
    ]
    out += _section(
        "Audit",
        f"{len(audit)} error-handling/logging finding(s): "
        f"{severities.get('error', 0)} error, {severities.get('warning', 0)} warning "
        f"(top {_TOP_N}).",
        audit_rows,
        _TOP_N,
    )

    out += _section(
        "Untraced",
        f"{len(untraced)} public def(s) lacking an @traces decorator (top {_TOP_N}).",
        untraced,
        _TOP_N,
    )

    cycles = tool.find_import_cycles(sources)
    out += ["## Import cycles", "", f"{len(cycles)} first-party import cycle(s)."]
    if cycles:
        out.append("")
        out += [f"- {' -> '.join(cycle)}" for cycle in cycles[:_TOP_N]]
    out.append("")

    dead = tool.find_dead_params(sources)
    out += _section(
        "Dead params",
        f"{len(dead)} function(s) with unused parameter(s) "
        f"(interface/abstract declarations excluded; top {_TOP_N}).",
        [{**r, "name": f"{r['name']} ({', '.join(r['params'])})"} for r in dead],
        _TOP_N,
    )

    untyped = tool.find_untyped(sources)
    out += _section(
        "Untyped",
        f"{len(untyped)} function(s) missing type hints "
        f"(precondition for type-resolved refactors; top {_TOP_N}).",
        [
            {
                **r,
                "name": r["name"]
                + (f" (params: {', '.join(r['params'])})" if r["params"] else "")
                + (" (return)" if r["return"] else ""),
            }
            for r in untyped
        ],
        _TOP_N,
    )

    return "\n".join(out).rstrip() + "\n"


def main(argv: list[str] | None = None) -> int:
    """CLI entry point. Always returns 0 — the report is advisory."""
    parser = argparse.ArgumentParser(
        description="Advisory AST health report (markdown). Always exits 0."
    )
    parser.add_argument(
        "--source",
        default=_DEFAULT_SOURCE,
        help="Comma-separated dirs/files to analyze (default: %(default)s).",
    )
    parser.add_argument(
        "--output",
        default=None,
        help="Write the markdown report here (default: stdout).",
    )
    args = parser.parse_args(argv)

    report = build_report(_sources(args.source))

    if args.output:
        Path(args.output).write_text(report, encoding="utf-8")
    else:
        sys.stdout.write(report)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())