orchestrata_ultimate.knowledge_graph.code_parser

Reference › orchestrata_ultimate › knowledge_graph › code_parser

CodeParser extracts structure from source files for the knowledge graph (Spec 008, REQ-008-017 and REQ-008-056 to -060). Python files are parsed with the standard ast module in _parse_python, which collects classes with their methods (_extract_class_with_methods), functions, imports and decorators; Prolog files are parsed with regular expressions in _parse_prolog into predicates and rules. parse_file dispatches by extension and returns a FileAnalysis including errors, line count and parse time. get_file_hash gives a SHA-256 of a file, which the synchronizer uses to skip unchanged files, and create_file_node builds a FileNode.

Three batch methods turn many analyses into edges. resolve_calls builds a name index of all parsed functions and matches each function's call names with file scope first, then package scope, then global scope, producing caller and callee ids with a confidence score from _call_confidence (same file above same package above global) and the unresolved call_name. resolve_inheritance maps base-class names to ClassNodes and resolve_imports maps import statements to File nodes, using _file_path_to_module. The parser reads files but writes nothing to the graph. ast_analyzer and the synchronizer import it; result types come from types.

At a glance

Class diagram

Dependency diagram

API

Classes

CodeParser

Decorators: traces(spec='008', story='US-004', reqs=['REQ-008-017']) · Line 31

Parses source files to extract structural information.

Functions

Source

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

"""Code parser using Python's ast module.

Parses Python and Prolog source files to extract structural
information for the knowledge graph. Uses Python's built-in
ast module for Python files, and regex-based parsing for Prolog.
"""

import ast
import hashlib
import logging
import re
import time
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.types import (
    ClassNode,
    FileAnalysis,
    FileNode,
    FunctionNode,
    ImportNode,
    PredicateNode,
    RuleNode,
)

logger = logging.getLogger(__name__)


