main-site/apps/blog/models.py

from __future__ import annotations

import re
from math import ceil
from typing import Any

from django.core.paginator import EmptyPage, PageNotAnInteger, Paginator
from django.db import models
from django.db.models import CASCADE, PROTECT, SET_NULL, Prefetch
from django.shortcuts import get_object_or_404
from django.utils.html import strip_tags
from django.utils.text import slugify
from modelcluster.contrib.taggit import ClusterTaggableManager
from modelcluster.fields import ParentalKey
from taggit.models import Tag, TaggedItemBase
from wagtail.admin.forms.pages import WagtailAdminPageForm
from wagtail.admin.panels import FieldPanel, ObjectList, PageChooserPanel, TabbedInterface
from wagtail.contrib.routable_page.models import RoutablePageMixin, route
from wagtail.fields import RichTextField, StreamField
from wagtail.models import Page
from wagtail.search import index
from wagtailseo.models import SeoMixin

from apps.authors.models import Author
from apps.blog.blocks import ARTICLE_BODY_BLOCKS


def _generate_summary_from_stream(body: Any, *, max_chars: int = 220) -> str:
    parts: list[str] = []
    if body is None:
        return ""
    for block in body:
        if getattr(block, "block_type", None) == "code":
            continue
        value = getattr(block, "value", block)
        text = value.source if hasattr(value, "source") else str(value)
        clean_text = strip_tags(text)
        if clean_text:
            parts.append(clean_text)
    summary = re.sub(r"\s+", " ", " ".join(parts)).strip()
    if len(summary) <= max_chars:
        return summary
    truncated = summary[:max_chars].rsplit(" ", 1)[0].strip()
    return truncated or summary[:max_chars].strip()


class HomePage(Page):
    featured_article = models.ForeignKey(
        "blog.ArticlePage", null=True, blank=True, on_delete=SET_NULL, related_name="+"
    )

    subpage_types = ["blog.ArticleIndexPage", "legal.LegalIndexPage", "blog.AboutPage"]

    content_panels = Page.content_panels + [
        PageChooserPanel("featured_article", "blog.ArticlePage"),
    ]

    def get_context(self, request, *args, **kwargs):
        ctx = super().get_context(request, *args, **kwargs)
        articles_qs = (
            ArticlePage.objects.live()
            .public()
            .select_related("author", "category")
            .prefetch_related("tags__metadata")
            .order_by("-published_date")
        )
        articles = list(articles_qs[:5])
        ctx["featured_article"] = self.featured_article
        ctx["latest_articles"] = articles
        ctx["more_articles"] = articles[:3]
        ctx["available_tags"] = (
            Tag.objects.filter(
                id__in=ArticlePage.objects.live().public().values_list("tags__id", flat=True)
            ).distinct().order_by("name")
        )
        ctx["available_categories"] = Category.objects.filter(show_in_nav=True).order_by("sort_order", "name")
        return ctx


class ArticleIndexPage(RoutablePageMixin, Page):
    parent_page_types = ["blog.HomePage"]
    subpage_types = ["blog.ArticlePage"]
    ARTICLES_PER_PAGE = 12

    def get_articles(self):
        return (
            ArticlePage.objects.child_of(self)
            .live()
            .select_related("author", "category")
            .prefetch_related("tags__metadata")
            .order_by("-published_date")
        )

    def get_category_url(self, category):
        return f"{self.url}category/{category.slug}/"

    def get_listing_context(self, request, active_category=None):
        tag_slug = request.GET.get("tag")
        articles = self.get_articles()
        available_categories = Category.objects.order_by("sort_order", "name")
        category_links = [
            {"category": category, "url": self.get_category_url(category)}
            for category in available_categories
        ]
        if active_category:
            articles = articles.filter(category=active_category)
        available_tags = (
            Tag.objects.filter(id__in=articles.values_list("tags__id", flat=True)).distinct().order_by("name")
        )
        if tag_slug:
            articles = articles.filter(tags__slug=tag_slug)
        paginator = Paginator(articles, self.ARTICLES_PER_PAGE)
        page_num = request.GET.get("page")
        try:
            page_obj = paginator.page(page_num)
        except PageNotAnInteger:
            page_obj = paginator.page(1)
        except EmptyPage:
            page_obj = paginator.page(paginator.num_pages)
        return {
            "articles": page_obj,
            "paginator": paginator,
            "active_tag": tag_slug,
            "available_tags": available_tags,
            "available_categories": available_categories,
            "category_links": category_links,
            "active_category": active_category,
            "active_category_url": self.get_category_url(active_category) if active_category else "",
        }

    @route(r"^category/(?P<category_slug>[-\w]+)/$")
    def category_listing(self, request, category_slug):
        category = get_object_or_404(Category, slug=category_slug)
        return self.render(request, context_overrides=self.get_listing_context(request, active_category=category))

    def get_context(self, request, *args, **kwargs):
        ctx = super().get_context(request, *args, **kwargs)
        ctx.update(self.get_listing_context(request))
        return ctx


class ArticleTag(TaggedItemBase):
    content_object = ParentalKey("blog.ArticlePage", related_name="tagged_items", on_delete=CASCADE)


class Category(models.Model):
    COLOUR_CHOICES = [("cyan", "Cyan"), ("pink", "Pink"), ("neutral", "Neutral")]

    name = models.CharField(max_length=100, unique=True)
    slug = models.SlugField(unique=True)
    description = models.TextField(blank=True)
    hero_image = models.ForeignKey(
        "wagtailimages.Image", null=True, blank=True, on_delete=SET_NULL, related_name="+"
    )
    colour = models.CharField(max_length=20, choices=COLOUR_CHOICES, default="neutral")
    sort_order = models.IntegerField(default=0)
    show_in_nav = models.BooleanField(default=True)

    panels = [
        FieldPanel("name"),
        FieldPanel("slug"),
        FieldPanel("description"),
        FieldPanel("hero_image"),
        FieldPanel("colour"),
        FieldPanel("sort_order"),
        FieldPanel("show_in_nav"),
    ]

    class Meta:
        ordering = ["sort_order", "name"]
        verbose_name_plural = "categories"

    def __str__(self):
        return self.name


class TagMetadata(models.Model):
    COLOUR_CHOICES = [("cyan", "Cyan"), ("pink", "Pink"), ("neutral", "Neutral")]

    tag = models.OneToOneField("taggit.Tag", on_delete=CASCADE, related_name="metadata")
    colour = models.CharField(max_length=20, choices=COLOUR_CHOICES, default="neutral")

    @classmethod
    def get_fallback_css(cls) -> dict[str, str]:
        return {
            "bg": "bg-zinc-800 dark:bg-zinc-100",
            "text": "text-white dark:text-black",
            "border": "border-zinc-600/20 dark:border-zinc-400/20",
        }

    def get_css_classes(self) -> dict[str, str]:
        mapping = {
            "cyan": {
                "bg": "bg-brand-cyan/10",
                "text": "text-brand-cyan",
                "border": "border-brand-cyan/20",
            },
            "pink": {
                "bg": "bg-brand-pink/10",
                "text": "text-brand-pink",
                "border": "border-brand-pink/20",
            },
            "neutral": self.get_fallback_css(),
        }
        return mapping.get(self.colour, self.get_fallback_css())


class ArticlePageAdminForm(WagtailAdminPageForm):
    SUMMARY_MAX_CHARS = 220

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        for name in ("slug", "author", "category", "summary"):
            if name in self.fields:
                self.fields[name].required = False

        default_author = self._get_default_author(create=False)
        if default_author and not self.initial.get("author"):
            self.initial["author"] = default_author.pk

        default_category = self._get_default_category(create=False)
        if default_category and not self.initial.get("category"):
            self.initial["category"] = default_category.pk

    def clean(self):
        cleaned_data = getattr(self, "cleaned_data", {})
        self._apply_defaults(cleaned_data)
        self.cleaned_data = cleaned_data

        cleaned_data = super().clean()
        self._apply_defaults(cleaned_data)

        if not cleaned_data.get("slug"):
            self.add_error("slug", "Slug is required.")
        if not cleaned_data.get("author"):
            self.add_error("author", "Author is required.")
        if not cleaned_data.get("category"):
            self.add_error("category", "Category is required.")
        if not cleaned_data.get("summary"):
            self.add_error("summary", "Summary is required.")
        return cleaned_data

    def _apply_defaults(self, cleaned_data: dict[str, Any]) -> dict[str, Any]:
        title = (cleaned_data.get("title") or "").strip()

        if not cleaned_data.get("slug") and title:
            cleaned_data["slug"] = self._build_unique_page_slug(title)
        if not cleaned_data.get("author"):
            cleaned_data["author"] = self._get_default_author(create=True)
        if not cleaned_data.get("category"):
            cleaned_data["category"] = self._get_default_category(create=True)
        if not cleaned_data.get("summary"):
            cleaned_data["summary"] = _generate_summary_from_stream(
                cleaned_data.get("body"),
                max_chars=self.SUMMARY_MAX_CHARS,
            ) or title

        return cleaned_data

    def _get_default_author(self, *, create: bool) -> Author | None:
        user = self.for_user
        if not user or not user.is_authenticated:
            return None
        existing = Author.objects.filter(user=user).first()
        if existing or not create:
            return existing

        base_name = (user.get_full_name() or user.get_username() or f"user-{user.pk}").strip()
        base_slug = slugify(base_name) or f"user-{user.pk}"
        slug = base_slug
        suffix = 2
        while Author.objects.filter(slug=slug).exists():
            slug = f"{base_slug}-{suffix}"
            suffix += 1
        return Author.objects.create(user=user, name=base_name, slug=slug)

    def _get_default_category(self, *, create: bool):
        existing = Category.objects.filter(slug="general").first()
        if existing or not create:
            return existing
        category, _ = Category.objects.get_or_create(
            slug="general",
            defaults={"name": "General", "description": "General articles", "colour": "neutral"},
        )
        return category

    def _build_unique_page_slug(self, title: str) -> str:
        base_slug = slugify(title) or "article"
        parent_page = self.parent_page
        if parent_page is None and self.instance.pk:
            parent_page = self.instance.get_parent()
        if parent_page is None:
            return base_slug

        sibling_pages = parent_page.get_children().exclude(pk=self.instance.pk)
        slug = base_slug
        suffix = 2
        while sibling_pages.filter(slug=slug).exists():
            slug = f"{base_slug}-{suffix}"
            suffix += 1
        return slug


