""" Inspect one product's stock movements at a branch; optionally set qty. Example: python manage.py reconcile_product_stock \\ --shop-name "Kabul Main Branch" \\ --product-name "BARLIN 5-LITER ENGINE OIL" python manage.py reconcile_product_stock \\ --shop-name "Kabul Main Branch" \\ --product-name "BARLIN 5-LITER ENGINE OIL" \\ --set-qty=0 --dry-run python manage.py reconcile_product_stock \\ --shop-name "Kabul Main Branch" \\ --product-name "BARLIN 5-LITER ENGINE OIL" \\ --set-qty=0 """ from django.core.management.base import BaseCommand from django.db import transaction from django.db.models import Q, Sum from shop_management.models.data.product import Product from shop_management.models.data.purchase import PurchaseItem from shop_management.models.data.sale import SaleItem from shop_management.models.data.shop import Shop from shop_management.models.data.stock import Stock from shop_management.models.data.stock_transfer import StockTransfer class Command(BaseCommand): help = ( "Show purchase/sale/transfer history for one product at a branch, " "and optionally set available stock qty (e.g. --set-qty=0)." ) def add_arguments(self, parser): parser.add_argument("--shop", type=int, help="Shop id") parser.add_argument("--shop-name", type=str, help="Shop name") parser.add_argument("--product-name", type=str, help="Exact or partial product name") parser.add_argument("--product-code", type=str, help="Product code (e.g. 591)") parser.add_argument( "--set-qty", type=int, help="Set all matching listable stock rows' quantity to this value.", ) parser.add_argument("--dry-run", action="store_true") def handle(self, *args, **options): shop = self._resolve_shop(options) if not shop: return products = self._resolve_products(shop, options) if not products: self.stderr.write(self.style.ERROR("No matching product found in this shop.")) return for product in products: self._audit_product(shop, product) if options.get("set_qty") is not None: self._set_qty(shop, products, options["set_qty"], dry_run=options["dry_run"]) def _resolve_shop(self, options): shop_id = options.get("shop") shop_name = (options.get("shop_name") or "").strip() if shop_id: shop = Shop.objects.filter(pk=shop_id).first() elif shop_name: shop = Shop.objects.filter(name__iexact=shop_name).first() if not shop: shop = Shop.objects.filter(name__icontains=shop_name).first() else: self.stderr.write(self.style.ERROR("Use --shop or --shop-name")) return None if not shop: self.stderr.write(self.style.ERROR("Shop not found.")) return None return shop def _resolve_products(self, shop, options): name = (options.get("product_name") or "").strip() code = (options.get("product_code") or "").strip() qs = Product.objects.filter(shop=shop) if code: qs = qs.filter(code__iexact=code) if name: exact = qs.filter(name__iexact=name) qs = exact if exact.exists() else qs.filter(name__icontains=name) if not name and not code: self.stderr.write(self.style.ERROR("Use --product-name or --product-code")) return [] return list(qs.order_by("id")[:10]) def _audit_product(self, shop, product): self.stdout.write("") self.stdout.write( self.style.MIGRATE_HEADING( f"=== {shop.name} | {product.name} (id={product.id}, code={product.code or '-'}) ===" ) ) purchase_qty = ( PurchaseItem.objects.filter( product=product, purchase__shop=shop, is_deleted=False, purchase__is_deleted=False, ).aggregate(total=Sum("quantity"))["total"] or 0 ) # Real customer sales only (exclude transfer-sales that do not move stock again) sale_qty = ( SaleItem.objects.filter( product=product, sale__shop=shop, ) .exclude(sale__notes__startswith="Stock transfer sale for ") .exclude(sale__customer__name__startswith="Transfer to ") .aggregate(total=Sum("quantity"))["total"] or 0 ) transfer_sale_qty = ( SaleItem.objects.filter( product=product, sale__shop=shop, ) .filter( Q(sale__notes__startswith="Stock transfer sale for ") | Q(sale__customer__name__startswith="Transfer to ") ) .aggregate(total=Sum("quantity"))["total"] or 0 ) transfer_out = ( StockTransfer.objects.filter( product=product, supplier_branch=shop, status=StockTransfer.Status.ACCEPTED, ).aggregate(total=Sum("transfer_quantity"))["total"] or 0 ) transfer_in = ( StockTransfer.objects.filter( product=product, customer_branch=shop, status=StockTransfer.Status.ACCEPTED, ).aggregate(total=Sum("transfer_quantity"))["total"] or 0 ) stocks = list( Stock.objects.filter( shop=shop, purchase_item__product=product, is_deleted=False, ) .select_related("purchase_item", "purchase_item__purchase__supplier") .order_by("id") ) current_qty = sum(int(s.quantity or 0) for s in stocks) expected = int(purchase_qty) + int(transfer_in) - int(sale_qty) - int(transfer_out) self.stdout.write(f"Purchased qty: {purchase_qty}") self.stdout.write(f"Transferred IN: {transfer_in}") self.stdout.write(f"Customer sold qty: {sale_qty}") self.stdout.write(f"Transferred OUT: {transfer_out}") self.stdout.write(f"Transfer-sale rows qty: {transfer_sale_qty} (info only, not deducted twice)") self.stdout.write(f"Expected stock qty: {expected} (purchase+in - sales - out)") self.stdout.write(f"Current stock qty: {current_qty}") if current_qty != expected: self.stdout.write( self.style.WARNING( f"MISMATCH: current ({current_qty}) != expected ({expected})" ) ) else: self.stdout.write(self.style.SUCCESS("Current qty matches expected.")) self.stdout.write("Stock rows:") for s in stocks: supplier = "-" if s.purchase_item and s.purchase_item.purchase and s.purchase_item.purchase.supplier: supplier = s.purchase_item.purchase.supplier.name self.stdout.write( f" stock#{s.id} qty={s.quantity} avg_buy={s.average_buy_price} " f"supplier={supplier}" ) sales = ( SaleItem.objects.filter(product=product, sale__shop=shop) .select_related("sale", "sale__customer") .order_by("-sale_id")[:15] ) self.stdout.write("Recent sale items:") for item in sales: notes = (item.sale.notes or "")[:60] cust = item.sale.customer.name if item.sale.customer else "-" self.stdout.write( f" sale#{item.sale_id} date={item.sale.sale_date} qty={item.quantity} " f"customer={cust} notes={notes}" ) transfers = ( StockTransfer.objects.filter( product=product, supplier_branch=shop, status=StockTransfer.Status.ACCEPTED, ) .select_related("customer_branch") .order_by("-id")[:15] ) self.stdout.write("Accepted transfers OUT:") for t in transfers: self.stdout.write( f" {t.transfer_id} -> {t.customer_branch.name if t.customer_branch else '-'} " f"qty={t.transfer_quantity} date={t.transfer_date} payment={t.show_payment_button}" ) def _set_qty(self, shop, products, qty, dry_run=False): product_ids = [p.id for p in products] stocks = list( Stock.objects.filter( shop=shop, purchase_item__product_id__in=product_ids, is_deleted=False, ) ) if not stocks: self.stderr.write(self.style.ERROR("No stock rows to update.")) return self.stdout.write("") self.stdout.write( self.style.MIGRATE_HEADING( f"{'DRY-RUN ' if dry_run else ''}Setting qty={qty} on {len(stocks)} stock row(s)..." ) ) for s in stocks: name = s.purchase_item.product.name if s.purchase_item and s.purchase_item.product else "?" self.stdout.write( f" {'Would set' if dry_run else 'Set'} stock#{s.id} '{name}' " f"{s.quantity} -> {qty}" ) if dry_run: continue with transaction.atomic(): s.quantity = qty s.save(update_fields=["quantity", "updated_at"]) self.stdout.write( self.style.SUCCESS( f"{'Would update' if dry_run else 'Updated'} {len(stocks)} row(s)." ) ) if not dry_run: self.stdout.write("Refresh Stock List — BARLIN should hide when qty is 0.")