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

import logging
from datetime import datetime
from typing import Any

from django.conf import settings
from django.contrib.auth.models import Permission
from django.core.management.base import BaseCommand
from django.db.models import Q

from engine.core.models import Vendor
from engine.vibes_auth.models import Group, User

logger = logging.getLogger(__name__)

user_support_permissions = [
    "add_chatmessage",
    "view_chatmessage",
    "change_chatthread",
    "view_chatthread",
    "view_order",
    "view_ordercrmlink",
    "view_orderproduct",
    "view_address",
    "view_user",
    "view_wishlist",
    "view_balance",
    "view_transaction",
]

stock_manager_permissions = [
    "add_stock",
    "change_stock",
    "view_stock",
    "add_category",
    "change_category",
    "view_category",
    "add_brand",
    "change_brand",
    "view_brand",
    "add_product",
    "change_product",
    "view_product",
    "add_attribute",
    "change_attribute",
    "view_attribute",
    "add_attributevalue",
    "change_attributevalue",
    "view_attributevalue",
    "add_attributegroup",
    "change_attributegroup",
    "view_attributegroup",
    "add_productimage",
    "change_productimage",
    "view_productimage",
]

head_stock_manager_permissions = [
    *stock_manager_permissions,
    "delete_brand",
    "delete_category",
    "delete_stock",
    "delete_product",
    "delete_attribute",
    "delete_attributevalue",
    "delete_attributegroup",
    "delete_productimage",
    "add_vendor",
    "change_vendor",
    "view_vendor",
    "view_user",
]

marketing_admin_permissions = [
    "view_feedback",
    "delete_feedback",
    "add_producttag",
    "change_producttag",
    "delete_producttag",
    "view_producttag",
    "add_categorytag",
    "change_categorytag",
    "delete_categorytag",
    "view_categorytag",
    "add_promocode",
    "change_promocode",
    "delete_promocode",
    "view_promocode",
    "add_promotion",
    "change_promotion",
    "delete_promotion",
    "view_promotion",
    "view_wishlist",
    "add_post",
    "change_post",
    "delete_post",
    "view_post",
    "add_posttag",
    "change_posttag",
    "delete_posttag",
    "view_posttag",
    "add_attachment",
    "change_attachment",
    "delete_attachment",
    "view_attachment",
    "add_messageattachment",
    "change_messageattachment",
    "delete_messageattachment",
    "view_messageattachment",
]

e_commerce_admin_permissions = [
    *user_support_permissions,
    *head_stock_manager_permissions,
    *marketing_admin_permissions,
    "delete_address",
    "change_balance",
    "delete_attachment",
    "add_constance",
    "change_constance",
    "delete_constance",
    "view_constance",
    "change_config",
    "view_config",
    "add_gateway",
    "change_gateway",
    "delete_gateway",
    "view_gateway",
    "view_periodictask",
    "view_periodictasks",
    "change_user",
    "view_group",
]


class Command(BaseCommand):
    def handle(self, *args: list[Any], **options: dict[Any, Any]) -> None:
        self.stdout.write("Initializing must-have instances...")

        initialized_path = settings.BASE_DIR / ".initialized"

        stored_date: datetime | None = None
        if initialized_path.exists():
            try:
                content = initialized_path.read_text(encoding="utf-8").strip()
                if content:
                    stored_date = datetime.fromisoformat(content)
            except Exception as exc:
                logger.warning("Failed to parse .initialized content: %s", exc)

        if stored_date is None:
            stored_date = datetime.min

        if not (settings.RELEASE_DATE > stored_date):
            self.stdout.write(self.style.WARNING("Initialization skipped: already up-to-date."))
            return

        Vendor.objects.get_or_create(name="INNER")

        user_support, is_user_support_created = Group.objects.get_or_create(name="User Support")
        if is_user_support_created:
            perms = Permission.objects.filter(codename__in=user_support_permissions)
            user_support.permissions.add(*perms)

        stock_manager, is_stock_manager_created = Group.objects.get_or_create(name="Stock Manager")
        if is_stock_manager_created:
            perms = Permission.objects.filter(codename__in=stock_manager_permissions)
            stock_manager.permissions.add(*perms)

        head_stock_manager, is_head_stock_manager_created = Group.objects.get_or_create(name="Head Stock Manager")
        if is_head_stock_manager_created:
            perms = Permission.objects.filter(codename__in=head_stock_manager_permissions)
            head_stock_manager.permissions.add(*perms)

        marketing_admin, is_marketing_admin_created = Group.objects.get_or_create(name="Marketing Admin")
        if is_marketing_admin_created:
            perms = Permission.objects.filter(codename__in=marketing_admin_permissions)
            marketing_admin.permissions.add(*perms)

        e_commerce_admin, is_e_commerce_admin_created = Group.objects.get_or_create(name="E-Commerce Admin")
        if is_e_commerce_admin_created:
            perms = Permission.objects.filter(codename__in=e_commerce_admin_permissions)
            e_commerce_admin.permissions.add(*perms)

        valid_codes = [code for code, _ in settings.LANGUAGES]
        (User.objects.filter(Q(language="") | ~Q(language__in=valid_codes)).update(language=settings.LANGUAGE_CODE))

        try:
            if not settings.DEBUG:
                initialized_path.write_text(settings.RELEASE_DATE.isoformat(), encoding="utf-8")
        except Exception as exc:
            logger.error("Failed to update .initialized file: %s", exc)

        self.stdout.write(self.style.SUCCESS("Successfully initialized must-have instances!"))