customerTicket/apps/accounts/models.py

import hashlib, secrets, string
from datetime import timedelta
from django.conf import settings
from django.contrib.auth.models import AbstractUser
from django.db import IntegrityError, models, transaction
from django.utils import timezone
from apps.core.models import PublicIdModel, TimeStampedModel

def normalize_mobile(value: str) -> str:
    value = "".join(c for c in (value or "").strip() if c.isdigit() or c == "+")
    if value.startswith("0098"): value = "+98" + value[4:]
    elif value.startswith("98") and not value.startswith("+98"): value = "+" + value
    elif value.startswith("0"): value = "+98" + value[1:]
    if not (value.startswith("+989") and len(value) == 13 and value[1:].isdigit()):
        raise ValueError("شماره موبایل معتبر نیست.")
    return value

class User(PublicIdModel, AbstractUser):
    class AccountType(models.TextChoices):
        CUSTOMER = "CUSTOMER", "مشتری"
        STAFF = "STAFF", "کارشناس"
    mobile = models.CharField("شماره موبایل حساب", max_length=16, unique=True, null=True, blank=True)
    account_type = models.CharField(max_length=10, choices=AccountType.choices, default=AccountType.CUSTOMER)
    voip_extension = models.CharField("شماره داخلی کارشناس", max_length=20, blank=True)
    voip_mobile = models.CharField(
        "شماره موبایل کارشناس برای VOIP",
        max_length=16,
        blank=True,
        help_text="مستقل از شماره حساب مشتری است و می‌تواند تکراری باشد.",
    )
    voip_call = models.BooleanField("امکان تماس VOIP", default=True)
    def save(self, *args, **kwargs):
        if self.mobile: self.mobile = normalize_mobile(self.mobile)
        if self.voip_mobile: self.voip_mobile = normalize_mobile(self.voip_mobile)
        super().save(*args, **kwargs)
    @property
    def local_mobile(self):
        return f"0{self.mobile[3:]}" if self.mobile and self.mobile.startswith("+98") else (self.mobile or "")
    @property
    def voip_local_mobile(self):
        return f"0{self.voip_mobile[3:]}" if self.voip_mobile.startswith("+98") else self.voip_mobile

class OTPChallenge(models.Model):
    class Purpose(models.TextChoices): LOGIN = "LOGIN", "ورود"
    mobile = models.CharField(max_length=16, db_index=True)
    purpose = models.CharField(max_length=20, choices=Purpose.choices, default=Purpose.LOGIN)
    code_hash = models.CharField(max_length=64)
    expires_at = models.DateTimeField()
    attempt_count = models.PositiveSmallIntegerField(default=0)
    max_attempts = models.PositiveSmallIntegerField(default=5)
    verified_at = models.DateTimeField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
    request_ip = models.GenericIPAddressField(null=True, blank=True)
    user_agent = models.CharField(max_length=300, blank=True)
    @staticmethod
    def hash_code(code): return hashlib.sha256(f"{settings.SECRET_KEY}:{code}".encode()).hexdigest()
    @classmethod
    def issue(cls, mobile, request=None):
        mobile = normalize_mobile(mobile)
        request_ip = request.META.get("REMOTE_ADDR") if request else None
        window = timezone.now() - timedelta(seconds=settings.OTP_RATE_LIMIT_WINDOW_SECONDS)
        if cls.objects.filter(mobile=mobile, created_at__gte=window).count() >= settings.OTP_RATE_LIMIT_PER_MOBILE:
            raise ValueError("تعداد درخواست کد برای این شماره بیش از حد مجاز است.")
        if request_ip and cls.objects.filter(request_ip=request_ip, created_at__gte=window).count() >= settings.OTP_RATE_LIMIT_PER_IP:
            raise ValueError("تعداد درخواست کد از این اتصال بیش از حد مجاز است.")
        cutoff = timezone.now() - timedelta(seconds=settings.OTP_RESEND_SECONDS)
        if cls.objects.filter(mobile=mobile, created_at__gt=cutoff).exists(): raise ValueError("برای ارسال مجدد کمی صبر کنید.")
        is_dev_login = settings.DEBUG and mobile == settings.DEV_OTP_MOBILE and bool(settings.DEV_OTP_CODE)
        code = settings.DEV_OTP_CODE if is_dev_login else "".join(secrets.choice("0123456789") for _ in range(settings.OTP_LENGTH))
        obj = cls.objects.create(mobile=mobile, code_hash=cls.hash_code(code), expires_at=timezone.now()+timedelta(seconds=settings.OTP_EXPIRE_SECONDS), max_attempts=settings.OTP_MAX_ATTEMPTS, request_ip=request_ip, user_agent=(request.META.get("HTTP_USER_AGENT", "")[:300] if request else ""))
        return obj, code
    def verify(self, code):
        if self.verified_at or timezone.now() >= self.expires_at or self.attempt_count >= self.max_attempts: return False
        self.attempt_count += 1
        ok = secrets.compare_digest(self.code_hash, self.hash_code(code))
        if ok: self.verified_at = timezone.now()
        self.save(update_fields=["attempt_count", "verified_at"])
        return ok

