# payments/management/commands/fix_wallet_usage_same_day_dupes.py from __future__ import annotations from decimal import Decimal from datetime import datetime, date, time, timedelta from typing import Optional from django.core.management.base import BaseCommand from django.db import transaction from django.db.models import Count, F from django.utils import timezone try: # Python 3.9+ stdlib from zoneinfo import ZoneInfo except Exception: # pragma: no cover ZoneInfo = None # type: ignore from accounts.models import User from payments.models import ViralPeWallet, ViralPeWalletUsage def signed_amount(u: ViralPeWalletUsage) -> Decimal: """credit -> +amount, debit -> -amount""" amt = Decimal(u.amount_used) return amt if u.transaction_type == "credit" else (amt * Decimal("-1")) def local_day_window_utc(target_date: date, tz: timezone.tzinfo) -> tuple[datetime, datetime]: """ Given a local calendar date + tz, return the [start_utc, end_utc) window that covers that entire local day when filtering a timezone-aware DateTimeField. """ start_local = datetime.combine(target_date, time.min).replace(tzinfo=tz) end_local = start_local + timedelta(days=1) return (start_local.astimezone(timezone.utc), end_local.astimezone(timezone.utc)) class Command(BaseCommand): help = ( "Find same-day duplicate wallet usages for a user and delete EXACTLY ONE (latest) per duplicate group, " "reversing its wallet effect IF doing so will NOT make the wallet negative.\n\n" "Duplicates are defined by: user + local DATE(used_on) + transaction_type + amount_used + purpose + purpose_note. " "order_id is intentionally ignored.\n\n" "DRY-RUN by default; pass --apply to execute." ) def add_arguments(self, parser): parser.add_argument("--mobile", required=True, help="User mobile number, e.g. 9640337477") parser.add_argument("--date", dest="on_date", default=None, help="Target local date in YYYY-MM-DD (default: today in selected timezone)") parser.add_argument("--apply", action="store_true", help="Apply changes (otherwise DRY RUN)") parser.add_argument( "--tz", dest="tzname", default=None, help="IANA timezone (e.g., Asia/Kolkata). Defaults to Django's current timezone.", ) def handle(self, *args, **opts): mobile = opts["mobile"] on_date_str = opts.get("on_date") apply_changes: bool = bool(opts.get("apply")) tzname: Optional[str] = opts.get("tzname") # Resolve timezone if tzname: if ZoneInfo is None: self.stderr.write("ERROR: zoneinfo not available; remove --tz or upgrade to Python 3.9+.") return tz = ZoneInfo(tzname) # type: ignore[arg-type] else: tz = timezone.get_current_timezone() # Parse date (default: today in tz) if on_date_str: try: target_date = datetime.strptime(on_date_str, "%Y-%m-%d").date() except ValueError: self.stderr.write("Invalid --date. Use YYYY-MM-DD") return else: # Use 'today' in the provided tz now_local = timezone.now().astimezone(tz) target_date = now_local.date() # Resolve user try: user = User.objects.get(mobile_number=mobile) except User.DoesNotExist: self.stderr.write(f"User with mobile {mobile} not found.") return self.stdout.write(f"User: {user.id} / {mobile}") self.stdout.write(f"Local date: {target_date} | Timezone: {tz}") self.stdout.write("Duplicate keys: type + amount + purpose + purpose_note (within that local day)\n") # Build UTC window for the local calendar day start_utc, end_utc = local_day_window_utc(target_date, tz) # Base queryset for that local day (converted to UTC) base_qs = ViralPeWalletUsage.objects.filter( user=user, used_on__gte=start_utc, used_on__lt=end_utc, ) # Find duplicate groups (order_id intentionally excluded) dup_groups = ( base_qs.values( "transaction_type", "amount_used", "purpose", "purpose_note", ) .annotate(n=Count("id")) .filter(n__gt=1) ) total_groups = dup_groups.count() if total_groups == 0: self.stdout.write("No duplicate groups for this date. Nothing to do.") return self.stdout.write(f"Found {total_groups} duplicate group(s).\n") groups_processed = 0 deletions = 0 skips_negative = 0 for g in dup_groups: groups_processed += 1 tx_type = g["transaction_type"] amount = g["amount_used"] purpose = g["purpose"] pnote = g["purpose_note"] count = g["n"] # All rows in this group (oldest -> newest; id tie-break) g_qs = ( base_qs.filter( transaction_type=tx_type, amount_used=amount, purpose=purpose, purpose_note=pnote, ) .order_by("used_on", "id") ) ids = list(g_qs.values_list("id", flat=True)) self.stdout.write( f"Group {groups_processed}: type={tx_type} amount={amount} " f"purpose={purpose!r} note={pnote!r} records={count} ids={ids}" ) # Candidate to delete: latest one cand = g_qs.last() self.stdout.write( f" Candidate -> usage#{cand.id} [{cand.transaction_type}] " f"amt={cand.amount_used} order_id={cand.order_id or ''} used_on={cand.used_on}" ) s = signed_amount(cand) reverse_delta = (s * Decimal("-1")) wallet = ViralPeWallet.objects.filter(user=user).first() current_balance = wallet.balance if wallet else Decimal("0.00") projected_balance = current_balance + reverse_delta self.stdout.write( f" Wallet now: {current_balance} | reverse_delta: {reverse_delta} " f"| projected: {projected_balance}" ) if projected_balance < 0: self.stdout.write(" SKIP: Would make wallet negative. Handle manually.\n") skips_negative += 1 continue if not apply_changes: self.stdout.write(" DRY RUN: Would delete this usage and set wallet to " f"{projected_balance}\n") deletions += 1 continue # APPLY atomically with row lock + re-check try: with transaction.atomic(): w, _ = ViralPeWallet.objects.select_for_update().get_or_create(user=user) fresh = w.balance new_projected = fresh + reverse_delta if new_projected < 0: self.stdout.write(" CONCURRENT SKIP: Now would go negative. Aborting this group.\n") skips_negative += 1 continue # Apply wallet change then delete usage row w.balance = F("balance") + reverse_delta w.save(update_fields=["balance"]) cand.delete() w.refresh_from_db(fields=["balance"]) self.stdout.write(f" APPLIED: Deleted usage#{cand.id}. New wallet: {w.balance}\n") deletions += 1 except Exception as e: self.stderr.write(f" ERROR applying group: {e}\n") # Summary self.stdout.write("=== Summary ===") self.stdout.write(f"Groups processed : {groups_processed}") self.stdout.write(f"Would delete / Deleted : {deletions}") self.stdout.write(f"Skipped (negative) : {skips_negative}") if not apply_changes: self.stdout.write("Mode: DRY RUN (use --apply to execute)")