rose-ash/shared/sexp/jinja_bridge.py

"""
Jinja ↔ s-expression bridge.

Provides two-way integration so s-expression components and Jinja templates
can coexist during incremental migration:

**Jinja → s-expression** (use s-expression components inside Jinja templates)::

    {{ sexp('(~link-card :slug "apple" :title "Apple")') | safe }}

**S-expression → Jinja** (embed Jinja output inside s-expressions)::

    (raw! (jinja "fragments/link_card.html" :slug "apple" :title "Apple"))

Setup::

    from shared.sexp.jinja_bridge import setup_sexp_bridge
    setup_sexp_bridge(app)  # call after setup_jinja(app)
"""

from __future__ import annotations

import glob
import os
from typing import Any

from .types import NIL, Component, Keyword, Symbol
from .parser import parse
from .html import render as html_render, _render_component


# ---------------------------------------------------------------------------
# Shared component environment
# ---------------------------------------------------------------------------

# Global component registry — populated at app startup by loading component
# definition files or calling register_components().
_COMPONENT_ENV: dict[str, Any] = {}


def get_component_env() -> dict[str, Any]:
    """Return the shared component environment."""
    return _COMPONENT_ENV


def load_sexp_dir(directory: str) -> None:
    """Load all .sexp and .sexpr files from a directory and register components."""
    for filepath in sorted(
        glob.glob(os.path.join(directory, "*.sexp"))
        + glob.glob(os.path.join(directory, "*.sexpr"))
    ):
        with open(filepath, encoding="utf-8") as f:
            register_components(f.read())


# ---------------------------------------------------------------------------
# Dev-mode auto-reload of sexp templates
# ---------------------------------------------------------------------------

_watched_dirs: list[str] = []
_file_mtimes: dict[str, float] = {}


def watch_sexp_dir(directory: str) -> None:
    """Register a directory for dev-mode file watching."""
    _watched_dirs.append(directory)
    # Seed mtimes
    for fp in sorted(
        glob.glob(os.path.join(directory, "*.sexp"))
        + glob.glob(os.path.join(directory, "*.sexpr"))
    ):
        _file_mtimes[fp] = os.path.getmtime(fp)


def reload_if_changed() -> None:
    """Re-read sexp files if any have changed on disk. Called per-request in dev."""
    changed = False
    for directory in _watched_dirs:
        for fp in sorted(
            glob.glob(os.path.join(directory, "*.sexp"))
            + glob.glob(os.path.join(directory, "*.sexpr"))
        ):
            mtime = os.path.getmtime(fp)
            if fp not in _file_mtimes or _file_mtimes[fp] != mtime:
                _file_mtimes[fp] = mtime
                changed = True
    if changed:
        _COMPONENT_ENV.clear()
        for directory in _watched_dirs:
            load_sexp_dir(directory)


def load_service_components(service_dir: str) -> None:
    """Load service-specific s-expression components from {service_dir}/sexp/."""
    sexp_dir = os.path.join(service_dir, "sexp")
    if os.path.isdir(sexp_dir):
        load_sexp_dir(sexp_dir)
        watch_sexp_dir(sexp_dir)


def register_components(sexp_source: str) -> None:
    """Parse and evaluate s-expression component definitions into the
    shared environment.

    Typically called at app startup::

        register_components('''
            (defcomp ~link-card (&key link title image icon)
              (a :href link :class "block rounded ..."
                (div :class "flex ..."
                  (if image
                    (img :src image :class "...")
                    (div :class "..." (i :class icon)))
                  (div :class "..." (div :class "..." title)))))
        ''')
    """
    from .evaluator import _eval
    from .parser import parse_all

    exprs = parse_all(sexp_source)
    for expr in exprs:
        _eval(expr, _COMPONENT_ENV)


# ---------------------------------------------------------------------------
# sexp() — render s-expression from Jinja template
# ---------------------------------------------------------------------------

def sexp(source: str, **kwargs: Any) -> str:
    """Render an s-expression string to HTML.

    Keyword arguments are merged into the evaluation environment,
    so Jinja context variables can be passed through::

        {{ sexp('(~link-card :title title :slug slug)',
                title=post.title, slug=post.slug) | safe }}

    This is a synchronous function — suitable for Jinja globals.
    For async resolution (with I/O primitives), use ``sexp_async()``.
    """
    env = dict(_COMPONENT_ENV)
    env.update(kwargs)
    expr = parse(source)
    return html_render(expr, env)


def render(component_name: str, **kwargs: Any) -> str:
    """Call a registered component by name with Python kwargs.

    Automatically converts Python snake_case to sexp kebab-case.
    No sexp strings needed — just a function call.
    """
    name = component_name if component_name.startswith("~") else f"~{component_name}"
    comp = _COMPONENT_ENV.get(name)
    if not isinstance(comp, Component):
        raise ValueError(f"Unknown component: {name}")

    env = dict(_COMPONENT_ENV)
    args: list[Any] = []
    for key, val in kwargs.items():
        kw_name = key.replace("_", "-")
        args.append(Keyword(kw_name))
        args.append(val)
        env[kw_name] = val

    return _render_component(comp, args, env)


async def sexp_async(source: str, **kwargs: Any) -> str:
    """Async version of ``sexp()`` — resolves I/O primitives (frag, query)
    before rendering.

    Use when the s-expression contains I/O nodes::

        {{ sexp_async('(frag "blog" "card" :slug "apple")') | safe }}
    """
    from .resolver import resolve, RequestContext

    env = dict(_COMPONENT_ENV)
    env.update(kwargs)
    expr = parse(source)

    # Try to get request context from Quart
    ctx = _get_request_context()
    return await resolve(expr, ctx=ctx, env=env)


def _get_request_context():
    """Build RequestContext from current Quart request, if available."""
    from .primitives_io import RequestContext
    try:
        from quart import g, request
        user = getattr(g, "user", None)
        is_htmx = bool(request.headers.get("HX-Request"))
        return RequestContext(user=user, is_htmx=is_htmx)
    except Exception:
        return RequestContext()


# ---------------------------------------------------------------------------
# Quart integration
# ---------------------------------------------------------------------------

def setup_sexp_bridge(app: Any) -> None:
    """Register s-expression helpers with a Quart app's Jinja environment.

    Call this in your app factory after ``setup_jinja(app)``::

        from shared.sexp.jinja_bridge import setup_sexp_bridge
        setup_sexp_bridge(app)

    This registers:
    - ``sexp(source, **kwargs)`` — sync render (components, pure HTML)
    - ``sexp_async(source, **kwargs)`` — async render (with I/O resolution)
    """
    app.jinja_env.globals["sexp"] = sexp
    app.jinja_env.globals["render"] = render
    app.jinja_env.globals["sexp_async"] = sexp_async