orchestrata_ultimate.knowledge_graph.ast_analyzer

Reference › orchestrata_ultimate › knowledge_graph › ast_analyzer

ASTAnalyzer is the live-analysis front door of the knowledge_graph package. It coordinates a code_parser CodeParser, a class_extractor ClassExtractor and a function_extractor FunctionExtractor, each injectable through the constructor. analyze_file returns a FileAnalysis for one file, and analyze_file_detailed does the same with detailed class-method extraction. analyze_project walks a directory recursively in sorted order, keeps files whose suffix is in the extension set (default .py and .pl), skips paths containing __pycache__, .git, .venv, node_modules or .tox, parses each file and totals classes, functions, predicates and lines into a ProjectAnalysis together with the elapsed milliseconds and collected per-file errors. A path that is not a directory yields a ProjectAnalysis whose errors list says so rather than raising. The module only reads source files and writes nothing to the graph; no other module imports it, and it is traced to Spec 008 (REQ-008-099).

At a glance

Class diagram

Dependency diagram

API

Classes

ASTAnalyzer

Line 20

Coordinates code analysis using parser and extractors.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/ast_analyzer.py

"""AST analyzer for the knowledge graph.

Orchestrates CodeParser, ClassExtractor, and FunctionExtractor
to produce complete file and project analyses.
"""

import logging
import time
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.class_extractor import ClassExtractor
from orchestrata_ultimate.knowledge_graph.code_parser import CodeParser
from orchestrata_ultimate.knowledge_graph.function_extractor import FunctionExtractor
from orchestrata_ultimate.knowledge_graph.types import FileAnalysis, ProjectAnalysis

logger = logging.getLogger(__name__)


class ASTAnalyzer:
    """Coordinates code analysis using parser and extractors.

    Implements ICodeAnalyzer protocol.
    """

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-092"])
    def __init__(
        self,
        parser: CodeParser | None = None,
        class_extractor: ClassExtractor | None = None,
        function_extractor: FunctionExtractor | None = None,
    ) -> None:
        self._parser = parser or CodeParser()
        self._class_extractor = class_extractor or ClassExtractor()
        self._function_extractor = function_extractor or FunctionExtractor()

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-098"])
    def analyze_file(self, file_path: str) -> FileAnalysis:
        """Analyze a single file."""
        return self._parser.parse_file(file_path)

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-099"])
    def analyze_project(
        self,
        project_path: str,
        extensions: set[str] | None = None,
        ignore_dirs: set[str] | None = None,
    ) -> ProjectAnalysis:
        """Analyze all supported files in a project directory."""
        start = time.perf_counter()
        exts = extensions or {".py", ".pl"}
        ignore = ignore_dirs or {"__pycache__", ".git", ".venv", "node_modules", ".tox"}

        project = Path(project_path)
        if not project.is_dir():
            return ProjectAnalysis(
                project_path=project_path,
                errors=[f"Not a directory: {project_path}"],
            )

        files: list[FileAnalysis] = []
        total_classes = 0
        total_functions = 0
        total_predicates = 0
        total_lines = 0
        errors: list[str] = []

        for source_file in sorted(project.rglob("*")):
            if source_file.suffix not in exts:
                continue
            if any(part in ignore for part in source_file.parts):
                continue

            analysis = self._parser.parse_file(str(source_file))
            files.append(analysis)
            total_classes += len(analysis.classes)
            total_functions += len(analysis.functions)
            total_predicates += len(analysis.predicates)
            total_lines += analysis.line_count
            errors.extend(analysis.errors)

        elapsed = (time.perf_counter() - start) * 1000

        return ProjectAnalysis(
            project_path=project_path,
            files=files,
            total_classes=total_classes,
            total_functions=total_functions,
            total_predicates=total_predicates,
            total_files=len(files),
            total_lines=total_lines,
            parse_time_ms=elapsed,
            errors=errors,
        )

    def analyze_file_detailed(self, file_path: str) -> FileAnalysis:
        """Analyze a file with detailed class method extraction."""
        import ast as ast_mod

        path = Path(file_path)
        if not path.exists() or path.suffix != ".py":
            return self._parser.parse_file(file_path)

        content = path.read_text(encoding="utf-8", errors="replace")
        try:
            tree = ast_mod.parse(content, filename=file_path)
        except SyntaxError:
            return self._parser.parse_file(file_path)

        analysis = self._parser.parse_file(file_path)

        # Enrich with detailed method info
        class_data = self._class_extractor.extract_with_methods(tree, file_path)
        for class_node, methods in class_data:
            # Replace the basic class node with enriched one
            for i, existing in enumerate(analysis.classes):
                if existing.name == class_node.name:
                    analysis.classes[i] = class_node
                    break

        return analysis