You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
470 lines
19 KiB
470 lines
19 KiB
import random
|
|
import base64
|
|
import hashlib
|
|
from cryptography.fernet import Fernet
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AbstractUser
|
|
from django.db import models
|
|
from django.utils.text import slugify
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.utils import timezone
|
|
from phonenumber_field.modelfields import PhoneNumberField
|
|
from utils.validators import validate_possible_number
|
|
from apps.account.manager import UserManager
|
|
|
|
|
|
class User(AbstractUser):
|
|
class DeviceOs(models.TextChoices):
|
|
android = 'android', _('Android')
|
|
apple = 'apple', _('Apple iOS')
|
|
web = 'web', _('Web')
|
|
|
|
class UserType(models.TextChoices):
|
|
CLIENT = 'client', _('Client / User')
|
|
STUDENT = 'student', _('Student / Member')
|
|
PROFESSOR = 'professor', _('Professor / Scholar')
|
|
CONSULTANT = 'consultant', _('Consultant')
|
|
ADMIN = 'admin', _('Admin')
|
|
SUPER_ADMIN = 'super_admin', _('Super Admin')
|
|
REGIONAL_ADMIN = 'regional_admin', _('Regional Admin')
|
|
INSTITUTION_ADMIN = 'institution_admin', _('Institution Admin')
|
|
EDITOR = 'editor', _('Editor')
|
|
VIEWER = 'viewer', _('Viewer')
|
|
|
|
class GenderChoices(models.TextChoices):
|
|
MALE = 'male', _('Male')
|
|
FEMALE = 'female', _('Female')
|
|
OTHER = 'other', _('Other')
|
|
|
|
first_name = models.CharField(
|
|
max_length=150,
|
|
verbose_name=_("First Name"),
|
|
null=True,
|
|
blank=True
|
|
)
|
|
last_name = models.CharField(
|
|
max_length=150,
|
|
verbose_name=_("Last Name"),
|
|
null=True,
|
|
blank=True
|
|
)
|
|
username = models.CharField(
|
|
unique=True,
|
|
null=True,
|
|
blank=True,
|
|
max_length=150,
|
|
verbose_name=_("Username"),
|
|
error_messages={'unique': _("A user with that username already exists.")}
|
|
)
|
|
email = models.EmailField(
|
|
unique=True,
|
|
verbose_name=_("Email Address"),
|
|
help_text=_("User primary email address."),
|
|
null=True,
|
|
blank=True,
|
|
error_messages={'unique': _("A user with that email already exists.")}
|
|
)
|
|
fullname = models.CharField(
|
|
max_length=255,
|
|
verbose_name=_("Full Name"),
|
|
help_text=_("Full name of the user."),
|
|
null=True,
|
|
blank=True
|
|
)
|
|
birthdate = models.DateField(verbose_name=_('Birthdate'), null=True, blank=True)
|
|
avatar = models.ImageField(
|
|
max_length=512,
|
|
null=True,
|
|
blank=True,
|
|
upload_to='users/avatars/%Y/%m/',
|
|
verbose_name=_('Avatar')
|
|
)
|
|
phone_number = PhoneNumberField(
|
|
validators=[validate_possible_number],
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_('Phone Number'),
|
|
help_text=_("e.g., +1 555 1234567")
|
|
)
|
|
language = models.CharField(
|
|
max_length=10,
|
|
default='en',
|
|
choices=settings.LANGUAGES,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_('Language')
|
|
)
|
|
gender = models.CharField(
|
|
max_length=20,
|
|
choices=GenderChoices.choices,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_('Gender')
|
|
)
|
|
user_type = models.CharField(
|
|
max_length=20,
|
|
choices=UserType.choices,
|
|
default=UserType.CLIENT,
|
|
verbose_name=_("User Type")
|
|
)
|
|
date_joined = models.DateTimeField(
|
|
auto_now_add=True,
|
|
verbose_name=_("Date Joined")
|
|
)
|
|
|
|
city = models.CharField(verbose_name=_('City'), max_length=255, null=True, blank=True)
|
|
country = models.CharField(max_length=255, verbose_name=_('Country'), null=True, blank=True)
|
|
title = models.CharField(max_length=255, null=True, blank=True, verbose_name=_('Title / Honorific'))
|
|
rank = models.CharField(max_length=255, null=True, blank=True, verbose_name=_('Academic / Clerical Rank'))
|
|
institution_name = models.CharField(max_length=255, null=True, blank=True, verbose_name=_('Affiliated Institution'))
|
|
website = models.CharField(max_length=512, null=True, blank=True, verbose_name=_('Personal Website'))
|
|
address = models.TextField(null=True, blank=True, verbose_name=_('Postal Address'))
|
|
|
|
device_id = models.CharField(verbose_name=_('Device ID'), max_length=255, null=True, blank=True)
|
|
device_os = models.CharField(choices=DeviceOs.choices, null=True, blank=True, max_length=16, verbose_name=_('Device OS'))
|
|
user_agent = models.TextField(verbose_name=_('User Agent'), null=True, blank=True)
|
|
client_ip = models.CharField(max_length=64, verbose_name=_('Client IP'), null=True, blank=True)
|
|
|
|
fcm = models.CharField(max_length=512, null=True, blank=True, verbose_name=_('FCM Token'))
|
|
slug = models.SlugField(max_length=255, unique=True, null=True, blank=True, verbose_name=_('Slug'))
|
|
is_staff = models.BooleanField(default=False, verbose_name=_('Is Staff'))
|
|
is_active = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Active"),
|
|
help_text=_("Designates whether this user should be treated as active.")
|
|
)
|
|
deleted_at = models.DateTimeField(null=True, blank=True, verbose_name=_('Deleted At'))
|
|
info = models.TextField(verbose_name=_("Bio / Info"), null=True, blank=True)
|
|
skill = models.CharField(max_length=512, null=True, blank=True, verbose_name=_('Skill / Role'))
|
|
languages = models.JSONField(default=list, blank=True, verbose_name=_('Languages'))
|
|
skills = models.JSONField(default=list, blank=True, verbose_name=_('Skills'))
|
|
password_enc = models.CharField(
|
|
max_length=512,
|
|
blank=True,
|
|
default='',
|
|
verbose_name=_("Encrypted Password"),
|
|
help_text=_("Encrypted copy of password for displaying in the admin panel if required.")
|
|
)
|
|
|
|
role = models.ForeignKey(
|
|
'account.Role',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='users',
|
|
verbose_name=_("Assigned Role")
|
|
)
|
|
custom_permissions = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
verbose_name=_("Custom Permission Overrides"),
|
|
help_text=_("Dictionary of permission overrides: {permission_name: True/False}")
|
|
)
|
|
managed_regions = models.ManyToManyField(
|
|
'profiles.Region',
|
|
blank=True,
|
|
related_name='regional_admins',
|
|
verbose_name=_("Managed Regions"),
|
|
help_text=_("Regions assigned to this regional administrator or staff member. Administrative actions and regional permissions apply only to these regions.")
|
|
)
|
|
regional_permissions = models.JSONField(
|
|
default=list,
|
|
blank=True,
|
|
verbose_name=_("Regional Permission Overrides"),
|
|
help_text=_("List of permission names that are restricted to the user's managed regions (e.g. ['can_manage_institutions', 'can_manage_tickets']).")
|
|
)
|
|
|
|
objects = UserManager()
|
|
|
|
EMAIL_FIELD = "email"
|
|
USERNAME_FIELD = "email"
|
|
REQUIRED_FIELDS = []
|
|
|
|
def __str__(self):
|
|
return self.email or self.fullname or self.username or f"User #{self.id}"
|
|
|
|
def soft_delete(self):
|
|
self.deleted_at = timezone.now()
|
|
self.is_active = False
|
|
self.fullname = f'{self.fullname}:deleted' if self.fullname else 'deleted_user'
|
|
number = str(random.randint(1000000000, 9999999999))
|
|
if self.phone_number:
|
|
self.phone_number = f'{self.phone_number}:deleted{number}'
|
|
if self.email:
|
|
self.email = f'{self.email}:deleted{number}'
|
|
if self.device_id:
|
|
self.device_id = f'{self.device_id}:deleted{number}'
|
|
self.save()
|
|
|
|
def save(self, *args, **kwargs):
|
|
if not self.username and self.email:
|
|
self.username = self.email
|
|
elif self.email and User.objects.filter(username=self.email).exclude(pk=self.pk).exists():
|
|
self.username = f'{self.email}:{self.id or random.randint(1000, 9999)}'
|
|
|
|
# Maintain first_name, last_name, and fullname synchronization
|
|
if self.first_name or self.last_name:
|
|
self.fullname = f"{self.first_name or ''} {self.last_name or ''}".strip()
|
|
elif self.fullname:
|
|
parts = self.fullname.strip().split(maxsplit=1)
|
|
if len(parts) == 2:
|
|
self.first_name = parts[0]
|
|
self.last_name = parts[1]
|
|
elif len(parts) == 1:
|
|
self.first_name = parts[0]
|
|
self.last_name = ""
|
|
|
|
if not self.role_id:
|
|
try:
|
|
from apps.account.models.role import Role
|
|
default_role = Role.objects.filter(is_default=True).first()
|
|
if default_role:
|
|
self.role = default_role
|
|
except Exception:
|
|
pass
|
|
|
|
return super().save(*args, **kwargs)
|
|
|
|
def get_full_name(self):
|
|
full = f"{self.first_name or ''} {self.last_name or ''}".strip()
|
|
return full or self.fullname or self.email or ""
|
|
|
|
def get_short_name(self):
|
|
return self.first_name or self.fullname or self.email or ""
|
|
|
|
@property
|
|
def is_guest(self):
|
|
return self.email is None
|
|
|
|
def has_role(self, role_name):
|
|
return self.groups.filter(name__iexact=f"{role_name} Group").exists() or self.user_type == role_name
|
|
|
|
def has_perm_flag(self, perm_name: str) -> bool:
|
|
"""
|
|
Check effective boolean permission flag:
|
|
1. Superuser always returns True.
|
|
2. Custom permission override on the user takes precedence if set.
|
|
3. Falls back to assigned Role value if user has a role.
|
|
4. Fallback when user has no explicit Role attached:
|
|
- Basic community actions ('can_create_posts', 'can_edit_own_posts', 'can_comment') default to True.
|
|
- 'can_publish_posts_directly' defaults to True if user is_editor.
|
|
5. Otherwise returns False.
|
|
"""
|
|
if not self.is_active:
|
|
return False
|
|
if self.is_superuser:
|
|
return True
|
|
if self.custom_permissions and perm_name in self.custom_permissions:
|
|
val = self.custom_permissions[perm_name]
|
|
if val is not None:
|
|
return bool(val)
|
|
if self.role and hasattr(self.role, perm_name):
|
|
return getattr(self.role, perm_name)
|
|
# Graceful fallback matching Role model defaults when no Role object is linked
|
|
if perm_name in ['can_create_posts', 'can_edit_own_posts', 'can_delete_own_posts', 'can_comment']:
|
|
return True
|
|
if perm_name == 'can_publish_posts_directly':
|
|
return bool(getattr(self, 'is_editor', False))
|
|
return False
|
|
|
|
@property
|
|
def is_super_admin(self):
|
|
return self.is_superuser or self.user_type == self.UserType.SUPER_ADMIN or (self.role and self.role.slug == 'super_admin')
|
|
|
|
@property
|
|
def is_regional_admin(self):
|
|
return self.is_super_admin or self.user_type == self.UserType.REGIONAL_ADMIN or (self.role and self.role.slug == 'regional_admin')
|
|
|
|
def can_manage_region(self, region) -> bool:
|
|
"""
|
|
Check if user has administrative rights over the specified region:
|
|
- Super admins have global authority over all regions.
|
|
- Regional admins / staff only have authority over their assigned managed_regions.
|
|
- If no region is provided or user has no region rights, returns False.
|
|
"""
|
|
if not self.is_authenticated or not self.is_active:
|
|
return False
|
|
if self.is_superuser or self.is_super_admin:
|
|
return True
|
|
if not (self.is_regional_admin or self.has_perm_flag('can_manage_institutions') or self.managed_regions.exists()):
|
|
return False
|
|
if not region:
|
|
return False
|
|
region_id = region.id if hasattr(region, 'id') else region
|
|
return self.managed_regions.filter(id=region_id).exists()
|
|
|
|
def has_regional_perm(self, perm_name: str, region=None) -> bool:
|
|
"""
|
|
Check if user has a permission, optionally scoped to a region:
|
|
- Super admin: always True.
|
|
- If region is None: True if user holds the permission flag in general.
|
|
- If region is provided:
|
|
* Must have the permission flag (or be a regional admin).
|
|
* Must have authority over that specific region via can_manage_region(region).
|
|
* If regional_permissions is configured, checks permission in dict/list.
|
|
"""
|
|
if not self.is_authenticated or not self.is_active:
|
|
return False
|
|
if self.is_superuser or self.is_super_admin:
|
|
return True
|
|
|
|
has_flag = self.has_perm_flag(perm_name) or self.is_regional_admin
|
|
if not has_flag:
|
|
return False
|
|
|
|
if self.regional_permissions:
|
|
allowed = False
|
|
# Check 1: direct key in dict (e.g. {"institutions.change": True} or {"can_manage_institutions": True})
|
|
if isinstance(self.regional_permissions, dict):
|
|
if perm_name in self.regional_permissions:
|
|
val = self.regional_permissions[perm_name]
|
|
allowed = bool(val)
|
|
# Check 2: category-action format (e.g. "institutions.change" -> category "institutions", action "change")
|
|
elif '.' in perm_name:
|
|
cat, action = perm_name.split('.', 1)
|
|
if cat in self.regional_permissions:
|
|
cat_perms = self.regional_permissions[cat]
|
|
if isinstance(cat_perms, (list, tuple, set)):
|
|
allowed = action in cat_perms
|
|
elif isinstance(cat_perms, dict):
|
|
allowed = bool(cat_perms.get(action))
|
|
elif isinstance(cat_perms, bool):
|
|
allowed = cat_perms
|
|
elif isinstance(self.regional_permissions, (list, tuple, set)):
|
|
allowed = perm_name in self.regional_permissions
|
|
|
|
if not allowed:
|
|
return False
|
|
|
|
if region is not None:
|
|
return self.can_manage_region(region)
|
|
|
|
return self.is_regional_admin or self.managed_regions.exists()
|
|
|
|
|
|
def get_managed_regions_queryset(self):
|
|
from apps.profiles.models import Region
|
|
if self.is_superuser or self.is_super_admin:
|
|
return Region.objects.all()
|
|
if self.is_regional_admin or self.managed_regions.exists():
|
|
return self.managed_regions.all()
|
|
return Region.objects.none()
|
|
|
|
@property
|
|
def is_institution_admin(self):
|
|
return self.is_regional_admin or self.user_type == self.UserType.INSTITUTION_ADMIN or (self.role and self.role.slug in ['institution_admin', 'institution_owner'])
|
|
|
|
@property
|
|
def is_editor(self):
|
|
return self.is_institution_admin or self.user_type == self.UserType.EDITOR or (self.role and self.role.slug == 'content_moderator')
|
|
|
|
def is_super_admin_panel_user(self):
|
|
return self.is_superuser or self.user_type == self.UserType.SUPER_ADMIN
|
|
|
|
def is_admin_panel_user(self):
|
|
return self.is_staff or self.is_superuser or self.has_perm_flag('can_access_admin_panel') or self.user_type in [
|
|
self.UserType.SUPER_ADMIN,
|
|
self.UserType.ADMIN,
|
|
self.UserType.REGIONAL_ADMIN,
|
|
self.UserType.INSTITUTION_ADMIN,
|
|
]
|
|
|
|
def is_professor_panel_user(self):
|
|
return self.user_type == self.UserType.PROFESSOR
|
|
|
|
def can_access_admin_panel(self):
|
|
return (
|
|
self.is_staff or
|
|
self.is_superuser or
|
|
self.has_perm_flag('can_access_admin_panel') or
|
|
self.user_type in [
|
|
self.UserType.SUPER_ADMIN,
|
|
self.UserType.ADMIN,
|
|
self.UserType.REGIONAL_ADMIN,
|
|
self.UserType.INSTITUTION_ADMIN,
|
|
self.UserType.EDITOR,
|
|
]
|
|
)
|
|
|
|
@property
|
|
def verification_status(self):
|
|
latest = self.verification_documents.first()
|
|
if latest:
|
|
return latest.status
|
|
return 'unverified'
|
|
|
|
|
|
def _get_fernet(self):
|
|
key_bytes = settings.SECRET_KEY.encode('utf-8')
|
|
hashed_key = hashlib.sha256(key_bytes).digest()
|
|
fernet_key = base64.urlsafe_b64encode(hashed_key)
|
|
return Fernet(fernet_key)
|
|
|
|
def set_plain_password(self, plain_password):
|
|
if not plain_password:
|
|
self.password_enc = ''
|
|
return
|
|
try:
|
|
password_bytes = plain_password.encode('utf-8')
|
|
fernet = self._get_fernet()
|
|
encrypted_token = fernet.encrypt(password_bytes)
|
|
self.password_enc = encrypted_token.decode('utf-8')
|
|
except Exception:
|
|
self.password_enc = ''
|
|
|
|
def get_plain_password(self):
|
|
if not self.password_enc:
|
|
return ''
|
|
try:
|
|
encrypted_bytes = self.password_enc.encode('utf-8')
|
|
fernet = self._get_fernet()
|
|
decrypted_bytes = fernet.decrypt(encrypted_bytes)
|
|
return decrypted_bytes.decode('utf-8')
|
|
except Exception:
|
|
return ''
|
|
|
|
class Meta:
|
|
ordering = ("-id",)
|
|
verbose_name = _("User")
|
|
verbose_name_plural = _("Users")
|
|
|
|
|
|
class LoginHistory(models.Model):
|
|
user = models.ForeignKey("account.User", on_delete=models.CASCADE, related_name='login_history', verbose_name=_('User'))
|
|
lat = models.FloatField(verbose_name=_('Latitude'), null=True, blank=True)
|
|
lon = models.FloatField(verbose_name=_('Longitude'), null=True, blank=True)
|
|
country = models.CharField(max_length=255, verbose_name=_('Country'), null=True, blank=True)
|
|
city = models.CharField(max_length=255, verbose_name=_('City'), null=True, blank=True)
|
|
ip = models.CharField(max_length=255, null=True, blank=True, verbose_name=_('IP Address'))
|
|
timezone = models.CharField(max_length=100, null=True, blank=True, verbose_name=_('Timezone'))
|
|
user_agent = models.TextField(verbose_name=_('User Agent'), null=True, blank=True)
|
|
device_os = models.CharField(max_length=16, null=True, blank=True, verbose_name=_('Device OS'))
|
|
at_time = models.DateTimeField(auto_now_add=True, verbose_name=_('Timestamp'))
|
|
|
|
class Meta:
|
|
verbose_name = _('Login History')
|
|
verbose_name_plural = _('Login Histories')
|
|
ordering = ('-at_time',)
|
|
|
|
def __str__(self):
|
|
return f"{self.user} - {self.ip} ({self.at_time})"
|
|
|
|
|
|
class LocationHistory(models.Model):
|
|
user = models.ForeignKey("account.User", on_delete=models.CASCADE, related_name='location_history', verbose_name=_('User'))
|
|
lat = models.FloatField(verbose_name=_('Latitude'), null=True, blank=True)
|
|
lon = models.FloatField(verbose_name=_('Longitude'), null=True, blank=True)
|
|
country = models.CharField(max_length=255, verbose_name=_('Country'), null=True, blank=True)
|
|
city = models.CharField(max_length=255, verbose_name=_('City'), null=True, blank=True)
|
|
selected_manually = models.BooleanField(default=False, verbose_name=_('Selected Manually'))
|
|
ip = models.CharField(max_length=255, null=True, blank=True, verbose_name=_('IP Address'))
|
|
timezone = models.CharField(null=True, blank=True, max_length=60, verbose_name=_('Timezone'))
|
|
at_time = models.DateTimeField(auto_now_add=True, verbose_name=_('Timestamp'))
|
|
|
|
class Meta:
|
|
verbose_name = _('Location History')
|
|
verbose_name_plural = _('Location History')
|
|
ordering = ('-at_time',)
|
|
|
|
def __str__(self):
|
|
return f"{self.user} - {self.city}, {self.country}"
|