class ArticlePage(SeoMixin, Page):
    category = models.ForeignKey("blog.Category", on_delete=PROTECT, related_name="+")
    author = models.ForeignKey("authors.Author", on_delete=PROTECT)
    hero_image = models.ForeignKey(
        "wagtailimages.Image", null=True, blank=True, on_delete=SET_NULL, related_name="+"
    )
    summary = models.TextField()
    body = StreamField(ARTICLE_BODY_BLOCKS, use_json_field=True)
    tags = ClusterTaggableManager(through="blog.ArticleTag", blank=True)
    read_time_mins = models.PositiveIntegerField(editable=False, default=1)
    comments_enabled = models.BooleanField(default=True)
    published_date = models.DateTimeField(
        null=True,
        blank=True,
        help_text="Display date for this article. Auto-set on first publish if left blank.",
    )

    parent_page_types = ["blog.ArticleIndexPage"]
    subpage_types: list[str] = []
    base_form_class = ArticlePageAdminForm

    content_panels = [
        FieldPanel("title"),
        FieldPanel("summary"),
        FieldPanel("body"),
    ]

    metadata_panels = [
        FieldPanel("category"),
        FieldPanel("author"),
        FieldPanel("tags"),
        FieldPanel("hero_image"),
        FieldPanel("comments_enabled"),
    ]

    publishing_panels = [
        FieldPanel("published_date"),
        FieldPanel("go_live_at"),
        FieldPanel("expire_at"),
    ]

    edit_handler = TabbedInterface(
        [
            ObjectList(content_panels, heading="Content"),
            ObjectList(metadata_panels, heading="Metadata"),
            ObjectList(publishing_panels, heading="Publishing"),
            ObjectList(SeoMixin.seo_panels, heading="SEO"),
        ]
    )

    search_fields = Page.search_fields + [
        index.SearchField("summary"),
        index.SearchField("body_text", es_extra={"analyzer": "english"}),
        index.AutocompleteField("title"),
        index.RelatedFields("tags", [
            index.SearchField("name"),
        ]),
        index.FilterField("category"),
        index.FilterField("published_date"),
    ]

    @property
    def body_text(self) -> str:
        """Extract prose text from body StreamField, excluding code blocks."""
        parts: list[str] = []
        for block in self.body:
            if block.block_type == "code":
                continue
            value = block.value
            text = value.source if hasattr(value, "source") else str(value)
            parts.append(text)
        return " ".join(parts)

    def save(self, *args: Any, **kwargs: Any) -> None:
        if not self.category_id:
            self.category, _ = Category.objects.get_or_create(
                slug="general",
                defaults={"name": "General", "description": "General articles", "colour": "neutral"},
            )
        if not (self.summary or "").strip():
            self.summary = _generate_summary_from_stream(self.body) or self.title
        if not self.published_date and self.first_published_at:
            self.published_date = self.first_published_at
        if self._should_refresh_read_time():
            self.read_time_mins = self._compute_read_time()
        return super().save(*args, **kwargs)

    def _should_refresh_read_time(self) -> bool:
        if not self.pk:
            return True

        previous = type(self).objects.only("body").filter(pk=self.pk).first()
        if previous is None:
            return True

        return previous.body_text != self.body_text

    def _compute_read_time(self) -> int:
        words = []
        for block in self.body:
            if block.block_type == "code":
                continue
            value = block.value
            text = value.source if hasattr(value, "source") else str(value)
            words.extend(re.findall(r"\w+", text))
        return max(1, ceil(len(words) / 200))

    def get_tags_with_metadata(self):
        tags = self.tags.all()
        return [(tag, getattr(tag, "metadata", None)) for tag in tags]

    def get_related_articles(self, count: int = 3):
        tag_ids = self.tags.values_list("id", flat=True)
        related = list(
            ArticlePage.objects.live()
            .filter(tags__in=tag_ids)
            .exclude(pk=self.pk)
            .distinct()
            .order_by("-published_date")[:count]
        )
        if len(related) < count:
            exclude_ids = [a.pk for a in related] + [self.pk]
            fallback = list(
                ArticlePage.objects.live()
                .exclude(pk__in=exclude_ids)
                .order_by("-published_date")[: count - len(related)]
            )
            return related + fallback
        return related

    def get_context(self, request, *args, **kwargs):
        ctx = super().get_context(request, *args, **kwargs)
        ctx["related_articles"] = self.get_related_articles()
        from django.conf import settings

        from apps.comments.models import Comment
        from apps.comments.views import _annotate_reaction_counts, _get_session_key

        approved_replies = Comment.objects.filter(is_approved=True).select_related("parent")
        comments = list(
            self.comments.filter(is_approved=True, parent__isnull=True).prefetch_related(
                Prefetch("replies", queryset=approved_replies)
            )
        )
        _annotate_reaction_counts(comments, _get_session_key(request))
        ctx["approved_comments"] = comments
        ctx["turnstile_site_key"] = getattr(settings, "TURNSTILE_SITE_KEY", "")
        return ctx


class AboutPage(Page):
    mission_statement = models.TextField()
    body = RichTextField(blank=True)
    featured_author = models.ForeignKey(
        "authors.Author", null=True, blank=True, on_delete=SET_NULL, related_name="+"
    )

    parent_page_types = ["blog.HomePage"]
    subpage_types: list[str] = []

    content_panels = Page.content_panels + [
        FieldPanel("mission_statement"),
        FieldPanel("body"),
        FieldPanel("featured_author"),
    ]