class TicketMagicLink(models.Model):
    ticket = models.ForeignKey("tickets.Ticket", on_delete=models.CASCADE, related_name="magic_links")
    customer = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    token_hash = models.CharField(max_length=64, unique=True)
    created_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        on_delete=models.PROTECT,
        null=True,
        blank=True,
        related_name="created_magic_links",
    )
    created_unit = models.ForeignKey(
        "staff.OrganizationalUnit",
        on_delete=models.PROTECT,
        null=True,
        blank=True,
        related_name="magic_links",
    )
    recipient_name = models.CharField(max_length=200, blank=True)
    recipient_mobile = models.CharField(max_length=16, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    expires_at = models.DateTimeField()
    used_at = models.DateTimeField(null=True, blank=True)
    revoked_at = models.DateTimeField(null=True, blank=True)
    created_reason = models.CharField(max_length=100, blank=True)
    last_used_ip = models.GenericIPAddressField(null=True, blank=True)
    last_user_agent = models.CharField(max_length=500, blank=True)
    view_count = models.PositiveIntegerField(default=0)
    class Meta:
        permissions = [("view_all_magic_links", "Can audit magic links across all units")]
        indexes = [models.Index(fields=["created_unit", "created_at"], name="magic_unit_created_idx"), models.Index(fields=["expires_at", "revoked_at"], name="magic_validity_idx")]
    @staticmethod
    def digest(token): return hashlib.sha256(token.encode()).hexdigest()
    @staticmethod
    def generate_secure_token(length=10):
        alphabet=string.ascii_lowercase+string.digits
        return "".join(secrets.SystemRandom().choices(alphabet,k=length))
    @classmethod
    def issue_with_record(cls,ticket,reason="staff_reply",max_retries=5,created_by=None):
        # Never add a preflight SELECT to token issuance; callers may select_related
        # the profile when they need a populated recipient-name snapshot.
        profile = ticket.customer._state.fields_cache.get("customer_profile")
        recipient_name = " ".join(
            value for value in [getattr(profile, "first_name", ""), getattr(profile, "last_name", "")] if value
        ).strip()
        for attempt in range(max_retries):
            token=cls.generate_secure_token(10)
            try:
                with transaction.atomic():
                    record=cls.objects.create(
                        ticket=ticket,
                        customer=ticket.customer,
                        token_hash=cls.digest(token),
                        expires_at=timezone.now()+timedelta(days=settings.MAGIC_LINK_TTL_DAYS),
                        created_reason=reason,
                        created_by=created_by or ticket.created_by,
                        created_unit=ticket.owning_unit,
                        recipient_name=recipient_name,
                        recipient_mobile=ticket.customer.mobile or "",
                    )
                return token,record
            except IntegrityError:
                if attempt==max_retries-1: raise RuntimeError("خطای نادر: عدم موفقیت در تولید توکن یکتا.")
        raise RuntimeError("خطای نادر: عدم موفقیت در تولید توکن یکتا.")
    @classmethod
    def issue(cls,ticket,reason="staff_reply"):
        token,_=cls.issue_with_record(ticket,reason)
        return token


class MagicLinkAccessLog(models.Model):
    """Immutable audit row for every valid magic-link opening."""

    magic_link = models.ForeignKey(
        TicketMagicLink,
        on_delete=models.CASCADE,
        related_name="access_logs",
    )
    opened_at = models.DateTimeField(auto_now_add=True, db_index=True)
    ip_address = models.GenericIPAddressField(null=True, blank=True)
    user_agent = models.CharField(max_length=500, blank=True)

    class Meta:
        ordering = ["-opened_at"]
        indexes = [models.Index(fields=["magic_link", "opened_at"], name="magic_opened_idx")]
        verbose_name = "بازدید لینک جادویی"
        verbose_name_plural = "بازدیدهای لینک جادویی"

    def __str__(self):
        return f"{self.magic_link_id} @ {self.opened_at}"