Nexum-Expanded/runtime/src/nexum_runtime/tooling/languages/analysis.py

"""Real parser-backed and lexical source inspection for language packs."""

from __future__ import annotations

import hashlib
import re
from pathlib import Path
from typing import Any, Callable, cast

from .catalog import (
    LANGUAGE_PACKS,
    LanguagePack,
    configure_language_parser_cache,
    resolve_language_pack,
)


_LEXICAL_TOKEN_RE = re.compile(
    r"""
    (?P<space>\s+)
    |(?P<line_comment>//[^\n]*|\#[^\n]*|--[^\n]*)
    |(?P<block_comment>/\*.*?\*/)
    |(?P<string>"(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*')
    |(?P<number>\b(?:0x[0-9a-fA-F]+|\d+(?:\.\d+)?)\b)
    |(?P<identifier>[A-Za-z_][A-Za-z0-9_]*)
    |(?P<delimiter>[\(\)\[\]\{\}])
    |(?P<operator>\S)
    """,
    re.DOTALL | re.VERBOSE,
)


def _source_digest(source: str) -> str:
    return hashlib.sha256(source.encode("utf-8", errors="replace")).hexdigest()


def _lexical_analysis(pack: LanguagePack, source: str) -> dict[str, Any]:
    counts = {
        "line_comment": 0,
        "block_comment": 0,
        "string": 0,
        "number": 0,
        "identifier": 0,
        "delimiter": 0,
        "operator": 0,
    }
    delimiter_balance = {"(": 0, "[": 0, "{": 0}
    closing = {")": "(", "]": "[", "}": "{"}
    delimiter_errors = 0
    for match in _LEXICAL_TOKEN_RE.finditer(source):
        kind = match.lastgroup or ""
        if kind in counts:
            counts[kind] += 1
        if kind != "delimiter":
            continue
        token = match.group(0)
        if token in delimiter_balance:
            delimiter_balance[token] += 1
        else:
            opening = closing[token]
            if delimiter_balance[opening] <= 0:
                delimiter_errors += 1
            else:
                delimiter_balance[opening] -= 1
    delimiter_errors += sum(delimiter_balance.values())
    return {
        "schema": "nexum.language-inspection.v1",
        "language_pack_id": pack.pack_id,
        "language": pack.name,
        "backend": "lexical",
        "parser_language": "",
        "source_sha256": _source_digest(source),
        "total_bytes": len(source.encode("utf-8", errors="replace")),
        "total_lines": source.count("\n") + (1 if source else 0),
        "node_count": sum(counts.values()),
        "error_count": delimiter_errors,
        "max_depth": 0,
        "root_type": "source",
        "token_counts": counts,
        "operational": True,
    }


def _parser_analysis(
    pack: LanguagePack,
    source: str,
    parser_language: str,
) -> dict[str, Any]:
    try:
        from tree_sitter_language_pack import get_parser
    except ImportError as exc:
        raise RuntimeError("language parser package is unavailable") from exc
    try:
        configure_language_parser_cache()
        parser_factory = cast(Callable[[str], Any], get_parser)
        parser = parser_factory(parser_language)
        tree = parser.parse(source.encode("utf-8", errors="replace"))
    except (LookupError, RuntimeError, ValueError) as exc:
        raise RuntimeError(
            f"language parser is unavailable: {parser_language}"
        ) from exc
    root = tree.root_node
    stack: list[tuple[Any, int]] = [(root, 0)]
    node_count = 0
    error_count = 0
    missing_count = 0
    max_depth = 0
    while stack:
        node, depth = stack.pop()
        node_count += 1
        max_depth = max(max_depth, depth)
        if str(getattr(node, "type", "")) == "ERROR":
            error_count += 1
        if bool(getattr(node, "is_missing", False)):
            missing_count += 1
        children = getattr(node, "children", ())
        stack.extend((child, depth + 1) for child in children)
    return {
        "schema": "nexum.language-inspection.v1",
        "language_pack_id": pack.pack_id,
        "language": pack.name,
        "backend": "parser",
        "parser_mode": pack.parser_mode,
        "parser_language": parser_language,
        "source_sha256": _source_digest(source),
        "total_bytes": len(source.encode("utf-8", errors="replace")),
        "total_lines": source.count("\n") + (1 if source else 0),
        "node_count": node_count,
        "error_count": error_count + missing_count,
        "missing_node_count": missing_count,
        "max_depth": max_depth,
        "root_type": str(getattr(root, "type", "")),
        "root_has_error": bool(getattr(root, "has_error", False)),
        "operational": bool(node_count > 0),
    }


def analyze_language_source(
    source: str,
    *,
    language: str,
    path: str = "",
) -> dict[str, Any]:
    """Inspect source through the selected release-local language pack."""

    pack = resolve_language_pack(language, path=path)
    if not pack.parser_languages:
        return _lexical_analysis(pack, source)
    return _parser_analysis(pack, source, pack.parser_languages[0])


def analyze_language_file(
    path: Path,
    *,
    language: str = "",
) -> dict[str, Any]:
    """Inspect one existing text file without returning its content."""

    source = path.read_text(encoding="utf-8", errors="replace")
    result = analyze_language_source(
        source,
        language=language,
        path=path.name,
    )
    return {**result, "path": path.name}


def verify_language_packs() -> dict[str, Any]:
    """Exercise every declared source-intelligence backend."""

    failures: list[dict[str, str]] = []
    parser_checks = 0
    lexical_checks = 0
    for pack in LANGUAGE_PACKS:
        try:
            if pack.parser_languages:
                for parser_language in pack.parser_languages:
                    result = _parser_analysis(pack, "", parser_language)
                    if not bool(result["operational"]):
                        raise RuntimeError("parser returned no syntax tree")
                    parser_checks += 1
            else:
                result = _lexical_analysis(pack, "x\n")
                if not bool(result["operational"]):
                    raise RuntimeError("lexical analyzer returned no structure")
                lexical_checks += 1
        except (OSError, RuntimeError, ValueError) as exc:
            failures.append(
                {
                    "id": pack.pack_id,
                    "reason": f"{type(exc).__name__}: {exc}",
                }
            )
    return {
        "schema": "nexum.language-pack-verification.v1",
        "ok": not failures,
        "language_pack_count": len(LANGUAGE_PACKS),
        "verified_language_pack_count": len(LANGUAGE_PACKS) - len(failures),
        "parser_checks": parser_checks,
        "lexical_checks": lexical_checks,
        "failures": failures,
    }


__all__ = [
    "analyze_language_file",
    "analyze_language_source",
    "verify_language_packs",
]