""" Shared helper functions for s-expression page rendering. These are used by per-service sxc/pages modules to build common page elements (headers, search, etc.) from template context. """ from __future__ import annotations import hashlib from pathlib import Path from typing import Any from markupsafe import escape from .page import SEARCH_HEADERS_MOBILE, SEARCH_HEADERS_DESKTOP from .parser import SxExpr def _sx_fragment(*parts: str) -> SxExpr: """Wrap pre-rendered SX wire format strings in a fragment. Infrastructure utility for composing already-serialized SX strings. NOT for building SX from Python data — use sx_call() or _render_to_sx(). """ joined = " ".join(p for p in parts if p) return SxExpr(f"(<> {joined})") if joined else SxExpr("") def call_url(ctx: dict, key: str, path: str = "/") -> str: """Call a URL helper from context (e.g., blog_url, account_url).""" fn = ctx.get(key) if callable(fn): return fn(path) return str(fn or "") + path def get_asset_url(ctx: dict) -> str: """Extract the asset URL base from context.""" au = ctx.get("asset_url") if callable(au): result = au("") return result.rsplit("/", 1)[0] if "/" in result else result return au or "" # --------------------------------------------------------------------------- # Sx-native helper functions — return sx source (not HTML) # --------------------------------------------------------------------------- def _as_sx(val: Any) -> SxExpr | None: """Coerce a fragment value to SxExpr. If *val* is already a ``SxExpr`` (from a ``text/sx`` fragment), return it as-is. If it's a non-empty string (HTML from a ``text/html`` fragment), wrap it in ``~rich-text``. Otherwise return ``None``. """ if not val: return None if isinstance(val, SxExpr): return val if val.source else None html = str(val) return sx_call("rich-text", html=html) async def root_header_sx(ctx: dict, *, oob: bool = False) -> str: """Build the root header row as sx wire format.""" rights = ctx.get("rights") or {} is_admin = rights.get("admin") if isinstance(rights, dict) else getattr(rights, "admin", False) settings_url = call_url(ctx, "blog_url", "/settings/") if is_admin else "" return await _render_to_sx("header-row-sx", cart_mini=_as_sx(ctx.get("cart_mini")), blog_url=call_url(ctx, "blog_url", ""), site_title=ctx.get("base_title", ""), app_label=ctx.get("app_label", ""), nav_tree=_as_sx(ctx.get("nav_tree")), auth_menu=_as_sx(ctx.get("auth_menu")), nav_panel=_as_sx(ctx.get("nav_panel")), settings_url=settings_url, is_admin=is_admin, oob=oob, ) def mobile_menu_sx(*sections: str) -> SxExpr: """Assemble mobile menu from pre-built sections (deepest first).""" parts = [s for s in sections if s] return _sx_fragment(*parts) if parts else SxExpr("") async def mobile_root_nav_sx(ctx: dict) -> str: """Root-level mobile nav via ~mobile-root-nav component.""" nav_tree = ctx.get("nav_tree") or "" auth_menu = ctx.get("auth_menu") or "" if not nav_tree and not auth_menu: return "" return await _render_to_sx("mobile-root-nav", nav_tree=_as_sx(nav_tree), auth_menu=_as_sx(auth_menu), ) # --------------------------------------------------------------------------- # Shared nav-item builders — used by BOTH desktop headers and mobile menus # --------------------------------------------------------------------------- async def _post_nav_items_sx(ctx: dict) -> SxExpr: """Build post-level nav items (container_nav + admin cog). Shared by ``post_header_sx`` (desktop) and ``post_mobile_nav_sx`` (mobile).""" post = ctx.get("post") or {} slug = post.get("slug", "") if not slug: return SxExpr("") parts: list[str] = [] page_cart_count = ctx.get("page_cart_count", 0) if page_cart_count and page_cart_count > 0: cart_href = call_url(ctx, "cart_url", f"/{slug}/") parts.append(await _render_to_sx("page-cart-badge", href=cart_href, count=str(page_cart_count))) container_nav = str(ctx.get("container_nav") or "").strip() # Skip empty fragment wrappers like "(<> )" if container_nav and container_nav.replace("(<>", "").replace(")", "").strip(): parts.append(await _render_to_sx("container-nav-wrapper", content=SxExpr(container_nav))) # Admin cog admin_nav = ctx.get("post_admin_nav") if not admin_nav: rights = ctx.get("rights") or {} has_admin = rights.get("admin") if isinstance(rights, dict) else getattr(rights, "admin", False) if has_admin and slug: from quart import request admin_href = call_url(ctx, "blog_url", f"/{slug}/admin/") is_admin_page = ctx.get("is_admin_section") or "/admin" in request.path admin_nav = await _render_to_sx("admin-cog-button", href=admin_href, is_admin_page=is_admin_page or None) if admin_nav: parts.append(admin_nav) return _sx_fragment(*parts) if parts else SxExpr("") async def _post_admin_nav_items_sx(ctx: dict, slug: str, selected: str = "") -> SxExpr: """Build post-admin nav items (calendars, markets, etc.). Shared by ``post_admin_header_sx`` (desktop) and mobile menu.""" select_colours = ctx.get("select_colours", "") parts: list[str] = [] items = [ ("events_url", f"/{slug}/admin/", "calendars"), ("market_url", f"/{slug}/admin/", "markets"), ("cart_url", f"/{slug}/admin/payments/", "payments"), ("blog_url", f"/{slug}/admin/entries/", "entries"), ("blog_url", f"/{slug}/admin/data/", "data"), ("blog_url", f"/{slug}/admin/preview/", "preview"), ("blog_url", f"/{slug}/admin/edit/", "edit"), ("blog_url", f"/{slug}/admin/settings/", "settings"), ] for url_key, path, label in items: url_fn = ctx.get(url_key) if not callable(url_fn): continue href = url_fn(path) is_sel = label == selected parts.append(await _render_to_sx("nav-link", href=href, label=label, select_colours=select_colours, is_selected=is_sel or None)) return _sx_fragment(*parts) if parts else SxExpr("") # --------------------------------------------------------------------------- # Mobile menu section builders — wrap shared nav items for hamburger panel # --------------------------------------------------------------------------- async def post_mobile_nav_sx(ctx: dict) -> str: """Post-level mobile menu section.""" nav = await _post_nav_items_sx(ctx) if not nav: return "" post = ctx.get("post") or {} slug = post.get("slug", "") title = (post.get("title") or slug)[:40] return await _render_to_sx("mobile-menu-section", label=title, href=call_url(ctx, "blog_url", f"/{slug}/"), level=1, items=nav, ) async def search_mobile_sx(ctx: dict) -> str: """Build mobile search input as sx wire format.""" return await _render_to_sx("search-mobile", current_local_href=ctx.get("current_local_href", "/"), search=ctx.get("search", ""), search_count=ctx.get("search_count", ""), hx_select=ctx.get("hx_select", "#main-panel"), search_headers_mobile=SEARCH_HEADERS_MOBILE, ) async def search_desktop_sx(ctx: dict) -> str: """Build desktop search input as sx wire format.""" return await _render_to_sx("search-desktop", current_local_href=ctx.get("current_local_href", "/"), search=ctx.get("search", ""), search_count=ctx.get("search_count", ""), hx_select=ctx.get("hx_select", "#main-panel"), search_headers_desktop=SEARCH_HEADERS_DESKTOP, ) async def post_header_sx(ctx: dict, *, oob: bool = False, child: str = "") -> str: """Build the post-level header row as sx wire format.""" post = ctx.get("post") or {} slug = post.get("slug", "") if not slug: return "" title = (post.get("title") or "")[:160] feature_image = post.get("feature_image") label_sx = await _render_to_sx("post-label", feature_image=feature_image, title=title) nav_sx = await _post_nav_items_sx(ctx) or None link_href = call_url(ctx, "blog_url", f"/{slug}/") return await _render_to_sx("menu-row-sx", id="post-row", level=1, link_href=link_href, link_label_content=label_sx, nav=nav_sx, child_id="post-header-child", child=SxExpr(child) if child else None, oob=oob, external=True, ) async def post_admin_header_sx(ctx: dict, slug: str, *, oob: bool = False, selected: str = "", admin_href: str = "") -> str: """Post admin header row as sx wire format.""" # Label label_sx = await _render_to_sx("post-admin-label", selected=str(escape(selected)) if selected else None) nav_sx = await _post_admin_nav_items_sx(ctx, slug, selected) or None if not admin_href: blog_fn = ctx.get("blog_url") admin_href = blog_fn(f"/{slug}/admin/") if callable(blog_fn) else f"/{slug}/admin/" return await _render_to_sx("menu-row-sx", id="post-admin-row", level=2, link_href=admin_href, link_label_content=label_sx, nav=nav_sx, child_id="post-admin-header-child", oob=oob, ) async def oob_header_sx(parent_id: str, child_id: str, row_sx: str) -> str: """Wrap a header row sx in an OOB swap. child_id is accepted for call-site compatibility but no longer used — the child placeholder is created by ~menu-row-sx itself. """ return await _render_to_sx("oob-header-sx", parent_id=parent_id, row=SxExpr(row_sx), ) async def header_child_sx(inner_sx: str, *, id: str = "root-header-child") -> str: """Wrap inner sx in a header-child div.""" return await _render_to_sx("header-child-sx", id=id, inner=_sx_fragment(inner_sx), ) async def oob_page_sx(*, oobs: str = "", filter: str = "", aside: str = "", content: str = "", menu: str = "") -> str: """Build OOB response as sx wire format.""" return await _render_to_sx("oob-sx", oobs=_sx_fragment(oobs) if oobs else None, filter=SxExpr(filter) if filter else None, aside=SxExpr(aside) if aside else None, menu=SxExpr(menu) if menu else None, content=SxExpr(content) if content else None, ) async def full_page_sx(ctx: dict, *, header_rows: str, filter: str = "", aside: str = "", content: str = "", menu: str = "", meta_html: str = "", meta: str = "") -> str: """Build a full page using sx_page() with ~app-body. meta_html: raw HTML injected into the
shell (legacy). meta: sx source for meta tags — auto-hoisted to by sx.js. """ # Auto-generate mobile nav from context when no menu provided if not menu: menu = await mobile_root_nav_sx(ctx) body_sx = await _render_to_sx("app-body", header_rows=_sx_fragment(header_rows) if header_rows else None, filter=SxExpr(filter) if filter else None, aside=SxExpr(aside) if aside else None, menu=SxExpr(menu) if menu else None, content=SxExpr(content) if content else None, ) if meta: # Wrap body + meta in a fragment so sx.js renders both; # auto-hoist moves meta/title/link elements to . body_sx = _sx_fragment(meta, body_sx) return sx_page(ctx, body_sx, meta_html=meta_html) def _build_component_ast(__name: str, **kwargs: Any) -> list: """Build an AST list for a component call from Python kwargs. Returns e.g. [Symbol("~card"), Keyword("title"), "hello", Keyword("count"), 3] No SX string generation — values stay as native Python objects. """ from .types import Symbol, Keyword, NIL comp_sym = Symbol(__name if __name.startswith("~") else f"~{__name}") ast: list = [comp_sym] for key, val in kwargs.items(): kebab = key.replace("_", "-") ast.append(Keyword(kebab)) if val is None: ast.append(NIL) elif isinstance(val, SxExpr): # SxExpr values need to be parsed into AST from .parser import parse if not val.source: ast.append(NIL) else: ast.append(parse(val.source)) else: ast.append(val) return ast async def _render_to_sx_with_env(__name: str, extra_env: dict, **kwargs: Any) -> str: """Like ``_render_to_sx`` but merges *extra_env* into the evaluation environment before eval. Used by ``register_sx_layout`` so .sx defcomps can read ctx values as free variables. Uses ``async_eval_slot_to_sx`` (not ``async_eval_to_sx``) so the top-level component body is expanded server-side — free variables from *extra_env* are resolved during expansion rather than being serialized as unresolved symbols for the client. **Private** — service code should use ``sx_call()`` or defmacros instead. """ from .jinja_bridge import get_component_env, _get_request_context import os if os.environ.get("SX_USE_REF") == "1": from .ref.async_eval_ref import async_eval_slot_to_sx else: from .async_eval import async_eval_slot_to_sx from .types import Symbol, Keyword, NIL as _NIL # Build AST with extra_env entries as keyword args so _aser_component # binds them as params (otherwise it defaults all params to NIL). comp_sym = Symbol(__name if __name.startswith("~") else f"~{__name}") ast: list = [comp_sym] for k, v in extra_env.items(): ast.append(Keyword(k)) ast.append(v if v is not None else _NIL) for k, v in kwargs.items(): ast.append(Keyword(k.replace("_", "-"))) ast.append(v if v is not None else _NIL) env = dict(get_component_env()) env.update(extra_env) ctx = _get_request_context() return SxExpr(await async_eval_slot_to_sx(ast, env, ctx)) async def _render_to_sx(__name: str, **kwargs: Any) -> str: """Call a defcomp and get SX wire format back. No SX string literals. Builds an AST from Python values and evaluates it through the SX evaluator, which resolves IO primitives and serializes component/tag calls as SX wire format. **Private** — service code should use ``sx_call()`` or defmacros instead. Only infrastructure code (helpers.py, layouts.py) should call this. """ from .jinja_bridge import get_component_env, _get_request_context import os if os.environ.get("SX_USE_REF") == "1": from .ref.async_eval_ref import async_eval_to_sx else: from .async_eval import async_eval_to_sx ast = _build_component_ast(__name, **kwargs) env = dict(get_component_env()) ctx = _get_request_context() return SxExpr(await async_eval_to_sx(ast, env, ctx)) # Backwards-compat alias — layout infrastructure still imports this. # Will be removed once all layouts use register_sx_layout(). render_to_sx_with_env = _render_to_sx_with_env async def render_to_html(__name: str, **kwargs: Any) -> str: """Call a defcomp and get HTML back. No SX string literals. Same as render_to_sx() but produces HTML output instead of SX wire format. Used by route renders that need HTML (full pages, fragments). """ from .jinja_bridge import get_component_env, _get_request_context import os if os.environ.get("SX_USE_REF") == "1": from .ref.async_eval_ref import async_render else: from .async_eval import async_render ast = _build_component_ast(__name, **kwargs) env = dict(get_component_env()) ctx = _get_request_context() return await async_render(ast, env, ctx) def sx_call(component_name: str, **kwargs: Any) -> str: """Build an s-expression component call string from Python kwargs. Converts snake_case to kebab-case automatically:: sx_call("test-row", nodeid="foo", outcome="passed") # => '(~test-row :nodeid "foo" :outcome "passed")' Values are serialized: strings are quoted, None becomes nil, bools become true/false, numbers stay as-is. List values use ``(list ...)`` so the client gets an iterable array rather than a rendered fragment. """ from .parser import serialize, SxExpr name = component_name if component_name.startswith("~") else f"~{component_name}" parts = [name] for key, val in kwargs.items(): parts.append(f":{key.replace('_', '-')}") if isinstance(val, list): items = [serialize(v) for v in val if v is not None] if not items: parts.append("nil") else: parts.append("(list " + " ".join(items) + ")") else: parts.append(serialize(val)) return SxExpr("(" + " ".join(parts) + ")") def components_for_request(source: str = "") -> str: """Return defcomp/defmacro source for definitions the client doesn't have yet. Reads the ``SX-Components`` header (comma-separated component names like ``~card,~nav-item``) and returns only the definitions the client is missing. If *source* is provided, only sends components needed for that source (plus transitive deps). If the header is absent, returns all needed defs. """ from quart import request from .jinja_bridge import _COMPONENT_ENV from .deps import components_needed from .types import Component, Macro from .parser import serialize # Determine which components the page needs if source: needed = components_needed(source, _COMPONENT_ENV) else: needed = None # all loaded_raw = request.headers.get("SX-Components", "") loaded = set(loaded_raw.split(",")) if loaded_raw else set() parts = [] for key, val in _COMPONENT_ENV.items(): if isinstance(val, Component): comp_name = f"~{val.name}" # Skip if not needed for this page if needed is not None and comp_name not in needed and key not in needed: continue # Skip components the client already has if comp_name in loaded or val.name in loaded: continue # Reconstruct defcomp source 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 val.name in loaded: 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})") return "\n".join(parts) def sx_response(source: str, status: int = 200, headers: dict | None = None): """Return an s-expression wire-format response. Takes a raw sx string:: return sx_response('(~test-row :nodeid "foo")') For SX requests, missing component definitions are prepended as a ``\n{body}') # On-demand CSS: scan source for classes, send only new rules from .css_registry import scan_classes_from_sx, lookup_rules, registry_loaded, lookup_css_hash, store_css_hash new_classes: set[str] = set() cumulative_classes: set[str] = set() if registry_loaded(): new_classes = scan_classes_from_sx(source) if comp_defs: # Scan only the component definitions actually being sent new_classes.update(scan_classes_from_sx(comp_defs)) # Resolve known classes from SX-Css header (hash or full list) known_classes: set[str] = set() known_raw = request.headers.get("SX-Css", "") if known_raw: if len(known_raw) <= 16: # Treat as hash looked_up = lookup_css_hash(known_raw) if looked_up is not None: known_classes = looked_up else: # Cache miss — send all classes (safe fallback) known_classes = set() else: known_classes = set(known_raw.split(",")) cumulative_classes = known_classes | new_classes new_classes -= known_classes if new_classes: new_rules = lookup_rules(new_classes) if new_rules: body = f'\n{body}' # Dev mode: pretty-print sx source for readable Network tab responses if _is_dev_mode(): body = _pretty_print_sx_body(body) resp = Response(body, status=status, content_type="text/sx") if new_classes: resp.headers["SX-Css-Add"] = ",".join(sorted(new_classes)) if cumulative_classes: resp.headers["SX-Css-Hash"] = store_css_hash(cumulative_classes) if headers: for k, v in headers.items(): resp.headers[k] = v return resp # --------------------------------------------------------------------------- # Sx wire-format full page shell # --------------------------------------------------------------------------- _SX_PAGE_TEMPLATE = """\