""" Service-specific page context IO handlers. These are application-specific (rose-ash), not part of the generic SX framework. Each handler builds a dict of template data from Quart request context for use by .sx page components. """ from __future__ import annotations from typing import Any from .primitives_io import register_io_handler # --------------------------------------------------------------------------- # Root / post headers # --------------------------------------------------------------------------- @register_io_handler("root-header-ctx") async def _io_root_header_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(root-header-ctx)`` → dict with all root header values. Fetches cart-mini, auth-menu, nav-tree fragments and computes settings-url / is-admin from rights. Result is cached on ``g`` per request so multiple calls (e.g. header + mobile) are free. """ from quart import g, current_app, request cached = getattr(g, "_root_header_ctx", None) if cached is not None: return cached from shared.infrastructure.fragments import fetch_fragments from shared.infrastructure.cart_identity import current_cart_identity from shared.infrastructure.urls import app_url from shared.config import config from .types import NIL user = getattr(g, "user", None) ident = current_cart_identity() cart_params: dict[str, Any] = {} if ident["user_id"] is not None: cart_params["user_id"] = ident["user_id"] if ident["session_id"] is not None: cart_params["session_id"] = ident["session_id"] auth_params: dict[str, Any] = {} if user and getattr(user, "email", None): auth_params["email"] = user.email nav_params = {"app_name": current_app.name, "path": request.path} cart_mini, auth_menu, nav_tree = await fetch_fragments([ ("cart", "cart-mini", cart_params or None), ("account", "auth-menu", auth_params or None), ("blog", "nav-tree", nav_params), ]) rights = getattr(g, "rights", None) or {} is_admin = ( rights.get("admin", False) if isinstance(rights, dict) else getattr(rights, "admin", False) ) result = { "cart-mini": cart_mini or NIL, "blog-url": app_url("blog", ""), "site-title": config()["title"], "app-label": current_app.name, "nav-tree": nav_tree or NIL, "auth-menu": auth_menu or NIL, "nav-panel": NIL, "settings-url": app_url("blog", "/settings/") if is_admin else "", "is-admin": is_admin, } g._root_header_ctx = result return result @register_io_handler("post-header-ctx") async def _io_post_header_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(post-header-ctx)`` → dict with post-level header values.""" from quart import g, request cached = getattr(g, "_post_header_ctx", None) if cached is not None: return cached from shared.infrastructure.urls import app_url from .types import NIL from .parser import SxExpr dctx = getattr(g, "_defpage_ctx", None) or {} post = dctx.get("post") or {} slug = post.get("slug", "") if not slug: result: dict[str, Any] = {"slug": ""} g._post_header_ctx = result return result title = (post.get("title") or "")[:160] feature_image = post.get("feature_image") or NIL # Container nav (pre-fetched by page helper into defpage ctx) raw_nav = dctx.get("container_nav") or "" container_nav: Any = NIL nav_str = str(raw_nav).strip() if nav_str and nav_str.replace("(<>", "").replace(")", "").strip(): if isinstance(raw_nav, SxExpr): container_nav = raw_nav else: container_nav = SxExpr(nav_str) page_cart_count = dctx.get("page_cart_count", 0) or 0 rights = getattr(g, "rights", None) or {} is_admin = ( rights.get("admin", False) if isinstance(rights, dict) else getattr(rights, "admin", False) ) is_admin_page = dctx.get("is_admin_section") or "/admin" in request.path from quart import current_app select_colours = current_app.jinja_env.globals.get("select_colours", "") result = { "slug": slug, "title": title, "feature-image": feature_image, "link-href": app_url("blog", f"/{slug}/"), "container-nav": container_nav, "page-cart-count": page_cart_count, "cart-href": app_url("cart", f"/{slug}/") if page_cart_count else "", "admin-href": app_url("blog", f"/{slug}/admin/"), "is-admin": is_admin, "is-admin-page": is_admin_page or NIL, "select-colours": select_colours, } g._post_header_ctx = result return result # --------------------------------------------------------------------------- # Cart # --------------------------------------------------------------------------- @register_io_handler("cart-page-ctx") async def _io_cart_page_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(cart-page-ctx)`` → dict with cart page header values.""" from quart import g from .types import NIL from shared.infrastructure.urls import app_url page_post = getattr(g, "page_post", None) if not page_post: return {"slug": "", "title": "", "feature-image": NIL, "cart-url": "/"} slug = getattr(page_post, "slug", "") or "" title = (getattr(page_post, "title", "") or "")[:160] feature_image = getattr(page_post, "feature_image", None) or NIL return { "slug": slug, "title": title, "feature-image": feature_image, "page-cart-url": app_url("cart", f"/{slug}/"), "cart-url": app_url("cart", "/"), } # --------------------------------------------------------------------------- # Events # --------------------------------------------------------------------------- @register_io_handler("events-calendar-ctx") async def _io_events_calendar_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(events-calendar-ctx)`` → dict with events calendar header values.""" from quart import g cal = getattr(g, "calendar", None) if not cal: dctx = getattr(g, "_defpage_ctx", None) or {} cal = dctx.get("calendar") if not cal: return {"slug": ""} return { "slug": getattr(cal, "slug", "") or "", "name": getattr(cal, "name", "") or "", "description": getattr(cal, "description", "") or "", } @register_io_handler("events-day-ctx") async def _io_events_day_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(events-day-ctx)`` → dict with events day header values.""" from quart import g, url_for from .types import NIL from .parser import SxExpr dctx = getattr(g, "_defpage_ctx", None) or {} cal = getattr(g, "calendar", None) or dctx.get("calendar") day_date = dctx.get("day_date") or getattr(g, "day_date", None) if not cal or not day_date: return {"date-str": ""} cal_slug = getattr(cal, "slug", "") or "" # Build confirmed entries nav confirmed = dctx.get("confirmed_entries") or [] rights = getattr(g, "rights", None) or {} is_admin = ( rights.get("admin", False) if isinstance(rights, dict) else getattr(rights, "admin", False) ) from .helpers import sx_call nav_parts: list[str] = [] if confirmed: entry_links = [] for entry in confirmed: href = url_for( "calendar.day.calendar_entries.calendar_entry.get", calendar_slug=cal_slug, year=day_date.year, month=day_date.month, day=day_date.day, entry_id=entry.id, ) start = entry.start_at.strftime("%H:%M") if entry.start_at else "" end = ( f" \u2013 {entry.end_at.strftime('%H:%M')}" if entry.end_at else "" ) entry_links.append(sx_call( "events-day-entry-link", href=href, name=entry.name, time_str=f"{start}{end}", )) inner = "".join(entry_links) nav_parts.append(sx_call( "events-day-entries-nav", inner=SxExpr(inner), )) if is_admin and day_date: admin_href = url_for( "defpage_day_admin", calendar_slug=cal_slug, year=day_date.year, month=day_date.month, day=day_date.day, ) nav_parts.append(sx_call("nav-link", href=admin_href, icon="fa fa-cog")) return { "date-str": day_date.strftime("%A %d %B %Y"), "year": day_date.year, "month": day_date.month, "day": day_date.day, "nav": SxExpr("".join(nav_parts)) if nav_parts else NIL, } @register_io_handler("events-entry-ctx") async def _io_events_entry_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(events-entry-ctx)`` → dict with events entry header values.""" from quart import g, url_for from .types import NIL from .parser import SxExpr dctx = getattr(g, "_defpage_ctx", None) or {} cal = getattr(g, "calendar", None) or dctx.get("calendar") entry = getattr(g, "entry", None) or dctx.get("entry") if not cal or not entry: return {"id": ""} cal_slug = getattr(cal, "slug", "") or "" day = dctx.get("day") month = dctx.get("month") year = dctx.get("year") # Times start = entry.start_at end = entry.end_at time_str = "" if start: time_str = start.strftime("%H:%M") if end: time_str += f" \u2192 {end.strftime('%H:%M')}" link_href = url_for( "calendar.day.calendar_entries.calendar_entry.get", calendar_slug=cal_slug, year=year, month=month, day=day, entry_id=entry.id, ) # Build nav: associated posts + admin link entry_posts = dctx.get("entry_posts") or [] rights = getattr(g, "rights", None) or {} is_admin = ( rights.get("admin", False) if isinstance(rights, dict) else getattr(rights, "admin", False) ) from .helpers import sx_call from shared.infrastructure.urls import app_url nav_parts: list[str] = [] if entry_posts: post_links = "" for ep in entry_posts: ep_slug = getattr(ep, "slug", "") ep_title = getattr(ep, "title", "") feat = getattr(ep, "feature_image", None) href = app_url("blog", f"/{ep_slug}/") if feat: img_html = sx_call("events-post-img", src=feat, alt=ep_title) else: img_html = sx_call("events-post-img-placeholder") post_links += sx_call( "events-entry-nav-post-link", href=href, img=SxExpr(img_html), title=ep_title, ) nav_parts.append( sx_call("events-entry-posts-nav-oob", items=SxExpr(post_links)) .replace(' :hx-swap-oob "true"', '') ) if is_admin: admin_url = url_for( "calendar.day.calendar_entries.calendar_entry.admin.admin", calendar_slug=cal_slug, day=day, month=month, year=year, entry_id=entry.id, ) nav_parts.append(sx_call("events-entry-admin-link", href=admin_url)) # Entry admin nav (ticket_types link) admin_href = url_for( "calendar.day.calendar_entries.calendar_entry.admin.admin", calendar_slug=cal_slug, day=day, month=month, year=year, entry_id=entry.id, ) if is_admin else "" ticket_types_href = url_for( "calendar.day.calendar_entries.calendar_entry.ticket_types.get", calendar_slug=cal_slug, entry_id=entry.id, year=year, month=month, day=day, ) from quart import current_app select_colours = current_app.jinja_env.globals.get("select_colours", "") return { "id": str(entry.id), "name": entry.name or "", "time-str": time_str, "link-href": link_href, "nav": SxExpr("".join(nav_parts)) if nav_parts else NIL, "admin-href": admin_href, "ticket-types-href": ticket_types_href, "is-admin": is_admin, "select-colours": select_colours, } @register_io_handler("events-slot-ctx") async def _io_events_slot_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(events-slot-ctx)`` → dict with events slot header values.""" from quart import g dctx = getattr(g, "_defpage_ctx", None) or {} slot = getattr(g, "slot", None) or dctx.get("slot") if not slot: return {"name": ""} return { "name": getattr(slot, "name", "") or "", "description": getattr(slot, "description", "") or "", } @register_io_handler("events-ticket-type-ctx") async def _io_events_ticket_type_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(events-ticket-type-ctx)`` → dict with ticket type header values.""" from quart import g, url_for dctx = getattr(g, "_defpage_ctx", None) or {} cal = getattr(g, "calendar", None) or dctx.get("calendar") entry = getattr(g, "entry", None) or dctx.get("entry") ticket_type = getattr(g, "ticket_type", None) or dctx.get("ticket_type") if not cal or not entry or not ticket_type: return {"id": ""} cal_slug = getattr(cal, "slug", "") or "" day = dctx.get("day") month = dctx.get("month") year = dctx.get("year") link_href = url_for( "calendar.day.calendar_entries.calendar_entry.ticket_types.ticket_type.get", calendar_slug=cal_slug, year=year, month=month, day=day, entry_id=entry.id, ticket_type_id=ticket_type.id, ) return { "id": str(ticket_type.id), "name": getattr(ticket_type, "name", "") or "", "link-href": link_href, } # --------------------------------------------------------------------------- # Market # --------------------------------------------------------------------------- @register_io_handler("market-header-ctx") async def _io_market_header_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(market-header-ctx)`` → dict with market header data.""" from quart import g, url_for from shared.config import config as get_config from .parser import SxExpr cfg = get_config() market_title = cfg.get("market_title", "") link_href = url_for("defpage_market_home") # Get categories if market is loaded market = getattr(g, "market", None) categories = {} if market: from bp.browse.services.nav import get_nav nav_data = await get_nav(g.s, market_id=market.id) categories = nav_data.get("cats", {}) # Build minimal ctx for existing helper functions select_colours = getattr(g, "select_colours", "") if not select_colours: from quart import current_app select_colours = current_app.jinja_env.globals.get("select_colours", "") rights = getattr(g, "rights", None) or {} mini_ctx: dict[str, Any] = { "market_title": market_title, "top_slug": "", "sub_slug": "", "categories": categories, "qs": "", "hx_select_search": "#main-panel", "select_colours": select_colours, "rights": rights, "category_label": "", } # Build header + mobile nav data via new data-driven helpers from sxc.pages.layouts import _market_header_data, _mobile_nav_panel_sx header_data = _market_header_data(mini_ctx) mobile_nav = _mobile_nav_panel_sx(mini_ctx) return { "market-title": market_title, "link-href": link_href, "top-slug": "", "sub-slug": "", "categories": header_data.get("categories", []), "hx-select": header_data.get("hx-select", "#main-panel"), "select-colours": header_data.get("select-colours", ""), "all-href": header_data.get("all-href", ""), "all-active": header_data.get("all-active", False), "admin-href": header_data.get("admin-href", ""), "mobile-nav": SxExpr(mobile_nav) if mobile_nav else "", } # --------------------------------------------------------------------------- # Federation # --------------------------------------------------------------------------- @register_io_handler("federation-actor-ctx") async def _io_federation_actor_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any] | None: """``(federation-actor-ctx)`` → serialized actor dict or None.""" from quart import g actor = getattr(g, "_social_actor", None) if not actor: return None return { "id": actor.id, "preferred_username": actor.preferred_username, "display_name": getattr(actor, "display_name", None), "icon_url": getattr(actor, "icon_url", None), "actor_url": getattr(actor, "actor_url", ""), } # --------------------------------------------------------------------------- # Misc UI contexts # --------------------------------------------------------------------------- @register_io_handler("select-colours") async def _io_select_colours( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> str: """``(select-colours)`` → the shared select/hover CSS class string.""" from quart import current_app return current_app.jinja_env.globals.get("select_colours", "") @register_io_handler("account-nav-ctx") async def _io_account_nav_ctx( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> Any: """``(account-nav-ctx)`` → account nav fragments as SxExpr, or NIL.""" from quart import g from .types import NIL from .parser import SxExpr from .helpers import sx_call val = getattr(g, "account_nav", None) if not val: return NIL if isinstance(val, SxExpr): return val return sx_call("rich-text", html=str(val)) @register_io_handler("app-rights") async def _io_app_rights( args: list[Any], kwargs: dict[str, Any], ctx: Any ) -> dict[str, Any]: """``(app-rights)`` → user rights dict from ``g.rights``.""" from quart import g return getattr(g, "rights", None) or {}