cobol-java-v3/cobol_testgen/pipeline_bridge.py

"""Bridge: procedure_parser -> BrSeq/BrIf/BrEval tree pipeline integration.

Primary: new procedure_parser (fast, deterministic, no path explosion).
Fallback: old BrParser (timeout-guarded for programs new parser can't handle).
"""

import logging
from .models import BrSeq, BrIf, BrEval, BrPerform, BrSearch, GoTo

logger = logging.getLogger(__name__)
from .procedure_parser import extract_branch_tree as new_parse, BranchNode


def build_branch_tree_fallback(proc_text, fields=None):
    """New parser primary with old parser timeout fallback."""
    from .core import build_branch_tree as old_build

    # 1. New parser (fast, 10-50ms, no DP cap limit)
    new_tree, new_assigns = None, {}
    try:
        root, assigns_list = new_parse(proc_text, fields)
        new_tree = _convert_to_model(root)
        new_assigns = _assigns_list_to_dict(assigns_list)
    except Exception as e:
        logger.warning(f"New parser failed: {e}")
        pass

    # New parser generates O(N) paths (not O(2^N)), so no cap needed.
    # Just use it directly when it works.
    if new_tree is not None:
        return new_tree, new_assigns

    # 2. Old parser with 3s timeout (fallback only)
    old_tree, old_assigns = None, {}
    try:
        import threading
        r, e, d = [None], [None], [False]
        def run():
            try:
                r[0] = old_build(proc_text, fields)
            except Exception as ex:
                e[0] = ex
            d[0] = True
        t = threading.Thread(target=run, daemon=True)
        t.start(); t.join(3.0)
        if d[0] and not e[0] and r[0]:
            ot, oa = r[0]
            old_tree, old_assigns = ot, oa
    except Exception as e:
        logger.warning(f"Old parser fallback failed: {e}")
        pass

    if old_tree is not None:
        return old_tree, old_assigns
    return BrSeq(), {}


def _convert_to_model(root: BranchNode) -> BrSeq:
    seq = BrSeq()
    for c in root.children:
        _convert_node(c, seq)
    return seq


def _convert_node(node: BranchNode, parent: BrSeq):
    k = node.kind

    if k in ("PARAGRAPH", "SECTION", "PERFORM_CALL", "GO_TO", "EXIT", "CALL"):
        for c in node.children:
            _convert_node(c, parent)
        return

    if k == "IF":
        br = BrIf(node.condition_text or " ".join(node.branch_names))
        for c in node.children:
            if c.kind == "ELSE":
                for ec in c.children: _convert_node(ec, br.false_seq)
            elif c.kind == "THEN":
                for sc in c.children: _convert_node(sc, br.true_seq)
            else:
                _convert_node(c, br.true_seq)
        parent.add(br)
        return

    if k == "EVALUATE":
        subj = (node.branch_names or [""])[0]
        subj = subj[5:-1] if subj.startswith("EVAL(") and subj.endswith(")") else subj
        br = BrEval(subj)
        for c in node.children:
            if c.kind == "WHEN":
                cond = (c.branch_names or [""])[0]
                cond = cond[5:-1] if cond.startswith("WHEN(") and cond.endswith(")") else cond
                # Strip trailing body text at first COBOL verb
                for verb in ('DISPLAY', 'MOVE', 'ADD', 'SUBTRACT', 'MULTIPLY', 'DIVIDE', 'COMPUTE',
                             'STRING', 'UNSTRING', 'SET', 'INSPECT', 'INITIALIZE', 'CONTINUE',
                             'PERFORM', 'CALL', 'EXIT', 'GOBACK', 'STOP',
                             'READ', 'WRITE', 'DELETE', 'REWRITE', 'ACCEPT', 'OPEN', 'CLOSE'):
                    idx = cond.upper().find(f' {verb} ')
                    if idx >= 0:
                        cond = cond[:idx].strip()
                        break
                ws = BrSeq()
                for wc in c.children: _convert_node(wc, ws)
                if cond.upper() == "OTHER":
                    br.has_other = True
                    for wc in c.children: _convert_node(wc, br.other_seq)
                else:
                    br.when_list.append((cond, ws))
        parent.add(br)
        return

    if k == "PERFORM":
        cond = node.condition_text or ""
        br_names = [b.upper() for b in node.branch_names] if node.branch_names else []
        if any('VARY' in b for b in br_names):
            br = BrPerform("varying", condition=cond)
        elif any('SKIP' in b or 'ENTER' in b for b in br_names):
            br = BrPerform("until", condition=cond)
        elif any('TIMES' in b for b in br_names):
            br = BrPerform("times", condition=cond)
        else:
            br = BrPerform("until", condition=cond)
        for c in node.children: _convert_node(c, br.body_seq)
        parent.add(br)
        return

    if k == "READ":
        for c in node.children: _convert_node(c, parent)
        return

    if k == "AT_END":
        br = BrIf("AT END")
        for c in node.children: _convert_node(c, br.true_seq)
        parent.add(br)
        return

    if k == "NOT_AT_END":
        for i in range(len(parent.children) - 1, -1, -1):
            if isinstance(parent.children[i], BrIf):
                for c in node.children:
                    _convert_node(c, parent.children[i].false_seq)
                break
        return

    if k in ("SORT", "MERGE", "WHEN"):
        name = " ".join(node.branch_names) if node.branch_names else k
        parent.add(BrPerform("sort", condition=name))
        return

    if k == "GO_TO_DEPENDING":
        parent.add(GoTo("DEPENDING"))
        return

    for c in node.children:
        _convert_node(c, parent)


def _count_br_nodes(node) -> int:
    count = 0
    if isinstance(node, (BrIf, BrEval, BrPerform, BrSearch)):
        count += 1
    if isinstance(node, BrSeq):
        for c in node.children: count += _count_br_nodes(c)
    if isinstance(node, BrIf):
        count += _count_br_nodes(node.true_seq) + _count_br_nodes(node.false_seq)
    if isinstance(node, BrEval):
        for _, s in node.when_list: count += _count_br_nodes(s)
        count += _count_br_nodes(node.other_seq)
    if isinstance(node, BrPerform):
        count += _count_br_nodes(node.body_seq)
    if isinstance(node, BrSearch):
        count += _count_br_nodes(node.at_end_seq)
        for _, s in node.when_list: count += _count_br_nodes(s)
    return count


def _assigns_list_to_dict(assigns_list: list) -> dict:
    result = {}
    for a in assigns_list:
        tgt = a.get("tgt", "")
        src = a.get("src") or a.get("source_vars")
        if tgt and src:
            a_norm = dict(a)
            if 'type' in a_norm:
                a_norm['type'] = a_norm['type'].lower()
            if 'src' in a_norm and not a_norm.get('source_vars'):
                a_norm['source_vars'] = [a_norm['src']]
            result[tgt] = [a_norm]
    return result