shihaam
c1f661cdb6
Build and Push Docker Images / Build and Push Docker Images (push) Failing after 5s
228 lines
8.1 KiB
Python
228 lines
8.1 KiB
Python
"""
|
|
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
|