giles
5344b382a5
Events: 3861 → 21 lines, split into 8 sub-modules (renders, helpers, layouts, calendar, entries, slots, tickets, utils). Updated 16 bp routes. SX Docs: 3224 → 27 lines, split into 5 sub-modules (renders, utils, essays, helpers, layouts). Updated 37 import sites in bp/pages/routes.py. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
397 lines
14 KiB
Python
397 lines
14 KiB
Python
"""Page helpers — _h_* helper functions + _register_events_helpers + _ensure_* context hydration."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared hydration helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _add_to_defpage_ctx(**kwargs: Any) -> None:
|
|
"""Add data to g._defpage_ctx for the app-level context_processor."""
|
|
from quart import g
|
|
if not hasattr(g, '_defpage_ctx'):
|
|
g._defpage_ctx = {}
|
|
g._defpage_ctx.update(kwargs)
|
|
|
|
|
|
async def _ensure_calendar(calendar_slug: str | None) -> None:
|
|
"""Load calendar into g.calendar if not already present."""
|
|
from quart import g, abort
|
|
if hasattr(g, 'calendar'):
|
|
_add_to_defpage_ctx(calendar=g.calendar)
|
|
return
|
|
from bp.calendar.services.calendar_view import (
|
|
get_calendar_by_post_and_slug, get_calendar_by_slug,
|
|
)
|
|
post_data = getattr(g, "post_data", None)
|
|
if post_data:
|
|
post_id = (post_data.get("post") or {}).get("id")
|
|
cal = await get_calendar_by_post_and_slug(g.s, post_id, calendar_slug)
|
|
else:
|
|
cal = await get_calendar_by_slug(g.s, calendar_slug)
|
|
if not cal:
|
|
abort(404)
|
|
g.calendar = cal
|
|
g.calendar_slug = calendar_slug
|
|
_add_to_defpage_ctx(calendar=cal)
|
|
|
|
|
|
async def _ensure_entry(entry_id: int | None) -> None:
|
|
"""Load calendar entry into g.entry if not already present."""
|
|
from quart import g, abort
|
|
if hasattr(g, 'entry'):
|
|
_add_to_defpage_ctx(entry=g.entry)
|
|
return
|
|
from sqlalchemy import select
|
|
from models.calendars import CalendarEntry
|
|
result = await g.s.execute(
|
|
select(CalendarEntry).where(
|
|
CalendarEntry.id == entry_id,
|
|
CalendarEntry.deleted_at.is_(None),
|
|
)
|
|
)
|
|
entry = result.scalar_one_or_none()
|
|
if entry is None:
|
|
abort(404)
|
|
g.entry = entry
|
|
_add_to_defpage_ctx(entry=entry)
|
|
|
|
|
|
async def _ensure_entry_context(entry_id: int | None) -> None:
|
|
"""Load full entry context (ticket data, posts) into g.* and _defpage_ctx."""
|
|
from quart import g
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
from models.calendars import CalendarEntry
|
|
from bp.tickets.services.tickets import (
|
|
get_available_ticket_count,
|
|
get_sold_ticket_count,
|
|
get_user_reserved_count,
|
|
)
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from bp.calendar_entry.services.post_associations import get_entry_posts
|
|
|
|
await _ensure_entry(entry_id)
|
|
|
|
# Reload with ticket_types eagerly loaded
|
|
stmt = (
|
|
select(CalendarEntry)
|
|
.where(CalendarEntry.id == entry_id, CalendarEntry.deleted_at.is_(None))
|
|
.options(selectinload(CalendarEntry.ticket_types))
|
|
)
|
|
result = await g.s.execute(stmt)
|
|
calendar_entry = result.scalar_one_or_none()
|
|
|
|
if calendar_entry and getattr(g, "calendar", None):
|
|
if calendar_entry.calendar_id != g.calendar.id:
|
|
calendar_entry = None
|
|
|
|
if calendar_entry:
|
|
await g.s.refresh(calendar_entry, ['slot'])
|
|
g.entry = calendar_entry
|
|
entry_posts = await get_entry_posts(g.s, calendar_entry.id)
|
|
ticket_remaining = await get_available_ticket_count(g.s, calendar_entry.id)
|
|
ticket_sold_count = await get_sold_ticket_count(g.s, calendar_entry.id)
|
|
ident = current_cart_identity()
|
|
user_ticket_count = await get_user_reserved_count(
|
|
g.s, calendar_entry.id,
|
|
user_id=ident["user_id"],
|
|
session_id=ident["session_id"],
|
|
)
|
|
user_ticket_counts_by_type = {}
|
|
if calendar_entry.ticket_types:
|
|
for tt in calendar_entry.ticket_types:
|
|
if tt.deleted_at is None:
|
|
user_ticket_counts_by_type[tt.id] = await get_user_reserved_count(
|
|
g.s, calendar_entry.id,
|
|
user_id=ident["user_id"],
|
|
session_id=ident["session_id"],
|
|
ticket_type_id=tt.id,
|
|
)
|
|
_add_to_defpage_ctx(
|
|
entry=calendar_entry,
|
|
entry_posts=entry_posts,
|
|
ticket_remaining=ticket_remaining,
|
|
ticket_sold_count=ticket_sold_count,
|
|
user_ticket_count=user_ticket_count,
|
|
user_ticket_counts_by_type=user_ticket_counts_by_type,
|
|
)
|
|
|
|
|
|
async def _ensure_day_data(year: int, month: int, day: int) -> None:
|
|
"""Load day-specific data for layout header functions."""
|
|
from quart import g, session as qsession
|
|
if hasattr(g, 'day_date'):
|
|
return
|
|
from datetime import date as date_cls, datetime, timezone, timedelta
|
|
from sqlalchemy import select
|
|
from bp.calendar.services import get_visible_entries_for_period
|
|
from models.calendars import CalendarSlot
|
|
|
|
calendar = getattr(g, "calendar", None)
|
|
if not calendar:
|
|
return
|
|
|
|
try:
|
|
day_date = date_cls(year, month, day)
|
|
except (ValueError, TypeError):
|
|
return
|
|
|
|
period_start = datetime(year, month, day, tzinfo=timezone.utc)
|
|
period_end = period_start + timedelta(days=1)
|
|
|
|
user = getattr(g, "user", None)
|
|
session_id = qsession.get("calendar_sid")
|
|
|
|
visible = await get_visible_entries_for_period(
|
|
sess=g.s,
|
|
calendar_id=calendar.id,
|
|
period_start=period_start,
|
|
period_end=period_end,
|
|
user=user,
|
|
session_id=session_id,
|
|
)
|
|
|
|
weekday_attr = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"][day_date.weekday()]
|
|
stmt = (
|
|
select(CalendarSlot)
|
|
.where(
|
|
CalendarSlot.calendar_id == calendar.id,
|
|
getattr(CalendarSlot, weekday_attr) == True, # noqa: E712
|
|
CalendarSlot.deleted_at.is_(None),
|
|
)
|
|
.order_by(CalendarSlot.time_start.asc(), CalendarSlot.id.asc())
|
|
)
|
|
result = await g.s.execute(stmt)
|
|
day_slots = list(result.scalars())
|
|
|
|
g.day_date = day_date
|
|
_add_to_defpage_ctx(
|
|
qsession=qsession,
|
|
day_date=day_date,
|
|
day=day,
|
|
year=year,
|
|
month=month,
|
|
day_entries=visible.merged_entries,
|
|
user_entries=visible.user_entries,
|
|
confirmed_entries=visible.confirmed_entries,
|
|
day_slots=day_slots,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Page helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _register_events_helpers() -> None:
|
|
from shared.sx.pages import register_page_helpers
|
|
|
|
register_page_helpers("events", {
|
|
"calendar-admin-content": _h_calendar_admin_content,
|
|
"day-admin-content": _h_day_admin_content,
|
|
"slots-content": _h_slots_content,
|
|
"slot-content": _h_slot_content,
|
|
"entry-content": _h_entry_content,
|
|
"entry-menu": _h_entry_menu,
|
|
"entry-admin-content": _h_entry_admin_content,
|
|
"admin-menu": _h_admin_menu,
|
|
"ticket-types-content": _h_ticket_types_content,
|
|
"ticket-type-content": _h_ticket_type_content,
|
|
"tickets-content": _h_tickets_content,
|
|
"ticket-detail-content": _h_ticket_detail_content,
|
|
"ticket-admin-content": _h_ticket_admin_content,
|
|
"markets-content": _h_markets_content,
|
|
})
|
|
|
|
|
|
async def _h_calendar_admin_content(calendar_slug=None, **kw):
|
|
await _ensure_calendar(calendar_slug)
|
|
from shared.sx.page import get_template_context
|
|
from .calendar import _calendar_admin_main_panel_html
|
|
ctx = await get_template_context()
|
|
return await _calendar_admin_main_panel_html(ctx)
|
|
|
|
|
|
async def _h_day_admin_content(calendar_slug=None, year=None, month=None, day=None, **kw):
|
|
await _ensure_calendar(calendar_slug)
|
|
if year is not None:
|
|
await _ensure_day_data(int(year), int(month), int(day))
|
|
from .calendar import _day_admin_main_panel_html
|
|
return await _day_admin_main_panel_html({})
|
|
|
|
|
|
async def _h_slots_content(calendar_slug=None, **kw):
|
|
from quart import g
|
|
await _ensure_calendar(calendar_slug)
|
|
calendar = getattr(g, "calendar", None)
|
|
from bp.slots.services.slots import list_slots as svc_list_slots
|
|
from .slots import render_slots_table
|
|
slots = await svc_list_slots(g.s, calendar.id) if calendar else []
|
|
_add_to_defpage_ctx(slots=slots)
|
|
return await render_slots_table(slots, calendar)
|
|
|
|
|
|
async def _h_slot_content(calendar_slug=None, slot_id=None, **kw):
|
|
from quart import g, abort
|
|
await _ensure_calendar(calendar_slug)
|
|
from bp.slot.services.slot import get_slot as svc_get_slot
|
|
from .slots import render_slot_main_panel
|
|
slot = await svc_get_slot(g.s, slot_id) if slot_id else None
|
|
if not slot:
|
|
abort(404)
|
|
g.slot = slot
|
|
_add_to_defpage_ctx(slot=slot)
|
|
calendar = getattr(g, "calendar", None)
|
|
return await render_slot_main_panel(slot, calendar)
|
|
|
|
|
|
async def _h_entry_content(calendar_slug=None, entry_id=None, **kw):
|
|
await _ensure_calendar(calendar_slug)
|
|
await _ensure_entry_context(entry_id)
|
|
from shared.sx.page import get_template_context
|
|
from .entries import _entry_main_panel_html
|
|
ctx = await get_template_context()
|
|
return await _entry_main_panel_html(ctx)
|
|
|
|
|
|
async def _h_entry_menu(calendar_slug=None, entry_id=None, **kw):
|
|
await _ensure_calendar(calendar_slug)
|
|
await _ensure_entry_context(entry_id)
|
|
from shared.sx.page import get_template_context
|
|
from .entries import _entry_nav_html
|
|
ctx = await get_template_context()
|
|
return await _entry_nav_html(ctx)
|
|
|
|
|
|
async def _h_entry_admin_content(calendar_slug=None, entry_id=None, **kw):
|
|
await _ensure_calendar(calendar_slug)
|
|
await _ensure_entry_context(entry_id)
|
|
from shared.sx.page import get_template_context
|
|
from .entries import _entry_admin_main_panel_html
|
|
ctx = await get_template_context()
|
|
return await _entry_admin_main_panel_html(ctx)
|
|
|
|
|
|
async def _h_admin_menu():
|
|
from shared.sx.helpers import render_to_sx
|
|
return await render_to_sx("events-admin-placeholder-nav")
|
|
|
|
|
|
async def _h_ticket_types_content(calendar_slug=None, entry_id=None,
|
|
year=None, month=None, day=None, **kw):
|
|
from quart import g
|
|
await _ensure_calendar(calendar_slug)
|
|
await _ensure_entry(entry_id)
|
|
entry = getattr(g, "entry", None)
|
|
calendar = getattr(g, "calendar", None)
|
|
from bp.ticket_types.services.tickets import list_ticket_types as svc_list_ticket_types
|
|
from .tickets import render_ticket_types_table
|
|
ticket_types = await svc_list_ticket_types(g.s, entry.id) if entry else []
|
|
_add_to_defpage_ctx(ticket_types=ticket_types)
|
|
return await render_ticket_types_table(ticket_types, entry, calendar, day, month, year)
|
|
|
|
|
|
async def _h_ticket_type_content(calendar_slug=None, entry_id=None,
|
|
ticket_type_id=None, year=None, month=None, day=None, **kw):
|
|
from quart import g, abort
|
|
await _ensure_calendar(calendar_slug)
|
|
await _ensure_entry(entry_id)
|
|
from bp.ticket_type.services.ticket import get_ticket_type as svc_get_ticket_type
|
|
from .tickets import render_ticket_type_main_panel
|
|
ticket_type = await svc_get_ticket_type(g.s, ticket_type_id) if ticket_type_id else None
|
|
if not ticket_type:
|
|
abort(404)
|
|
g.ticket_type = ticket_type
|
|
_add_to_defpage_ctx(ticket_type=ticket_type)
|
|
entry = getattr(g, "entry", None)
|
|
calendar = getattr(g, "calendar", None)
|
|
return await render_ticket_type_main_panel(ticket_type, entry, calendar, day, month, year)
|
|
|
|
|
|
async def _h_tickets_content(**kw):
|
|
from quart import g
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from bp.tickets.services.tickets import get_user_tickets
|
|
from .tickets import _tickets_main_panel_html
|
|
ident = current_cart_identity()
|
|
tickets = await get_user_tickets(
|
|
g.s,
|
|
user_id=ident["user_id"],
|
|
session_id=ident["session_id"],
|
|
)
|
|
from shared.sx.page import get_template_context
|
|
ctx = await get_template_context()
|
|
return await _tickets_main_panel_html(ctx, tickets)
|
|
|
|
|
|
async def _h_ticket_detail_content(code=None, **kw):
|
|
from quart import g, abort
|
|
from shared.infrastructure.cart_identity import current_cart_identity
|
|
from bp.tickets.services.tickets import get_ticket_by_code
|
|
from .tickets import _ticket_detail_panel_html
|
|
ticket = await get_ticket_by_code(g.s, code) if code else None
|
|
if not ticket:
|
|
abort(404)
|
|
# Verify ownership
|
|
ident = current_cart_identity()
|
|
if ident["user_id"] is not None:
|
|
if ticket.user_id != ident["user_id"]:
|
|
abort(404)
|
|
elif ident["session_id"] is not None:
|
|
if ticket.session_id != ident["session_id"]:
|
|
abort(404)
|
|
else:
|
|
abort(404)
|
|
from shared.sx.page import get_template_context
|
|
ctx = await get_template_context()
|
|
return await _ticket_detail_panel_html(ctx, ticket)
|
|
|
|
|
|
async def _h_ticket_admin_content(**kw):
|
|
from quart import g
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.orm import selectinload
|
|
from models.calendars import CalendarEntry, Ticket
|
|
from .tickets import _ticket_admin_main_panel_html
|
|
|
|
result = await g.s.execute(
|
|
select(Ticket)
|
|
.options(
|
|
selectinload(Ticket.entry).selectinload(CalendarEntry.calendar),
|
|
selectinload(Ticket.ticket_type),
|
|
)
|
|
.order_by(Ticket.created_at.desc())
|
|
.limit(50)
|
|
)
|
|
tickets = result.scalars().all()
|
|
|
|
total = await g.s.scalar(select(func.count(Ticket.id)))
|
|
confirmed = await g.s.scalar(
|
|
select(func.count(Ticket.id)).where(Ticket.state == "confirmed")
|
|
)
|
|
checked_in = await g.s.scalar(
|
|
select(func.count(Ticket.id)).where(Ticket.state == "checked_in")
|
|
)
|
|
reserved = await g.s.scalar(
|
|
select(func.count(Ticket.id)).where(Ticket.state == "reserved")
|
|
)
|
|
stats = {
|
|
"total": total or 0,
|
|
"confirmed": confirmed or 0,
|
|
"checked_in": checked_in or 0,
|
|
"reserved": reserved or 0,
|
|
}
|
|
|
|
from shared.sx.page import get_template_context
|
|
ctx = await get_template_context()
|
|
return await _ticket_admin_main_panel_html(ctx, tickets, stats)
|
|
|
|
|
|
async def _h_markets_content(**kw):
|
|
from shared.sx.page import get_template_context
|
|
from .calendar import _markets_main_panel_html
|
|
ctx = await get_template_context()
|
|
return await _markets_main_panel_html(ctx)
|