@traces(spec="008", story="US-004", reqs=["REQ-008-017"])
class CodeParser:
    """Parses source files to extract structural information.

    Supports Python (.py) and Prolog (.pl) files.
    """

    SUPPORTED_EXTENSIONS = {".py", ".pl"}

    @traces(spec="008", story="US-004", reqs=["REQ-008-017", "REQ-008-018"])
    def parse_file(self, file_path: str) -> FileAnalysis:
        """Parse a single file and return analysis results."""
        start = time.perf_counter()
        path = Path(file_path)

        if path.suffix not in self.SUPPORTED_EXTENSIONS:
            return FileAnalysis(
                file_path=file_path,
                language="unknown",
                errors=[f"Unsupported file type: {path.suffix}"],
            )

        if not path.exists():
            return FileAnalysis(
                file_path=file_path,
                errors=[f"File not found: {file_path}"],
            )

        try:
            content = path.read_text(encoding="utf-8", errors="replace")
        except Exception as exc:
            return FileAnalysis(
                file_path=file_path,
                errors=[f"Read error: {exc}"],
            )

        line_count = content.count("\n") + 1

        if path.suffix == ".py":
            analysis = self._parse_python(file_path, content)
        elif path.suffix == ".pl":
            analysis = self._parse_prolog(file_path, content)
        else:
            analysis = FileAnalysis(file_path=file_path)

        analysis.line_count = line_count
        analysis.parse_time_ms = (time.perf_counter() - start) * 1000
        return analysis

    @traces(spec="008", story="US-004", reqs=["REQ-008-021"])
    def get_file_hash(self, file_path: str) -> str:
        """Get SHA256 hash of a file."""
        path = Path(file_path)
        if not path.exists():
            return ""
        content = path.read_bytes()
        return hashlib.sha256(content).hexdigest()

    def create_file_node(self, file_path: str) -> FileNode:
        """Create a FileNode from a file path."""
        path = Path(file_path)
        lang = (
            "python"
            if path.suffix == ".py"
            else "prolog" if path.suffix == ".pl" else "unknown"
        )
        try:
            stat = path.stat()
            content = path.read_text(encoding="utf-8", errors="replace")
            line_count = content.count("\n") + 1
        except Exception as exc:
            logger.warning(
                "Failed to read file stats for %s: %s", path, exc, exc_info=True
            )
            stat = None
            line_count = 0

        return FileNode(
            path=str(path),
            name=path.name,
            language=lang,
            line_count=line_count,
            size_bytes=stat.st_size if stat else 0,
            hash=self.get_file_hash(file_path),
        )

    # ── Python Parsing ────────────────────────────────────────────

    @traces(spec="008", story="US-004", reqs=["REQ-008-017"])
    def _parse_python(self, file_path: str, content: str) -> FileAnalysis:
        """Parse a Python file using ast module."""
        analysis = FileAnalysis(file_path=file_path, language="python")

        try:
            tree = ast.parse(content, filename=file_path)
        except SyntaxError as exc:
            analysis.errors.append(f"Syntax error: {exc}")
            return analysis

        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                class_node, method_nodes = self._extract_class_with_methods(
                    node, file_path
                )
                analysis.classes.append(class_node)
                analysis.functions.extend(method_nodes)
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                # Only top-level functions (not methods inside classes)
                if not self._is_method(node, tree):
                    func_node = self._extract_function(node, file_path)
                    analysis.functions.append(func_node)
            elif isinstance(node, (ast.Import, ast.ImportFrom)):
                import_node = self._extract_import(node, file_path)
                if import_node:
                    analysis.imports.append(import_node)

        return analysis

    @traces(
        spec="008",
        story="US-004",
        reqs=[
            "REQ-008-072",
            "REQ-008-073",
            "REQ-008-075",
        ],
    )
    def _extract_class_with_methods(
        self, node: ast.ClassDef, file_path: str
    ) -> tuple[ClassNode, list[FunctionNode]]:
        """Extract class and its methods as separate FunctionNodes.

        Returns:
            Tuple of (ClassNode, list of method FunctionNodes with class_id set).
        """
        class_node = self._extract_class(node, file_path)
        method_nodes: list[FunctionNode] = []

        for item in node.body:
            if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                func_node = self._extract_function(item, file_path)
                func_node.class_id = class_node.id
                method_nodes.append(func_node)

        return class_node, method_nodes

    @traces(spec="008", story="US-004", reqs=["REQ-008-065"])
    def _extract_class(self, node: ast.ClassDef, file_path: str) -> ClassNode:
        """Extract class information from an AST node."""
        methods = []
        for item in node.body:
            if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                methods.append(item.name)

        base_classes = []
        for base in node.bases:
            if isinstance(base, ast.Name):
                base_classes.append(base.id)
            elif isinstance(base, ast.Attribute):
                base_classes.append(ast.dump(base))

        decorators = []
        for dec in node.decorator_list:
            if isinstance(dec, ast.Name):
                decorators.append(dec.id)
            elif isinstance(dec, ast.Attribute):
                decorators.append(ast.dump(dec))

        docstring = ast.get_docstring(node) or ""

        return ClassNode(
            name=node.name,
            file_id=file_path,
            line_start=node.lineno,
            line_end=node.end_lineno or node.lineno,
            docstring=docstring,
            methods=methods,
            base_classes=base_classes,
            decorators=decorators,
        )

    @traces(spec="008", story="US-004", reqs=["REQ-008-074", "REQ-008-101"])
    def _extract_function(
        self, node: ast.FunctionDef | ast.AsyncFunctionDef, file_path: str
    ) -> FunctionNode:
        """Extract function information from an AST node."""
        params = [arg.arg for arg in node.args.args]

        return_type = ""
        if node.returns:
            try:
                return_type = ast.unparse(node.returns)
            except Exception as exc:
                logger.warning(
                    "Failed to unparse return type annotation: %s", exc, exc_info=True
                )

        decorators = []
        for dec in node.decorator_list:
            if isinstance(dec, ast.Name):
                decorators.append(dec.id)

        calls = []
        for child in ast.walk(node):
            if isinstance(child, ast.Call):
                if isinstance(child.func, ast.Name):
                    calls.append(child.func.id)
                elif isinstance(child.func, ast.Attribute):
                    calls.append(child.func.attr)

        docstring = ast.get_docstring(node) or ""

        return FunctionNode(
            name=node.name,
            file_id=file_path,
            line_start=node.lineno,
            line_end=node.end_lineno or node.lineno,
            signature=f"def {node.name}({', '.join(params)})",
            parameters=params,
            return_type=return_type,
            decorators=decorators,
            calls=list(set(calls)),
            is_async=isinstance(node, ast.AsyncFunctionDef),
            docstring=docstring,
        )

    @traces(spec="008", story="US-004", reqs=["REQ-008-017"])
    def _extract_import(
        self, node: ast.Import | ast.ImportFrom, file_path: str
    ) -> ImportNode | None:
        """Extract import information."""
        if isinstance(node, ast.Import):
            for alias in node.names:
                return ImportNode(
                    source_file_id=file_path,
                    target_module=alias.name,
                    imported_names=[alias.asname or alias.name],
                    line_number=node.lineno,
                )
        elif isinstance(node, ast.ImportFrom):
            module = node.module or ""
            names = [alias.name for alias in node.names]
            return ImportNode(
                source_file_id=file_path,
                target_module=module,
                imported_names=names,
                is_relative=node.level > 0,
                line_number=node.lineno,
            )
        return None

    @traces(spec="008", story="US-004", reqs=["REQ-008-017"])
    def _is_method(self, node: ast.AST, tree: ast.Module) -> bool:
        """Check if a function is a method (inside a class)."""
        for cls_node in ast.walk(tree):
            if isinstance(cls_node, ast.ClassDef):
                for item in cls_node.body:
                    if item is node:
                        return True
        return False

    # ── Prolog Parsing ────────────────────────────────────────────

    @traces(spec="008", story="US-004", reqs=["REQ-008-018"])
    def _parse_prolog(self, file_path: str, content: str) -> FileAnalysis:
        """Parse a Prolog file using regex-based extraction."""
        analysis = FileAnalysis(file_path=file_path, language="prolog")

        # Extract module declaration
        module_match = re.search(r":- module\((\w+),", content)
        module_name = module_match.group(1) if module_match else ""

        # Extract exported predicates from module declaration
        exports_match = re.search(r":- module\(\w+,\s*\[(.*?)\]\)", content, re.DOTALL)
        exported = set()
        if exports_match:
            exported = {
                f"{m.group(1)}/{m.group(2)}"
                for m in re.finditer(r"(\w+)/(\d+)", exports_match.group(1))
            }

        # Extract dynamic declarations
        dynamics = {
            f"{m.group(1)}/{m.group(2)}"
            for m in re.finditer(r":- dynamic\s+(\w+)/(\d+)", content)
        }

        # Extract predicates (head :- body or head.)
        predicate_map: dict[str, PredicateNode] = {}
        lines = content.split("\n")

        for i, line in enumerate(lines, 1):
            stripped = line.strip()
            if not stripped or stripped.startswith("%") or stripped.startswith(":-"):
                continue

            # Match predicate head
            head_match = re.match(r"(\w+)\(([^)]*)\)", stripped)
            if head_match:
                name = head_match.group(1)
                args = head_match.group(2)
                arity = len([a.strip() for a in args.split(",")]) if args.strip() else 0
                functor = f"{name}/{arity}"

                if functor not in predicate_map:
                    pred = PredicateNode(
                        name=name,
                        arity=arity,
                        file_id=file_path,
                        module=module_name,
                        line_start=i,
                        line_end=i,
                        is_dynamic=functor in dynamics,
                        is_exported=functor in exported,
                    )
                    predicate_map[functor] = pred
                else:
                    predicate_map[functor].line_end = i

                # Extract rule body
                if ":-" in stripped:
                    body_part = stripped.split(":-", 1)[1].strip()
                    rule = RuleNode(
                        head=stripped.split(":-")[0].strip(),
                        body=body_part.rstrip("."),
                        predicate_id=predicate_map[functor].id,
                        file_id=file_path,
                        line_start=i,
                        line_end=i,
                    )
                    analysis.rules.append(rule)

        analysis.predicates = list(predicate_map.values())
        return analysis

    # ── Graph Resolution (FR-014, FR-015, FR-016) ────────────────

    @traces(
        spec="008",
        story="US-004",
        reqs=[
            "REQ-008-056",
            "REQ-008-057",
            "REQ-008-058",
            "REQ-008-060",
        ],
    )
    def resolve_calls(self, all_analyses: list[FileAnalysis]) -> list[dict[str, Any]]:
        """Resolve function call names to target FunctionNodes.

        Builds a name index from all parsed functions, then matches each
        function's ``calls[]`` list against the index using file scope →
        package scope → global scope resolution.

        Args:
            all_analyses: List of FileAnalysis from a batch sync.

        Returns:
            List of dicts with keys: caller_id, callee_id, confidence,
            call_name (the unresolved name that produced this edge — lets
            the graph-write layer narrow its own name-match fallback to
            only the pairs this scoped resolution could not reach).
        """
        # Build name → [(function_id, file_path)] index
        name_index: dict[str, list[tuple[str, str]]] = {}
        for analysis in all_analyses:
            for func in analysis.functions:
                name_index.setdefault(func.name, []).append(
                    (func.id, analysis.file_path)
                )

        edges: list[dict[str, Any]] = []
        seen: set[tuple[str, str]] = set()

        for analysis in all_analyses:
            for func in analysis.functions:
                for call_name in func.calls:
                    candidates = name_index.get(call_name, [])
                    if not candidates:
                        continue
                    edges.extend(
                        self._call_edges_for(
                            func, analysis.file_path, call_name, candidates, seen
                        )
                    )

        return edges

    @staticmethod
    @traces(spec="008", story="US-004", reqs=["REQ-008-057", "REQ-008-058"])
    def _call_edges_for(
        func: Any,
        caller_file: str,
        call_name: str,
        candidates: list[tuple[str, str]],
        seen: set[tuple[str, str]],
    ) -> list[dict[str, Any]]:
        """Build deduplicated CALLS edges from one caller to its candidates.

        Scopes confidence by call site: same file > same package > global,
        further attenuated when a name is ambiguous (multiple candidates).
        ``seen`` is mutated to track caller→callee pairs already emitted.
        """
        edges: list[dict[str, Any]] = []
        for cand_id, cand_file in candidates:
            if cand_id == func.id:
                continue  # skip self-calls
            pair = (func.id, cand_id)
            if pair in seen:
                continue
            seen.add(pair)

            confidence = _call_confidence(caller_file, cand_file, len(candidates))
            edges.append(
                {
                    "caller_id": func.id,
                    "callee_id": cand_id,
                    "confidence": round(confidence, 3),
                    "call_name": call_name,
                }
            )
        return edges

    @traces(
        spec="008",
        story="US-004",
        reqs=[
            "REQ-008-062",
            "REQ-008-063",
            "REQ-008-064",
            "REQ-008-066",
        ],
    )
    def resolve_inheritance(
        self, all_analyses: list[FileAnalysis]
    ) -> list[dict[str, str]]:
        """Resolve base class names to target ClassNodes.

        Args:
            all_analyses: List of FileAnalysis from a batch sync.

        Returns:
            List of dicts with keys: child_id, parent_id, base_name (the
            unresolved name that produced this edge — lets the graph-write
            layer narrow its own name-match fallback to only the pairs
            this scoped resolution could not reach).
        """
        # Build class name → [(class_id, file_path)] index
        name_index: dict[str, list[tuple[str, str]]] = {}
        for analysis in all_analyses:
            for cls in analysis.classes:
                name_index.setdefault(cls.name, []).append((cls.id, analysis.file_path))

        edges: list[dict[str, str]] = []
        seen: set[tuple[str, str]] = set()

        for analysis in all_analyses:
            for cls in analysis.classes:
                for base_name in cls.base_classes:
                    # Strip ast.dump output for attribute access (e.g. "Attribute(...)")
                    clean_name = (
                        base_name.split("(")[0] if "(" in base_name else base_name
                    )
                    candidates = name_index.get(clean_name, [])
                    for cand_id, _ in candidates:
                        if cand_id == cls.id:
                            continue
                        pair = (cls.id, cand_id)
                        if pair in seen:
                            continue
                        seen.add(pair)
                        edges.append(
                            {
                                "child_id": cls.id,
                                "parent_id": cand_id,
                                "base_name": clean_name,
                            }
                        )

        return edges

    @traces(
        spec="008",
        story="US-004",
        reqs=[
            "REQ-008-067",
            "REQ-008-068",
            "REQ-008-069",
            "REQ-008-071",
        ],
    )
    def resolve_imports(self, all_analyses: list[FileAnalysis]) -> list[dict[str, Any]]:
        """Resolve import statements to target File nodes.

        Builds a module path → file path mapping, then resolves each
        ImportNode's target_module to a known file.

        Args:
            all_analyses: List of FileAnalysis from a batch sync.

        Returns:
            List of dicts with keys: source_file, target_file,
            imported_names, line_number, is_relative.
        """
        # Build module → file mapping
        module_to_file: dict[str, str] = {}
        for analysis in all_analyses:
            module_path = _file_path_to_module(analysis.file_path)
            if module_path:
                module_to_file[module_path] = analysis.file_path

        edges: list[dict[str, Any]] = []
        seen: set[tuple[str, str]] = set()

        for analysis in all_analyses:
            for imp in analysis.imports:
                target_file = module_to_file.get(imp.target_module)
                if not target_file:
                    # Try partial match (import orchestrata_ultimate.cxp matches orchestrata.cxp.__init__)
                    for mod, fpath in module_to_file.items():
                        if mod.startswith(
                            imp.target_module + "."
                        ) or imp.target_module.startswith(mod + "."):
                            target_file = fpath
                            break

                if not target_file or target_file == analysis.file_path:
                    continue

                pair = (analysis.file_path, target_file)
                if pair in seen:
                    continue
                seen.add(pair)

                edges.append(
                    {
                        "source_file": analysis.file_path,
                        "target_file": target_file,
                        "imported_names": imp.imported_names,
                        "line_number": imp.line_number,
                        "is_relative": imp.is_relative,
                    }
                )

        return edges


