text: import datetime
  from typing import TYPE_CHECKING, Iterable, Optional, Union
  from uuid import uuid4
  
  from django.conf import settings
  from django.contrib.postgres.aggregates import StringAgg
  from django.contrib.postgres.indexes import GinIndex
  from django.contrib.postgres.search import SearchVectorField
  from django.db import models
  from django.db.models import JSONField  # type: ignore
  from django.db.models import (
      BooleanField,
      Case,
      Count,
      DateField,
      ExpressionWrapper,
      F,
      FilteredRelation,
      OuterRef,
      Q,
      Subquery,
      Sum,
      TextField,
      Value,
      When,
  )
  from django.db.models.functions import Coalesce
  from django.urls import reverse
  from django.utils.encoding import smart_text
  from django_measurement.models import MeasurementField
  from django_prices.models import MoneyField
  from measurement.measures import Weight
  from mptt.managers import TreeManager
  from mptt.models import MPTTModel
  from versatileimagefield.fields import PPOIField, VersatileImageField
  
  from ..account.utils import requestor_is_staff_member_or_app
  from ..channel.models import Channel
  from ..core.db.fields import SanitizedJSONField
  from ..core.models import ModelWithMetadata, PublishableModel, SortableModel
  from ..core.permissions import ProductPermissions, ProductTypePermissions
  from ..core.utils import build_absolute_uri
  from ..core.utils.draftjs import json_content_to_raw_text
  from ..core.utils.editorjs import clean_editor_js
  from ..core.utils.translations import TranslationProxy
  from ..core.weight import WeightUnits, zero_weight
  from ..discount import DiscountInfo
  from ..discount.utils import calculate_discounted_price
  from ..seo.models import SeoModel, SeoModelTranslation
  from . import ProductMediaTypes
  
  if TYPE_CHECKING:
      # flake8: noqa
      from django.db.models import OrderBy
      from prices import Money
  
      from ..account.models import User
      from ..app.models import App
  
  
  class Category(ModelWithMetadata, MPTTModel, SeoModel):
      name = models.CharField(max_length=250)
      slug = models.SlugField(max_length=255, unique=True, allow_unicode=True)
      description = SanitizedJSONField(blank=True, null=True, sanitizer=clean_editor_js)
      parent = models.ForeignKey(
          "self", null=True, blank=True, related_name="children", on_delete=models.CASCADE
      )
      background_image = VersatileImageField(
          upload_to="category-backgrounds", blank=True, null=True
      )
      background_image_alt = models.CharField(max_length=128, blank=True)
  
      objects = models.Manager()
      tree = TreeManager()
      translated = TranslationProxy()
  
      def __str__(self) -> str:
          return self.name
  
  
  class CategoryTranslation(SeoModelTranslation):
      language_code = models.CharField(max_length=10)
      category = models.ForeignKey(
          Category, related_name="translations", on_delete=models.CASCADE
      )
      name = models.CharField(max_length=128)
      description = SanitizedJSONField(blank=True, null=True, sanitizer=clean_editor_js)
  
      class Meta:
          unique_together = (("language_code", "category"),)
  
      def __str__(self) -> str:
          return self.name
  
      def __repr__(self) -> str:
          class_ = type(self)
          return "%s(pk=%r, name=%r, category_pk=%r)" % (
              class_.__name__,
              self.pk,
              self.name,
              self.category_id,
          )
  
  
  class ProductType(ModelWithMetadata):
      name = models.CharField(max_length=250)
      slug = models.SlugField(max_length=255, unique=True, allow_unicode=True)
      has_variants = models.BooleanField(default=True)
      is_shipping_required = models.BooleanField(default=True)
      is_digital = models.BooleanField(default=False)
      weight = MeasurementField(
          measurement=Weight, unit_choices=WeightUnits.CHOICES, default=zero_weight
      )
  
      class Meta(ModelWithMetadata.Meta):
          ordering = ("slug",)
          app_label = "product"
          permissions = (
              (
                  ProductTypePermissions.MANAGE_PRODUCT_TYPES_AND_ATTRIBUTES.codename,
                  "Manage product types and attributes.",
              ),
          )
  
      def __str__(self) -> str:
          return self.name
  
      def __repr__(self) -> str:
          class_ = type(self)
          return "<%s.%s(pk=%r, name=%r)>" % (
              class_.__module__,
              class_.__name__,
              self.pk,
              self.name,
          )
  
  
  class ProductsQueryset(models.QuerySet):
      def published(self, channel_slug: str):
          today = datetime.date.today()
          return self.filter(
              Q(channel_listings__publication_date__lte=today)
              | Q(channel_listings__publication_date__isnull=True),
              channel_listings__channel__slug=str(channel_slug),
              channel_listings__channel__is_active=True,
              channel_listings__is_published=True,
          )
  
      def not_published(self, channel_slug: str):
          today = datetime.date.today()
          return self.annotate_publication_info(channel_slug).filter(
              Q(publication_date__gt=today) & Q(is_published=True)
              | Q(is_published=False)
              | Q(is_published__isnull=True)
          )
  
      def published_with_variants(self, channel_slug: str):
          published = self.published(channel_slug)
          query = ProductVariantChannelListing.objects.filter(
              variant_id=OuterRef("variants__id"),
              channel__slug=str(channel_slug),
              price_amount__isnull=False,
          ).values_list("variant", flat=True)
          return published.filter(variants__in=query).distinct()
  
      def visible_to_user(self, requestor: Union["User", "App"], channel_slug: str):
          if requestor_is_staff_member_or_app(requestor):
              if channel_slug:
                  return self.filter(channel_listings__channel__slug=str(channel_slug))
              return self.all()
          return self.published_with_variants(channel_slug)
  
      def annotate_publication_info(self, channel_slug: str):
          return self.annotate_is_published(channel_slug).annotate_publication_date(
              channel_slug
          )
  
      def annotate_is_published(self, channel_slug: str):
          query = Subquery(
              ProductChannelListing.objects.filter(
                  product_id=OuterRef("pk"), channel__slug=str(channel_slug)
              ).values_list("is_published")[:1]
          )
          return self.annotate(
              is_published=ExpressionWrapper(query, output_field=BooleanField())
          )
  
      def annotate_publication_date(self, channel_slug: str):
          query = Subquery(
              ProductChannelListing.objects.filter(
                  product_id=OuterRef("pk"), channel__slug=str(channel_slug)
              ).values_list("publication_date")[:1]
          )
          return self.annotate(
              publication_date=ExpressionWrapper(query, output_field=DateField())
          )
  
      def annotate_visible_in_listings(self, channel_slug):
          query = Subquery(
              ProductChannelListing.objects.filter(
                  product_id=OuterRef("pk"), channel__slug=str(channel_slug)
              ).values_list("visible_in_listings")[:1]
          )
          return self.annotate(
              visible_in_listings=ExpressionWrapper(query, output_field=BooleanField())
          )
  
      def sort_by_attribute(
          self, attribute_pk: Union[int, str], descending: bool = False
      ):
          """Sort a query set by the values of the given product attribute.
  
          :param attribute_pk: The database ID (must be a numeric) of the attribute
                               to sort by.
          :param descending: The sorting direction.
          """
          from ..attribute.models import AttributeProduct, AttributeValue
  
          qs: models.QuerySet = self
          # If the passed attribute ID is valid, execute the sorting
          if not (isinstance(attribute_pk, int) or attribute_pk.isnumeric()):
              return qs.annotate(
                  concatenated_values_order=Value(
                      None, output_field=models.IntegerField()
                  ),
                  concatenated_values=Value(None, output_field=models.CharField()),
              )
  
          # Retrieve all the products' attribute data IDs (assignments) and
          # product types that have the given attribute associated to them
          associated_values = tuple(
              AttributeProduct.objects.filter(attribute_id=attribute_pk).values_list(
                  "pk", "product_type_id"
              )
          )
  
          if not associated_values:
              qs = qs.annotate(
                  concatenated_values_order=Value(
                      None, output_field=models.IntegerField()
                  ),
                  concatenated_values=Value(None, output_field=models.CharField()),
              )
  
          else:
              attribute_associations, product_types_associated_to_attribute = zip(
                  *associated_values
              )
  
              qs = qs.annotate(
                  # Contains to retrieve the attribute data (singular) of each product
                  # Refer to `AttributeProduct`.
                  filtered_attribute=FilteredRelation(
                      relation_name="attributes",
                      condition=Q(attributes__assignment_id__in=attribute_associations),
                  ),
                  # Implicit `GROUP BY` required for the `StringAgg` aggregation
                  grouped_ids=Count("id"),
                  # String aggregation of the attribute's values to efficiently sort them
                  concatenated_values=Case(
                      # If the product has no association data but has
                      # the given attribute associated to its product type,
                      # then consider the concatenated values as empty (non-null).
                      When(
                          Q(product_type_id__in=product_types_associated_to_attribute)
                          & Q(filtered_attribute=None),
                          then=models.Value(""),
                      ),
                      default=StringAgg(
                          F("filtered_attribute__values__name"),
                          delimiter=",",
                          ordering=(
                              [
                                  f"filtered_attribute__values__{field_name}"
                                  for field_name in AttributeValue._meta.ordering or []
                              ]
                          ),
                      ),
                      output_field=models.CharField(),
                  ),
                  concatenated_values_order=Case(
                      # Make the products having no such attribute be last in the sorting
                      When(concatenated_values=None, then=2),
                      # Put the products having an empty attribute value at the bottom of
                      # the other products.
                      When(concatenated_values="", then=1),
                      # Put the products having an attribute value to be always at the top
                      default=0,
                      output_field=models.IntegerField(),
                  ),
              )
  
          # Sort by concatenated_values_order then
          # Sort each group of products (0, 1, 2, ...) per attribute values
          # Sort each group of products by name,
          # if they have the same values or not values
          ordering = "-" if descending else ""
          return qs.order_by(
              f"{ordering}concatenated_values_order",
              f"{ordering}concatenated_values",
              f"{ordering}name",
          )
  
      def prefetched_for_webhook(self, single_object=True):
          if single_object:
              return self.prefetch_related(
                  "attributes__values",
                  "attributes__assignment__attribute",
                  "media",
              )
          return self.prefetch_related(
              "attributes__values",
              "attributes__assignment__attribute",
              "collections",
              "variants__stocks__allocations",
              "media",
              "category",
          )
  
  
  class Product(SeoModel, ModelWithMetadata):
      product_type = models.ForeignKey(
          ProductType, related_name="products", on_delete=models.CASCADE
      )
      name = models.CharField(max_length=250)
      slug = models.SlugField(max_length=255, unique=True, allow_unicode=True)
      description = SanitizedJSONField(blank=True, null=True, sanitizer=clean_editor_js)
      description_plaintext = TextField(blank=True)
      search_vector = SearchVectorField(null=True, blank=True)
  
      category = models.ForeignKey(
          Category,
          related_name="products",
          on_delete=models.SET_NULL,
          null=True,
          blank=True,
      )
      updated_at = models.DateTimeField(auto_now=True, null=True)
      charge_taxes = models.BooleanField(default=True)
      weight = MeasurementField(
          measurement=Weight, unit_choices=WeightUnits.CHOICES, blank=True, null=True
      )
      default_variant = models.OneToOneField(
          "ProductVariant",
          blank=True,
          null=True,
          on_delete=models.SET_NULL,
          related_name="+",
      )
      rating = models.FloatField(null=True, blank=True)
  
      objects = ProductsQueryset.as_manager()
      translated = TranslationProxy()
  
      class Meta:
          app_label = "product"
          ordering = ("slug",)
          permissions = (
              (ProductPermissions.MANAGE_PRODUCTS.codename, "Manage products."),
          )
          indexes = [GinIndex(fields=["search_vector"])]
          indexes.extend(ModelWithMetadata.Meta.indexes)
  
      def __iter__(self):
          if not hasattr(self, "__variants"):
              setattr(self, "__variants", self.variants.all())
          return iter(getattr(self, "__variants"))
  
      def __repr__(self) -> str:
          class_ = type(self)
          return "<%s.%s(pk=%r, name=%r)>" % (
              class_.__module__,
              class_.__name__,
              self.pk,
              self.name,
          )
  
      def __str__(self) -> str:
          return self.name
  
      @property
      def plain_text_description(self) -> str:
          return json_content_to_raw_text(self.description)
  
      def get_first_image(self):
          all_media = self.media.all()
          images = [media for media in all_media if media.type == ProductMediaTypes.IMAGE]
          return images[0] if images else None
  
      @staticmethod
      def sort_by_attribute_fields() -> list:
          return ["concatenated_values_order", "concatenated_values", "name"]
  
  
  class ProductTranslation(SeoModelTranslation):
      language_code = models.CharField(max_length=10)
      product = models.ForeignKey(
          Product, related_name="translations", on_delete=models.CASCADE
      )
      name = models.CharField(max_length=250)
      description = SanitizedJSONField(blank=True, null=True, sanitizer=clean_editor_js)
  
      class Meta:
          unique_together = (("language_code", "product"),)
  
      def __str__(self) -> str:
          return self.name
  
      def __repr__(self) -> str:
          class_ = type(self)
          return "%s(pk=%r, name=%r, product_pk=%r)" % (
              class_.__name__,
              self.pk,
              self.name,
              self.product_id,
          )
  
  
  class ProductVariantQueryset(models.QuerySet):
      def annotate_quantities(self):
          return self.annotate(
              quantity=Coalesce(Sum("stocks__quantity"), 0),
              quantity_allocated=Coalesce(
                  Sum("stocks__allocations__quantity_allocated"), 0
              ),
          )
  
      def available_in_channel(self, channel_slug):
          return self.filter(
              channel_listings__price_amount__isnull=False,
              channel_listings__channel__slug=channel_slug,
          )
  
      def prefetched_for_webhook(self):
          return self.prefetch_related(
              "attributes__values",
              "attributes__assignment__attribute",
              "variant_media__media",
          )
  
  
  class ProductChannelListing(PublishableModel):
      product = models.ForeignKey(
          Product,
          null=False,
          blank=False,
          related_name="channel_listings",
          on_delete=models.CASCADE,
      )
      channel = models.ForeignKey(
          Channel,
          null=False,
          blank=False,
          related_name="product_listings",
          on_delete=models.CASCADE,
      )
      visible_in_listings = models.BooleanField(default=False)
      available_for_purchase = models.DateField(blank=True, null=True)
      currency = models.CharField(max_length=settings.DEFAULT_CURRENCY_CODE_LENGTH)
      discounted_price_amount = models.DecimalField(
          max_digits=settings.DEFAULT_MAX_DIGITS,
          decimal_places=settings.DEFAULT_DECIMAL_PLACES,
          blank=True,
          null=True,
      )
      discounted_price = MoneyField(
          amount_field="discounted_price_amount", currency_field="currency"
      )
  
      class Meta:
          unique_together = [["product", "channel"]]
          ordering = ("pk",)
  
      def is_available_for_purchase(self):
          return (
              self.available_for_purchase is not None
              and datetime.date.today() >= self.available_for_purchase
          )
  
  
  class ProductVariant(SortableModel, ModelWithMetadata):
      sku = models.CharField(max_length=255, unique=True)
      name = models.CharField(max_length=255, blank=True)
      product = models.ForeignKey(
          Product, related_name="variants", on_delete=models.CASCADE
      )
      media = models.ManyToManyField("ProductMedia", through="VariantMedia")
      track_inventory = models.BooleanField(default=True)
  
      weight = MeasurementField(
          measurement=Weight, unit_choices=WeightUnits.CHOICES, blank=True, null=True
      )
  
      objects = ProductVariantQueryset.as_manager()
      translated = TranslationProxy()
  
      class Meta(ModelWithMetadata.Meta):
          ordering = ("sort_order", "sku")
          app_label = "product"
  
      def __str__(self) -> str:
          return self.name or self.sku
  
      def get_price(
          self,
          product: Product,
          collections: Iterable["Collection"],
          channel: Channel,
          channel_listing: "ProductVariantChannelListing",
          discounts: Optional[Iterable[DiscountInfo]] = None,
      ) -> "Money":
          return calculate_discounted_price(
              product=product,
              price=channel_listing.price,
              discounts=discounts,
              collections=collections,
              channel=channel,
          )
  
      def get_weight(self):
          return self.weight or self.product.weight or self.product.product_type.weight
  
      def is_shipping_required(self) -> bool:
          return self.product.product_type.is_shipping_required
  
      def is_digital(self) -> bool:
          is_digital = self.product.product_type.is_digital
          return not self.is_shipping_required() and is_digital
  
      def display_product(self, translated: bool = False) -> str:
          if translated:
              product = self.product.translated
              variant_display = str(self.translated)
          else:
              variant_display = str(self)
              product = self.product
          product_display = (
              f"{product} ({variant_display})" if variant_display else str(product)
          )
          return smart_text(product_display)
  
      def get_ordering_queryset(self):
          return self.product.variants.all()
  
  
  class ProductVariantTranslation(models.Model):
      language_code = models.CharField(max_length=10)
      product_variant = models.ForeignKey(
          ProductVariant, related_name="translations", on_delete=models.CASCADE
      )
      name = models.CharField(max_length=255, blank=True)
  
      translated = TranslationProxy()
  
      class Meta:
          unique_together = (("language_code", "product_variant"),)
  
      def __repr__(self):
          class_ = type(self)
          return "%s(pk=%r, name=%r, variant_pk=%r)" % (
              class_.__name__,
              self.pk,
              self.name,
              self.product_variant_id,
          )
  
      def __str__(self):
          return self.name or str(self.product_variant)
  
  
  class ProductVariantChannelListing(models.Model):
      variant = models.ForeignKey(
          ProductVariant,
          null=False,
          blank=False,
          related_name="channel_listings",
          on_delete=models.CASCADE,
      )
      channel = models.ForeignKey(
          Channel,
          null=False,
          blank=False,
          related_name="variant_listings",
          on_delete=models.CASCADE,
      )
      currency = models.CharField(max_length=settings.DEFAULT_CURRENCY_CODE_LENGTH)
      price_amount = models.DecimalField(
          max_digits=settings.DEFAULT_MAX_DIGITS,
          decimal_places=settings.DEFAULT_DECIMAL_PLACES,
          blank=True,
          null=True,
      )
      price = MoneyField(amount_field="price_amount", currency_field="currency")
  
      cost_price_amount = models.DecimalField(
          max_digits=settings.DEFAULT_MAX_DIGITS,
          decimal_places=settings.DEFAULT_DECIMAL_PLACES,
          blank=True,
          null=True,
      )
      cost_price = MoneyField(amount_field="cost_price_amount", currency_field="currency")
  
      class Meta:
          unique_together = [["variant", "channel"]]
          ordering = ("pk",)
  
  
  class DigitalContent(ModelWithMetadata):
      FILE = "file"
      TYPE_CHOICES = ((FILE, "digital_product"),)
      use_default_settings = models.BooleanField(default=True)
      automatic_fulfillment = models.BooleanField(default=False)
      content_type = models.CharField(max_length=128, default=FILE, choices=TYPE_CHOICES)
      product_variant = models.OneToOneField(
          ProductVariant, related_name="digital_content", on_delete=models.CASCADE
      )
      content_file = models.FileField(upload_to="digital_contents", blank=True)
      max_downloads = models.IntegerField(blank=True, null=True)
      url_valid_days = models.IntegerField(blank=True, null=True)
  
      def create_new_url(self) -> "DigitalContentUrl":
          return self.urls.create()
  
  
  class DigitalContentUrl(models.Model):
      token = models.UUIDField(editable=False, unique=True)
      content = models.ForeignKey(
          DigitalContent, related_name="urls", on_delete=models.CASCADE
      )
      created = models.DateTimeField(auto_now_add=True)
      download_num = models.IntegerField(default=0)
      line = models.OneToOneField(
          "order.OrderLine",
          related_name="digital_content_url",
          blank=True,
          null=True,
          on_delete=models.CASCADE,
      )
  
      def save(
          self, force_insert=False, force_update=False, using=None, update_fields=None
      ):
          if not self.token:
              self.token = str(uuid4()).replace("-", "")
          super().save(force_insert, force_update, using, update_fields)
  
      def get_absolute_url(self) -> Optional[str]:
          url = reverse("digital-product", kwargs={"token": str(self.token)})
          return build_absolute_uri(url)
  
  
  class ProductMedia(SortableModel):
      product = models.ForeignKey(Product, related_name="media", on_delete=models.CASCADE)
      image = VersatileImageField(
          upload_to="products", ppoi_field="ppoi", blank=True, null=True
      )
      ppoi = PPOIField()
      alt = models.CharField(max_length=128, blank=True)
      type = models.CharField(
          max_length=32,
          choices=ProductMediaTypes.CHOICES,
          default=ProductMediaTypes.IMAGE,
      )
      external_url = models.CharField(max_length=256, blank=True, null=True)
      oembed_data = JSONField(blank=True, default=dict)
  
      class Meta:
          ordering = ("sort_order", "pk")
          app_label = "product"
  
      def get_ordering_queryset(self):
          return self.product.media.all()
  
  
  class VariantMedia(models.Model):
      variant = models.ForeignKey(
          "ProductVariant", related_name="variant_media", on_delete=models.CASCADE
      )
      media = models.ForeignKey(
          ProductMedia, related_name="variant_media", on_delete=models.CASCADE
      )
  
      class Meta:
          unique_together = ("variant", "media")
  
  
  class CollectionProduct(SortableModel):
      collection = models.ForeignKey(
          "Collection", related_name="collectionproduct", on_delete=models.CASCADE
      )
      product = models.ForeignKey(
          Product, related_name="collectionproduct", on_delete=models.CASCADE
      )
  
      class Meta:
          unique_together = (("collection", "product"),)
  
      def get_ordering_queryset(self):
          return self.product.collectionproduct.all()
  
  
  class CollectionsQueryset(models.QuerySet):
      def published(self, channel_slug: str):
          today = datetime.date.today()
          return self.filter(
              Q(channel_listings__publication_date__lte=today)
              | Q(channel_listings__publication_date__isnull=True),
              channel_listings__channel__slug=str(channel_slug),
              channel_listings__channel__is_active=True,
              channel_listings__is_published=True,
          )
  
      def visible_to_user(self, requestor: Union["User", "App"], channel_slug: str):
          if requestor_is_staff_member_or_app(requestor):
              if channel_slug:
                  return self.filter(channel_listings__channel__slug=str(channel_slug))
              return self.all()
          return self.published(channel_slug)
  
  
  class Collection(SeoModel, ModelWithMetadata):
      name = models.CharField(max_length=250, unique=True)
      slug = models.SlugField(max_length=255, unique=True, allow_unicode=True)
      products = models.ManyToManyField(
          Product,
          blank=True,
          related_name="collections",
          through=CollectionProduct,
          through_fields=("collection", "product"),
      )
      background_image = VersatileImageField(
          upload_to="collection-backgrounds", blank=True, null=True
      )
      background_image_alt = models.CharField(max_length=128, blank=True)
      description = SanitizedJSONField(blank=True, null=True, sanitizer=clean_editor_js)
  
      objects = CollectionsQueryset.as_manager()
  
      translated = TranslationProxy()
  
      class Meta(ModelWithMetadata.Meta):
          ordering = ("slug",)
  
      def __str__(self) -> str:
          return self.name
  
  
  class CollectionChannelListing(PublishableModel):
      collection = models.ForeignKey(
          Collection,
          null=False,
          blank=False,
          related_name="channel_listings",
          on_delete=models.CASCADE,
      )
      channel = models.ForeignKey(
          Channel,
          null=False,
          blank=False,
          related_name="collection_listings",
          on_delete=models.CASCADE,
      )
  
      class Meta:
          unique_together = [["collection", "channel"]]
          ordering = ("pk",)
  
  
  class CollectionTranslation(SeoModelTranslation):
      language_code = models.CharField(max_length=10)
      collection = models.ForeignKey(
          Collection, related_name="translations", on_delete=models.CASCADE
      )
      name = models.CharField(max_length=128)
      description = SanitizedJSONField(blank=True, null=True, sanitizer=clean_editor_js)
  
      class Meta:
          unique_together = (("language_code", "collection"),)
  
      def __repr__(self):
          class_ = type(self)
          return "%s(pk=%r, name=%r, collection_pk=%r)" % (
              class_.__name__,
              self.pk,
              self.name,
              self.collection_id,
          )
  
      def __str__(self) -> str:
          return self.name
  