Function _try_pythonmonkey_parse

codewiki/src/be/utils.py:161–210

Attempt to parse via PythonMonkey/mermaid-parser-py. Returns the extracted parse-error message, "" on success, or None when PythonMonkey itself is unusable (broken JS event loop binding on Python 3.13+) so the caller can fall back to mermaid-py.

(diagram_content: str)

async def _try_pythonmonkey_parse(diagram_content: str) -> str | None:
    """Attempt to parse via PythonMonkey/mermaid-parser-py.

    Returns the extracted parse-error message, "" on success, or None when
    PythonMonkey itself is unusable (broken JS event loop binding on
    Python 3.13+) so the caller can fall back to mermaid-py.
    """
    global _PYTHONMONKEY_BROKEN
    if _PYTHONMONKEY_BROKEN:
        return None

    import os

    try:
        from mermaid_parser.parser import parse_mermaid_py
    except Exception:
        _PYTHONMONKEY_BROKEN = True
        return None

    old_stderr = sys.stderr
    sys.stderr = open(os.devnull, 'w')
    try:
        if (
            _main_loop is not None
            and _main_loop.is_running()
            and threading.get_ident() != _main_loop_thread_ident
        ):
            fut = asyncio.run_coroutine_threadsafe(
                parse_mermaid_py(diagram_content), _main_loop
            )
            await asyncio.wrap_future(fut)
        else:
            await parse_mermaid_py(diagram_content)
        return ""
    except Exception as e:
        error_str = str(e)
        # PythonMonkey 1.3.1 only supports Python 3.8-3.11; on newer Pythons
        # every JS call raises this. Latch the failure once so subsequent
        # diagrams skip the broken path and go straight to mermaid-py.
        if "cannot find a running Python event-loop" in error_str:
            _PYTHONMONKEY_BROKEN = True
            return None
        match = re.search(r"Error:(.*?)(?=Stack Trace:|$)", error_str, re.DOTALL)
        if match:
            return match.group(0).strip()
        # Unknown error from the JS parser — fall back rather than surface it.
        return None
    finally:
        sys.stderr.close()
        sys.stderr = old_stderr


def _parse_via_mermaid_py(diagram_content: str) -> str: