# payments/management/commands/import_wallets_from_excel.py from __future__ import annotations import re from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Optional, Dict, Tuple from django.core.management.base import BaseCommand, CommandError from django.db import transaction from openpyxl import load_workbook from users.models import CustomUser from payments.models import ViralPeWallet, ViralPeWalletUsage # ---- Constants -------------------------------------------------------------- DEFAULT_SHEET = "Data" # We’ll accept a few header variants (case-insensitive) HEADER_ALIASES: Dict[str, Tuple[str, ...]] = { "mobile_number": ("mobile_number", "Mobile_number", "mobile", "Mobile"), "amount": ("amount", "Amount", "amt", "AMOUNT"), "purpose_note": ("purpose_note", "Purpose_Note", "purpose note", "Purpose Note"), } BASE_PURPOSE = "One Apportunity to Viralpe (withdrawl request)" PURPOSE_CODE = "oneapp_withdrawal" # Choices field on the model # ---- Helpers ---------------------------------------------------------------- def clean_mobile(val) -> Optional[str]: if val is None: return None s = re.sub(r"\D+", "", str(val)) return s or None def parse_decimal(val) -> Optional[Decimal]: if val is None or str(val).strip() == "": return None try: d = Decimal(str(val)).quantize(Decimal("0.01")) return d except (InvalidOperation, ValueError): return None def build_header_index(header_row_values) -> Dict[str, int]: """ Build a mapping {logical_name -> column_index} using HEADER_ALIASES. Header matching is case-insensitive and ignores extra spaces. Raises if a required logical header isn't found. """ raw_headers = [ (str(c).strip() if c is not None else "") for c in header_row_values ] normalized = [h.lower().strip() for h in raw_headers] result: Dict[str, int] = {} for logical, aliases in HEADER_ALIASES.items(): found_idx = None for alias in aliases: alias_norm = alias.lower().strip() if alias_norm in normalized: found_idx = normalized.index(alias_norm) break if found_idx is None: raise CommandError( f"Missing required column for '{logical}'. " f"Accepted headers: {', '.join(aliases)}. " f"Found headers: {', '.join(raw_headers)}" ) result[logical] = found_idx return result # ---- Command ---------------------------------------------------------------- class Command(BaseCommand): help = ( "Import credits from 1Apportunity withdrawal Excel into ViralPeWallet & ViralPeWalletUsage. " f"Expected sheet '{DEFAULT_SHEET}' with columns: Mobile_number, amount, purpose_note. " "Creates usage with purpose_code='oneapp_withdrawal'." ) def add_arguments(self, parser): parser.add_argument("excel_path", type=str, help="Path to the Excel file") parser.add_argument( "--sheet", type=str, default=DEFAULT_SHEET, help=f"Worksheet name (default: {DEFAULT_SHEET})", ) parser.add_argument( "--dry-run", action="store_true", help="Validate and show stats without writing to the database.", ) parser.add_argument( "--skip-if-existing", action="store_true", help=( "Idempotency: skip creating a credit if a matching ViralPeWalletUsage " "(same user, amount_used, purpose_code, purpose, purpose_note) already exists." ), ) def handle(self, *args, **opts): excel_path = Path(opts["excel_path"]) sheet_name = opts["sheet"] dry = bool(opts["dry_run"]) skip_existing = bool(opts["skip_if_existing"]) if not excel_path.exists(): raise CommandError(f"Excel not found: {excel_path}") wb = load_workbook(excel_path, data_only=True, read_only=True) if sheet_name not in wb.sheetnames: raise CommandError(f"Sheet '{sheet_name}' not found. Available: {', '.join(wb.sheetnames)}") ws = wb[sheet_name] # header first_row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True)) header_idx = build_header_index(first_row) # Stats total_rows = 0 skipped_mobile_missing = 0 skipped_user_not_found = 0 skipped_invalid_or_zero = 0 skipped_existing = 0 wallets_created = 0 wallets_updated = 0 usages_created = 0 @transaction.atomic def do_import(): nonlocal total_rows, skipped_mobile_missing, skipped_user_not_found nonlocal skipped_invalid_or_zero, skipped_existing, wallets_created, wallets_updated, usages_created # Preload map for speed mobile_to_user = dict(CustomUser.objects.values_list("mobile_number", "id")) for row in ws.iter_rows(min_row=2, values_only=True): total_rows += 1 mobile = clean_mobile(row[header_idx["mobile_number"]]) if not mobile: skipped_mobile_missing += 1 continue user_id = mobile_to_user.get(mobile) if not user_id: skipped_user_not_found += 1 continue amount = parse_decimal(row[header_idx["amount"]]) if not amount or amount <= 0: skipped_invalid_or_zero += 1 continue purpose_note_val = row[header_idx["purpose_note"]] purpose_note_str = (str(purpose_note_val).strip() if purpose_note_val is not None else "") # Idempotency check if skip_existing: exists = ViralPeWalletUsage.objects.filter( user_id=user_id, transaction_type="credit", purpose_code=PURPOSE_CODE, purpose=BASE_PURPOSE, purpose_note=purpose_note_str, amount_used=amount, ).exists() if exists: skipped_existing += 1 continue # Ensure wallet wallet, created = ViralPeWallet.objects.select_for_update().get_or_create(user_id=user_id) if created: wallets_created += 1 if not dry: # Update wallet and log usage wallet.balance = (wallet.balance or Decimal("0.00")) + amount wallet.save(update_fields=["balance", "last_updated"]) wallets_updated += 1 ViralPeWalletUsage.objects.create( user_id=user_id, transaction_type="credit", amount_used=amount, purpose=BASE_PURPOSE, purpose_code=PURPOSE_CODE, purpose_note=purpose_note_str, ) usages_created += 1 do_import() self.stdout.write(self.style.SUCCESS( "Import complete.\n" f" Sheet: {sheet_name}\n" f" Rows read: {total_rows}\n" f" Wallets created: {wallets_created}\n" f" Wallets updated (balance increased): {wallets_updated}\n" f" Usage rows created: {usages_created}\n" f" Skipped (no/invalid/zero amount): {skipped_invalid_or_zero}\n" f" Skipped (mobile missing): {skipped_mobile_missing}\n" f" Skipped (user not found): {skipped_user_not_found}\n" f" Skipped (already existed; idempotency): {skipped_existing}\n" f"{'(dry-run)' if dry else ''}" ))