"""Company-owned business parties, separate from login identities.""" import re import uuid from django.core.exceptions import ValidationError from django.core.validators import RegexValidator, validate_email from django.db import models, transaction from apps.organization.model_base import TimeStampedModel from apps.organization.models import Company NUMBER_PATTERN = r"^CUS-[0-9]{8,18}$" MAX_NUMBER = 999999999999999999 def format_customer_number(value): if not 1 <= value <= MAX_NUMBER: raise ValidationError("Customer number capacity exceeded.") return f"CUS-{value:08d}" def normalize_mobile(value): """Optional +, ASCII digits, cosmetic spaces/hyphens/parentheses only.""" value = value.strip() if not value: return "" if not re.fullmatch(r"[+0-9 ()-]+", value): raise ValidationError("Use digits with an optional leading + and phone separators.") value = re.sub(r"[ ()-]", "", value) if not re.fullmatch(r"\+?[0-9]{7,15}", value): raise ValidationError("Use 7 to 15 digits with an optional leading +.") if value.startswith("+0"): raise ValidationError("An international number cannot start with +0.") return value class CustomerNumberSequence(models.Model): """Internal allocator; never exposed in admin. Lock Company before this row.""" company = models.OneToOneField(Company, primary_key=True, on_delete=models.PROTECT, related_name="customer_number_sequence") next_value = models.PositiveBigIntegerField(default=1) class Meta: constraints = [models.CheckConstraint(condition=models.Q(next_value__gte=1), name="customers_sequence_positive")] class Customer(TimeStampedModel): class Type(models.TextChoices): INDIVIDUAL = "INDIVIDUAL", "Individual" ORGANIZATION = "ORGANIZATION", "Organization" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) company = models.ForeignKey(Company, on_delete=models.PROTECT, related_name="customers") customer_number = models.CharField(max_length=22, editable=False, blank=True) customer_type = models.CharField(max_length=12, choices=Type.choices) full_name = models.CharField(max_length=200, blank=True) organization_name = models.CharField(max_length=200, blank=True) primary_mobile = models.CharField(max_length=32, blank=True) primary_email = models.EmailField(max_length=254, blank=True) is_active = models.BooleanField(default=True) class Meta: ordering = ["customer_number", "id"] constraints = [ models.UniqueConstraint(fields=["company", "customer_number"], name="customers_company_number_uniq"), models.CheckConstraint(condition=models.Q(customer_type__in=["INDIVIDUAL", "ORGANIZATION"]), name="customers_type_valid"), models.CheckConstraint(condition=models.Q(customer_number__regex=NUMBER_PATTERN), name="customers_number_format"), models.CheckConstraint(condition=( models.Q(customer_type="INDIVIDUAL", organization_name="") & ~models.Q(full_name="") | models.Q(customer_type="ORGANIZATION") & ~models.Q(organization_name="") ), name="customers_type_name_valid"), ] @property def display_name(self): return self.organization_name if self.customer_type == self.Type.ORGANIZATION else self.full_name def __str__(self): # Admin logs/reference labels need the stable identifier, not contact PII. return self.customer_number or "New customer" def clean_fields(self, exclude=None): for name in ("full_name", "organization_name", "primary_email"): value = getattr(self, name) if isinstance(value, str): setattr(self, name, value.strip()) if isinstance(self.primary_email, str) and "@" in self.primary_email: local, domain = self.primary_email.rsplit("@", 1) self.primary_email = local + "@" + domain.lower() if isinstance(self.primary_mobile, str): try: self.primary_mobile = normalize_mobile(self.primary_mobile) except ValidationError as error: raise ValidationError({"primary_mobile": error.messages}) from error super().clean_fields(exclude=exclude) def clean(self): super().clean() errors = {} if self.customer_type == self.Type.INDIVIDUAL: if not self.full_name: errors["full_name"] = "An individual requires a full name." if self.organization_name: errors["organization_name"] = "Leave organization name blank for an individual." elif self.customer_type == self.Type.ORGANIZATION and not self.organization_name: errors["organization_name"] = "An organization requires an organization name." previous = type(self).objects.filter(pk=self.pk).first() if previous: if self.company_id != previous.company_id: errors["company"] = "Customer company ownership cannot change." if self.customer_number != previous.customer_number: errors["customer_number"] = "Customer number cannot change." # Existing active flags survive company deactivation. Metadata correction # is still allowed; only creation/reactivation requires an active company. if self.is_active and (previous is None or not previous.is_active): if not Company.objects.filter(pk=self.company_id, is_active=True).exists(): errors["company"] = "Creating or reactivating a customer requires an active company." if errors: raise ValidationError(errors) def save(self, *, force_insert=False, force_update=False, using=None, update_fields=None): if using not in (None, "default") or self._state.db not in (None, "default"): raise ValueError("Customer writes use the default PostgreSQL database.") fields = None if update_fields is None else set(update_fields) if fields == set(): return with transaction.atomic(): previous = type(self).objects.filter(pk=self.pk).first() company_id = previous.company_id if previous else self.company_id company = Company.objects.select_for_update().filter(pk=company_id).first() if company is None: raise ValidationError({"company": "Select a persisted company."}) previous = type(self).objects.select_for_update().filter(pk=self.pk).first() if previous: if self.company_id != previous.company_id or self.customer_number != previous.customer_number: raise ValidationError("Customer company and number are immutable.") if fields is not None: for field in self._meta.concrete_fields: if field.name not in fields and field.attname not in fields: setattr(self, field.attname, getattr(previous, field.attname)) fields.add("updated_at") else: if not self._state.adding or force_update or fields is not None: raise ValidationError("Customer no longer exists.") if self.customer_number: raise ValidationError({"customer_number": "Customer numbers are allocated automatically."}) sequence, _ = CustomerNumberSequence.objects.select_for_update().get_or_create(company=company) self.customer_number = format_customer_number(sequence.next_value) sequence.next_value += 1 sequence.save(update_fields=["next_value"]) try: self.full_clean() return super().save(force_insert=force_insert, force_update=force_update, using="default", update_fields=fields) except Exception: if previous is None: self.customer_number = "" raise class CustomerDetail(TimeStampedModel): """Validated detail writes; lifecycle/primary workflows live in services.""" id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) customer = models.ForeignKey(Customer, on_delete=models.PROTECT, related_name="%(class)s_records") label = models.CharField(max_length=100, blank=True) is_primary = models.BooleanField(default=False) is_active = models.BooleanField(default=True) class Meta: abstract = True ordering = ["created_at", "id"] constraints = [models.CheckConstraint( condition=models.Q(is_primary=False) | models.Q(is_active=True), name="%(class)s_primary_active")] def __str__(self): return f"{self._meta.verbose_name} {self.pk}" def clean_fields(self, exclude=None): for field in self._meta.concrete_fields: if isinstance(field, models.CharField): value = getattr(self, field.name) if isinstance(value, str): setattr(self, field.name, value.strip()) super().clean_fields(exclude=exclude) def clean(self): super().clean() previous = type(self).objects.filter(pk=self.pk).first() if previous and previous.customer_id != self.customer_id: raise ValidationError("Customer ownership cannot change; create a new record.") if self.is_primary and not self.is_active: raise ValidationError("An inactive record cannot be primary.") newly_operational = self.is_active and (previous is None or not previous.is_active) newly_primary = self.is_primary and (previous is None or not previous.is_primary) if newly_operational or newly_primary: from .detail_locks import require_operational customer = Customer.objects.filter(pk=self.customer_id).first() if customer is None: raise ValidationError("A persisted customer is required.") require_operational(customer) def save(self, *, force_insert=False, force_update=False, using=None, update_fields=None): from .detail_locks import locked_customer if using not in (None, "default") or self._state.db not in (None, "default"): raise ValueError("Customer detail writes use the default PostgreSQL database.") fields = None if update_fields is None else set(update_fields) if fields == set(): return previous = type(self).objects.filter(pk=self.pk).first() customer_id = previous.customer_id if previous else self.customer_id if customer_id is None or not Customer.objects.filter(pk=customer_id).exists(): raise ValidationError("A persisted customer is required.") with locked_customer(customer_id): previous = type(self).objects.select_for_update().filter(pk=self.pk).first() if previous: if self.customer_id != previous.customer_id: raise ValidationError("Customer ownership cannot change.") if isinstance(self, CustomerContact) and self.contact_type != previous.contact_type: raise ValidationError("Contact type cannot change; create a new contact.") if fields is not None: for field in self._meta.concrete_fields: if field.name not in fields and field.attname not in fields: setattr(self, field.attname, getattr(previous, field.attname)) fields.add("updated_at") elif not self._state.adding or force_update or fields is not None: raise ValidationError("Customer detail no longer exists.") if isinstance(self, CustomerContact) and fields is not None: fields.add("normalized_value") self.full_clean() return super().save(force_insert=force_insert, force_update=force_update, using="default", update_fields=fields) class CustomerAddress(CustomerDetail): class Type(models.TextChoices): HOME = "HOME", "Home" WORK = "WORK", "Work" BILLING = "BILLING", "Billing" SHIPPING = "SHIPPING", "Shipping" OTHER = "OTHER", "Other" address_type = models.CharField(max_length=8, choices=Type.choices) address_line_1 = models.CharField(max_length=255) address_line_2 = models.CharField(max_length=255, blank=True) area = models.CharField(max_length=100, blank=True) city = models.CharField(max_length=100) district = models.CharField(max_length=100, blank=True) postal_code = models.CharField(max_length=20, blank=True) country_code = models.CharField(max_length=2, validators=[RegexValidator(r"\A[A-Z]{2}\Z", "Use two ASCII letters.")]) class Meta(CustomerDetail.Meta): abstract = False constraints = CustomerDetail.Meta.constraints + [ models.CheckConstraint(condition=models.Q(address_type__in=["HOME", "WORK", "BILLING", "SHIPPING", "OTHER"]), name="customeraddress_type_valid"), models.CheckConstraint(condition=models.Q(country_code__regex=r"^[A-Z]{2}$"), name="customeraddress_country_format"), models.CheckConstraint(condition=~models.Q(address_line_1="") & ~models.Q(city=""), name="customeraddress_required_text"), models.UniqueConstraint(fields=["customer"], condition=models.Q(is_primary=True, is_active=True), name="customeraddress_active_primary_uniq"), ] def clean_fields(self, exclude=None): if isinstance(self.country_code, str): self.country_code = self.country_code.strip().upper() super().clean_fields(exclude=exclude) class CustomerContact(CustomerDetail): class Type(models.TextChoices): MOBILE = "MOBILE", "Mobile" PHONE = "PHONE", "Phone" EMAIL = "EMAIL", "Email" WHATSAPP = "WHATSAPP", "WhatsApp" OTHER = "OTHER", "Other" contact_type = models.CharField(max_length=8, choices=Type.choices) value = models.CharField(max_length=254) normalized_value = models.CharField(max_length=254, editable=False) class Meta(CustomerDetail.Meta): abstract = False constraints = CustomerDetail.Meta.constraints + [ models.CheckConstraint(condition=models.Q(contact_type__in=["MOBILE", "PHONE", "EMAIL", "WHATSAPP", "OTHER"]), name="customercontact_type_valid"), models.CheckConstraint(condition=~models.Q(value="") & ~models.Q(normalized_value=""), name="customercontact_required_value"), models.UniqueConstraint(fields=["customer", "contact_type"], condition=models.Q(is_primary=True, is_active=True), name="customercontact_active_primary_uniq"), models.UniqueConstraint(fields=["customer", "contact_type", "normalized_value"], condition=models.Q(is_active=True), name="customercontact_active_value_uniq"), ] def clean_fields(self, exclude=None): if isinstance(self.value, str): self.value = self.value.strip() normalized = self.value try: if self.contact_type in (self.Type.MOBILE, self.Type.PHONE, self.Type.WHATSAPP): normalized = normalize_mobile(self.value) elif self.contact_type == self.Type.EMAIL: validate_email(self.value) local, domain = self.value.rsplit("@", 1) normalized = local + "@" + domain.lower() except ValidationError as error: raise ValidationError({"value": error.messages}) from error self.normalized_value = normalized super().clean_fields(exclude=exclude) def clean(self): super().clean() previous = type(self).objects.filter(pk=self.pk).first() if previous and previous.contact_type != self.contact_type: raise ValidationError("Contact type cannot change; create a new contact.") if self.is_active and self.normalized_value and type(self).objects.filter( customer_id=self.customer_id, contact_type=self.contact_type, normalized_value=self.normalized_value, is_active=True, ).exclude(pk=self.pk).exists(): # Also surface this as an editable-field error in ModelForms, where # the derived normalized_value is excluded from constraint validation. raise ValidationError({"value": "This customer already has this active contact of the same type."})