orchestrata_ultimate.verification.review.helpers

Reference › orchestrata_ultimate › verification › review › helpers

Shared utilities for the review checks, used by the check modules, the junior volk and the review package. _parse_file parses a Python file into an AST and returns None when it cannot; _read_lines reads a file's lines. _is_public tests for a leading underscore, _function_line_count counts a function body's logical lines excluding the docstring, _collect_imported_modules lists imported top-level modules with line numbers, and _get_pyproject_deps extracts dependency names from pyproject.toml. _STDLIB_MODULES is taken from sys.stdlib_module_names. read_workspace_files reads workspace files into one source listing. The review engine calls these before running each check.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/review/helpers.py

"""Helper functions for deterministic code review checks.

Provides file parsing, line reading, and module introspection
utilities used by the individual check modules.
"""

import ast
import os
import re
import sys

from orchestrata_ultimate.core.base.traces import traces

# Stdlib top-level module names (common subset for fast lookup)
_STDLIB_MODULES: set[str] = set(sys.stdlib_module_names)


@traces(spec="027", story="validate_dependencies", reqs=["REQ-027-065"])
def _parse_file(filepath: str) -> ast.Module | None:
    """Parse a Python file into an AST.

    Args:
        filepath: Absolute path to the Python file.

    Returns:
        Parsed AST module, or None on failure.
    """
    try:
        with open(filepath, encoding="utf-8") as f:
            source = f.read()
        return ast.parse(source, filename=filepath)
    except SyntaxError, OSError:
        return None


def _read_lines(filepath: str) -> list[str]:
    """Read file lines.

    Args:
        filepath: Absolute path to the file.

    Returns:
        List of lines, or empty list on failure.
    """
    try:
        with open(filepath, encoding="utf-8") as f:
            return f.readlines()
    except OSError:
        return []


def _is_public(name: str) -> bool:
    """Check if a name is public (no leading underscore)."""
    return not name.startswith("_")


def _function_line_count(
    node: ast.FunctionDef | ast.AsyncFunctionDef,
) -> int:
    """Count logical lines in a function body (excludes docstring).

    Args:
        node: AST function definition node.

    Returns:
        Number of logical lines in the function body.
    """
    body = node.body
    start = 0
    # Skip docstring
    if (
        body
        and isinstance(body[0], ast.Expr)
        and isinstance(body[0].value, ast.Constant)
    ):
        start = 1
    if not body[start:]:
        return 0
    first = body[start].lineno
    last = body[-1].end_lineno or body[-1].lineno
    return last - first + 1


@traces(spec="027", story="validate_dependencies", reqs=["REQ-027-063"])
def _get_pyproject_deps(workspace: str) -> set[str]:
    """Extract dependency names from pyproject.toml.

    Args:
        workspace: Path to project workspace root.

    Returns:
        Set of normalized dependency names.
    """
    deps: set[str] = set()
    toml_path = os.path.join(workspace, "pyproject.toml")
    if not os.path.exists(toml_path):
        return deps
    try:
        with open(toml_path, encoding="utf-8") as f:
            content = f.read()
    except OSError:
        return deps

    in_deps = False
    for line in content.splitlines():
        stripped = line.strip()
        if stripped in ("dependencies = [", "dependencies=["):
            in_deps = True
            continue
        if in_deps:
            if stripped == "]":
                break
            m = re.match(r'["\']([a-zA-Z0-9_-]+)', stripped)
            if m:
                deps.add(m.group(1).lower().replace("-", "_"))
    return deps


def _collect_imported_modules(
    tree: ast.Module,
) -> list[tuple[str, int, str]]:
    """Extract all imported top-level module names with line numbers.

    Only checks module-level imports (tree.body), not function-level
    inline imports. isort only reorders top-level imports, and
    function-level lazy imports are a valid pattern.

    Args:
        tree: Parsed AST module.

    Returns:
        List of (top_level_module, line_number, full_import_string).
    """
    imports: list[tuple[str, int, str]] = []
    for node in tree.body:
        if isinstance(node, ast.Import):
            for alias in node.names:
                top = alias.name.split(".")[0]
                imports.append((top, node.lineno, alias.name))
        elif isinstance(node, ast.ImportFrom):
            if node.module and node.level == 0:
                top = node.module.split(".")[0]
                imports.append((top, node.lineno, node.module))
    return imports


def read_workspace_files(
    workspace: str,
    files: list[str],
) -> str:
    """Read file contents from workspace into a source listing.

    Args:
        workspace: Root directory of the project.
        files: List of relative file paths.

    Returns:
        Concatenated source listing string.
    """
    parts: list[str] = []
    for rel_path in files:
        filepath = os.path.join(workspace, rel_path)
        if not os.path.exists(filepath):
            parts.append(f"\n=== {rel_path} ===\n(file not found)\n")
            continue
        try:
            with open(filepath, encoding="utf-8") as fh:
                parts.append(f"\n=== {rel_path} ===\n{fh.read()}\n")
        except OSError, UnicodeDecodeError:
            parts.append(f"\n=== {rel_path} ===\n(could not read)\n")
    return "".join(parts)