DocForge2/src/docforge/treesitter_logic.py

"""Tree-sitter-backed control-flow extraction for JavaScript and C++.

Tree-sitter supplies concrete syntax trees.  This module adds the small amount
of language-aware control-flow interpretation needed to emit DocForge's
language-neutral ``LogicProjection`` contract.  Project code is parsed as
data; it is never imported, compiled, or executed.
"""

from __future__ import annotations

import hashlib
from collections.abc import Iterable
from dataclasses import dataclass
from functools import lru_cache

import tree_sitter_cpp
import tree_sitter_javascript
from tree_sitter import Language, Node, Parser

from .errors import DocForgeError
from .models import LogicEdge, LogicNode, LogicProjection

_TRIVIA_NODE_TYPES = frozenset({"comment"})


@dataclass(frozen=True)
class TreeSitterLogicOwner:
    """One named function or method that should receive a Logic projection."""

    owner_node_id: str
    qualified_name: str
    line: int


@dataclass(frozen=True)
class DiscoveredFunction:
    """One parser-identified callable available to a project adapter."""

    qualified_name: str
    name: str
    line: int
    kind: str


@dataclass(frozen=True)
class _Tail:
    source_id: str
    relation: str = "next"
    label: str | None = None


@dataclass(frozen=True)
class _Condition:
    entry_id: str
    when_true: tuple[_Tail, ...]
    when_false: tuple[_Tail, ...]


@dataclass(frozen=True)
class _Control:
    break_id: str | None = None
    continue_id: str | None = None


@dataclass(frozen=True)
class _LanguageProfile:
    name: str
    language: Language
    root_type: str
    block_types: frozenset[str]
    function_types: frozenset[str]
    loop_types: frozenset[str]
    return_types: frozenset[str]
    raise_types: frozenset[str]
    switch_case_types: frozenset[str]


@lru_cache(maxsize=1)
def _javascript_profile() -> _LanguageProfile:
    return _LanguageProfile(
        name="javascript",
        language=Language(tree_sitter_javascript.language()),
        root_type="program",
        block_types=frozenset({"program", "statement_block"}),
        function_types=frozenset(
            {"function_declaration", "generator_function_declaration", "method_definition"}
        ),
        loop_types=frozenset(
            {"while_statement", "do_statement", "for_statement", "for_in_statement"}
        ),
        return_types=frozenset({"return_statement"}),
        raise_types=frozenset({"throw_statement"}),
        switch_case_types=frozenset({"switch_case", "switch_default"}),
    )


@lru_cache(maxsize=1)
def _cpp_profile() -> _LanguageProfile:
    return _LanguageProfile(
        name="cpp",
        language=Language(tree_sitter_cpp.language()),
        root_type="translation_unit",
        block_types=frozenset({"translation_unit", "compound_statement"}),
        function_types=frozenset({"function_definition"}),
        loop_types=frozenset(
            {
                "while_statement",
                "do_statement",
                "for_statement",
                "for_range_loop",
            }
        ),
        return_types=frozenset({"return_statement", "co_return_statement"}),
        raise_types=frozenset({"throw_statement"}),
        switch_case_types=frozenset({"case_statement"}),
    )


def analyze_javascript_source(
    source: str,
    *,
    source_id: str,
    owners: Iterable[TreeSitterLogicOwner],
    filename: str = "<javascript-source>",
    max_nodes_per_function: int = 2_000,
) -> tuple[LogicProjection, ...]:
    """Build control-flow projections for named JavaScript functions and methods."""

    return _analyze_tree_sitter_source(
        source,
        source_id=source_id,
        owners=owners,
        filename=filename,
        profile=_javascript_profile(),
        max_nodes_per_function=max_nodes_per_function,
    )


def discover_javascript_functions(source: str) -> tuple[DiscoveredFunction, ...]:
    """Return named JavaScript functions, methods, and assigned arrow functions."""

    return _discover_functions(source, _javascript_profile())


def discover_cpp_functions(source: str) -> tuple[DiscoveredFunction, ...]:
    """Return named C++ functions and methods."""

    return _discover_functions(source, _cpp_profile())


def analyze_cpp_source(
    source: str,
    *,
    source_id: str,
    owners: Iterable[TreeSitterLogicOwner],
    filename: str = "<cpp-source>",
    max_nodes_per_function: int = 2_000,
) -> tuple[LogicProjection, ...]:
    """Build control-flow projections for named C++ functions and methods."""

    return _analyze_tree_sitter_source(
        source,
        source_id=source_id,
        owners=owners,
        filename=filename,
        profile=_cpp_profile(),
        max_nodes_per_function=max_nodes_per_function,
    )


