rose-ash/shared/sx/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)::

    {{ sx('(~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.sx.jinja_bridge import setup_sx_bridge
    setup_sx_bridge(app)  # call after setup_jinja(app)
"""

from __future__ import annotations

import glob
import hashlib
import os
from typing import Any

from .types import NIL, Component, Keyword, Macro, 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] = {}

# SHA-256 hash (12 hex chars) of all component definitions — used for
# client-side localStorage caching.
_COMPONENT_HASH: str = ""


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


def get_component_hash() -> str:
    """Return the current component definitions hash."""
    return _COMPONENT_HASH


def _compute_component_hash() -> None:
    """Recompute _COMPONENT_HASH from all registered Component and Macro definitions."""
    global _COMPONENT_HASH
    from .parser import serialize
    parts = []
    for key in sorted(_COMPONENT_ENV):
        val = _COMPONENT_ENV[key]
        if isinstance(val, Component):
            param_strs = ["&key"] + list(val.params)
            if val.has_children:
                param_strs.extend(["&rest", "children"])
            params_sx = "(" + " ".join(param_strs) + ")"
            body_sx = serialize(val.body)
            parts.append(f"(defcomp ~{val.name} {params_sx} {body_sx})")
        elif isinstance(val, Macro):
            param_strs = list(val.params)
            if val.rest_param:
                param_strs.extend(["&rest", val.rest_param])
            params_sx = "(" + " ".join(param_strs) + ")"
            body_sx = serialize(val.body)
            parts.append(f"(defmacro {val.name} {params_sx} {body_sx})")
    if parts:
        digest = hashlib.sha256("\n".join(parts).encode()).hexdigest()[:12]
        _COMPONENT_HASH = digest
    else:
        _COMPONENT_HASH = ""


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


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

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


def watch_sx_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, "*.sx"))
    ):
        _file_mtimes[fp] = os.path.getmtime(fp)


def reload_if_changed() -> None:
    """Re-read sx 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, "*.sx"))
        ):
            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_sx_dir(directory)


def load_service_components(service_dir: str, service_name: str | None = None) -> None:
    """Load service-specific s-expression components and handlers.

    Components from ``{service_dir}/sx/`` and handlers from
    ``{service_dir}/sx/handlers/`` or ``{service_dir}/sx/handlers.sx``.
    """
    sx_dir = os.path.join(service_dir, "sx")
    if os.path.isdir(sx_dir):
        load_sx_dir(sx_dir)
        watch_sx_dir(sx_dir)

    # Load handler definitions if service_name is provided
    if service_name:
        load_handler_dir(os.path.join(sx_dir, "handlers"), service_name)
        # Also check for a single handlers.sx file
        handlers_file = os.path.join(sx_dir, "handlers.sx")
        if os.path.isfile(handlers_file):
            from .handlers import load_handler_file
            load_handler_file(handlers_file, service_name)


def load_handler_dir(directory: str, service_name: str) -> None:
    """Load handler .sx files from a directory if it exists."""
    if os.path.isdir(directory):
        from .handlers import load_handler_dir as _load
        _load(directory, service_name)


def register_components(sx_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
    from .css_registry import scan_classes_from_sx

    # Snapshot existing component names before eval
    existing = set(_COMPONENT_ENV.keys())

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

    # Pre-scan CSS classes for newly registered components.
    # Scan the full source once — components from the same file share the set.
    # Slightly over-counts per component but safe and avoids re-scanning at request time.
    all_classes: set[str] | None = None
    for key, val in _COMPONENT_ENV.items():
        if key not in existing and isinstance(val, Component):
            if all_classes is None:
                all_classes = scan_classes_from_sx(sx_source)
            val.css_classes = set(all_classes)

    _compute_component_hash()


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

def sx(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::

        {{ sx('(~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 ``sx_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 sx kebab-case.
    No sx 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 sx_async(source: str, **kwargs: Any) -> str:
    """Async version of ``sx()`` — resolves I/O primitives (frag, query)
    before rendering.

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

        {{ sx_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("SX-Request") or request.headers.get("HX-Request"))
        return RequestContext(user=user, is_htmx=is_htmx)
    except Exception:
        return RequestContext()


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

def client_components_tag(*names: str) -> str:
    """Emit a <script type="text/sx"> tag with component definitions.

    Reads the source definitions from loaded .sx files and sends them
    to the client so sx.js can render them identically.

    Usage in Python::

        body_end_html = client_components_tag("test-filter-card", "test-row")

    Or send all loaded components::

        body_end_html = client_components_tag()
    """
    from .parser import serialize
    parts = []
    for key, val in _COMPONENT_ENV.items():
        if isinstance(val, Component):
            if names and val.name not in names and key.lstrip("~") not in names:
                continue
            # Reconstruct defcomp source from the Component object
            param_strs = ["&key"] + list(val.params)
            if val.has_children:
                param_strs.extend(["&rest", "children"])
            params_sx = "(" + " ".join(param_strs) + ")"
            body_sx = serialize(val.body, pretty=True)
            parts.append(f"(defcomp ~{val.name} {params_sx} {body_sx})")
        elif isinstance(val, Macro):
            if names and val.name not in names:
                continue
            param_strs = list(val.params)
            if val.rest_param:
                param_strs.extend(["&rest", val.rest_param])
            params_sx = "(" + " ".join(param_strs) + ")"
            body_sx = serialize(val.body, pretty=True)
            parts.append(f"(defmacro {val.name} {params_sx} {body_sx})")
    if not parts:
        return ""
    source = "\n".join(parts)
    return f'<script type="text/sx" data-components>{source}</script>'


def sx_css_all() -> str:
    """Return all CSS rules (preamble + utilities) for Jinja fallback pages."""
    from .css_registry import get_all_css
    return get_all_css()


def setup_sx_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.sx.jinja_bridge import setup_sx_bridge
        setup_sx_bridge(app)

    This registers:
    - ``sx(source, **kwargs)`` — sync render (components, pure HTML)
    - ``sx_async(source, **kwargs)`` — async render (with I/O resolution)
    - ``sx_css_all()`` — full CSS dump for non-sx pages
    """
    app.jinja_env.globals["sx"] = sx
    app.jinja_env.globals["render"] = render
    app.jinja_env.globals["sx_async"] = sx_async
    app.jinja_env.globals["sx_css_all"] = sx_css_all