mandelstudio/mandelstudio/management/commands/fix_no_credit_card_text.py

from __future__ import annotations

from typing import Any
from urllib.parse import unquote

from django.core.management.base import BaseCommand, CommandError
from django.db import transaction

from wagtail.blocks import StreamValue
from wagtail.models import Locale, Page

NO_CC_TEXT = {
    "nl": "No credit card required",
    "en": "No credit card required",
    "de": "Keine Kreditkarte erforderlich",
    "fr": "Aucune carte bancaire requise",
    "es": "No se requiere tarjeta",
    "it": "Nessuna carta richiesta",
    "pt": "Não é necessário cartão",
    "ru": "Карта не требуется",
}


def _localized_url(source_id: int, locale: Locale) -> str | None:
    source = Page.objects.get(id=source_id)
    translated = (
        Page.objects.filter(translation_key=source.translation_key, locale=locale)
        .specific()
        .first()
    )
    chosen = translated or source
    return getattr(chosen, "url", None)


def _fix_cta_footer(
    value: dict[str, Any],
    *,
    no_cc_text: str,
    primary_url: str | None,
    secondary_url: str | None,
) -> bool:
    changed = False

    raw = value.get("no_credit_card_text")
    if isinstance(raw, str):
        cleaned = raw.replace(""", '"').strip()
        # Remove any accidental "language detector" message fragments.
        if "is not Dutch" in cleaned or "translation from" in cleaned:
            cleaned = no_cc_text
        if cleaned != no_cc_text:
            value["no_credit_card_text"] = no_cc_text
            changed = True
    else:
        value["no_credit_card_text"] = no_cc_text
        changed = True

    # Ensure CTA URLs are readable and not percent-encoded.
    for key in ("primary_cta_url", "secondary_cta_url"):
        url = value.get(key)
        if isinstance(url, str) and "%" in url:
            decoded = unquote(url)
            if decoded != url:
                value[key] = decoded
                changed = True

    # Align CTA URLs to the translated contact/services pages when available.
    if primary_url and value.get("primary_cta_url") != primary_url:
        value["primary_cta_url"] = primary_url
        changed = True
    if secondary_url and value.get("secondary_cta_url") != secondary_url:
        value["secondary_cta_url"] = secondary_url
        changed = True

    return changed


class Command(BaseCommand):
    help = (
        "Fix corrupted 'no credit card' copy in saas_cta_footer blocks across locales"
    )

    def add_arguments(self, parser):
        parser.add_argument(
            "--page-id",
            type=int,
            help="Optional Wagtail page id to fix (overrides capabilities lookup).",
        )
        parser.add_argument(
            "--apply",
            action="store_true",
            help="Persist and publish changes (default is dry-run).",
        )

    def handle(self, *args, **options):
        apply_changes = options["apply"]
        page_id = options.get("page_id")

        with transaction.atomic():
            if page_id:
                page = Page.objects.filter(id=page_id).specific().first()
                if page is None:
                    raise CommandError(f"Page id={page_id} not found")
                self._fix_page(page, apply_changes=apply_changes)
            else:
                nl_locale = Locale.objects.filter(language_code="nl").first()
                if nl_locale is None:
                    raise CommandError("Locale nl not found")

                source = (
                    Page.objects.filter(locale=nl_locale, slug="mogelijkheden")
                    .specific()
                    .first()
                )
                if source is None:
                    raise CommandError(
                        "Could not find source page nl/slug=mogelijkheden"
                    )

                for locale in Locale.objects.all().order_by("language_code"):
                    code = locale.language_code
                    page = (
                        Page.objects.filter(
                            translation_key=source.translation_key, locale=locale
                        )
                        .specific()
                        .first()
                    )
                    if page is None:
                        self.stdout.write(
                            f"SKIP {code}: no translation for mogelijkheden"
                        )
                        continue
                    self._fix_page(page, apply_changes=apply_changes)

            if not apply_changes:
                raise CommandError(
                    "Dry-run complete. Re-run with --apply to persist changes."
                )

    def _fix_page(self, page: Page, *, apply_changes: bool) -> None:
        locale = page.locale
        code = locale.language_code

        specific = page.specific
        if not hasattr(specific, "body"):
            self.stdout.write(f"SKIP {code}: no body streamfield")
            return

        body = specific.body
        raw_data = list(body.raw_data)
        no_cc = NO_CC_TEXT.get(code) or NO_CC_TEXT["en"]
        contact_url = _localized_url(131, locale)  # contact
        services_url = _localized_url(129, locale)  # services
        changed = False

        for block in raw_data:
            if block.get("type") != "saas_cta_footer":
                continue
            value = block.get("value")
            if not isinstance(value, dict):
                continue
            if _fix_cta_footer(
                value,
                no_cc_text=no_cc,
                primary_url=contact_url,
                secondary_url=services_url,
            ):
                block["value"] = value
                changed = True

        if not changed:
            self.stdout.write(f"OK   {code}: no cta footer changes needed")
            return

        self.stdout.write(
            f"CHG  {code}: fixed saas_cta_footer no_credit_card_text/urls"
        )
        specific.body = StreamValue(body.stream_block, raw_data, is_lazy=True)
        if apply_changes:
            rev = specific.save_revision()
            rev.publish()