"""Add related demonstration records without deleting existing business data.""" from datetime import date, datetime, time, timedelta from decimal import Decimal from importlib import import_module from pathlib import Path import sqlite3 from django.apps import apps from django.contrib.auth import get_user_model from django.core.management.base import BaseCommand, CommandError from django.db import connection, transaction from django.utils import timezone from rest_framework.request import Request from rest_framework.test import APIRequestFactory from account.audit import audit_event from account.permissions import TAB_KEYS from shop_management.services.automatic_journals import create_system_journal from shop_management.services.customer_payments import post_customer_payment from shop_management.services.expenses import post_expense from shop_management.services.finance_transfers import post_finance_transfer from shop_management.services.notifications import sync_operational_alerts from shop_management.services.opening_balances import post_financial_opening from shop_management.services.purchase_returns import post_purchase_return from shop_management.services.purchases import post_purchase from shop_management.services.sale_returns import post_sale_return from shop_management.services.sales import post_sale from shop_management.services.stock_adjustments import post_stock_adjustment from shop_management.services.supplier_payments import post_supplier_payment from shop_management.services.warehouse_transfers import dispatch_transfer, receive_transfer MARKER = "System demo seed v1" class Command(BaseCommand): help = "Add 10 linked samples per business table, with posted inventory and finance documents. Safe to rerun." def add_arguments(self, parser): parser.add_argument("--count", type=int, default=10) parser.add_argument("--owner", type=int, help="Existing staff user who can see the demo branches.") parser.add_argument("--date", type=date.fromisoformat, default=timezone.localdate()) parser.add_argument("--dry-run", action="store_true", help="Validate the complete seed, then roll it back.") def handle(self, *args, **options): count = options["count"] if count < 2: raise CommandError("Use a count of at least 2 for transfers.") User = get_user_model() owner = User.objects.filter(pk=options["owner"], is_staff=True).first() if options["owner"] else User.objects.filter(is_superuser=True, is_active=True).first() if not owner: raise CommandError("An existing staff owner is required; use --owner ID.") if self.model("Shop").objects.filter(description=MARKER).exists(): self.stdout.write("Demo data already exists; no records were added.") return self.owner = owner self.day = options["date"] self.moment = timezone.make_aware(datetime.combine(self.day, time(10))) models = list(apps.get_app_config("shop_management").get_models()) before = {model: model.objects.count() for model in models} if not options["dry_run"] and connection.vendor == "sqlite": source = Path(connection.settings_dict["NAME"]) if source.is_file(): backup = source.with_name(f"db.sqlite3-before-demo-{timezone.now():%Y%m%d-%H%M%S}") with sqlite3.connect(source) as src, sqlite3.connect(backup) as dest: src.backup(dest) self.stdout.write(f"Database backup: {backup}") with transaction.atomic(): self.seed(count) missing = [model.__name__ for model in models if model.objects.count() - before[model] < count] if missing: raise CommandError(f"Seed incomplete; rolling back. Missing samples: {', '.join(missing)}") for model in models: self.stdout.write(f"{model.__name__}: +{model.objects.count() - before[model]}") if options["dry_run"]: transaction.set_rollback(True) self.stdout.write(self.style.SUCCESS("Dry run passed; all changes rolled back." if options["dry_run"] else f"Demo data added. Main branch: Demo Main. Existing records preserved.")) @staticmethod def model(name): return apps.get_model("shop_management", name) def create(self, model, **values): Model = self.model(model) if any(field.name == "created_by" for field in Model._meta.fields): values.setdefault("created_by", self.owner) return Model.objects.create(**values) def serialized(self, module, serializer_name, **values): Serializer = getattr(import_module(f"shop_management.serializers.data.{module}"), serializer_name) request = Request(APIRequestFactory().post("/demo-seed/", values, format="json")) request.user = self.owner serializer = Serializer(data=values, context={"request": request}) serializer.is_valid(raise_exception=True) return serializer.save(created_by=self.owner) def seed(self, count): shops, currencies, units = [], [], [] for i in range(count): branch = self.create("Shop", owner=self.owner, name="Demo Main" if i == 0 else f"Demo Branch {i + 1:02}", description=MARKER) currency = self.create("Currency", shop=branch, title="AFN", description="Afghan Afghani") unit = self.create("Unit", shop=branch, title="Piece", description=MARKER) self.create("CompanySetting", shop=branch, name=branch.name, legal_name=branch.name, base_currency=currency, address="Kabul, Afghanistan", invoice_footer_note="Demonstration data") shops.append(branch) currencies.append(currency) units.append(unit) shop, currency, unit = shops[0], currencies[0], units[0] usd = self.create("Currency", shop=shop, title="USD", description="US Dollar") warehouses = [self.create("Warehouse", shop=shop, code=f"DEMO-{i + 1:02}", name=f"Demo Warehouse {i + 1:02}", address="Kabul") for i in range(count)] accounts = [] for i in range(count): account = self.create("Account", shop=shop, name=f"Demo {'Bank' if i % 2 else 'Cash'} {i + 1:02}", account_type="BANK" if i % 2 else "CASH", currency=currency, note=MARKER) opening = self.create("FinancialOpeningBalance", shop=shop, balance_type="ACCOUNT", financial_account=account, currency=currency, amount=Decimal("100000"), opening_date=self.day, notes=MARKER) post_financial_opening(opening.pk, user=self.owner) accounts.append(account) self.create("ExchangeRate", shop=shop, from_currency=usd, to_currency=currency, rate=Decimal("70"), rate_date=self.day - timedelta(days=i)) product_names = ["Poultry Vitamins", "Medical Gloves", "Poultry Vaccine", "Disinfectant", "Feed Supplement", "Vitamin C", "Syringes", "Mineral Mix", "Electrolyte Powder", "Hand Sprayer"] for i in range(count): self.stdout.write(f"Creating related sample group {i + 1}/{count}…") suffix = f"{i + 1:02}" user, created = get_user_model().objects.get_or_create(username=f"demo_staff_{suffix}", defaults={"is_staff": True, "allowed_tabs": TAB_KEYS, "first_name": "Demo", "last_name": f"Staff {suffix}"}) if not created: raise CommandError(f"Username {user.username} already exists; seed rolled back.") user.set_unusable_password() user.save(update_fields=["password"]) shop.assigned_users.add(user) audit_event("CREATE", user, actor=self.owner, target_user=user, reference=f"{MARKER}: staff {suffix}", branch_id=shop.pk) category = self.create("Category", shop=shop, name=f"Demo Category {suffix}", description=MARKER) product = self.create("Product", shop=shop, category=category, name=f"Demo {product_names[i % len(product_names)]} {suffix}", code=f"DEMO-P{suffix}", barcode=f"99000000{i + 1:05}", default_unit=unit, requires_batch=True, requires_expiry=True, minimum_stock_level=100) price = Decimal(150 + i * 10) cost = Decimal(100 + i * 5) self.create("ProductPrice", product=product, currency=currency, price_type="RETAIL", amount=price) customer = self.create("Customer", shop=shop, name=f"Demo Customer {suffix}", contact=f"+9370000{i + 1:04}", address="Kabul", email=f"demo.customer{suffix}@example.com") supplier = self.create("Supplier", shop=shop, name=f"Demo Supplier {suffix}", contact=f"+9371000{i + 1:04}", address="Kabul", email=f"demo.supplier{suffix}@example.com") employee = self.create("Employee", shop=shop, name=f"Demo Employee {suffix}", salary=Decimal("15000"), age=25 + i, contract_date=self.day, contract_durations="12 months") expense_category = self.create("ExpenseCategory", shop=shop, name=f"Demo Operating Expense {suffix}", description=MARKER) purchase = self.create("Purchase", shop=shop, supplier=supplier, warehouse=warehouses[0], currency=currency, purchase_date=self.day, due_date=self.day + timedelta(days=10), exchange_rate_used=1, uses_inventory_engine=True, supplier_invoice_number=f"DEMO-SUP-{suffix}", notes=MARKER) purchase_item = self.create("PurchaseItem", purchase=purchase, product=product, currency=currency, unit=unit, quantity=100, buy_price=cost, retail_price=price, sale_price=price, batch_number=f"DEMO-B{suffix}", manufacture_date=self.day - timedelta(days=30), expire_date=self.day + timedelta(days=30)) post_purchase(purchase.pk, user=self.owner) batch = self.model("InventoryBatch").objects.get(product=product, batch_number=f"DEMO-B{suffix}") # An advance receipt becomes a real credit application when the sale posts. prepayment = self.create("CustomerPayment", shop=shop, customer=customer, financial_account=accounts[0], currency=currency, amount=Decimal("100"), payment_date=self.day, reference=f"DEMO-ADV-{suffix}", notes=MARKER) post_customer_payment(prepayment.pk, user=self.owner) sale = self.create("Sale", shop=shop, customer=customer, warehouse=warehouses[0], currency=currency, sale_date=self.day, invoice_date=self.day, due_date=self.day + timedelta(days=10), sales_type="CREDIT", uses_inventory_engine=True, notes=MARKER) sale_item = self.create("SaleItem", sale=sale, product=product, unit=unit, quantity=10, sale_price=price) post_sale(sale.pk, user=self.owner) payment = self.create("CustomerPayment", shop=shop, customer=customer, financial_account=accounts[0], currency=currency, amount=Decimal("100"), payment_date=self.day, reference=f"DEMO-REC-{suffix}", notes=MARKER) self.create("CustomerPaymentAllocation", payment=payment, sale=sale, allocated_amount=Decimal("100")) post_customer_payment(payment.pk, user=self.owner) payment = self.create("SupplierPayment", shop=shop, supplier=supplier, financial_account=accounts[0], currency=currency, amount=Decimal("100"), payment_date=self.day, reference=f"DEMO-PAY-{suffix}", notes=MARKER) self.create("SupplierPaymentAllocation", payment=payment, purchase=purchase, allocated_amount=Decimal("100")) post_supplier_payment(payment.pk, user=self.owner) self.serialized("customer_balance_payment", "CustomerBalancePaymentSerializer", customer=customer.pk, account=accounts[0].pk, currency=currency.pk, amount="50.00", payment_date=self.moment.isoformat(), notes=MARKER) self.serialized("supplier_balance_payment", "SupplierBalancePaymentSerializer", supplier=supplier.pk, account=accounts[0].pk, currency=currency.pk, amount="50.00", payment_date=self.moment.isoformat(), notes=MARKER) # Legacy payment lists remain useful without duplicating modern payments. self.serialized("purchase_payment", "PurchasePaymentSerializer", purchase=purchase.pk, account=accounts[0].pk, currency=currency.pk, amount="10.00", payment_date=self.moment.isoformat(), notes=MARKER) self.create("SalePayment", sale=sale, currency=currency, amount=Decimal("10"), payment_date=self.moment, notes=MARKER) self.create("AccountTransaction", account=accounts[0], account_currency=currency, transaction_currency=currency, transaction_type="credit", amount=Decimal("10"), exchange_rate=1, converted_amount=Decimal("10"), transaction_date=self.day, description=MARKER, source_type="DEMO_SALE_PAYMENT", source_id=str(sale.pk)) create_system_journal(shop=shop, entry_date=self.day, source_type="DEMO_SALE_PAYMENT", source_id=sale.pk, description=MARKER, currency=currency, exchange_rate=1, user=self.owner, lines=[{"code": "1100", "side": "debit", "amount": 10, "financial_account": accounts[0]}, {"code": "1400", "side": "credit", "amount": 10, "customer": customer}]) sale_return = self.create("SaleReturn", sale=sale, shop=shop, uses_inventory_engine=True, reason=MARKER) self.create("SaleReturnItem", sale_return=sale_return, sale_item=sale_item, return_quantity=1, refund_amount=price) post_sale_return(sale_return.pk, user=self.owner) purchase_return = self.create("PurchaseReturn", purchase=purchase, shop=shop, uses_inventory_engine=True, reason=MARKER) self.create("PurchaseReturnItem", purchase_return=purchase_return, purchase_item=purchase_item, return_quantity=1, refund_amount=cost, batch=batch) post_purchase_return(purchase_return.pk, user=self.owner) adjustment = self.create("StockAdjustment", warehouse=warehouses[0], adjustment_date=self.day, reason="Demo damaged package", notes=MARKER) self.create("StockAdjustmentItem", adjustment=adjustment, product=product, batch=batch, direction="OUT", quantity=1, currency=currency, unit_cost=cost, remarks=MARKER) post_stock_adjustment(adjustment.pk, user=self.owner) transfer = self.create("WarehouseTransfer", shop=shop, source_warehouse=warehouses[0], destination_warehouse=warehouses[1], transfer_date=self.day, notes=MARKER) self.create("WarehouseTransferItem", transfer=transfer, product=product, batch=batch, quantity=2) dispatch_transfer(transfer.pk, user=self.owner) receive_transfer(transfer.pk, user=self.owner) stock = self.create("Stock", shop=shop, purchase_item=purchase_item, quantity=89, average_buy_price=cost, sale_price=price, retail_price=price, average_retail_price=price, comments=MARKER) self.create("StockTransfer", transfer_id=f"DEMO-TR-{suffix}", product=product, source_stock=stock, supplier_branch=shop, customer_branch=shops[1], currency=currency, transfer_quantity=1, old_quantity=89, new_quantity=88, sale_price=price, transfer_date=self.day, notes=MARKER) expense = self.create("Expense", shop=shop, category=expense_category, account=accounts[0], currency=currency, amount=Decimal("25"), exchange_rate=1, converted_amount=Decimal("25"), payment_date=self.moment, expense_by=employee.name, notes=MARKER) post_expense(expense.pk, user=self.owner) transfer = self.create("AccountTransfer", from_account=accounts[0], to_account=accounts[1], from_currency=currency, to_currency=currency, amount=Decimal("50"), exchange_rate=1, converted_amount=Decimal("50"), uses_controlled_engine=True, status="DRAFT", transfer_date=self.day, reference=f"DEMO-FIN-{suffix}", note=MARKER) post_finance_transfer(transfer.pk, user=self.owner) for module, serializer, amount in [("advance", "AdvanceSerializer", "100"), ("payroll", "PayrollSerializer", "1500")]: self.serialized(module, serializer, shop=shop.pk, employee=employee.pk, account=accounts[0].pk, currency=currency.pk, amount=amount, year=str(self.day.year), month=str(self.day.month), payment_date=self.moment.isoformat(), notes=MARKER) self.serialized("passive_income", "PassiveIncomeSerializer", shop=shop.pk, name=f"Demo Other Income {suffix}", account=accounts[0].pk, currency=currency.pk, amount="75", income_date=self.moment.isoformat(), notes=MARKER) self.serialized("customer_debtor", "CustomerDebtorSerializer", customer=customer.pk, account=accounts[0].pk, currency=currency.pk, amount="100", loan_date=self.moment.isoformat(), notes=MARKER) quote = self.create("Quotation", shop=shop, customer=customer, currency=currency, quotation_date=self.day, valid_until=self.day + timedelta(days=7), notes=MARKER) self.create("QuotationItem", quotation=quote, product=product, unit=unit, quantity=5, unit_price=price) self.create("DailyDiaryEntry", shop=shop, entry_date=self.day - timedelta(days=i), title=f"Demo daily note {suffix}", content=f"{MARKER}: reviewed purchases, sales, and stock.") self.create("DailyDiaryLog", shop=shop, entry_date=self.day, entry_datetime=self.moment, title=f"Demo stock review {suffix}", description=f"{MARKER}: reviewed {product.name} after posting purchases, sales, and transfers.", activity_type="manual", category="manual", is_auto=False) sync_operational_alerts(self.owner)