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) def save(self, *args, **kwargs): if self.mobile: self.mobile = normalize_mobile(self.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 "") 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}"