def _discover_functions(
    source: str,
    profile: _LanguageProfile,
) -> tuple[DiscoveredFunction, ...]:
    raw = source.encode("utf-8")
    parser = Parser(profile.language)
    tree = parser.parse(raw)
    root = tree.root_node
    if root.has_error:
        return ()
    definitions = _function_definitions(root, raw, profile)
    result: list[DiscoveredFunction] = []
    for (qualified_name, line), node in definitions.items():
        normalized = qualified_name.replace("::", ".")
        name = normalized.split(".")[-1]
        result.append(
            DiscoveredFunction(
                qualified_name=qualified_name,
                name=name,
                line=line,
                kind=(
                    "method"
                    if node.type == "method_definition" or "." in normalized
                    else "function"
                ),
            )
        )
    return tuple(
        sorted(
            result,
            key=lambda item: (item.qualified_name, item.line, item.kind),
        )
    )


def _analyze_tree_sitter_source(
    source: str,
    *,
    source_id: str,
    owners: Iterable[TreeSitterLogicOwner],
    filename: str,
    profile: _LanguageProfile,
    max_nodes_per_function: int,
) -> tuple[LogicProjection, ...]:
    if max_nodes_per_function < 2:
        raise ValueError("max_nodes_per_function must allow entry and exit nodes")
    raw = source.encode("utf-8")
    parser = Parser(profile.language)
    tree = parser.parse(raw)
    if tree.root_node.type != profile.root_type or tree.root_node.has_error:
        error = _first_error(tree.root_node)
        raise DocForgeError(
            "invalid_logic_source",
            f"{profile.name.title()} source cannot be parsed for logic analysis",
            source=filename,
            line=(error.start_point.row + 1) if error is not None else 1,
        )
    definitions = _function_definitions(tree.root_node, raw, profile)
    requested = tuple(sorted(owners, key=lambda item: item.owner_node_id))
    if len({owner.owner_node_id for owner in requested}) != len(requested):
        raise DocForgeError("invalid_logic_owner", "Logic owner IDs must be unique")
    projections: list[LogicProjection] = []
    for owner in requested:
        function = _resolve_owner(owner, definitions)
        if function is None:
            raise DocForgeError(
                "missing_logic_owner",
                f"A requested {profile.name} logic owner was not found in its source",
                owner_node_id=owner.owner_node_id,
                qualified_name=owner.qualified_name,
                line=owner.line,
            )
        projections.append(
            _TreeSitterFunctionBuilder(
                raw=raw,
                source_id=source_id,
                owner_node_id=owner.owner_node_id,
                function=function,
                profile=profile,
                max_nodes=max_nodes_per_function,
            ).build()
        )
    return tuple(projections)


def _first_error(node: Node) -> Node | None:
    if node.is_error or node.is_missing:
        return node
    for child in node.named_children:
        error = _first_error(child)
        if error is not None:
            return error
    return None


def _function_definitions(
    root: Node,
    raw: bytes,
    profile: _LanguageProfile,
) -> dict[tuple[str, int], Node]:
    definitions: dict[tuple[str, int], Node] = {}

    def visit(node: Node, scopes: tuple[str, ...]) -> None:
        next_scopes = scopes
        scope_name = _scope_name(node, raw, profile)
        if scope_name:
            next_scopes = (*scopes, scope_name)
        function_name = _function_name(node, raw, profile)
        if function_name:
            qualified = (
                function_name if "::" in function_name else ".".join((*scopes, function_name))
            )
            function_node = node
            if node.type == "variable_declarator":
                function_node = node.child_by_field_name("value") or node
            definitions[(qualified, node.start_point.row + 1)] = function_node
            next_scopes = (*scopes, function_name)
        for child in node.named_children:
            visit(child, next_scopes)

    visit(root, ())
    return definitions


def _scope_name(node: Node, raw: bytes, profile: _LanguageProfile) -> str | None:
    if profile.name == "javascript" and node.type in {"class_declaration", "class"}:
        return _field_text(node, "name", raw)
    if profile.name == "cpp" and node.type in {
        "namespace_definition",
        "class_specifier",
        "struct_specifier",
        "union_specifier",
    }:
        return _field_text(node, "name", raw)
    return None


