""" Check every accepted stock transfer: - Qty moves once (sending branch down, receiving branch up) - If Payment is ON, one Purchase (Transfer from …) and one Sale (Transfer to …) exist for money. Those must not add qty again. Example: python manage.py audit_all_transfers python manage.py audit_all_transfers --problems-only """ from collections import defaultdict from django.core.management.base import BaseCommand from shop_management.management.commands.heal_stock_quantities import ( Command as HealCommand, _is_transfer_sale, _is_transfer_supplier, _norm_name, ) from shop_management.models.data.purchase import PurchaseItem from shop_management.models.data.sale import SaleItem from shop_management.models.data.stock_transfer import StockTransfer class Command(BaseCommand): help = ( "Audit all accepted transfers: purchase/sale created once for money, " "stock qty counted once. Report only." ) def add_arguments(self, parser): parser.add_argument( "--problems-only", action="store_true", help="Hide OK transfer lines. Still print the summary.", ) def handle(self, *args, **options): problems_only = bool(options.get("problems_only")) healer = HealCommand() transfers = ( StockTransfer.objects.filter(status=StockTransfer.Status.ACCEPTED) .select_related( "product", "supplier_branch", "customer_branch", "source_stock__purchase_item__product", "target_stock__purchase_item__product", ) .order_by("transfer_date", "id") ) dest_skip = {} dest_purchases = defaultdict(list) source_sales = defaultdict(list) self.stdout.write( self.style.NOTICE( "Every transfer: qty once. Payment ON also creates 1 purchase + 1 sale for money." ) ) self.stdout.write("") counts = defaultdict(int) for transfer in transfers: dest = transfer.customer_branch source = transfer.supplier_branch if not dest or not source: counts["missing_branch"] += 1 continue if dest.id == source.id: continue name = healer._preferred_name(transfer, "in") or getattr( transfer.product, "name", "" ) name_key = _norm_name(name) qty = int(transfer.transfer_quantity or 0) date = transfer.transfer_date payment_on = bool(transfer.show_payment_button) counts["transfers"] += 1 if payment_on: counts["payment_on"] += 1 else: counts["payment_off"] += 1 if dest.id not in dest_skip: dest_skip[dest.id] = healer._transfer_generated_purchase_item_ids(dest) skip_ids = dest_skip[dest.id] if dest.id not in dest_purchases: items = ( PurchaseItem.objects.filter( purchase__shop=dest, is_deleted=False, purchase__is_deleted=False, ) .select_related("product", "purchase", "purchase__supplier") ) dest_purchases[dest.id] = list(items) if source.id not in source_sales: items = ( SaleItem.objects.filter(sale__shop=source) .select_related("product", "sale", "sale__customer") ) source_sales[source.id] = list(items) purchase_matches = [] for item in dest_purchases[dest.id]: item_name = _norm_name(getattr(item.product, "name", None)) if item_name != name_key: continue supplier = getattr(getattr(item.purchase, "supplier", None), "name", "") if not _is_transfer_supplier(supplier) and item.id not in skip_ids: continue if int(item.quantity or 0) != qty: continue if date and item.purchase.purchase_date and item.purchase.purchase_date != date: continue purchase_matches.append(item) sale_matches = [] for item in source_sales[source.id]: item_name = _norm_name(getattr(item.product, "name", None)) if item_name != name_key: continue if not _is_transfer_sale(item.sale): continue if int(item.quantity or 0) != qty: continue if date and item.sale.sale_date and item.sale.sale_date != date: continue sale_matches.append(item) problems = [] if payment_on: if not purchase_matches: problems.append("missing Transfer-from purchase") elif len(purchase_matches) > 1: problems.append(f"{len(purchase_matches)} Transfer-from purchases (should be 1)") if not sale_matches: problems.append("missing Transfer-to sale") elif len(sale_matches) > 1: problems.append(f"{len(sale_matches)} Transfer-to sales (should be 1)") else: if purchase_matches: problems.append("Payment OFF but Transfer-from purchase exists") if problems: counts["problem"] += 1 status = "PROBLEM" style = self.style.ERROR else: counts["ok"] += 1 status = "OK" style = self.style.SUCCESS if payment_on: extra = f"purchase=1 sale=1 money-only" else: extra = "no purchase (Payment OFF)" if problems_only and not problems: continue pay = "Payment ON" if payment_on else "Payment OFF" extra = "; ".join(problems) if problems else extra self.stdout.write( style( f"{status:8} | {date} | {qty:>4} | {name} | " f"{source.name} → {dest.name} | {pay} | {extra}" ) ) self.stdout.write("") self.stdout.write(self.style.MIGRATE_HEADING("PRODUCT STOCK (qty counted once?)")) self._print_double_products(healer) self.stdout.write("") self.stdout.write(self.style.SUCCESS("Summary:")) self.stdout.write(f" accepted transfers: {counts['transfers']}") self.stdout.write(f" Payment ON (should have 1 purchase + 1 sale): {counts['payment_on']}") self.stdout.write(f" Payment OFF (no transfer purchase): {counts['payment_off']}") self.stdout.write(f" transfer lines OK: {counts['ok']}") self.stdout.write(f" transfer lines with a problem: {counts['problem']}") self.stdout.write("") self.stdout.write( "PROBLEM = missing or extra purchase/sale row. " "That is paperwork. Stock qty is in the PRODUCT STOCK section." ) def _print_double_products(self, healer): from collections import defaultdict as dd from shop_management.management.commands.heal_stock_quantities import _Bucket from shop_management.models.data.shop import Shop shops = list(Shop.objects.all().order_by("id")) double_count = 0 ok_count = 0 high_count = 0 for shop in shops: buckets = dd(_Bucket) skip_ids = healer._transfer_generated_purchase_item_ids(shop) healer._add_stocks(shop, buckets, None, None) healer._add_purchases(shop, buckets, None, skip_ids) healer._add_purchase_returns(shop, buckets, None, skip_ids) healer._add_sales(shop, buckets, None) healer._add_sale_returns(shop, buckets, None) healer._add_transfers(shop, buckets, None) healer._add_pending_out(shop, buckets, None) printed = False for name in sorted(buckets.keys()): bucket = buckets[name] current = bucket.current expected = bucket.expected transfer_in = int(bucket["transfer_in"] or 0) if transfer_in <= 0: continue extra = current - expected if expected < 0: continue if extra == transfer_in and extra > 0: double_count += 1 if not printed: self.stdout.write(f"=== {shop.name} ===") printed = True self.stdout.write( self.style.ERROR( f" DOUBLE | {bucket['name'] or name} | stock={current} " f"should={expected} extra={extra} (= transfer in {transfer_in})" ) ) elif current == expected: ok_count += 1 elif extra > 0: high_count += 1 if double_count == 0: self.stdout.write( self.style.SUCCESS( f" No transferred product has qty added twice. " f"OK={ok_count}, extra opening leftover={high_count}, DOUBLE=0" ) ) else: self.stdout.write(self.style.ERROR(f" DOUBLE products: {double_count}"))