"""Clear all Employee rows and import from employee.xlsx. Usage: uv run python manage.py import_employees_xlsx [--path PATH] [--dry-run] Default path: /employee.xlsx Behavior: - Deletes all Employee rows (CASCADE wipes Attendance, Payroll, Onboarding, Contract, Family, AssetAssignment for those employees). - Deletes auto-bridged Contacts that pointed to deleted Employees' Users. - For each xlsx row, get_or_create User by email and create fresh Employee. - The hr.signals.ensure_contact_for_employee post_save handler re-creates Contact rows. """ from datetime import datetime from pathlib import Path from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.contrib.auth import get_user_model from apps.hr.models import Employee from apps.contacts.models import Contact from apps.companies.models import Organization, OrganizationMembership User = get_user_model() STATUS_MAP = { "permanent": "active", "contract": "active", "intern": "active", "internship": "active", "probation": "active", "resigned": "terminated", "terminated": "terminated", "inactive": "inactive", "on leave": "on_leave", } DATE_FORMATS = ["%d %B %Y", "%d-%m-%Y", "%Y-%m-%d", "%d/%m/%Y"] def parse_date(value): if value is None or value == "" or value == "-": return None if isinstance(value, datetime): return value.date() if hasattr(value, "year") and hasattr(value, "month"): return value s = str(value).strip() if not s or s == "-": return None for fmt in DATE_FORMATS: try: return datetime.strptime(s, fmt).date() except ValueError: continue return None def split_name(full_name: str): parts = (full_name or "").strip().split() if not parts: return "", "" if len(parts) == 1: return parts[0], "" return parts[0], " ".join(parts[1:]) def join_address(*parts): return ", ".join(p.strip() for p in parts if p and str(p).strip() and str(p).strip().lower() != "null") def clean(value): if value is None: return "" s = str(value).strip() if s.lower() == "null": return "" return s class Command(BaseCommand): help = "Wipe Employee table and reimport from employee.xlsx" def add_arguments(self, parser): parser.add_argument( "--path", default=None, help="Path to xlsx (default: backend/employee.xlsx)", ) parser.add_argument( "--dry-run", action="store_true", help="Parse only, no DB writes", ) def handle(self, *args, **options): try: import openpyxl except ImportError as exc: raise CommandError("openpyxl is required: uv add openpyxl") from exc from django.conf import settings as dj_settings path = Path(options["path"]) if options["path"] else Path(dj_settings.BASE_DIR) / "employee.xlsx" if not path.exists(): raise CommandError(f"File not found: {path}") wb = openpyxl.load_workbook(path, data_only=True) ws = wb.active rows = list(ws.iter_rows(values_only=True)) # Filter: rows where col[3] (employee_id) is non-empty and not the header word records = [] for raw in rows: if not raw or len(raw) < 27: continue emp_id = clean(raw[3]) if not emp_id or emp_id == "EMPLOYEE ID": continue records.append({ "name": clean(raw[2]), "employee_id": emp_id, "email": clean(raw[4]).lower(), "phone": clean(raw[5]), "address": join_address(clean(raw[8]), clean(raw[9]), clean(raw[10]), clean(raw[11])), "position": clean(raw[19]), "job_level": clean(raw[20]), "employment_status": clean(raw[24]), "hire_date": parse_date(raw[25]), "termination_date": parse_date(raw[26]), }) # Dedupe by employee_id (last wins) unique = {} for r in records: unique[r["employee_id"]] = r records = list(unique.values()) self.stdout.write(self.style.NOTICE(f"Parsed {len(records)} unique employee rows from {path.name}")) # Validate: every record must have email + hire_date (Employee model requires hire_date) missing = [r for r in records if not r["email"] or not r["hire_date"]] if missing: for r in missing: self.stderr.write(f"Skipping {r['employee_id']} {r['name']} — missing email or hire_date") records = [r for r in records if r["email"] and r["hire_date"]] if options["dry_run"]: self.stdout.write(self.style.WARNING("Dry run — no DB changes")) for r in records[:5]: self.stdout.write(repr(r)) self.stdout.write(f"... ({len(records)} total)") return with transaction.atomic(): old_user_ids = list(Employee.objects.values_list("user_id", flat=True)) deleted_emp = Employee.objects.all().delete() self.stdout.write(self.style.WARNING(f"Deleted Employees: {deleted_emp[0]} rows + cascades")) deleted_contacts = Contact.objects.filter(user_id__in=old_user_ids).delete() self.stdout.write(self.style.WARNING(f"Deleted bridged Contacts: {deleted_contacts[0]}")) org = Organization.objects.first() if org is None: raise CommandError("No Organization exists to import employees into.") created_users = 0 updated_users = 0 created_emps = 0 for r in records: first, last = split_name(r["name"]) user, was_created = User.objects.get_or_create( email=r["email"], defaults={ "first_name": first, "last_name": last, "phone": r["phone"], }, ) if was_created: user.set_unusable_password() user.save() created_users += 1 else: user.first_name = first or user.first_name user.last_name = last or user.last_name if r["phone"]: user.phone = r["phone"] user.save() updated_users += 1 OrganizationMembership.objects.get_or_create(organization=org, user=user) status = STATUS_MAP.get(r["employment_status"].lower(), "active") Employee.objects.create( organization=org, user=user, employee_id=r["employee_id"], position=r["position"], department=r["job_level"], # xlsx has no department col; use job level as fallback status=status, hire_date=r["hire_date"], termination_date=r["termination_date"], address=r["address"], ) created_emps += 1 self.stdout.write(self.style.SUCCESS( f"Imported {created_emps} employees. Users created: {created_users}, updated: {updated_users}" )) contact_count = Contact.objects.filter(user__isnull=False).count() self.stdout.write(self.style.SUCCESS(f"Bridged Contacts now: {contact_count}"))