from django.shortcuts import render,redirect from django.contrib import messages from visitors_app.models import user,roles,location,company,department,areas,designation from django.contrib.auth.hashers import make_password from visitors_app.context_processors import my_constants from django.shortcuts import get_object_or_404 from django.http import JsonResponse from visitors_app.views import date_time from django.core.files.images import get_image_dimensions from django.db import connection from visitors_app.views import generate_vms_string from django.views.decorators.csrf import csrf_exempt from django.core.exceptions import ValidationError from django.core.validators import validate_email from django.db import transaction import csv import io import base64 from openpyxl import Workbook from openpyxl.styles import PatternFill, Font from visitors_app.check_subscription import check_subscription from django.db.models import Q from django.db import IntegrityError @check_subscription def admin_user_add(request): constants = my_constants(request) username = request.session.get('admin') if not username: return redirect('admin') user_data = constants.get('admin_data', {}) user_id = user_data.get('id') locationes = location.objects.filter(status=1) companyes = company.objects.filter(status=1) departmentes = department.objects.all() role = roles.objects.all() designationes = designation.objects.all() # area = areas.objects.all() if request.method == 'POST': first_name = request.POST['first_name'] last_name = request.POST['last_name'] email = request.POST['email'] employee_code = request.POST['employee_code'] image = request.FILES.get('image') if employee_code == '': messages.error(request, "Error: Employee Code is Required.", extra_tags='danger') return redirect('admin_user_add') else: pass if user.objects.filter(employee_code=employee_code).exists(): messages.error(request, 'All ready Employee Code exists......', extra_tags='danger') return redirect('admin_user_add') else: pass # if user.objects.filter(email=email).exists(): # messages.error(request, 'All Ready Email Id Exists', extra_tags='danger') # return redirect('admin_user_add') check_email_id= user.objects.filter(type__in=['employee', 'admin', 'gate_keeper','visitors'],email=email).exists() if check_email_id : messages.error(request, 'All Ready Email Id Exists', extra_tags='danger') return redirect('admin_user_add') if image: if 'image' in request.FILES and request.FILES['image'] and len(request.FILES['image']) != 0: image = request.FILES['image'] try: width, height = get_image_dimensions(image) except AttributeError: messages.error(request, "Error: Uploaded file is not an image.", extra_tags='danger') return redirect('admin_user_add') # Check if the image format is valid valid_formats = ['image/jpeg', 'image/png', 'image/jpg'] if image.content_type not in valid_formats: messages.error(request, "Error: Invalid image format.", extra_tags='danger') return redirect('admin_user_add') else: messages.error(request, "Error: Invalid image format.", extra_tags='danger') return redirect('admin_user_add') else: image = image password = request.POST['password'] mobile = request.POST['mobile'] if mobile != '': if user.objects.filter(mobile=mobile).exists(): messages.error(request, 'Already Mobile exists') return redirect('admin_user_add') else: pass gender = request.POST['gender'] address = request.POST['address'] type = request.POST.getlist('roles_id[]') type = ','.join([role.lower().replace(' ', '_') for role in type]) company_id = request.POST['company_id'] department_id = request.POST['department_id'] location_id = request.POST['location_id'] designation_id = request.POST['designation_id'] status = request.POST['status'] approval_require_value = request.POST['approval_require'] if approval_require_value == '1': approval_require = True elif approval_require_value == '0': approval_require = False else: approval_require = False password = make_password(password) created_at = date_time() useres = user(first_name=first_name, last_name=last_name, email=email, password=password, mobile=mobile, gender=gender, address=address, type=type, company_id=company_id, department_id=department_id, location_id=location_id, is_active=status, created_at=created_at, uni_id=generate_vms_string(), image=image, employee_code=employee_code, created_by=user_id, designation_id=designation_id, approval_require=approval_require) try: useres.save() except Exception as e: messages.error(request, f"{e}", extra_tags="danger") return redirect('admin_user_add') messages.success(request, "user successfully created.", extra_tags="success") return redirect('admin_user') return render(request, 'dashboard/admin_dashboard/admin_user/admin_user_add.html', {'locationes': locationes, 'companyes': companyes, 'departmentes': departmentes, 'roles': role, 'designationes': designationes,'function_name': 'Admin','username': username}) @check_subscription def admin_user(request): username = request.session.get('admin') if not username: return redirect('admin') return render(request, 'dashboard/admin_dashboard/admin_user/admin_user.html',{'function_name': 'Admin','username': username}) @check_subscription def admin_user_ajax_page_ajax(request): constants = my_constants(request) username = request.session.get('admin') if not username: return redirect('admin') database_name = constants['database_name'] start = int(request.POST.get('start', 0)) length = int(request.POST.get('length', 10)) search_value = request.POST.get('search[value]', '') search = '' if search_value != '': search = f"""AND ( {database_name}.users.first_name LIKE "%{search_value}%" OR {database_name}.users.last_name LIKE "%{search_value}%" OR {database_name}.users.email LIKE "%{search_value}%" OR {database_name}.users.mobile LIKE "%{search_value}%" OR {database_name}.company.company_name LIKE "%{search_value}%" OR {database_name}.location.location_name LIKE "%{search_value}%" OR {database_name}.department.department_name LIKE "%{search_value}%" OR {database_name}.users.employee_code LIKE "%{search_value}%" OR {database_name}.users.type LIKE "%{search_value}%" )""" # First SQL query to get the count of users sql_query = f""" SELECT COUNT(*) AS user_count FROM {database_name}.users LEFT JOIN {database_name}.company ON {database_name}.users.company_id = {database_name}.company.id LEFT JOIN {database_name}.location ON {database_name}.users.location_id = {database_name}.location.id LEFT JOIN {database_name}.department ON {database_name}.users.department_id = {database_name}.department.id LEFT JOIN {database_name}.designation ON {database_name}.users.designation_id = {database_name}.designation.id WHERE {database_name}.users.type != 'visitors' {search}; """ with connection.cursor() as cursor: cursor.execute(sql_query) database_all_data = cursor.fetchall() recordsTotal = database_all_data[0][0] if database_all_data else 0 # Second SQL query to get detailed user data sql_query = f""" SELECT {database_name}.users.*, {database_name}.company.company_name, {database_name}.location.location_name, {database_name}.department.department_name, {database_name}.users.approval_require FROM {database_name}.users LEFT JOIN {database_name}.company ON {database_name}.users.company_id = {database_name}.company.id LEFT JOIN {database_name}.location ON {database_name}.users.location_id = {database_name}.location.id LEFT JOIN {database_name}.department ON {database_name}.users.department_id = {database_name}.department.id LEFT JOIN {database_name}.designation ON {database_name}.users.designation_id = {database_name}.designation.id WHERE {database_name}.users.type != 'visitors' {search} ORDER BY {database_name}.users.id DESC LIMIT {length} OFFSET {start}; """ with connection.cursor() as cursor: cursor.execute(sql_query) database_all_data = cursor.fetchall() columns = [col[0] for col in cursor.description] all_user_dtl = [dict(zip(columns, row)) for row in database_all_data] return JsonResponse({ "recordsTotal": recordsTotal, "recordsFiltered": recordsTotal, 'data': all_user_dtl }) @check_subscription def admin_user_edit(request, id): constants = my_constants(request) username = request.session.get('admin') if not username: return redirect('admin') user_data = constants.get('admin_data', {}) all_user = get_object_or_404(user, id=id) email = username['user_email'] mobile = all_user.mobile all_email = all_user.email all_location = location.objects.filter(status=1) all_company = company.objects.filter(status=1) all_department = department.objects.all() all_roles = roles.objects.all() designationes = designation.objects.all() # employee_codes = all_user.employee_code role_mapping = {role.name.lower().replace(' ', '_'): role.name for role in all_roles} if request.method == "POST": all_user.first_name = request.POST['first_name'] employee_codes = request.POST['employee_code'] # employee_codes = int(employee_codes) all_user.last_name = request.POST['last_name'] all_user.email = request.POST['email'] all_user.company_id = request.POST['company_id'] all_user.department_id = request.POST['department_id'] all_user.location_id = request.POST['location_id'] all_user.gender = request.POST['gender'] all_user.address = request.POST['address'] all_user.is_active = request.POST['status'] all_user.mobile = request.POST['mobile'] all_user.designation_id = request.POST['designation_id'] if request.POST['mobile'] != '': if mobile != all_user.mobile: user_mobile_is_exist = user.objects.filter(mobile=all_user.mobile).exists() if user_mobile_is_exist == True: messages.error(request, 'User mobile already exists.', extra_tags='danger') return redirect('admin_user_edit', id=id) if request.POST['email'] != '': if all_email != all_user.email: user_email_is_exist = user.objects.filter(Q(type='admin') | Q(type='employee') | Q(type='gate_keeper') | Q(type='visitors'),email=all_user.email,).exists() if user_email_is_exist == True: messages.error(request, 'User Email Id Already Exists.', extra_tags='danger') return redirect('admin_user_edit', id=id) if int(employee_codes) != int(all_user.employee_code): user_employee_codes_is_exist = user.objects.filter(employee_code=employee_codes).exists() if user_employee_codes_is_exist == True: messages.error(request, "User Employee Codes already exists.", extra_tags='danger') return redirect('admin_user_edit', id=id) all_user.employee_code = employee_codes # Handle multiple role selections selected_roles = request.POST.getlist('roles_id[]') all_user.type = ','.join(selected_roles) # Save roles as a comma-separated string approval_require = request.POST.get('approval_require', '') all_user.approval_require = bool(int(approval_require)) if approval_require != '' else False password = request.POST['Password'] if password: all_user.password = make_password(password) # Consider hashing the password if 'image' in request.FILES: file = request.FILES['image'] file_extension = file.name.split('.')[-1].lower() if file_extension in ['png', 'jpg', 'jpeg']: all_user.image = file else: messages.error(request, 'Error: Invalid image format.', extra_tags='danger') return redirect('admin_user_edit', id=id) all_user.created_at = date_time() all_user.save() all_user.save() messages.success(request, "User successfully updated.", extra_tags="success") return redirect('admin_user') # For editing, fetch the user's existing roles user_role_keys = all_user.type.split(',') # Split roles back into a list return render(request, 'dashboard/admin_dashboard/admin_user/admin_user_edit.html', { 'all_user': all_user, 'locationes': all_location, 'companyes': all_company, 'departmentes': all_department, 'roles': role_mapping, 'user_role_keys': user_role_keys, # Pass the user's existing role keys 'user_data': user_data, 'designationes': designationes, 'function_name': 'Admin', 'username': username }) # @check_subscription # def admin_user_edit(request, id): # constants = my_constants(request) # # username = request.session.get('admin') # if not username: # return redirect('admin') # # user_data = constants.get('admin_data', {}) # # all_user = get_object_or_404(user, id=id) # # old_mobile = all_user.mobile # old_email = all_user.email # old_employee_code = all_user.employee_code # # all_location = location.objects.filter(status=1) # all_company = company.objects.filter(status=1) # all_department = department.objects.all() # all_roles = roles.objects.all() # designationes = designation.objects.all() # # role_mapping = { # role.name.lower().replace(' ', '_'): role.name # for role in all_roles # } # # if request.method == "POST": # # try: # # ========================= # # Get Form Data # # ========================= # first_name = request.POST.get('first_name') # last_name = request.POST.get('last_name') # # new_email = request.POST.get('email') # new_mobile = request.POST.get('mobile') # new_employee_code = request.POST.get('employee_code') # # company_id = request.POST.get('company_id') # department_id = request.POST.get('department_id') # location_id = request.POST.get('location_id') # # gender = request.POST.get('gender') # address = request.POST.get('address') # status = request.POST.get('status') # designation_id = request.POST.get('designation_id') # # # ========================= # # Mobile Validation # # ========================= # if new_mobile: # if str(old_mobile) != str(new_mobile): # # user_mobile_is_exist = user.objects.filter( # mobile=new_mobile # ).exclude(id=id).exists() # # if user_mobile_is_exist: # messages.error( # request, # 'User mobile already exists.', # extra_tags='danger' # ) # return redirect('admin_user_edit', id=id) # # # ========================= # # Email Validation # # ========================= # if new_email: # if old_email != new_email: # # user_email_is_exist = user.objects.filter( # Q(type='admin') | # Q(type='employee') | # Q(type='gate_keeper'), # email=new_email # ).exclude(id=id).exists() # # if user_email_is_exist: # messages.error( # request, # 'User Email Id Already Exists.', # extra_tags='danger' # ) # return redirect('admin_user_edit', id=id) # # # ========================= # # Employee Code Validation # # ========================= # if new_employee_code: # # if str(old_employee_code) != str(new_employee_code): # # employee_code_exists = user.objects.filter( # employee_code=new_employee_code # ).exclude(id=id).exists() # # if employee_code_exists: # messages.error( # request, # "User Employee Code already exists.", # extra_tags='danger' # ) # return redirect('admin_user_edit', id=id) # # # ========================= # # Assign Values # # ========================= # all_user.first_name = first_name # all_user.last_name = last_name # all_user.email = new_email # all_user.mobile = new_mobile # all_user.employee_code = new_employee_code # # all_user.company_id = company_id # all_user.department_id = department_id # all_user.location_id = location_id # # all_user.gender = gender # all_user.address = address # all_user.is_active = status # all_user.designation_id = designation_id # # # ========================= # # Multiple Roles # # ========================= # selected_roles = request.POST.getlist('roles_id[]') # # if selected_roles: # all_user.type = ",".join(selected_roles) # # # ========================= # # Approval Require # # ========================= # approval_require = request.POST.get('approval_require', '') # # all_user.approval_require = ( # bool(int(approval_require)) # if approval_require != '' # else False # ) # # # ========================= # # Password Update # # ========================= # password = request.POST.get('Password') # # if password: # all_user.password = make_password(password) # # # ========================= # # Image Upload # # ========================= # if 'image' in request.FILES: # # file = request.FILES['image'] # # file_extension = file.name.split('.')[-1].lower() # # if file_extension in ['png', 'jpg', 'jpeg']: # all_user.image = file # # else: # messages.error( # request, # 'Invalid image format. Only PNG, JPG, JPEG allowed.', # extra_tags='danger' # ) # return redirect('admin_user_edit', id=id) # # # ========================= # # Update Time # # ========================= # all_user.created_at = date_time() # # # ========================= # # Save # # ========================= # all_user.save() # # messages.success( # request, # "User successfully updated.", # extra_tags="success" # ) # # return redirect('admin_user') # # except IntegrityError as e: # # messages.error( # request, # f"IntegrityError occurred: {str(e)}", # extra_tags='danger' # ) # # return redirect('admin_user_edit', id=id) # # except Exception as e: # # messages.error( # request, # f"Error: {str(e)}", # extra_tags='danger' # ) # # return redirect('admin_user_edit', id=id) # # # ========================= # # Existing User Roles # # ========================= # user_role_keys = all_user.type.split(',') if all_user.type else [] # # return render( # request, # 'dashboard/admin_dashboard/admin_user/admin_user_edit.html', # { # 'all_user': all_user, # 'locationes': all_location, # 'companyes': all_company, # 'departmentes': all_department, # 'roles': role_mapping, # 'user_role_keys': user_role_keys, # 'user_data': user_data, # 'designationes': designationes, # 'function_name': 'Admin', # 'username': username # } # ) @check_subscription def admin_user_delete(request, id): username = request.session.get('admin') if not username: return redirect('admin') all_user = get_object_or_404(user, id=id) all_user.delete() data = { 'status': 1, 'message': f'User Deleted Successfully.', } messages.success(request, f"successfully User Deleted!", extra_tags="success") return JsonResponse(data) @csrf_exempt def admin_user_csv_upload(request): constants = my_constants(request) username = request.session.get('admin') if not username: return redirect('admin') user_data = constants.get('admin_data', {}) user_id = user_data.get('id') if not user_id: return JsonResponse({'status': 'error', 'message': 'Invalid user data'}) if request.method == 'POST': errors = [] failed_rows = [] success_count = 0 try: if 'csv_file' not in request.FILES: return JsonResponse({'status': 'error', 'message': 'No CSV file uploaded'}) csv_file = request.FILES['csv_file'] if not csv_file.name.endswith('.csv'): return JsonResponse({'status': 'error', 'message': 'File is not a CSV'}) if csv_file.size > 5 * 1024 * 1024: return JsonResponse({'status': 'error', 'message': 'File size exceeds 5MB limit'}) raw_bytes = csv_file.read() for encoding in ('utf-8-sig', 'utf-8', 'latin-1', 'cp1252'): try: decoded_file = raw_bytes.decode(encoding).splitlines() break except (UnicodeDecodeError, LookupError): continue else: return JsonResponse( {'status': 'error', 'message': 'Could not decode file. Please save the CSV as UTF-8.'}) reader = csv.DictReader(decoded_file) fieldnames = reader.fieldnames + ['Error'] if reader.fieldnames else [] required_columns = { 'First_Name', 'Last_Name', 'Password', 'Mobile', 'Gender', 'Address', 'Company_Name', 'Department_Name', 'Location_Name', 'Employee_Code', 'Designation_Name' } # Email is no longer required if not required_columns.issubset(reader.fieldnames or []): missing_cols = required_columns - set(reader.fieldnames or []) return JsonResponse({ 'status': 'error', 'message': f'CSV is missing required columns: {", ".join(missing_cols)}' }) with transaction.atomic(): for row_num, row in enumerate(reader, start=2): try: # row = {k: v.strip() for k, v in row.items()} row = {k: v.strip() if v else '' for k, v in row.items()} email = row.get('Email', '').strip() employee_code_str = row['Employee_Code'] # Email validation only if provided if email: try: validate_email(email) except ValidationError: row['Error'] = f"Invalid email format: '{email}'" failed_rows.append(row) continue # Check for duplicates if email provided # email_exists = user.objects.filter(email=email).exists() if email else None employee_code_exists = user.objects.filter(employee_code=employee_code_str).exists() # Lookups company_obj = company.objects.filter(company_name__iexact=row['Company_Name']).first() department_obj = department.objects.filter(department_name__iexact=row['Department_Name']).first() location_obj = location.objects.filter(location_name__iexact=row['Location_Name']).first() designation_obj = designation.objects.filter(name__iexact=row['Designation_Name']).first() # Validate related objects and duplicates missing = [] if not company_obj: missing.append('Company_Name') if not department_obj: missing.append('Department_Name') if not location_obj: missing.append('Location_Name') if not designation_obj: missing.append('Designation_Name') # if email_exists: # missing.append('Duplicate Email') if employee_code_exists: missing.append('Duplicate Employee Code') if missing: row['Error'] = f"Invalid or Duplicate Data: {', '.join(missing)}" failed_rows.append(row) continue # Create the user user.objects.create( first_name=row['First_Name'], last_name=row['Last_Name'], email=email, mobile=row['Mobile'], gender=row['Gender'], address=row['Address'], employee_code=employee_code_str, password=make_password(row['Password']), company_id=company_obj.id, department_id=department_obj.id, location_id=location_obj.id, designation_id=designation_obj.id, approval_require=0, # Approval require default false is_superuser=0, is_active=1, confirmed=0, type='employee', created_by=user_id, uni_id=generate_vms_string(), created_at=date_time() ) success_count += 1 except Exception as e: row['Error'] = f"Unexpected error: {str(e)}" failed_rows.append(row) continue # If there are failed rows, export to Excel if failed_rows: wb = Workbook() ws = wb.active ws.title = "Failed Uploads" # Write headers headers = fieldnames ws.append(headers) error_col_index = headers.index('Error') + 1 error_header_cell = ws.cell(row=1, column=error_col_index) error_header_cell.fill = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid") error_header_cell.font = Font(color="FFFFFF") # Write failed rows for row_data in failed_rows: row_values = [row_data.get(field, '') for field in headers] ws.append(row_values) error_cell = ws.cell(row=ws.max_row, column=error_col_index) error_cell.fill = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid") error_cell.font = Font(color="FFFFFF") output = io.BytesIO() wb.save(output) output.seek(0) excel_base64 = base64.b64encode(output.getvalue()).decode('utf-8') return JsonResponse({ 'status': 'success', 'message': f'{success_count} users imported successfully', 'count': success_count, 'failed_rows_excel': excel_base64, 'failed_rows_filename': 'failed_uploads.xlsx' }) # If no failed rows return JsonResponse({ 'status': 'success', 'message': f'{success_count} Designation imported successfully', 'count': success_count }) except Exception as e: return JsonResponse({'status': 'error', 'message': f"File processing error: {str(e)}"}) return JsonResponse({'status': 'error', 'message': 'Invalid request method'})