import logging from typing import TYPE_CHECKING, Self import requests from constance import config from django.contrib.gis.geos import Point from django.db import models from django.db.models import QuerySet from modeltranslation.manager import MultilingualManager, MultilingualQuerySet if TYPE_CHECKING: from engine.core.models import AttributeValue, Product, Stock, Vendor # noqa: F401 logger = logging.getLogger(__name__) class AddressManager(models.Manager): def create(self, **kwargs): if not kwargs.get("raw_data"): raise ValueError("'raw_data' (address string) must be provided.") params: dict[str, str | int | None] = { "format": "json", "addressdetails": 1, "q": kwargs.get("raw_data"), } resp = requests.get(config.NOMINATIM_URL.rstrip("/") + "/search", params=params) resp.raise_for_status() results = resp.json() if not results: raise ValueError( f"No geocoding result for address: {kwargs.get('raw_data')}" ) data = results[0] addr = data.get("address", {}) street = f"{addr.get('road', '') or addr.get('pedestrian', '')}, {addr.get('house_number', '')}" district = addr.get("city_district") or addr.get("suburb") or "" city = addr.get("city") or addr.get("town") or addr.get("village") or "" region = addr.get("state") or addr.get("region") or "" postal_code = addr.get("postcode") or "" country = addr.get("country") or "" try: lat = float(data.get("lat")) lon = float(data.get("lon")) location = Point(lon, lat, srid=4326) except (TypeError, ValueError): location = None try: address_line_1 = kwargs.pop("address_line_1") except KeyError as e: raise ValueError("Missing required field 'address_line_1'") from e try: address_line_2 = kwargs.pop("address_line_2") except KeyError: address_line_2 = "" return super().get_or_create( raw_data=kwargs.get("raw_data"), address_line=f"{address_line_1}, {address_line_2}", street=street, district=district, city=city, region=region, postal_code=postal_code, country=country, user=kwargs.pop("user"), defaults={"api_response": data, "location": location}, )[0] class ProductQuerySet(MultilingualQuerySet["Product"]): """Custom QuerySet for Product with vendor-related operations.""" def available(self) -> Self: """Filter to only available products.""" return self.filter( is_active=True, brand__is_active=True, category__is_active=True, stocks__isnull=False, stocks__vendor__is_active=True, ) def available_in_stock(self) -> Self: """Filter to available products with stock quantity > 0.""" return self.filter( is_active=True, brand__is_active=True, category__is_active=True, stocks__isnull=False, stocks__vendor__is_active=True, stocks__quantity__gt=0, ) def with_related(self) -> Self: """Prefetch related objects for performance.""" return self.select_related("category", "brand").prefetch_related( "tags", "stocks", "images", "attributes__attribute__group" ) def stale(self) -> Self: """Filter products with no orders and no stocks.""" return self.filter(orderproduct__isnull=True, stocks__isnull=True) def for_vendor(self, vendor: "Vendor") -> Self: """Filter products that have stocks from a specific vendor.""" return self.filter(stocks__vendor=vendor) def not_in_orders(self) -> Self: """Filter products that are not part of any order.""" return self.filter(orderproduct__isnull=True) def for_vendor_not_in_orders(self, vendor: "Vendor") -> Self: """Filter products for a vendor that are not in any orders.""" return self.for_vendor(vendor).filter(orderproduct__isnull=True) def mark_inactive(self) -> int: """Bulk update products to inactive status. Returns count updated.""" return self.update(is_active=False) def mark_for_deletion(self, marker: str = "SCHON_DELETED_PRODUCT") -> int: """Mark products for deletion by setting description. Returns count updated.""" return self.update(description=marker) def marked_for_deletion(self, marker: str = "SCHON_DELETED_PRODUCT") -> Self: """Get products marked for deletion.""" return self.filter(description__exact=marker) def delete_with_related(self, batch_size: int = 5000) -> int: """ Delete products in batches along with their related objects. Returns the total number of products deleted. """ from engine.core.models import AttributeValue, ProductImage total_deleted = 0 while True: batch_ids = list(self.values_list("pk", flat=True)[:batch_size]) if not batch_ids: break AttributeValue.objects.filter(product_id__in=batch_ids).delete() ProductImage.objects.filter(product_id__in=batch_ids).delete() deleted_count, _ = self.model.objects.filter(pk__in=batch_ids).delete() total_deleted += deleted_count return total_deleted class ProductManager(MultilingualManager["Product"]): """Manager for Product model with custom queryset methods.""" _queryset_class = ProductQuerySet def get_queryset(self) -> ProductQuerySet: return ProductQuerySet(self.model, using=self._db) def available(self) -> ProductQuerySet: return self.get_queryset().available() def available_in_stock(self) -> ProductQuerySet: return self.get_queryset().available_in_stock() def with_related(self) -> ProductQuerySet: return self.get_queryset().with_related() def stale(self) -> ProductQuerySet: return self.get_queryset().stale() def for_vendor(self, vendor: "Vendor") -> ProductQuerySet: return self.get_queryset().for_vendor(vendor) def not_in_orders(self) -> ProductQuerySet: return self.get_queryset().not_in_orders() def for_vendor_not_in_orders(self, vendor: "Vendor") -> ProductQuerySet: return self.get_queryset().for_vendor_not_in_orders(vendor) class StockQuerySet(QuerySet["Stock"]): """Custom QuerySet for Stock with vendor-related operations.""" def for_vendor(self, vendor: "Vendor") -> Self: """Filter stocks for a specific vendor.""" return self.filter(vendor=vendor) def not_in_orders(self) -> Self: """Filter stocks whose products are not in any orders.""" return self.filter(product__orderproduct__isnull=True) def for_vendor_not_in_orders(self, vendor: "Vendor") -> Self: """Filter stocks for a vendor whose products are not in orders.""" return self.for_vendor(vendor).filter(product__orderproduct__isnull=True) class StockManager(models.Manager["Stock"]): """Manager for Stock model with custom queryset methods.""" def get_queryset(self) -> StockQuerySet: return StockQuerySet(self.model, using=self._db) def for_vendor(self, vendor: "Vendor") -> StockQuerySet: return self.get_queryset().for_vendor(vendor) def not_in_orders(self) -> StockQuerySet: return self.get_queryset().not_in_orders() def for_vendor_not_in_orders(self, vendor: "Vendor") -> StockQuerySet: return self.get_queryset().for_vendor_not_in_orders(vendor) class AttributeValueQuerySet(QuerySet["AttributeValue"]): """Custom QuerySet for AttributeValue with vendor-related operations.""" def for_vendor(self, vendor: "Vendor") -> Self: """Filter attribute values for products from a specific vendor.""" return self.filter(product__stocks__vendor=vendor) def not_in_orders(self) -> Self: """Filter attribute values whose products are not in any orders.""" return self.filter(product__orderproduct__isnull=True) def for_vendor_not_in_orders(self, vendor: "Vendor") -> Self: """Filter attribute values for a vendor whose products are not in orders.""" return self.for_vendor(vendor).filter(product__orderproduct__isnull=True) class AttributeValueManager(models.Manager["AttributeValue"]): """Manager for AttributeValue model with custom queryset methods.""" def get_queryset(self) -> AttributeValueQuerySet: return AttributeValueQuerySet(self.model, using=self._db) def for_vendor(self, vendor: "Vendor") -> AttributeValueQuerySet: return self.get_queryset().for_vendor(vendor) def not_in_orders(self) -> AttributeValueQuerySet: return self.get_queryset().not_in_orders() def for_vendor_not_in_orders(self, vendor: "Vendor") -> AttributeValueQuerySet: return self.get_queryset().for_vendor_not_in_orders(vendor)