rose-ash/blog/bp/blog/admin/routes.py

from __future__ import annotations

import re
from quart import (
    render_template,
    make_response,
    Blueprint,
    redirect,
    url_for,
    request,
    g,
)
from sqlalchemy import select, delete

from shared.browser.app.authz import require_admin
from shared.browser.app.utils.htmx import is_htmx_request
from shared.browser.app.redis_cacher import invalidate_tag_cache
from shared.sx.helpers import sx_response

from models.tag_group import TagGroup, TagGroupTag
from models.ghost_content import Tag


def _slugify(name: str) -> str:
    s = name.strip().lower()
    s = re.sub(r"[^\w\s-]", "", s)
    s = re.sub(r"[\s_]+", "-", s)
    return s.strip("-")


async def _unassigned_tags(session):
    """Return public, non-deleted tags not assigned to any group."""
    assigned_sq = select(TagGroupTag.tag_id).subquery()
    q = (
        select(Tag)
        .where(
            Tag.deleted_at.is_(None),
            (Tag.visibility == "public") | (Tag.visibility.is_(None)),
            Tag.id.notin_(select(assigned_sq)),
        )
        .order_by(Tag.name)
    )
    return list((await session.execute(q)).scalars())


def register():
    bp = Blueprint("tag_groups_admin", __name__, url_prefix="/settings/tag-groups")

    @bp.before_request
    async def _prepare_page_data():
        ep = request.endpoint or ""
        if "defpage_tag_groups_page" in ep:
            groups = list(
                (await g.s.execute(
                    select(TagGroup).order_by(TagGroup.sort_order, TagGroup.name)
                )).scalars()
            )
            unassigned = await _unassigned_tags(g.s)
            from shared.sx.page import get_template_context
            from sx.sx_components import _tag_groups_main_panel_sx
            tctx = await get_template_context()
            tctx.update({"groups": groups, "unassigned_tags": unassigned})
            g.tag_groups_content = _tag_groups_main_panel_sx(tctx)
        elif "defpage_tag_group_edit" in ep:
            tag_id = (request.view_args or {}).get("id")
            tg = await g.s.get(TagGroup, tag_id)
            if not tg:
                from quart import abort
                abort(404)
            assigned_rows = list(
                (await g.s.execute(
                    select(TagGroupTag.tag_id).where(TagGroupTag.tag_group_id == tag_id)
                )).scalars()
            )
            all_tags = list(
                (await g.s.execute(
                    select(Tag).where(
                        Tag.deleted_at.is_(None),
                        (Tag.visibility == "public") | (Tag.visibility.is_(None)),
                    ).order_by(Tag.name)
                )).scalars()
            )
            from shared.sx.page import get_template_context
            from sx.sx_components import _tag_groups_edit_main_panel_sx
            tctx = await get_template_context()
            tctx.update({
                "group": tg,
                "all_tags": all_tags,
                "assigned_tag_ids": set(assigned_rows),
            })
            g.tag_group_edit_content = _tag_groups_edit_main_panel_sx(tctx)

    from shared.sx.pages import mount_pages
    mount_pages(bp, "blog", names=["tag-groups-page", "tag-group-edit"])

    @bp.post("/")
    @require_admin
    async def create():
        form = await request.form
        name = (form.get("name") or "").strip()
        if not name:
            return redirect(url_for("blog.tag_groups_admin.defpage_tag_groups_page"))

        slug = _slugify(name)
        feature_image = (form.get("feature_image") or "").strip() or None
        colour = (form.get("colour") or "").strip() or None
        sort_order = int(form.get("sort_order") or 0)

        tg = TagGroup(
            name=name, slug=slug,
            feature_image=feature_image, colour=colour,
            sort_order=sort_order,
        )
        g.s.add(tg)
        await g.s.flush()

        await invalidate_tag_cache("blog")
        return redirect(url_for("blog.tag_groups_admin.defpage_tag_groups_page"))

    @bp.post("/<int:id>/")
    @require_admin
    async def save(id: int):
        tg = await g.s.get(TagGroup, id)
        if not tg:
            return redirect(url_for("blog.tag_groups_admin.defpage_tag_groups_page"))

        form = await request.form
        name = (form.get("name") or "").strip()
        if name:
            tg.name = name
            tg.slug = _slugify(name)
        tg.feature_image = (form.get("feature_image") or "").strip() or None
        tg.colour = (form.get("colour") or "").strip() or None
        tg.sort_order = int(form.get("sort_order") or 0)

        # Update tag assignments
        selected_tag_ids = set()
        for val in form.getlist("tag_ids"):
            try:
                selected_tag_ids.add(int(val))
            except (ValueError, TypeError):
                pass

        # Remove old assignments
        await g.s.execute(
            delete(TagGroupTag).where(TagGroupTag.tag_group_id == id)
        )
        await g.s.flush()

        # Add new assignments
        for tid in selected_tag_ids:
            g.s.add(TagGroupTag(tag_group_id=id, tag_id=tid))
        await g.s.flush()

        await invalidate_tag_cache("blog")
        return redirect(url_for("blog.tag_groups_admin.defpage_tag_group_edit", id=id))

    @bp.post("/<int:id>/delete/")
    @require_admin
    async def delete_group(id: int):
        tg = await g.s.get(TagGroup, id)
        if tg:
            await g.s.delete(tg)
            await g.s.flush()
            await invalidate_tag_cache("blog")
        return redirect(url_for("blog.tag_groups_admin.defpage_tag_groups_page"))

    return bp