def _function_name(node: Node, raw: bytes, profile: _LanguageProfile) -> str | None:
    if node.type in profile.function_types:
        if profile.name == "javascript":
            return _field_text(node, "name", raw)
        declarator = node.child_by_field_name("declarator")
        return _declarator_name(declarator, raw) if declarator is not None else None
    if profile.name != "javascript" or node.type != "variable_declarator":
        return None
    value = node.child_by_field_name("value")
    if value is None or value.type not in {"arrow_function", "function_expression"}:
        return None
    return _field_text(node, "name", raw)


def _declarator_name(node: Node, raw: bytes) -> str | None:
    if node.type in {
        "identifier",
        "field_identifier",
        "operator_name",
        "destructor_name",
        "qualified_identifier",
    }:
        return _text(node, raw)
    for field in ("declarator", "name"):
        child = node.child_by_field_name(field)
        if child is not None:
            result = _declarator_name(child, raw)
            if result:
                return result
    for child in node.named_children:
        result = _declarator_name(child, raw)
        if result:
            return result
    return None


def _resolve_owner(
    owner: TreeSitterLogicOwner,
    definitions: dict[tuple[str, int], Node],
) -> Node | None:
    exact = definitions.get((owner.qualified_name, owner.line))
    if exact is not None:
        return exact
    leaf = owner.qualified_name.replace("::", ".").split(".")[-1]
    candidates = [
        node
        for (qualified_name, line), node in definitions.items()
        if line == owner.line and qualified_name.replace("::", ".").split(".")[-1] == leaf
    ]
    return candidates[0] if len(candidates) == 1 else None


