# users/management/commands/import_users_from_excel.py import os import re from pathlib import Path from datetime import datetime from typing import Optional from django.core.files import File from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.utils import timezone from django.contrib.auth.hashers import make_password from openpyxl import load_workbook from users.models import CustomUser # adjust import BOOL_TRUE = {"1", "true", "yes", "y", "t", "on"} ROLE_CHOICES = {c[0] for c in CustomUser.ROLE_CHOICES} USER_TYPE_CHOICES = {c[0] for c in CustomUser.USER_TYPE_CHOICES} def parse_bool(val) -> bool: if val is None: return False if isinstance(val, bool): return val return str(val).strip().lower() in BOOL_TRUE def parse_str(val) -> Optional[str]: if val is None: return None s = str(val).strip() return s or None def parse_date(val) -> datetime: """ Excel cells might be datetime, date, or string. Fallback to now(). """ if not val: return timezone.now() if isinstance(val, datetime): # assume timezone-aware or naive -> make aware if timezone.is_naive(val): return timezone.make_aware(val, timezone.get_current_timezone()) return val s = str(val).strip() # try common formats for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%d-%m-%Y", "%d/%m/%Y", "%m/%d/%Y"): try: dt = datetime.strptime(s, fmt) return timezone.make_aware(dt, timezone.get_current_timezone()) except Exception: pass return timezone.now() def random_6_digit_pin(): # 6-digit numeric string; not secret, just initial login PIN import secrets, string return "".join(secrets.choice(string.digits) for _ in range(6)) def clean_mobile(m): # keep digits, common in India to store with spaces/+91 etc. if m is None: return None digits = re.sub(r"\D+", "", str(m)) return digits or None class Command(BaseCommand): help = "Import users from Excel 'Data' sheet (creates/updates users without setting referred_by)." def add_arguments(self, parser): parser.add_argument("excel_path", type=str, help="Path to the Excel file") parser.add_argument( "--profile-pic-base", type=str, default="", help="Optional base folder for profile_pic relative paths in Excel", ) parser.add_argument( "--dry-run", action="store_true", help="Parse and validate but do not write to DB", ) def handle(self, *args, **opts): excel_path = Path(opts["excel_path"]) base_pic = Path(opts["profile_pic_base"]) if opts["profile_pic_base"] else None dry_run = opts["dry_run"] 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 "Data" not in wb.sheetnames: raise CommandError("Sheet 'Data' not found.") ws = wb["Data"] header = [str(c.value).strip() if c.value is not None else "" for c in next(ws.iter_rows(min_row=1, max_row=1))[0:]] # Expected columns # mobile_number, email, first_name, profile_pic, user_type, role, referral_code, # referred_by, agreed_terms, is_active, is_staff, date_joined idx = {name: header.index(name) for name in header} created, updated, skipped = 0, 0, 0 # Use a transaction for consistency; chunking not strictly necessary for 13k @transaction.atomic def do_import(): nonlocal created, updated, skipped for row in ws.iter_rows(min_row=2, values_only=True): row_data = {col: row[idx[col]] if col in idx else None for col in header} mobile = clean_mobile(row_data.get("mobile_number")) if not mobile: skipped += 1 continue email = parse_str(row_data.get("email")) first_name = parse_str(row_data.get("first_name")) or "" last_name = parse_str(row_data.get("last_name")) or "" profile_pic_cell = parse_str(row_data.get("profile_pic")) user_type = (parse_str(row_data.get("user_type")) or "").lower() or "customer" role = (parse_str(row_data.get("role")) or "user").lower() referral_code = parse_str(row_data.get("referral_code")) # referred_by is deliberately ignored in this command agreed_terms = parse_bool(row_data.get("agreed_terms")) is_active = parse_bool(row_data.get("is_active")) if row_data.get("is_active") is not None else True is_staff = parse_bool(row_data.get("is_staff")) if row_data.get("is_staff") is not None else False date_joined = parse_date(row_data.get("date_joined")) # Validate choices if user_type not in USER_TYPE_CHOICES: # fallback user_type = "customer" if role not in ROLE_CHOICES: role = "user" # Upsert by mobile_number try: user = CustomUser.objects.select_for_update().filter(mobile_number=mobile).first() is_new = user is None if is_new: user = CustomUser(mobile_number=mobile) # unusable password, set initial login pin user.set_unusable_password() user.login_pin = random_6_digit_pin() user.date_joined = date_joined # Update common fields user.email = email user.first_name = first_name user.last_name = last_name user.user_type = user_type user.role = role user.referral_code = referral_code # can be None user.agreed_terms = agreed_terms user.is_active = is_active user.is_staff = is_staff # Don’t touch: transaction_pin, referred_by, pins_set_at, pins_email_sent_at if not dry_run: user.save() # Handle profile pic if local path given if profile_pic_cell: pic_path = Path(profile_pic_cell) if base_pic and not pic_path.is_absolute(): pic_path = base_pic / pic_path if pic_path.exists() and pic_path.is_file(): with pic_path.open("rb") as f: user.profile_pic.save(pic_path.name, File(f), save=True) if is_new: created += 1 else: updated += 1 except Exception as e: # Don’t abort everything for one bad row—log and continue self.stderr.write(f"[SKIP] {mobile}: {e}") skipped += 1 do_import() self.stdout.write(self.style.SUCCESS( f"Done. Created: {created}, Updated: {updated}, Skipped: {skipped}. " f"{'(dry-run)' if dry_run else ''}" ))