import datetime import random import string from collections.abc import Iterable # from manage_wallets.models import Wallet, Transaction, TransactionStatus, TransactionType from django.conf import settings from django.contrib.auth import get_user_model from django.contrib.auth.models import AbstractUser, BaseUserManager from django.core.cache import cache # from .utils import UserContext from django.core.validators import (MaxValueValidator, MinValueValidator, RegexValidator) from django.db import models from django.utils import timezone from django.utils.text import slugify from module_project.utils import RandomGenerator from .iam_constant import (PRINCIPAL_SOURCE_APP, PRINCIPAL_SOURCE_APPLE, PRINCIPAL_SOURCE_GOOGLE, PRINCIPAL_SOURCE_WEB, PRINCIPAL_TYPE_ADMIN, PRINCIPAL_TYPE_SUBADMIN, PRINCIPAL_TYPE_USER) # from phonenumber_field.modelfields import PhoneNumberField class BaseModel(models.Model): active = models.BooleanField(default=True) deleted = models.BooleanField(default=False) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, related_name="%(class)s_created", on_delete=models.CASCADE, blank=True, null=True, ) created_on = models.DateTimeField(auto_now_add=True) modified_by = models.ForeignKey( settings.AUTH_USER_MODEL, related_name="%(class)s_modified", on_delete=models.CASCADE, blank=True, null=True, ) modified_on = models.DateTimeField(auto_now=True) class Meta: abstract = True @classmethod def get_deleted(cls): return cls.objects.filter(deleted=True) @classmethod def get_all_except_deleted(cls): return cls.objects.filter(deleted=False) @classmethod def mark_deleted(cls, pk): try: obj = cls.objects.get(pk=pk) obj.active = False obj.deleted = True obj.save() return obj except cls.DoesNotExist: return None def delete(self, *args, **kwargs): self.active = False # Set active to False when deleting self.deleted = True self.save() def save(self, *args, **kwargs): if self.deleted: self.active = False # Ensure active is False if record is marked as deleted super().save(*args, **kwargs) class MasterModel(models.Model): name = models.CharField(max_length=255) label = models.CharField(max_length=255, null=True, blank=True) slug = models.SlugField(max_length=255, null=True, blank=True) sort_order = models.IntegerField(blank=True, null=True) small_image_url = models.ImageField(blank=True, null=True) large_image_url = models.ImageField(blank=True, null=True) active = models.BooleanField(default=True) deleted = models.BooleanField(default=False) created_by = models.SmallIntegerField(blank=True, null=True) created_on = models.DateTimeField(auto_now_add=True) modified_by = models.SmallIntegerField(blank=True, null=True) modified_on = models.DateTimeField(auto_now=True) class Meta: abstract = True def __str__(self): return f"{self.name}" def save(self, *args, **kwargs): # Generate a slug from the name field self.slug = slugify(self.name) return super().save(*args, **kwargs) class IAmPrincipalType(MasterModel): class Meta: db_table = "iam_principal_type" @classmethod def get_principal_type(cls, name): cache_key = f"principal_{name}" principal = cache.get(cache_key) if not principal: principal = cls.objects.filter(name=name).first() cache.set(cache_key, principal, timeout=60 * 15) # Cache for 15 minutes return principal @classmethod def get_principal_user(cls): return cls.get_principal_type(PRINCIPAL_TYPE_USER) @classmethod def get_principal_admin(cls): return cls.get_principal_type(PRINCIPAL_TYPE_ADMIN) @classmethod def get_principal_subadmin(cls): return cls.get_principal_type(PRINCIPAL_TYPE_SUBADMIN) class IAmPrincipalSource(MasterModel): class Meta: db_table = "iam_principal_source" @classmethod def get_principal_source(cls, name): cache_key = f"principal_{name}" principal = cache.get(cache_key) if not principal: principal = cls.objects.filter(name=name).first() cache.set(cache_key, principal, timeout=60 * 15) # Cache for 15 minutes return principal @classmethod def get_principal_web(cls): return cls.get_principal_source(PRINCIPAL_SOURCE_WEB) @classmethod def get_principal_app(cls): return cls.get_principal_source(PRINCIPAL_SOURCE_APP) @classmethod def get_principal_google(cls): return cls.get_principal_source(PRINCIPAL_SOURCE_GOOGLE) @classmethod def get_principal_apple(cls): return cls.get_principal_source(PRINCIPAL_SOURCE_APPLE) class IAmAppAction(MasterModel): class Meta: db_table = "iam_app_action" class IAmAppResource(MasterModel): action = models.ManyToManyField( IAmAppAction, through="IAmAppResourceActionLink", related_name="app_resource_action", ) class Meta: db_table = "iam_app_resource" class IAmRoleAppResourceActionLinkManager(models.Manager): def generate_app_resource_action_data(self): """ Generate a dictionary mapping resource names to associated actions. Returns: dict: A dictionary with resource names as keys and nested dictionaries where action IDs are keys and action names are values. Example: { "res1": {1: "a1", 2: "a2"}, "res2": {3: "a1", 4: "a2"} } """ app_resource_action = self.select_related("app_resource", "app_action").all() resource_action_link = {} for item in app_resource_action: resource = item.app_resource.name action = item.app_action.name id = item.id if resource in resource_action_link: resource_action_link[resource][id] = action else: resource_action_link[resource] = {id: action} # print(resource_action_link) return resource_action_link class IAmAppResourceActionLink(models.Model): app_resource = models.ForeignKey( IAmAppResource, related_name="resource_action_link_app_resource", on_delete=models.CASCADE, ) app_action = models.ForeignKey( IAmAppAction, related_name="resource_action_link_app_action", on_delete=models.CASCADE, ) objects = IAmRoleAppResourceActionLinkManager() class Meta: db_table = "iam_app_resource_action_link" def __str__(self): return f"{self.app_resource.name}: {self.app_action.name}" class IAmRole(MasterModel): app_resource_action = models.ManyToManyField( IAmAppResourceActionLink, through="IAmRoleAppResourceActionLink", related_name="role_app_resource_action", ) class Meta: db_table = "iam_role" class IAmRoleAppResourceActionLink(models.Model): role = models.ForeignKey( IAmRole, related_name="role_app_resource_action_link_role", on_delete=models.CASCADE, ) app_resource_action = models.ForeignKey( IAmAppResourceActionLink, related_name="role_app_resource_action_link_app_resource_action", on_delete=models.CASCADE, ) class Meta: db_table = "iam_role_app_resource_action_link" class IAmPrincipalGroup(MasterModel): role = models.ManyToManyField( IAmRole, through="IAmPricipalGroupRoleLink", related_name="principal_group_role" ) class Meta: db_table = "iam_principal_group" class IAmPricipalGroupRoleLink(models.Model): principal_group = models.ForeignKey( IAmPrincipalGroup, related_name="role_link_principal_group", on_delete=models.CASCADE, ) role = models.ForeignKey( IAmRole, related_name="role_link_role", on_delete=models.CASCADE ) class Meta: db_table = "iam_principal_group_role_link" class IAmPrincipalManager(BaseUserManager): def create_user(self, email, password=None, **extra_fields): if not email: raise ValueError("The Email field must be set") email = self.normalize_email(email) user = self.model(email=email, **extra_fields) user.set_password(password) user.save(using=self._db) return user def create_superuser(self, email, password=None, **extra_fields): extra_fields.setdefault("username", email) extra_fields.setdefault("is_staff", True) extra_fields.setdefault("is_superuser", True) extra_fields.setdefault("phone_no", "+919978895465") extra_fields.setdefault("gender", "Male") extra_fields.setdefault("date_of_birth", timezone.now()) extra_fields.setdefault("created_by", None) extra_fields.setdefault("created_on", timezone.now()) extra_fields.setdefault("modified_by", None) extra_fields.setdefault("modified_on", timezone.now()) return self.create_user(email, password, **extra_fields) class IAmPrincipal(AbstractUser): principal_type = models.ForeignKey( IAmPrincipalType, related_name="principals_type", null=True, on_delete=models.PROTECT, ) principal_source = models.ForeignKey( IAmPrincipalSource, related_name="principals_source", on_delete=models.CASCADE, null=True, ) email = models.EmailField(unique=True) gender = models.CharField(max_length=6, blank=True, null=True) date_of_birth = models.DateField(blank=True, null=True) # phone_no = PhoneNumberField() phone_no = models.CharField(max_length=15, blank=True, null=True) address_line1 = models.TextField(blank=True, null=True) address_line2 = models.TextField(blank=True, null=True) city = models.CharField(max_length=100, blank=True, null=True) state = models.CharField(max_length=100, blank=True, null=True) country = models.CharField(max_length=100, blank=True, null=True) post_code = models.CharField(max_length=100, blank=True, null=True) profile_photo = models.ImageField(upload_to="profile", blank=True, null=True) phone_verified = models.BooleanField(default=False) email_verified = models.BooleanField(default=False) created_by = models.ForeignKey( "self", null=True, blank=True, related_name="creations", on_delete=models.SET_NULL, ) created_on = models.DateTimeField(auto_now_add=True) modified_by = models.ForeignKey( "self", null=True, blank=True, related_name="modifications", on_delete=models.SET_NULL, ) modified_on = models.DateTimeField(auto_now=True) deleted = models.BooleanField(default=False) principal_group = models.ManyToManyField( IAmPrincipalGroup, through="IAmPrincipalGroupLink", related_name="principal_groups", ) register_complete = models.BooleanField(default=False) player_id = models.CharField( max_length=255, null=True, blank=True, help_text="OneSignal player id for push notification", ) USERNAME_FIELD = "email" REQUIRED_FIELDS = [] objects = IAmPrincipalManager() class Meta: db_table = "iam_principal" def __str__(self): return f"{self.email}" class IAmPrincipalGroupLink(models.Model): principal = models.ForeignKey( IAmPrincipal, related_name="principal_group_link_principal", on_delete=models.CASCADE, ) principal_group = models.ForeignKey( IAmPrincipalGroup, related_name="principal_group_link_group", on_delete=models.CASCADE, ) class Meta: db_table = "iam_principal_principal_group_link" class IAmPrincipalOtp(models.Model): principal = models.ForeignKey( IAmPrincipal, related_name="principal_otp", on_delete=models.CASCADE ) otp_code = models.CharField(max_length=4) otp_purpose = models.CharField(max_length=50, null=True, blank=True) valid_till = models.DateTimeField() is_used = models.BooleanField(default=False) class Meta: db_table = "iam_principal_otp" def __str__(self): return f"{self.principal.phone_no}:{self.otp_code} : {self.otp_purpose}" def save(self, *args, **kwargs): if not self.pk: self.otp_code = RandomGenerator.random_otp() self.valid_till = timezone.now() + timezone.timedelta( minutes=settings.OTP_EXPIRE_TIME ) super(IAmPrincipalOtp, self).save(*args, **kwargs) def is_expired(self): return timezone.now() >= self.valid_till class IAmPrincipalBiometric(BaseModel): principal = models.ForeignKey( IAmPrincipal, related_name="principal_biometric", on_delete=models.CASCADE ) biometric_type = models.CharField(max_length=100) biometric_data = models.CharField(max_length=255) class Meta: db_table = "iam_principal_biometric" def __str__(self): return f"{self.principal.first_name}:{self.biometric_type}" class AppVersion(models.Model): version = models.CharField(max_length=10, validators=[RegexValidator(r'^\d+\.\d+\.\d+$')]) force_upgrade = models.BooleanField(default=False, help_text='Indicates whether a force upgrade is needed for this app version.') recommend_upgrade = models.BooleanField(default=False, help_text='Indicates whether a recommend upgrade is needed for this app version.') class Meta: db_table = "app_version" def __str__(self): return self.version