class _TreeSitterFunctionBuilder:
    def __init__(
        self,
        *,
        raw: bytes,
        source_id: str,
        owner_node_id: str,
        function: Node,
        profile: _LanguageProfile,
        max_nodes: int,
    ) -> None:
        self.raw = raw
        self.source_id = source_id
        self.owner_node_id = owner_node_id
        self.function = function
        self.profile = profile
        self.max_nodes = max_nodes
        self.nodes: list[LogicNode] = []
        self.edges: list[LogicEdge] = []
        self._edge_ordinals: dict[str, int] = {}
        self._sequence = 0
        self._owner_digest = hashlib.sha256(owner_node_id.encode()).hexdigest()[:12]
        name = _function_name(function, raw, profile) or owner_node_id.rsplit(".", 1)[-1]
        self.entry_id = self._node("entry", f"Enter {name}", function)
        self.exit_id = self._node("exit", f"Exit {name}", function)

    def build(self) -> LogicProjection:
        body = self.function.child_by_field_name("body")
        incoming = (_Tail(self.entry_id),)
        if body is None:
            tails = incoming
        elif body.type in self.profile.block_types:
            tails = self._statements(body.named_children, incoming, control=None)
        else:
            tails = self._expression_body(body, incoming)
        self._connect(tails, self.exit_id)
        if not self._has_incoming(self.exit_id):
            self._edge(self.entry_id, "next", self.exit_id, "END")
        return LogicProjection(
            owner_node_id=self.owner_node_id,
            source_id=self.source_id,
            nodes=tuple(sorted(self.nodes, key=lambda item: item.logic_id)),
            edges=tuple(
                sorted(
                    self.edges,
                    key=lambda item: (
                        item.source_id,
                        item.ordinal,
                        item.relation,
                        item.target_id,
                        item.label or "",
                    ),
                )
            ),
        )

    def _statements(
        self,
        statements: Iterable[Node],
        incoming: tuple[_Tail, ...],
        *,
        control: _Control | None,
    ) -> tuple[_Tail, ...]:
        tails = incoming
        for statement in statements:
            if not tails:
                break
            tails = self._statement(statement, tails, control=control)
        return tails

    def _statement(
        self,
        statement: Node,
        incoming: tuple[_Tail, ...],
        *,
        control: _Control | None,
    ) -> tuple[_Tail, ...]:
        if statement.type in _TRIVIA_NODE_TYPES:
            return incoming
        if statement.type in self.profile.block_types:
            return self._statements(statement.named_children, incoming, control=control)
        if statement.type == "if_statement":
            return self._if(statement, incoming, control=control)
        if statement.type in self.profile.loop_types:
            return self._loop(statement, incoming)
        if statement.type == "switch_statement":
            return self._switch(statement, incoming, control=control)
        if statement.type == "try_statement":
            return self._try(statement, incoming, control=control)
        if statement.type in self.profile.return_types:
            value = next(iter(statement.named_children), None)
            label = "return" if value is None else f"return {_compact(_text(value, self.raw))}"
            node_id = self._node("return", label, statement)
            self._connect(incoming, node_id)
            self._edge(node_id, "return", self.exit_id, "RETURN")
            return ()
        if statement.type in self.profile.raise_types:
            value = next(iter(statement.named_children), None)
            keyword = "throw" if self.profile.name in {"javascript", "cpp"} else "raise"
            label = keyword if value is None else f"{keyword} {_compact(_text(value, self.raw))}"
            node_id = self._node("raise", label, statement)
            self._connect(incoming, node_id)
            self._edge(node_id, "raise", self.exit_id, keyword.upper())
            return ()
        if statement.type == "break_statement":
            node_id = self._node("break", "break", statement)
            self._connect(incoming, node_id)
            target = control.break_id if control is not None else None
            self._edge(node_id, "break" if target else "next", target or self.exit_id, "BREAK")
            return ()
        if statement.type == "continue_statement":
            node_id = self._node("continue", "continue", statement)
            self._connect(incoming, node_id)
            target = control.continue_id if control is not None else None
            self._edge(
                node_id,
                "continue" if target else "next",
                target or self.exit_id,
                "CONTINUE",
            )
            return ()
        if statement.type in {"function_declaration", "function_definition", "method_definition"}:
            return incoming
        if statement.type in {"else_clause", "finally_clause", "catch_clause"}:
            body = statement.child_by_field_name("body")
            return (
                self._statement(body, incoming, control=control) if body is not None else incoming
            )
        node_id = self._node(
            "call" if _contains_type(statement, "call_expression") else "action",
            _compact(_text(statement, self.raw)),
            statement,
        )
        self._connect(incoming, node_id)
        return (_Tail(node_id),)

    def _if(
        self,
        statement: Node,
        incoming: tuple[_Tail, ...],
        *,
        control: _Control | None,
    ) -> tuple[_Tail, ...]:
        expression = statement.child_by_field_name("condition")
        if expression is None:
            expression = _first_named(statement)
        condition = self._condition(_unwrap_condition(expression), incoming)
        consequence = statement.child_by_field_name("consequence")
        alternative = statement.child_by_field_name("alternative")
        body_tails = (
            self._statement(consequence, condition.when_true, control=control)
            if consequence is not None
            else condition.when_true
        )
        else_tails = (
            self._statement(alternative, condition.when_false, control=control)
            if alternative is not None
            else condition.when_false
        )
        return self._converge("Decision convergence", (*body_tails, *else_tails), statement)

    def _loop(self, statement: Node, incoming: tuple[_Tail, ...]) -> tuple[_Tail, ...]:
        after_id = self._node("convergence", "Loop exit", statement)
        condition_node = statement.child_by_field_name("condition")
        body = statement.child_by_field_name("body")
        if statement.type in {"for_in_statement", "for_range_loop"}:
            loop_id = self._node(
                "loop", _compact(_header_text(statement, body, self.raw)), statement
            )
            self._connect(incoming, loop_id)
            condition = _Condition(
                loop_id,
                (_Tail(loop_id, "when_true", "ITEM"),),
                (_Tail(loop_id, "when_false", "EXHAUSTED"),),
            )
        elif condition_node is not None:
            condition = self._condition(_unwrap_condition(condition_node), incoming)
            loop_id = condition.entry_id
        else:
            loop_id = self._node(
                "loop", _compact(_header_text(statement, body, self.raw)), statement
            )
            self._connect(incoming, loop_id)
            condition = _Condition(
                loop_id,
                (_Tail(loop_id, "when_true", "ITERATE"),),
                (_Tail(loop_id, "when_false", "EXIT"),),
            )
        control = _Control(break_id=after_id, continue_id=loop_id)
        body_tails = (
            self._statement(body, condition.when_true, control=control)
            if body is not None
            else condition.when_true
        )
        for tail in body_tails:
            self._edge(tail.source_id, "loop", loop_id, "NEXT ITERATION")
        self._connect(condition.when_false, after_id)
        return (_Tail(after_id),) if self._has_incoming(after_id) else ()

    def _switch(
        self,
        statement: Node,
        incoming: tuple[_Tail, ...],
        *,
        control: _Control | None,
    ) -> tuple[_Tail, ...]:
        expression = (
            statement.child_by_field_name("value")
            or statement.child_by_field_name("condition")
            or _first_named(statement)
        )
        switch_id = self._node(
            "condition",
            f"switch {_compact(_text(_unwrap_condition(expression), self.raw))}",
            statement,
        )
        self._connect(incoming, switch_id)
        body = statement.child_by_field_name("body")
        cases = [
            child
            for child in (body.named_children if body is not None else ())
            if child.type in self.profile.switch_case_types
        ]
        convergence_id = self._node("convergence", "Case convergence", statement)
        switch_control = _Control(
            break_id=convergence_id,
            continue_id=control.continue_id if control is not None else None,
        )
        completed: list[_Tail] = []
        for case in cases:
            case_value = case.child_by_field_name("value")
            label = (
                "default" if case_value is None else f"case {_compact(_text(case_value, self.raw))}"
            )
            case_id = self._node("case", label, case)
            self._edge(switch_id, "case", case_id, label.upper())
            body_nodes = tuple(
                child
                for child in case.named_children
                if case_value is None or child.id != case_value.id
            )
            completed.extend(
                self._statements(body_nodes, (_Tail(case_id),), control=switch_control)
            )
        self._connect(tuple(completed), convergence_id)
        return (_Tail(convergence_id),) if self._has_incoming(convergence_id) else ()

    def _try(
        self,
        statement: Node,
        incoming: tuple[_Tail, ...],
        *,
        control: _Control | None,
    ) -> tuple[_Tail, ...]:
        try_id = self._node("try", "try", statement)
        self._connect(incoming, try_id)
        body = statement.child_by_field_name("body")
        normal = (
            self._statement(body, (_Tail(try_id),), control=control)
            if body is not None
            else (_Tail(try_id),)
        )
        branches: list[_Tail] = list(normal)
        handlers = [child for child in statement.named_children if child.type == "catch_clause"]
        handler = statement.child_by_field_name("handler")
        if handler is not None and handler not in handlers:
            handlers.append(handler)
        for catch in handlers:
            parameter = catch.child_by_field_name("parameter") or catch.child_by_field_name(
                "parameters"
            )
            label = (
                "catch" if parameter is None else f"catch {_compact(_text(parameter, self.raw))}"
            )
            catch_id = self._node("except", label, catch)
            self._edge(try_id, "exception", catch_id, label.upper())
            catch_body = catch.child_by_field_name("body")
            branches.extend(
                self._statement(catch_body, (_Tail(catch_id),), control=control)
                if catch_body is not None
                else (_Tail(catch_id),)
            )
        converged = self._converge("Exception convergence", tuple(branches), statement)
        finalizer = statement.child_by_field_name("finalizer")
        if finalizer is None:
            finalizer = next(
                (child for child in statement.named_children if child.type == "finally_clause"),
                None,
            )
        if finalizer is None:
            return converged
        final_id = self._node("finally", "finally", finalizer)
        self._connect(converged, final_id)
        final_body = finalizer.child_by_field_name("body")
        return (
            self._statement(final_body, (_Tail(final_id),), control=control)
            if final_body is not None
            else (_Tail(final_id),)
        )

    def _condition(
        self,
        expression: Node,
        incoming: tuple[_Tail, ...],
    ) -> _Condition:
        expression = _unwrap_condition(expression)
        text = _text(expression, self.raw).strip()
        if expression.type == "unary_expression" and text.startswith("!"):
            operand = next(iter(expression.named_children), None)
            if operand is not None:
                inner = self._condition(operand, incoming)
                return _Condition(inner.entry_id, inner.when_false, inner.when_true)
        if expression.type == "binary_expression":
            left = expression.child_by_field_name("left")
            right = expression.child_by_field_name("right")
            operator = _operator_between(left, right, self.raw)
            if left is not None and right is not None and operator in {"&&", "||"}:
                first = self._condition(left, incoming)
                if operator == "&&":
                    second = self._condition(right, first.when_true)
                    return _Condition(
                        first.entry_id,
                        second.when_true,
                        (*first.when_false, *second.when_false),
                    )
                second = self._condition(right, first.when_false)
                return _Condition(
                    first.entry_id,
                    (*first.when_true, *second.when_true),
                    second.when_false,
                )
        node_id = self._node("condition", _compact(text), expression)
        self._connect(incoming, node_id)
        return _Condition(
            node_id,
            (_Tail(node_id, "when_true", "TRUE"),),
            (_Tail(node_id, "when_false", "FALSE"),),
        )

    def _expression_body(
        self,
        expression: Node,
        incoming: tuple[_Tail, ...],
    ) -> tuple[_Tail, ...]:
        if expression.type == "ternary_expression":
            condition_node = expression.child_by_field_name("condition")
            consequence = expression.child_by_field_name("consequence")
            alternative = expression.child_by_field_name("alternative")
            if condition_node is not None and consequence is not None and alternative is not None:
                condition = self._condition(condition_node, incoming)
                true_id = self._node(
                    "return",
                    f"return {_compact(_text(consequence, self.raw))}",
                    consequence,
                )
                false_id = self._node(
                    "return",
                    f"return {_compact(_text(alternative, self.raw))}",
                    alternative,
                )
                self._connect(condition.when_true, true_id)
                self._connect(condition.when_false, false_id)
                self._edge(true_id, "return", self.exit_id, "RETURN")
                self._edge(false_id, "return", self.exit_id, "RETURN")
                return ()
        node_id = self._node(
            "return",
            f"return {_compact(_text(expression, self.raw))}",
            expression,
        )
        self._connect(incoming, node_id)
        self._edge(node_id, "return", self.exit_id, "RETURN")
        return ()

    def _converge(
        self,
        label: str,
        incoming: tuple[_Tail, ...],
        source: Node,
    ) -> tuple[_Tail, ...]:
        if not incoming:
            return ()
        convergence_id = self._node("convergence", label, source)
        self._connect(incoming, convergence_id)
        return (_Tail(convergence_id),)

    def _node(self, kind: str, label: str, source: Node) -> str:
        if len(self.nodes) >= self.max_nodes:
            raise DocForgeError(
                "logic_too_large",
                "A function exceeds the configured logic-node safety boundary",
                owner_node_id=self.owner_node_id,
                maximum=self.max_nodes,
            )
        self._sequence += 1
        line = source.start_point.row + 1
        column = source.start_point.column
        logic_id = f"logic.{self._owner_digest}.{kind}.{line}.{column}.{self._sequence}"
        self.nodes.append(
            LogicNode(
                logic_id=logic_id,
                kind=kind,
                label=label,
                source_anchor=f"L{line}",
            )
        )
        return logic_id

    def _connect(self, incoming: tuple[_Tail, ...], target_id: str) -> None:
        for tail in incoming:
            self._edge(tail.source_id, tail.relation, target_id, tail.label)

    def _edge(
        self,
        source_id: str,
        relation: str,
        target_id: str,
        label: str | None,
    ) -> None:
        ordinal = self._edge_ordinals.get(source_id, 0)
        self._edge_ordinals[source_id] = ordinal + 1
        self.edges.append(
            LogicEdge(
                source_id=source_id,
                relation=relation,
                target_id=target_id,
                label=label,
                ordinal=ordinal,
            )
        )

    def _has_incoming(self, node_id: str) -> bool:
        return any(edge.target_id == node_id for edge in self.edges)


