""" This is the models module for api. """ import secrets from datetime import timedelta from django.contrib.auth.models import AbstractUser from django.db import models from .managers import CustomUserManager from .storages import idcard_storage from django.utils import timezone import pyotp from billing.models import WalletTransaction class User(AbstractUser): # Registration workflow states. `verified` (below) stays the login gate and # is kept in sync: verified is True iff status == STATUS_VERIFIED. STATUS_PENDING = "pending" # created, awaiting/undergoing auto-verify STATUS_VERIFIED = "verified" # approved, may log in STATUS_ID_REQUIRED = "id_required" # must (re)upload an ID/passport photo STATUS_ID_SUBMITTED = "id_submitted" # photo uploaded, awaiting admin review STATUS_CHOICES = [ (STATUS_PENDING, "Pending"), (STATUS_VERIFIED, "Verified"), (STATUS_ID_REQUIRED, "ID required"), (STATUS_ID_SUBMITTED, "ID submitted"), ] address = models.CharField(max_length=255, blank=True) email = models.EmailField(blank=True, null=True, unique=True) mobile = models.CharField( max_length=255, blank=True, unique=True, null=True, db_index=True ) designation = models.CharField(max_length=255, blank=True) acc_no = models.CharField(max_length=255, blank=True) id_card = models.CharField( max_length=255, blank=True, unique=True, null=True, db_index=True ) verified = models.BooleanField(default=False) status = models.CharField( max_length=20, choices=STATUS_CHOICES, default=STATUS_PENDING, db_index=True ) id_card_photo = models.FileField( upload_to="", storage=idcard_storage, blank=True, null=True, help_text="ID card / passport (JPG, PNG or PDF) uploaded by the user for manual review.", ) is_admin = models.BooleanField(default=False) dob = models.DateField(blank=True, null=True) terms_accepted = models.BooleanField(default=False) policy_accepted = models.BooleanField(default=False) wallet_balance = models.FloatField(default=0.0) ninja_user_id = models.CharField(max_length=255, blank=True) atoll = models.ForeignKey( "Atoll", on_delete=models.SET_NULL, null=True, blank=True, related_name="users" ) island = models.ForeignKey( "Island", on_delete=models.SET_NULL, null=True, blank=True, related_name="users" ) agreement = models.FileField( upload_to="agreements/", blank=True, null=True, help_text="Upload the agreement file signed by the user.", ) created_at = models.DateTimeField(default=timezone.now) updated_at = models.DateTimeField(auto_now=True) def get_all_fields(self, instance): return [field.name for field in instance.get_fields()] def add_wallet_funds(self, amount, description="", reference_id=None): self.wallet_balance += amount self.save(update_fields=["wallet_balance"]) WalletTransaction.objects.create( user=self, amount=amount, transaction_type="TOPUP", description=description, reference_id=reference_id, ) def deduct_wallet_funds(self, amount, description="", reference_id=None): if self.wallet_balance >= amount: self.wallet_balance -= amount self.save(update_fields=["wallet_balance"]) WalletTransaction.objects.create( user=self, amount=amount, transaction_type="DEBIT", description=description, reference_id=reference_id, ) return True return False objects = CustomUserManager() # Log in with username + password only; email is optional and not prompted # by createsuperuser. REQUIRED_FIELDS = [] def set_status(self, status, *, save=True): """Set workflow status and keep the `verified` login gate in sync.""" self.status = status self.verified = status == self.STATUS_VERIFIED if save: self.save(update_fields=["status", "verified", "updated_at"]) class IdUploadToken(models.Model): """ One-time-ish magic-link token letting an unverified user reach the public ID-upload page without logging in. Issuing a fresh token supersedes any outstanding ones for the same user. """ key = models.CharField(max_length=64, unique=True, db_index=True) user = models.ForeignKey( "User", on_delete=models.CASCADE, related_name="id_upload_tokens" ) created_at = models.DateTimeField(default=timezone.now) expires_at = models.DateTimeField() used_at = models.DateTimeField(null=True, blank=True) @classmethod def issue(cls, user, ttl_hours=48): cls.objects.filter(user=user, used_at__isnull=True).update( used_at=timezone.now() ) return cls.objects.create( key=secrets.token_urlsafe(32), user=user, expires_at=timezone.now() + timedelta(hours=ttl_hours), ) def is_valid(self): return self.used_at is None and self.expires_at > timezone.now() def __str__(self) -> str: return f"IdUploadToken(user={self.user_id}, valid={self.is_valid()})" class TemporaryUser(models.Model): t_id = models.AutoField(primary_key=True) t_username = models.CharField(max_length=255, unique=True, blank=True, null=True) t_first_name = models.CharField(max_length=255, blank=True) t_last_name = models.CharField(max_length=255, blank=True) t_address = models.CharField(max_length=255, blank=True) t_email = models.EmailField(blank=True, null=True, unique=True) t_mobile = models.CharField( max_length=255, blank=True, unique=True, null=True, db_index=True ) t_designation = models.CharField(max_length=255, blank=True) t_acc_no = models.CharField(max_length=255, blank=True) t_id_card = models.CharField( max_length=255, blank=True, unique=True, null=True, db_index=True ) t_verified = models.BooleanField(default=False) t_dob = models.DateField(blank=True, null=True) t_terms_accepted = models.BooleanField(default=False) t_policy_accepted = models.BooleanField(default=False) t_wallet_balance = models.FloatField(default=0.0) t_ninja_user_id = models.CharField(max_length=255, blank=True) t_atoll = models.ForeignKey( "Atoll", on_delete=models.SET_NULL, null=True, blank=True, related_name="temp_users", ) t_island = models.ForeignKey( "Island", on_delete=models.SET_NULL, null=True, blank=True, related_name="temp_users", ) created_at = models.DateTimeField(default=timezone.now) updated_at = models.DateTimeField(auto_now=True) def get_all_fields(self, instance): return [field.name for field in instance.get_fields()] otp_secret = models.CharField(max_length=50, default=pyotp.random_base32) otp_verified = models.BooleanField(default=False) def generate_otp(self): totp = pyotp.TOTP(self.otp_secret, interval=1800) return totp.now() def verify_otp(self, otp): totp = pyotp.TOTP(self.otp_secret, interval=1800) return totp.verify(otp) def is_expired(self): return self.created_at < timezone.now() - timedelta(minutes=5) class Meta: verbose_name = "Temporary User" verbose_name_plural = "Temporary Users" def __str__(self) -> str: return f"{self.t_username}" class Atoll(models.Model): name = models.CharField(max_length=255, unique=True) created_at = models.DateTimeField(default=timezone.now) updated_at = models.DateTimeField(auto_now=True) def __str__(self): return self.name class Island(models.Model): atoll = models.ForeignKey(Atoll, on_delete=models.CASCADE, related_name="islands") name = models.CharField(max_length=255, unique=True) created_at = models.DateTimeField(default=timezone.now) updated_at = models.DateTimeField(auto_now=True) def __str__(self): return self.name