schon/engine/core/management/commands/rebuild_slugs.py

56 lines
2 KiB
Python

from typing import Any
from django.core.management.base import BaseCommand
from django.db import transaction
from django.db.models import QuerySet
from django.utils.crypto import get_random_string
from engine.core.models import Brand, Category, Product
# noinspection PyProtectedMember
class Command(BaseCommand):
help = "Rebuild slug field for all slugified instances"
def reset_em(self, queryset: QuerySet[Any]) -> None:
total = queryset.count()
self.stdout.write(
f"Starting slug rebuilding for {total} {queryset.model._meta.verbose_name_plural}"
)
for idx, instance in enumerate(queryset.iterator(), start=1):
try:
while (
queryset.filter(name=instance.name)
.exclude(uuid=instance.uuid)
.count()
>= 1
):
instance.name = f"{instance.name} - {get_random_string(length=3, allowed_chars='0123456789')}"
instance.save()
instance.slug = None
with transaction.atomic():
instance.save(update_fields=["slug"])
self.stdout.write(
self.style.SUCCESS(
f"[{idx}/{total}] ({queryset.model._meta.verbose_name_plural} UUID:"
f" {instance.pk}) slug set to '{instance.slug}'"
)
)
except Exception as e:
self.stderr.write(
self.style.ERROR(
f"[{idx}/{total}] ({queryset.model._meta.verbose_name_plural}: {instance.name}/{instance.uuid})"
f" ERROR: {e}"
)
)
def handle(self, *args: list[Any], **options: dict[Any, Any]) -> None:
for queryset in [
Brand.objects.all(),
Category.objects.all(),
Product.objects.all(),
]:
self.reset_em(queryset)
self.stdout.write(self.style.SUCCESS("Slug rebuild complete."))