def _field_text(node: Node, field: str, raw: bytes) -> str | None:
    child = node.child_by_field_name(field)
    return _text(child, raw) if child is not None else None


def _first_named(node: Node) -> Node:
    return node.named_children[0] if node.named_children else node


def _unwrap_condition(node: Node) -> Node:
    current = node
    while current.type in {"parenthesized_expression", "condition_clause"}:
        value = current.child_by_field_name("value")
        current = value or _first_named(current)
    return current


def _operator_between(left: Node | None, right: Node | None, raw: bytes) -> str:
    if left is None or right is None:
        return ""
    return raw[left.end_byte : right.start_byte].decode("utf-8", errors="replace").strip()


def _header_text(statement: Node, body: Node | None, raw: bytes) -> str:
    end = body.start_byte if body is not None else statement.end_byte
    return raw[statement.start_byte : end].decode("utf-8", errors="replace").strip()


def _contains_type(node: Node, node_type: str) -> bool:
    if node.type == node_type:
        return True
    return any(_contains_type(child, node_type) for child in node.named_children)


def _text(node: Node, raw: bytes) -> str:
    return raw[node.start_byte : node.end_byte].decode("utf-8", errors="replace")


def _compact(value: str, limit: int = 240) -> str:
    compact = " ".join(value.strip().split())
    return compact if len(compact) <= limit else f"{compact[: limit - 1]}…"