from django.shortcuts import render, redirect from django.contrib import messages from visitors_app.models import location, company,designation 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.db import connection from django.views.decorators.csrf import csrf_exempt 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 @check_subscription def admin_designation_add(request): username = request.session.get('admin') if not username: return redirect('admin') locationes = location.objects.filter(status=1) companyes = company.objects.filter(status=1) if request.method == 'POST': designation_name = request.POST['designation_name'] # designation_allow_check_in = request.POST['allow_check_in'] created_at = date_time() designationes = designation(name=designation_name, created_at=created_at) designationes.save() messages.success(request, "Designation successfully created.", extra_tags="success") return redirect('admin_designation') return render(request, 'dashboard/admin_dashboard/admin_designation/admin_designation_add.html', {'locationes': locationes, 'companyes': companyes,'function_name': 'Admin','username':username}) @check_subscription def admin_designation(request): username = request.session.get('admin') if not username: return redirect('admin') return render(request, 'dashboard/admin_dashboard/admin_designation/admin_designation.html',{'function_name': 'Admin','username':username}) @check_subscription def admin_designation_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 = request.POST.get('start', 0) length = request.POST.get('length', 10) search_value = request.POST.get('search[value]', '') search = '' if search_value != '': # search = f'WHERE visitors.first_name LIKE "%{search_value}%" or employees.first_name LIKE "%{search_value}%" or appointment.id LIKE "%{search_value}%" or appointment.date LIKE "%{search_value}%" or appointment.time LIKE "%{search_value}%" or appointment.purpose LIKE "%{search_value}%"' search = f'AND ( name LIKE "%{search_value}%" )' # First SQL query to get the count of appointments sql_query = f""" SELECT COUNT(*) AS designation_count FROM {database_name}.designation where 1=1 {search}; """ with connection.cursor() as cursor: cursor.execute(sql_query) database_all_data = cursor.fetchall() # Check if database_all_data is not empty before accessing its elements if database_all_data: recordsTotal = database_all_data[0][0] else: recordsTotal = 0 # Second SQL query to get detailed appointment data sql_query = f""" select {database_name}.designation.* from {database_name}.designation where 1=1 {search} ORDER BY designation.id DESC LIMIT {length} OFFSET {start}; """ # Execute the second SQL query with connection.cursor() as cursor: cursor.execute(sql_query) database_all_data = cursor.fetchall() columns = [col[0] for col in cursor.description] all_coupon_dtl = [dict(zip(columns, row)) for row in database_all_data] return JsonResponse({"recordsTotal": recordsTotal, "recordsFiltered": recordsTotal, 'data': all_coupon_dtl}) @check_subscription def admin_designation_delete(request, id): username = request.session.get('admin') if not username: return redirect('admin') all_designation = get_object_or_404(designation, id=id) all_designation.delete() data = { 'status': 1, 'message': f'Designation Deleted Successfully.', } messages.success(request, f"successfully Designation Deleted!", extra_tags="success") return JsonResponse(data) @check_subscription def admin_designation_edit(request, id): 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') all_designation = get_object_or_404(designation, id=id) if request.method == "POST": all_designation.name = request.POST['designation_name'] # all_designation.allow_check_in = request.POST['allow_check_in'] all_designation.updated_at = date_time() all_designation.save() messages.success(request, "Designation successfully updated.", extra_tags="success") return redirect('admin_designation') return render(request, 'dashboard/admin_dashboard/admin_designation/all_designation_edit.html', {'all_designation': all_designation,'function_name': 'Admin','username':username}) @csrf_exempt def admin_designation_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'}) decoded_file = csv_file.read().decode('utf-8').splitlines() reader = csv.DictReader(decoded_file) fieldnames = reader.fieldnames + ['Error'] if reader.fieldnames else [] required_columns = {'Designation_Name', } 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()} designation_name = row.get('Designation_Name') # Check for duplicate Designation if designation.objects.filter(name__iexact=designation_name).exists(): row['Error'] = f"Duplicate Designation: {designation_name}" failed_rows.append(row) continue # Create the Designation designation.objects.create( name=designation_name, 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) # Highlight error column 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") 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} designations imported successfully', 'count': success_count, 'failed_rows_excel': excel_base64, 'failed_rows_filename': 'failed_uploads.xlsx' }) return JsonResponse({ 'status': 'success', 'message': f'{success_count} designations 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'})