# users/management/commands/import_users_from_excel.py import io import os import re import mimetypes from pathlib import Path from datetime import datetime from urllib.parse import urlparse from typing import Optional from django.core.files import File from django.core.files.base import ContentFile from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.utils import timezone from openpyxl import load_workbook try: import requests except ImportError: requests = None # ---- Adjust this import to your app if needed ---- from users.models import CustomUser # from accounts.models import CustomUser # <- use this instead if applicable # ------------------------------------------------- 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} ALLOWED_IMAGE_MIMES = {"image/jpeg", "image/png", "image/webp", "image/gif"} 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): return timezone.make_aware(val, timezone.get_current_timezone()) if timezone.is_naive(val) else val s = str(val).strip() 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(): import secrets, string return "".join(secrets.choice(string.digits) for _ in range(6)) def clean_mobile(m): # Keep only digits (handles +91, spaces, hyphens, etc.) if m is None: return None digits = re.sub(r"\D+", "", str(m)) return digits or None def is_url(s: str) -> bool: try: u = urlparse(s) return u.scheme in ("http", "https") and bool(u.netloc) except Exception: return False def filename_from_url(url: str, fallback: str = "profile.jpg") -> str: name = urlparse(url).path.rsplit("/", 1)[-1] or fallback if "." not in name: ext = mimetypes.guess_extension(mimetypes.guess_type(url)[0] or "") or ".jpg" name = name + ext return name def fetch_image(url: str, timeout: int, max_bytes: int) -> tuple[str, io.BytesIO, str]: """ Returns (filename, bytes_io, mime) after downloading the image from URL. Raises on errors or oversized/non-image responses. """ if requests: r = requests.get(url, stream=True, timeout=timeout, allow_redirects=True) r.raise_for_status() ctype = (r.headers.get("Content-Type") or "").split(";")[0].strip().lower() bio = io.BytesIO() total = 0 for chunk in r.iter_content(chunk_size=64 * 1024): if not chunk: continue total += len(chunk) if total > max_bytes: raise ValueError(f"image too large (> {max_bytes} bytes)") bio.write(chunk) bio.seek(0) # Validate content type (fallback to URL-based guess) guess = mimetypes.guess_type(url)[0] or "" is_image = ctype.startswith("image/") or guess.startswith("image/") if not is_image or (ctype and ctype not in ALLOWED_IMAGE_MIMES and not guess.startswith("image/")): raise ValueError(f"unsupported content-type: {ctype or 'unknown'}") return filename_from_url(url), bio, ctype or guess else: # urllib fallback if requests not installed from urllib.request import urlopen, Request req = Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urlopen(req, timeout=timeout) as resp: ctype = (resp.headers.get("Content-Type") or "").split(";")[0].strip().lower() bio = io.BytesIO() total = 0 while True: chunk = resp.read(64 * 1024) if not chunk: break total += len(chunk) if total > max_bytes: raise ValueError(f"image too large (> {max_bytes} bytes)") bio.write(chunk) bio.seek(0) guess = mimetypes.guess_type(url)[0] or "" is_image = ctype.startswith("image/") or guess.startswith("image/") if not is_image: raise ValueError(f"unsupported content-type: {ctype or 'unknown'}") return filename_from_url(url), bio, ctype or guess class Command(BaseCommand): help = "Import users from Excel 'Data' sheet (create/update) 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 local profile_pic relative paths in Excel", ) parser.add_argument( "--default-pic", type=str, default="", help="Absolute path to a default image to use when profile_pic cell is empty or fails.", ) parser.add_argument( "--pic-timeout", type=int, default=10, help="HTTP timeout (seconds) for profile picture downloads." ) parser.add_argument( "--store-url-instead", action="store_true", help="Do not download profile_pic URLs; store the URL string directly in the ImageField. " "If empty, use --default-pic file." ) parser.add_argument( "--pic-max-mb", type=int, default=5, help="Max download size in MB for a profile picture (default 5MB)." ) parser.add_argument( "--overwrite-pics", action="store_true", help="If set, overwrite existing user.profile_pic even if already present." ) 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"]) if not excel_path.exists(): raise CommandError(f"Excel not found: {excel_path}") base_pic = Path(opts["profile_pic_base"]).resolve() if opts.get("profile_pic_base") else None default_pic_path = Path(opts["default_pic"]).resolve() if opts.get("default_pic") else None if default_pic_path and not default_pic_path.exists(): raise CommandError(f"--default-pic not found: {default_pic_path}") timeout_s = int(opts.get("pic_timeout", 10)) max_bytes = int(opts.get("pic_max_mb", 5)) * 1024 * 1024 overwrite_pics = bool(opts.get("overwrite_pics")) dry_run = bool(opts.get("dry_run")) # Load workbook 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_row = next(ws.iter_rows(min_row=1, max_row=1))[0:] header = [str(c.value).strip() if c.value is not None else "" for c in header_row] # Required columns (create/update core) required = { "mobile_number", "email", "first_name", "last_name", # ok if empty, just present in header "profile_pic", # URL or path; can be empty per row "user_type", "role", "referral_code", # optional per row, but header needed "referred_by", # ignored in this command, but header needed "agreed_terms", "is_active", "is_staff", "date_joined", } missing = [c for c in required if c not in header] if missing: raise CommandError(f"Missing columns in 'Data' header: {', '.join(missing)}") idx = {name: header.index(name) for name in header} created, updated, skipped = 0, 0, 0 store_url_instead = bool(opts.get("store_url_instead")) def _attach_profile_pic(user, profile_pic_cell: Optional[str]): # Skip if already present unless we are overwriting if user.profile_pic and not overwrite_pics: return # (A) URL handling if profile_pic_cell and is_url(profile_pic_cell): if store_url_instead: # Just set the ImageField "name" to the URL string (no download) # This stores the URL in DB; your front-end can read/display it. user.profile_pic.name = profile_pic_cell user.save(update_fields=["profile_pic"]) return else: # Existing download behavior try: fname, bio, _mime = fetch_image(profile_pic_cell, timeout_s, max_bytes) user.profile_pic.save(fname, ContentFile(bio.read()), save=True) return except Exception as e: if default_pic_path: with default_pic_path.open("rb") as f: user.profile_pic.save(default_pic_path.name, File(f), save=True) else: self.stderr.write(f"[PIC-URL-FAIL] {profile_pic_cell} -> {e}") return # (B) Local path in Excel 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).resolve() 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) return else: if default_pic_path: with default_pic_path.open("rb") as f: user.profile_pic.save(default_pic_path.name, File(f), save=True) else: self.stderr.write(f"[PIC-NOT-FOUND] {pic_path}") return # (C) Empty -> use default if provided if default_pic_path: with default_pic_path.open("rb") as f: user.profile_pic.save(default_pic_path.name, File(f), save=True) # def _attach_profile_pic(user, profile_pic_cell: Optional[str]): # # Skip if already present unless we are overwriting # if user.profile_pic and not overwrite_pics: # return # # (A) URL -> download & attach # if profile_pic_cell and is_url(profile_pic_cell): # try: # fname, bio, _mime = fetch_image(profile_pic_cell, timeout_s, max_bytes) # user.profile_pic.save(fname, ContentFile(bio.read()), save=True) # return # except Exception as e: # if default_pic_path: # with default_pic_path.open("rb") as f: # user.profile_pic.save(default_pic_path.name, File(f), save=True) # else: # self.stderr.write(f"[PIC-URL-FAIL] {profile_pic_cell} -> {e}") # return # # (B) Local path -> attach if exists # 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).resolve() # 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) # return # else: # if default_pic_path: # with default_pic_path.open("rb") as f: # user.profile_pic.save(default_pic_path.name, File(f), save=True) # else: # self.stderr.write(f"[PIC-NOT-FOUND] {pic_path}") # return # # (C) Empty -> use default if provided # if default_pic_path: # with default_pic_path.open("rb") as f: # user.profile_pic.save(default_pic_path.name, File(f), save=True) @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 column intentionally 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: user_type = "customer" if role not in ROLE_CHOICES: role = "user" 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) user.set_unusable_password() # you can require PIN flow later 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/blank user.agreed_terms = agreed_terms user.is_active = is_active user.is_staff = is_staff # Leave: transaction_pin, referred_by, pins_set_at, pins_email_sent_at if not dry_run: user.save() _attach_profile_pic(user, profile_pic_cell) if is_new: created += 1 else: updated += 1 except Exception as e: 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 ''}" ))