giles
e53e8cc1f7
Convert 6 blog settings pages (settings-home, cache, snippets, menu-items, tag-groups, tag-group-edit) from Python page helpers to .sx defpages with (service "blog-page" ...) IO primitives. Create data-driven defcomps that handle iteration via (map ...) instead of Python loops. Post-related page helpers (editor, post-admin/data/preview/entries/settings/edit) remain as Python helpers — they depend on _ensure_post_data and sx_components rendering functions that need separate conversion. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
468 lines
18 KiB
Python
468 lines
18 KiB
Python
"""Blog defpage setup — registers layouts, page helpers, and loads .sx pages."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
def setup_blog_pages() -> None:
|
|
"""Register blog-specific layouts, page helpers, and load page definitions."""
|
|
_register_blog_layouts()
|
|
_register_blog_helpers()
|
|
_load_blog_page_files()
|
|
|
|
|
|
def _load_blog_page_files() -> None:
|
|
import os
|
|
from shared.sx.pages import load_page_dir
|
|
load_page_dir(os.path.dirname(__file__), "blog")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared hydration helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _add_to_defpage_ctx(**kwargs: Any) -> None:
|
|
from quart import g
|
|
if not hasattr(g, '_defpage_ctx'):
|
|
g._defpage_ctx = {}
|
|
g._defpage_ctx.update(kwargs)
|
|
|
|
|
|
async def _ensure_post_data(slug: str | None) -> None:
|
|
"""Load post data and set g.post_data + defpage context.
|
|
|
|
Replicates post bp's hydrate_post_data + context_processor.
|
|
"""
|
|
from quart import g, abort
|
|
|
|
if hasattr(g, 'post_data') and g.post_data:
|
|
await _inject_post_context(g.post_data)
|
|
return
|
|
|
|
if not slug:
|
|
abort(404)
|
|
|
|
from bp.post.services.post_data import post_data
|
|
|
|
is_admin = bool((g.get("rights") or {}).get("admin"))
|
|
p_data = await post_data(slug, g.s, include_drafts=True)
|
|
if not p_data:
|
|
abort(404)
|
|
|
|
# Draft access control
|
|
if p_data["post"].get("status") != "published":
|
|
if is_admin:
|
|
pass
|
|
elif g.user and p_data["post"].get("user_id") == g.user.id:
|
|
pass
|
|
else:
|
|
abort(404)
|
|
|
|
g.post_data = p_data
|
|
g.post_slug = slug
|
|
await _inject_post_context(p_data)
|
|
|
|
|
|
async def _inject_post_context(p_data: dict) -> None:
|
|
"""Add post context_processor data to defpage context."""
|
|
from shared.config import config
|
|
from shared.infrastructure.fragments import fetch_fragment
|
|
from shared.infrastructure.data_client import fetch_data
|
|
from shared.contracts.dtos import CartSummaryDTO, dto_from_dict
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
|
|
db_post_id = p_data["post"]["id"]
|
|
post_slug = p_data["post"]["slug"]
|
|
|
|
container_nav = await fetch_fragment("relations", "container-nav", params={
|
|
"container_type": "page",
|
|
"container_id": str(db_post_id),
|
|
"post_slug": post_slug,
|
|
})
|
|
|
|
ctx: dict = {
|
|
**p_data,
|
|
"base_title": config()["title"],
|
|
"container_nav": container_nav,
|
|
}
|
|
|
|
if p_data["post"].get("is_page"):
|
|
ident = current_cart_identity()
|
|
summary_params: dict = {"page_slug": post_slug}
|
|
if ident["user_id"] is not None:
|
|
summary_params["user_id"] = ident["user_id"]
|
|
if ident["session_id"] is not None:
|
|
summary_params["session_id"] = ident["session_id"]
|
|
raw_summary = await fetch_data(
|
|
"cart", "cart-summary", params=summary_params, required=False,
|
|
)
|
|
page_summary = dto_from_dict(CartSummaryDTO, raw_summary) if raw_summary else CartSummaryDTO()
|
|
ctx["page_cart_count"] = (
|
|
page_summary.count + page_summary.calendar_count + page_summary.ticket_count
|
|
)
|
|
ctx["page_cart_total"] = float(
|
|
page_summary.total + page_summary.calendar_total + page_summary.ticket_total
|
|
)
|
|
|
|
_add_to_defpage_ctx(**ctx)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Layouts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_blog_layouts() -> None:
|
|
from shared.sx.layouts import register_custom_layout
|
|
# :blog — root + blog header (for new-post, new-page)
|
|
register_custom_layout("blog", _blog_full, _blog_oob)
|
|
# :blog-settings — root + settings header (with settings nav menu)
|
|
register_custom_layout("blog-settings", _settings_full, _settings_oob,
|
|
mobile_fn=_settings_mobile)
|
|
# Sub-settings layouts (root + settings + sub header)
|
|
register_custom_layout("blog-cache", _cache_full, _cache_oob)
|
|
register_custom_layout("blog-snippets", _snippets_full, _snippets_oob)
|
|
register_custom_layout("blog-menu-items", _menu_items_full, _menu_items_oob)
|
|
register_custom_layout("blog-tag-groups", _tag_groups_full, _tag_groups_oob)
|
|
register_custom_layout("blog-tag-group-edit",
|
|
_tag_group_edit_full, _tag_group_edit_oob)
|
|
|
|
|
|
# --- Blog layout (root + blog header) ---
|
|
|
|
async def _blog_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx
|
|
from sx.sx_components import _blog_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
blog_hdr = await _blog_header_sx(ctx)
|
|
return "(<> " + root_hdr + " " + blog_hdr + ")"
|
|
|
|
|
|
async def _blog_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, oob_header_sx
|
|
from sx.sx_components import _blog_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
blog_hdr = await _blog_header_sx(ctx)
|
|
rows = "(<> " + root_hdr + " " + blog_hdr + ")"
|
|
return await oob_header_sx("root-header-child", "blog-header-child", rows)
|
|
|
|
|
|
# --- Settings layout (root + settings header) ---
|
|
|
|
async def _settings_full(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx
|
|
from sx.sx_components import _settings_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
return "(<> " + root_hdr + " " + settings_hdr + ")"
|
|
|
|
|
|
async def _settings_oob(ctx: dict, **kw: Any) -> str:
|
|
from shared.sx.helpers import root_header_sx, oob_header_sx
|
|
from sx.sx_components import _settings_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
rows = "(<> " + root_hdr + " " + settings_hdr + ")"
|
|
return await oob_header_sx("root-header-child", "root-settings-header-child", rows)
|
|
|
|
|
|
async def _settings_mobile(ctx: dict, **kw: Any) -> str:
|
|
from sx.sx_components import _settings_nav_sx
|
|
return await _settings_nav_sx(ctx)
|
|
|
|
|
|
# --- Sub-settings helpers ---
|
|
|
|
async def _sub_settings_full(ctx: dict, row_id: str, child_id: str,
|
|
endpoint: str, icon: str, label: str) -> str:
|
|
from shared.sx.helpers import root_header_sx
|
|
from sx.sx_components import _settings_header_sx, _sub_settings_header_sx
|
|
from quart import url_for as qurl
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
sub_hdr = await _sub_settings_header_sx(row_id, child_id,
|
|
qurl(endpoint), icon, label, ctx)
|
|
return "(<> " + root_hdr + " " + settings_hdr + " " + sub_hdr + ")"
|
|
|
|
|
|
async def _sub_settings_oob(ctx: dict, row_id: str, child_id: str,
|
|
endpoint: str, icon: str, label: str) -> str:
|
|
from shared.sx.helpers import oob_header_sx
|
|
from sx.sx_components import _settings_header_sx, _sub_settings_header_sx
|
|
from quart import url_for as qurl
|
|
settings_hdr_oob = await _settings_header_sx(ctx, oob=True)
|
|
sub_hdr = await _sub_settings_header_sx(row_id, child_id,
|
|
qurl(endpoint), icon, label, ctx)
|
|
sub_oob = await oob_header_sx("root-settings-header-child", child_id, sub_hdr)
|
|
return "(<> " + settings_hdr_oob + " " + sub_oob + ")"
|
|
|
|
|
|
# --- Cache ---
|
|
|
|
async def _cache_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "cache-row", "cache-header-child",
|
|
"defpage_cache_page", "refresh", "Cache")
|
|
|
|
|
|
async def _cache_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "cache-row", "cache-header-child",
|
|
"defpage_cache_page", "refresh", "Cache")
|
|
|
|
|
|
# --- Snippets ---
|
|
|
|
async def _snippets_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "snippets-row", "snippets-header-child",
|
|
"defpage_snippets_page", "puzzle-piece", "Snippets")
|
|
|
|
|
|
async def _snippets_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "snippets-row", "snippets-header-child",
|
|
"defpage_snippets_page", "puzzle-piece", "Snippets")
|
|
|
|
|
|
# --- Menu Items ---
|
|
|
|
async def _menu_items_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "menu_items-row", "menu_items-header-child",
|
|
"defpage_menu_items_page", "bars", "Menu Items")
|
|
|
|
|
|
async def _menu_items_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "menu_items-row", "menu_items-header-child",
|
|
"defpage_menu_items_page", "bars", "Menu Items")
|
|
|
|
|
|
# --- Tag Groups ---
|
|
|
|
async def _tag_groups_full(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_full(ctx, "tag-groups-row", "tag-groups-header-child",
|
|
"defpage_tag_groups_page", "tags", "Tag Groups")
|
|
|
|
|
|
async def _tag_groups_oob(ctx: dict, **kw: Any) -> str:
|
|
return await _sub_settings_oob(ctx, "tag-groups-row", "tag-groups-header-child",
|
|
"defpage_tag_groups_page", "tags", "Tag Groups")
|
|
|
|
|
|
# --- Tag Group Edit ---
|
|
|
|
async def _tag_group_edit_full(ctx: dict, **kw: Any) -> str:
|
|
from quart import request
|
|
g_id = (request.view_args or {}).get("id")
|
|
from quart import url_for as qurl
|
|
from shared.sx.helpers import root_header_sx
|
|
from sx.sx_components import _settings_header_sx, _sub_settings_header_sx
|
|
root_hdr = await root_header_sx(ctx)
|
|
settings_hdr = await _settings_header_sx(ctx)
|
|
sub_hdr = await _sub_settings_header_sx("tag-groups-row", "tag-groups-header-child",
|
|
qurl("defpage_tag_group_edit", id=g_id),
|
|
"tags", "Tag Groups", ctx)
|
|
return "(<> " + root_hdr + " " + settings_hdr + " " + sub_hdr + ")"
|
|
|
|
|
|
async def _tag_group_edit_oob(ctx: dict, **kw: Any) -> str:
|
|
from quart import request
|
|
g_id = (request.view_args or {}).get("id")
|
|
from quart import url_for as qurl
|
|
from shared.sx.helpers import oob_header_sx
|
|
from sx.sx_components import _settings_header_sx, _sub_settings_header_sx
|
|
settings_hdr_oob = await _settings_header_sx(ctx, oob=True)
|
|
sub_hdr = await _sub_settings_header_sx("tag-groups-row", "tag-groups-header-child",
|
|
qurl("defpage_tag_group_edit", id=g_id),
|
|
"tags", "Tag Groups", ctx)
|
|
sub_oob = await oob_header_sx("root-settings-header-child", "tag-groups-header-child", sub_hdr)
|
|
return "(<> " + settings_hdr_oob + " " + sub_oob + ")"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page helpers (async functions available in .sx defpage expressions)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_blog_helpers() -> None:
|
|
from shared.sx.pages import register_page_helpers
|
|
register_page_helpers("blog", {
|
|
"editor-content": _h_editor_content,
|
|
"editor-page-content": _h_editor_page_content,
|
|
"post-admin-content": _h_post_admin_content,
|
|
"post-data-content": _h_post_data_content,
|
|
"post-preview-content": _h_post_preview_content,
|
|
"post-entries-content": _h_post_entries_content,
|
|
"post-settings-content": _h_post_settings_content,
|
|
"post-edit-content": _h_post_edit_content,
|
|
})
|
|
|
|
|
|
# --- Editor helpers ---
|
|
|
|
async def _h_editor_content(**kw):
|
|
from sx.sx_components import render_editor_panel
|
|
return await render_editor_panel()
|
|
|
|
|
|
async def _h_editor_page_content(**kw):
|
|
from sx.sx_components import render_editor_panel
|
|
return await render_editor_panel(is_page=True)
|
|
|
|
|
|
# --- Post admin helpers ---
|
|
|
|
async def _h_post_admin_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from quart import g
|
|
from sqlalchemy import select
|
|
from shared.models.page_config import PageConfig
|
|
post = (g.post_data or {}).get("post", {})
|
|
features = {}
|
|
sumup_configured = False
|
|
sumup_merchant_code = ""
|
|
sumup_checkout_prefix = ""
|
|
if post.get("is_page"):
|
|
pc = (await g.s.execute(
|
|
select(PageConfig).where(
|
|
PageConfig.container_type == "page",
|
|
PageConfig.container_id == post["id"],
|
|
)
|
|
)).scalar_one_or_none()
|
|
if pc:
|
|
features = pc.features or {}
|
|
sumup_configured = bool(pc.sumup_api_key)
|
|
sumup_merchant_code = pc.sumup_merchant_code or ""
|
|
sumup_checkout_prefix = pc.sumup_checkout_prefix or ""
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _post_admin_main_panel_sx
|
|
tctx = await get_template_context()
|
|
tctx.update({
|
|
"features": features,
|
|
"sumup_configured": sumup_configured,
|
|
"sumup_merchant_code": sumup_merchant_code,
|
|
"sumup_checkout_prefix": sumup_checkout_prefix,
|
|
})
|
|
return _post_admin_main_panel_sx(tctx)
|
|
|
|
|
|
async def _h_post_data_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _post_data_content_sx
|
|
tctx = await get_template_context()
|
|
return _post_data_content_sx(tctx)
|
|
|
|
|
|
async def _h_post_preview_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from quart import g
|
|
from models.ghost_content import Post
|
|
from sqlalchemy import select as sa_select
|
|
post_id = g.post_data["post"]["id"]
|
|
post = (await g.s.execute(
|
|
sa_select(Post).where(Post.id == post_id)
|
|
)).scalar_one_or_none()
|
|
preview_ctx: dict = {}
|
|
sx_content = getattr(post, "sx_content", None) or ""
|
|
if sx_content:
|
|
from shared.sx.prettify import sx_to_pretty_sx
|
|
preview_ctx["sx_pretty"] = sx_to_pretty_sx(sx_content)
|
|
lexical_raw = getattr(post, "lexical", None) or ""
|
|
if lexical_raw:
|
|
from shared.sx.prettify import json_to_pretty_sx
|
|
preview_ctx["json_pretty"] = json_to_pretty_sx(lexical_raw)
|
|
if sx_content:
|
|
from shared.sx.parser import parse as sx_parse
|
|
from shared.sx.html import render as sx_html_render
|
|
from shared.sx.jinja_bridge import _COMPONENT_ENV
|
|
try:
|
|
parsed = sx_parse(sx_content)
|
|
preview_ctx["sx_rendered"] = sx_html_render(parsed, dict(_COMPONENT_ENV))
|
|
except Exception:
|
|
preview_ctx["sx_rendered"] = "<em>Error rendering sx</em>"
|
|
if lexical_raw:
|
|
from bp.blog.ghost.lexical_renderer import render_lexical
|
|
try:
|
|
preview_ctx["lex_rendered"] = render_lexical(lexical_raw)
|
|
except Exception:
|
|
preview_ctx["lex_rendered"] = "<em>Error rendering lexical</em>"
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _preview_main_panel_sx
|
|
tctx = await get_template_context()
|
|
tctx.update(preview_ctx)
|
|
return await _preview_main_panel_sx(tctx)
|
|
|
|
|
|
async def _h_post_entries_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from quart import g
|
|
from sqlalchemy import select
|
|
from shared.models.calendars import Calendar
|
|
from bp.post.services.entry_associations import get_post_entry_ids
|
|
post_id = g.post_data["post"]["id"]
|
|
associated_entry_ids = await get_post_entry_ids(post_id)
|
|
result = await g.s.execute(
|
|
select(Calendar)
|
|
.where(Calendar.deleted_at.is_(None))
|
|
.order_by(Calendar.name.asc())
|
|
)
|
|
all_calendars = result.scalars().all()
|
|
for calendar in all_calendars:
|
|
await g.s.refresh(calendar, ["entries", "post"])
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _post_entries_content_sx
|
|
tctx = await get_template_context()
|
|
tctx["all_calendars"] = all_calendars
|
|
tctx["associated_entry_ids"] = associated_entry_ids
|
|
return await _post_entries_content_sx(tctx)
|
|
|
|
|
|
async def _h_post_settings_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from quart import g, request
|
|
from models.ghost_content import Post
|
|
from sqlalchemy import select as sa_select
|
|
from sqlalchemy.orm import selectinload
|
|
from bp.post.admin.routes import _post_to_edit_dict
|
|
post_id = g.post_data["post"]["id"]
|
|
post = (await g.s.execute(
|
|
sa_select(Post)
|
|
.where(Post.id == post_id)
|
|
.options(selectinload(Post.tags))
|
|
)).scalar_one_or_none()
|
|
ghost_post = _post_to_edit_dict(post) if post else {}
|
|
save_success = request.args.get("saved") == "1"
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _post_settings_content_sx
|
|
tctx = await get_template_context()
|
|
tctx["ghost_post"] = ghost_post
|
|
tctx["save_success"] = save_success
|
|
return _post_settings_content_sx(tctx)
|
|
|
|
|
|
async def _h_post_edit_content(slug=None, **kw):
|
|
await _ensure_post_data(slug)
|
|
from quart import g, request
|
|
from models.ghost_content import Post
|
|
from sqlalchemy import select as sa_select
|
|
from sqlalchemy.orm import selectinload
|
|
from shared.infrastructure.data_client import fetch_data
|
|
from bp.post.admin.routes import _post_to_edit_dict
|
|
post_id = g.post_data["post"]["id"]
|
|
post = (await g.s.execute(
|
|
sa_select(Post)
|
|
.where(Post.id == post_id)
|
|
.options(selectinload(Post.tags))
|
|
)).scalar_one_or_none()
|
|
ghost_post = _post_to_edit_dict(post) if post else {}
|
|
save_success = request.args.get("saved") == "1"
|
|
save_error = request.args.get("error", "")
|
|
raw_newsletters = await fetch_data("account", "newsletters", required=False) or []
|
|
from types import SimpleNamespace
|
|
newsletters = [SimpleNamespace(**nl) for nl in raw_newsletters]
|
|
from shared.sx.page import get_template_context
|
|
from sx.sx_components import _post_edit_content_sx
|
|
tctx = await get_template_context()
|
|
tctx["ghost_post"] = ghost_post
|
|
tctx["save_success"] = save_success
|
|
tctx["save_error"] = save_error
|
|
tctx["newsletters"] = newsletters
|
|
return await _post_edit_content_sx(tctx)
|
|
|
|
|