from collections import defaultdict from decimal import Decimal from django.db import models from django.utils import timezone from shop_management.models.base import DataRoot from shop_management.models.data.supplier import Supplier from shop_management.models.data.product import Product from shop_management.models.data.shop import Shop from shop_management.models.data.utils import Unit, Currency, PaymentStatus from shop_management.models.data.organization import Warehouse class Purchase(DataRoot): class DocumentStatus(models.TextChoices): DRAFT = "DRAFT", "Draft" POSTED = "POSTED", "Posted" VOID = "VOID", "Cancelled" shop = models.ForeignKey(Shop, on_delete=models.CASCADE, related_name='purchases') supplier = models.ForeignKey( Supplier, on_delete=models.SET_NULL, null=True, blank=True, related_name='purchases' ) # Moved currency from items to purchase currency = models.ForeignKey( Currency, on_delete=models.SET_NULL, null=True, blank=True, related_name='purchases' ) # Purchase date defaults to current time if not provided. purchase_date = models.DateField(default=timezone.now) # Optional amount and received amount amount = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True, default=0) received_amount = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True, default=0) status = models.IntegerField(choices=PaymentStatus.choices, default=PaymentStatus.PENDING) is_deleted = models.BooleanField(default=False) purchase_number = models.CharField(max_length=50, null=True, blank=True) supplier_invoice_number = models.CharField(max_length=100, blank=True) warehouse = models.ForeignKey( Warehouse, null=True, blank=True, on_delete=models.PROTECT, related_name="purchases", ) due_date = models.DateField(null=True, blank=True) exchange_rate_used = models.DecimalField( max_digits=24, decimal_places=10, null=True, blank=True ) document_status = models.CharField( max_length=10, choices=DocumentStatus.choices, default=DocumentStatus.DRAFT ) uses_inventory_engine = models.BooleanField(default=False) posted_at = models.DateTimeField(null=True, blank=True) posted_by = models.ForeignKey( "account.User", null=True, blank=True, on_delete=models.PROTECT, related_name="purchases_posted", ) voided_at = models.DateTimeField(null=True, blank=True) voided_by = models.ForeignKey( "account.User", null=True, blank=True, on_delete=models.PROTECT, related_name="purchases_voided", ) void_reason = models.TextField(blank=True) notes = models.TextField(blank=True) class Meta: constraints = [ models.UniqueConstraint( fields=["shop", "purchase_number"], name="unique_purchase_number_per_shop" ) ] def __str__(self): supplier_name = self.supplier.name if self.supplier else "Unknown Supplier" return f"Purchase for {self.shop.name} from {supplier_name}" def _is_stock_transfer_purchase(self): supplier_name = str(getattr(getattr(self, "supplier", None), "name", "") or "").strip().lower() return supplier_name.startswith("transfer from ") @property def total_price(self): """ Group totals by item currency so mixed-currency transfer purchases stay separate. Purchase-level discount (`amount`) is applied only for normal purchases. For "Transfer from …" purchases, `amount` stores the transfer invoice total (not a discount) — subtracting it would wipe Purchase Unpaid on the dashboard. """ totals = defaultdict(Decimal) for item in self.purchase_items.all(): item_currency = item.currency or self.currency key = item_currency.title if item_currency else "Unknown Currency" quantity = Decimal(str(item.quantity or 0)) buy_price = Decimal(str(item.buy_price or 0)) totals[key] += quantity * buy_price is_stock_transfer_purchase = self._is_stock_transfer_purchase() discount = Decimal(str(self.amount or 0)) if discount and not is_stock_transfer_purchase: discount_currency = self.currency or next( (item.currency for item in self.purchase_items.all() if item.currency), None, ) discount_key = discount_currency.title if discount_currency else "Unknown Currency" totals[discount_key] -= discount if totals[discount_key] < 0: totals[discount_key] = Decimal("0") return dict(totals) if totals else {"Unknown Currency": Decimal("0")} def _payment_totals_by_invoice_currency(self, include_auto_stock_transfer=True): total_due_map = self.total_price or {} totals = defaultdict(Decimal) for payment in self.payments.all(): notes = str(getattr(payment, "notes", "") or "").lower() if not include_auto_stock_transfer and "automatic stock transfer payment" in notes: continue payment_currency = payment.currency or self.currency key = payment_currency.title if payment_currency else "Unknown Currency" totals[key] += Decimal(str(payment.amount or 0)) received_amount = Decimal(str(self.received_amount or 0)) has_recorded_payment = any(Decimal(str(amount or 0)) > Decimal("0") for amount in totals.values()) or received_amount > 0 if not has_recorded_payment and self.status == PaymentStatus.COMPLETED: return dict(total_due_map) positive_invoice_currencies = [ currency for currency, amount in total_due_map.items() if Decimal(str(amount or 0)) > Decimal("0") ] if len(positive_invoice_currencies) == 1: from shop_management.models.data.account_transaction import AccountTransaction invoice_currency = positive_invoice_currencies[0] invoice_currency_obj = None for item in self.purchase_items.all(): item_currency = item.currency or self.currency if item_currency and item_currency.title == invoice_currency: invoice_currency_obj = item_currency break if invoice_currency_obj is None and self.currency and self.currency.title == invoice_currency: invoice_currency_obj = self.currency paid_total = Decimal("0") for payment in self.payments.all(): notes = str(getattr(payment, "notes", "") or "").lower() if not include_auto_stock_transfer and "automatic stock transfer payment" in notes: continue payment_amount = Decimal(str(payment.amount or 0)) payment_currency = payment.currency or self.currency if not payment_currency or not invoice_currency_obj: paid_total += payment_amount elif payment_currency.id == invoice_currency_obj.id: paid_total += payment_amount else: rate = AccountTransaction.resolve_exchange_rate( self.shop, payment_currency.id, invoice_currency_obj.id ) if rate: paid_total += AccountTransaction.compute_converted_amount( payment_amount, rate ) elif getattr(payment, "converted_amount", None) is not None: paid_total += Decimal(str(payment.converted_amount or 0)) elif getattr(payment, "exchange_rate", None): paid_total += AccountTransaction.compute_converted_amount( payment_amount, Decimal(str(payment.exchange_rate)) ) else: # Last resort: keep payment visible in invoice currency bucket paid_total += payment_amount if received_amount > paid_total: paid_total = received_amount return {invoice_currency: paid_total} if received_amount and self.currency: key = self.currency.title totals[key] = max(Decimal(str(totals.get(key, 0) or 0)), received_amount) return dict(totals) def _related_stock_transfers(self): from shop_management.models.data.stock_transfer import StockTransfer related_select = ( "currency", "supplier_branch", "target_stock__purchase_item", "target_stock__purchase_item__currency", "target_stock__purchase_item__purchase__currency", ) linked = list( StockTransfer.objects.filter( target_stock__purchase_item__purchase_id=self.pk, status=StockTransfer.Status.ACCEPTED, ).select_related(*related_select) ) if linked: return linked # Destination stock merges often retarget StockTransfer.target_stock to a # different purchase, so fall back to same-day source→destination matches. if not self._is_stock_transfer_purchase() or not self.shop_id: return [] supplier_name = str(getattr(getattr(self, "supplier", None), "name", "") or "").strip() source_name = supplier_name[len("Transfer from "):].strip() if supplier_name.lower().startswith("transfer from ") else "" if not source_name: return [] product_ids = [ item.product_id for item in self.purchase_items.all() if getattr(item, "product_id", None) ] qs = StockTransfer.objects.filter( customer_branch_id=self.shop_id, status=StockTransfer.Status.ACCEPTED, supplier_branch__name__iexact=source_name, transfer_date=self.purchase_date, ).select_related(*related_select) if product_ids: qs = qs.filter(product_id__in=product_ids) return list(qs) def _stock_transfer_paid_totals(self): related_transfers = self._related_stock_transfers() if not related_transfers: return None if not any(transfer.show_payment_button for transfer in related_transfers): # Payment OFF = stock move only, no branch debt. return dict(self.total_price or {}) paid_totals = defaultdict(Decimal) payable_currencies = set() for transfer in related_transfers: purchase_item = getattr(getattr(transfer, "target_stock", None), "purchase_item", None) currency_obj = ( transfer.currency or (purchase_item.currency if purchase_item else None) or (purchase_item.purchase.currency if purchase_item and purchase_item.purchase else None) ) currency_key = currency_obj.title if currency_obj else "Unknown Currency" quantity = Decimal(str(transfer.transfer_quantity or 0)) raw_unit_price = ( purchase_item.buy_price if purchase_item and purchase_item.buy_price is not None else transfer.sale_price ) unit_price = Decimal(str(raw_unit_price or 0)) line_total = quantity * unit_price if transfer.show_payment_button: payable_currencies.add(currency_key) else: # Payment OFF lines are stock-only — treat as already settled. paid_totals[currency_key] += line_total # Count ALL real purchase payments on this purchase. # (Do not require transfer ids in notes — Purchase Payments UI does not always set them.) for payment in self.payments.all(): notes_lower = str(getattr(payment, "notes", "") or "").lower() if "automatic stock transfer payment" in notes_lower: continue payment_currency = payment.currency or self.currency currency_key = payment_currency.title if payment_currency else "Unknown Currency" if payable_currencies and currency_key not in payable_currencies: # Still count unknown/other currency payments so Paid is visible. payable_currencies.add(currency_key) paid_totals[currency_key] += Decimal(str(payment.amount or 0)) received_amount = Decimal(str(self.received_amount or 0)) if received_amount > 0: if self.currency: key = self.currency.title elif payable_currencies: key = next(iter(payable_currencies)) else: key = "Unknown Currency" paid_totals[key] = max(Decimal(str(paid_totals.get(key, 0) or 0)), received_amount) return dict(paid_totals) @property def total_paid(self): if self._is_stock_transfer_purchase(): transfer_paid_totals = self._stock_transfer_paid_totals() if transfer_paid_totals is not None: # Merge with normal payment totals so Paid never disappears when # transfer links are incomplete but PurchasePayment rows exist. normal_paid = self._payment_totals_by_invoice_currency( include_auto_stock_transfer=False ) or {} merged = defaultdict(Decimal) for currency, amount in (transfer_paid_totals or {}).items(): merged[currency] += Decimal(str(amount or 0)) for currency, amount in normal_paid.items(): current = Decimal(str(merged.get(currency, 0) or 0)) incoming = Decimal(str(amount or 0)) merged[currency] = max(current, incoming) return dict(merged) totals = defaultdict(Decimal, self._payment_totals_by_invoice_currency()) for allocation in self.supplier_payment_allocations.filter( payment__status="POSTED" ).select_related("payment__currency"): currency = allocation.payment.currency key = currency.title if currency else "Unknown Currency" totals[key] += Decimal(str(allocation.allocated_amount or 0)) return dict(totals) def _allocate_transfer_overpayment(self, remained): """ Mixed-currency transfer purchases are often paid in one currency for the combined AFG+USD invoice. Overpay in one bucket must settle unpaid in the others, otherwise Dashboard Purchase Unpaid stays positive while Stock Transfer remained is already 0. """ overpay = Decimal("0") unpaid_currencies = [] for currency, amount in remained.items(): value = Decimal(str(amount or 0)) if value < 0: overpay += -value elif value > 0: unpaid_currencies.append(currency) allocated = dict(remained) for currency in unpaid_currencies: unpaid = Decimal(str(allocated.get(currency, 0) or 0)) if unpaid <= 0 or overpay <= 0: continue take = min(unpaid, overpay) allocated[currency] = unpaid - take overpay -= take return { currency: max(Decimal(str(amount or 0)), Decimal("0")) for currency, amount in allocated.items() } @property def total_remained_amount(self): # No supplier means no payable party — stock can be received without # creating supplier debt (same idea as sales unpaid needing a customer). if not self.supplier_id: total_due_map = self.total_price or {} return {currency: Decimal("0") for currency in total_due_map.keys()} total_due_map = self.total_price or {} total_paid_map = self.total_paid or {} currencies = set(total_due_map.keys()) | set(total_paid_map.keys()) remained = {} for currency in currencies: remained[currency] = Decimal(str(total_due_map.get(currency, 0))) - Decimal( str(total_paid_map.get(currency, 0)) ) if self._is_stock_transfer_purchase(): return self._allocate_transfer_overpayment(remained) return remained def update_status(self): remained = self.total_remained_amount # Empty remained (no currencies) or all zero → completed new_status = ( PaymentStatus.COMPLETED if (not remained) or all(Decimal(str(amount or 0)) <= Decimal("0") for amount in remained.values()) else PaymentStatus.PENDING ) if self.status != new_status: self.status = new_status self.save(update_fields=['status']) def get_discount_percentage(self): """Calculate amount as percentage of subtotal""" subtotal = sum(item.quantity * item.buy_price for item in self.purchase_items.all()) if subtotal > 0 and self.amount: return (self.amount / subtotal) * 100 return 0 def set_discount_percentage(self, percentage): """Set amount based on percentage of subtotal""" subtotal = sum(item.quantity * item.buy_price for item in self.purchase_items.all()) if subtotal > 0: self.amount = (percentage / 100) * subtotal self.save(update_fields=['amount']) class PurchaseItem(DataRoot): purchase = models.ForeignKey(Purchase, on_delete=models.CASCADE, related_name='purchase_items') currency = models.ForeignKey( Currency, on_delete=models.SET_NULL, null=True, blank=True, related_name='purchase_items', ) product = models.ForeignKey( Product, on_delete=models.SET_NULL, null=True, blank=True, related_name='purchase_items' ) unit = models.ForeignKey(Unit, on_delete=models.SET_NULL, null=True, blank=True, related_name='purchase_items') quantity = models.PositiveIntegerField() buy_price = models.DecimalField(max_digits=14, decimal_places=4, help_text="Purchase/cost price") wholesale_price = models.DecimalField(max_digits=14, decimal_places=4, null=True, blank=True, help_text="Wholesale selling price for bulk buyers") retail_price = models.DecimalField(max_digits=14, decimal_places=4, null=True, blank=True, help_text="Retail selling price for regular customers") sale_price = models.DecimalField(max_digits=14, decimal_places=4, null=True, blank=True, help_text="Active selling price used in sales (set from retail_price by default)") # Remove default_date; add optional expire_date expire_date = models.DateField(null=True, blank=True) batch_number = models.CharField(max_length=100, blank=True) manufacture_date = models.DateField(null=True, blank=True) is_deleted = models.BooleanField(default=False) def __str__(self): prod_name = self.product.name if self.product else "Unknown Product" return f"{self.quantity} x {prod_name}"