@traces(spec="008", story="US-004", reqs=["REQ-008-060"])
def _same_package(file_a: str, file_b: str) -> bool:
    """Check if two files are in the same package directory."""
    parent_a = str(Path(file_a).parent)
    parent_b = str(Path(file_b).parent)
    return parent_a == parent_b


@traces(spec="008", story="US-004", reqs=["REQ-008-060"])
def _call_confidence(caller_file: str, callee_file: str, num_candidates: int) -> float:
    """Score a resolved call edge: same file > same package > global.

    Ambiguous names (multiple candidates) are attenuated so no single
    edge claims high confidence when the resolution is uncertain.
    """
    if callee_file == caller_file:
        confidence = 1.0
    elif _same_package(callee_file, caller_file):
        confidence = 0.9
    else:
        confidence = 0.8

    if num_candidates > 1:
        confidence = min(confidence, 0.8 / num_candidates)
    return confidence


@traces(spec="008", story="US-004", reqs=["REQ-008-067"])
def _file_path_to_module(file_path: str) -> str:
    """Convert a file path to a Python module path.

    Strips common prefixes (packages/*/src/, src/) and converts
    directory separators to dots. Removes .py and __init__.

    Examples:
        packages/orchestrata-cxp/src/orchestrata/cxp/packet.py
        → orchestrata.cxp.packet

        src/orchestrata/knowledge/types.py
        → orchestrata.knowledge.types
    """
    p = file_path.replace("\\", "/")

    # Strip packages/*/src/ prefix or leading src/
    if "/src/" in p:
        p = p.split("/src/", 1)[1]
    elif p.startswith("src/"):
        p = p[4:]

    # Remove .py extension
    if p.endswith(".py"):
        p = p[:-3]

    # Remove trailing __init__
    if p.endswith("/__init__"):
        p = p[: -len("/__init__")]

    # Convert path separators to dots
    module = p.replace("/", ".")

    return module