# payments/management/commands/import_commissions.py import sys from decimal import Decimal, InvalidOperation from typing import Optional, Tuple from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.db.models import Q from openpyxl import load_workbook from payments.models import CommissionConfig from recharge.models import Operator class Command(BaseCommand): help = ( "Import CommissionConfig rows from an Excel .xlsx file.\n" "Template columns (sheet 'CommissionConfig'):\n" " operator_id | operator_name | provider_name | commission_type | commission_value\n" "commission_type ∈ {'PERCENT','FLAT'}; commission_value per type." ) def add_arguments(self, parser): parser.add_argument("--file", required=True, help="Path to .xlsx file") parser.add_argument( "--mode", choices=["add", "upsert", "replace"], default="upsert", help=( "add = create only; error if a matching row exists\n" "upsert = create or update existing (default)\n" "replace = delete all existing CommissionConfig rows, then insert all from file" ), ) parser.add_argument( "--strict-provider", action="store_true", help=( "Fail if provider_name not normalized by settings.PROVIDER_SOURCE_MAP; " "otherwise, model.clean() will still validate source on save." ), ) parser.add_argument( "--dry-run", action="store_true", help="Validate and show a summary without writing to DB.", ) parser.add_argument( "--sheet", default="CommissionConfig", help="Worksheet name to read (default: CommissionConfig)", ) def handle(self, *args, **opts): path = opts["file"] sheet = opts["sheet"] mode = opts["mode"] strict = opts["strict-provider"] dry = opts["dry-run"] try: wb = load_workbook(path, data_only=True) except Exception as e: raise CommandError(f"Failed to open workbook: {e}") if sheet not in wb.sheetnames: raise CommandError(f"Sheet '{sheet}' not found. Available: {wb.sheetnames}") ws = wb[sheet] headers = self._read_headers(ws) required = ["operator_id", "operator_name", "provider_name", "commission_type", "commission_value"] self._ensure_required(headers, required) rows = list(ws.iter_rows(min_row=2, values_only=True)) total = len(rows) if total == 0: self.stdout.write(self.style.WARNING("No data rows found.")) return stats = dict(processed=0, created=0, updated=0, skipped=0, errors=0) errors = [] if mode == "replace" and not dry: self.stdout.write(self.style.WARNING("Mode=replace → deleting existing CommissionConfig rows...")) CommissionConfig.objects.all().delete() # Main import with transaction.atomic(): for idx, values in enumerate(rows, start=2): row = dict(zip(headers, values)) try: res = self._process_row(row, strict=strict, dry=dry) if res == "created": stats["created"] += 1 elif res == "updated": stats["updated"] += 1 elif res == "skipped": stats["skipped"] += 1 stats["processed"] += 1 except Exception as e: stats["errors"] += 1 msg = f"Row {idx}: {e}" errors.append(msg) # keep going to report all issues if dry: transaction.set_rollback(True) # ensure nothing is committed # Summary self.stdout.write("\n=== Import Summary ===") for k in ["processed", "created", "updated", "skipped", "errors"]: self.stdout.write(f"{k:>10}: {stats[k]}") if errors: self.stdout.write("\nErrors:") for e in errors[:50]: self.stdout.write(f" - {e}") if len(errors) > 50: self.stdout.write(f" ... and {len(errors)-50} more") if stats["errors"] > 0: raise CommandError("Import completed with errors (see above).") # ---------------- helpers ---------------- def _read_headers(self, ws): headers = [] for c in ws.iter_rows(min_row=1, max_row=1, values_only=True): headers = [str(h).strip() if h is not None else "" for h in c] return headers def _ensure_required(self, headers, required): missing = [h for h in required if h not in headers] if missing: raise CommandError(f"Missing required columns: {missing}") def _parse_decimal(self, val) -> Decimal: if val is None or val == "": return Decimal("0") try: return Decimal(str(val)) except (InvalidOperation, ValueError) as e: raise ValueError(f"Invalid decimal value: {val!r}") from e def _resolve_operator(self, operator_id, operator_name) -> Operator: if operator_id not in (None, ""): try: return Operator.objects.get(id=int(operator_id)) except (Operator.DoesNotExist, ValueError): raise ValueError(f"operator_id not found: {operator_id}") if operator_name and str(operator_name).strip(): name = str(operator_name).strip() q = Q(name__iexact=name) # If you also have 'code' field, uncomment to allow matching by code # q |= Q(code__iexact=name) qs = Operator.objects.filter(q) if qs.count() == 1: return qs.first() elif qs.count() > 1: raise ValueError(f"Multiple operators match name '{name}'. Use operator_id instead.") else: raise ValueError(f"operator_name not found: {name}") raise ValueError("Either operator_id or operator_name is required.") def _process_row(self, row: dict, strict: bool, dry: bool) -> str: """ Returns: 'created' | 'updated' | 'skipped' """ operator_id = row.get("operator_id") operator_name = row.get("operator_name") provider_name = (row.get("provider_name") or "").strip() commission_type = (row.get("commission_type") or "").strip().upper() commission_value = self._parse_decimal(row.get("commission_value")) if not provider_name: return "skipped" if commission_type not in ("PERCENT", "FLAT"): raise ValueError(f"commission_type must be 'PERCENT' or 'FLAT', got {commission_type!r}") if commission_type == "PERCENT": if commission_value < 0 or commission_value > 100: raise ValueError(f"Percentage out of range [0..100]: {commission_value}") else: # FLAT if commission_value < 0: raise ValueError(f"Flat commission cannot be negative: {commission_value}") operator = self._resolve_operator(operator_id, operator_name) # Match unique key (operator, provider_name, commission_type) lookup = dict(operator=operator, provider_name=provider_name, commission_type=commission_type) obj: Optional[CommissionConfig] = CommissionConfig.objects.filter(**lookup).first() if obj is None: obj = CommissionConfig(**lookup) obj.commission_value = commission_value # Legacy safety: ensure old fields don't block save (in case ModelForm logic differs) if hasattr(obj, "provider_percentage"): obj.provider_percentage = float(commission_value) if commission_type == "PERCENT" else 0.0 if hasattr(obj, "fixed_percentage"): obj.fixed_percentage = float(commission_value) if commission_type == "PERCENT" else 0.0 # Validate provider/operator source mapping in model.clean() obj.full_clean() if dry: # no DB writes return "created" if obj.pk is None else "updated" obj.save() return "created" if obj._state.adding else "updated"