import logging from django.shortcuts import render from human_resource.utils import calculate_basic_salary logger = logging.getLogger(__name__) from system_administration.utils import check_if_system_administrator, checkIfStaffCanEdit, convertTo24HRFormat, create_notifications, get_available_leave_days, get_staff_finance_data, get_staff_profile_data, send_email_with_attachment, send_error_messages, send_general_mail from access_control.permissions import user_has_permission from system_administration.email_settings import parse_staff_date from .models import * from .serializers import * from rest_framework.permissions import IsAuthenticated from rest_framework.authentication import TokenAuthentication from rest_framework.decorators import api_view, authentication_classes, permission_classes from rest_framework.response import Response from django.contrib.auth.models import User from django.contrib.auth import authenticate, login, logout from django.shortcuts import get_object_or_404 from datetime import datetime, timedelta, date import calendar from dateutil.relativedelta import relativedelta import json from django.http import JsonResponse from system_administration.models import * import pytz # Create your views here. MONTH_NAME_TO_NUMBER = { "january": 1, "february": 2, "march": 3, "april": 4, "may": 5, "june": 6, "july": 7, "august": 8, "september": 9, "october": 10, "november": 11, "december": 12, } def _parse_summary_month(raw_month): key = str(raw_month or "").strip().lower() if key.isdigit(): month_num = int(key) if 1 <= month_num <= 12: return month_num return MONTH_NAME_TO_NUMBER.get(key) def _staff_ref_map(staff_profile): if staff_profile is None: return {"staff_id": "", "staff_name": "", "staff_position": ""} return { "staff_id": str(staff_profile.id), "staff_name": f"{staff_profile.first_name} {staff_profile.last_name}".strip(), "staff_position": ( staff_profile.staff_position.position_title if staff_profile.staff_position is not None else "" ), } def _serialize_payroll_sheet_list_item(payroll_sheet, company_profile, target_timezone, date_format): return { "payroll_sheet_id": str(payroll_sheet.id), "payroll_sheet_title": payroll_sheet.payroll_sheet_title, "payroll_sheet_number": payroll_sheet.payroll_sheet_number, "payroll_sheet_description": payroll_sheet.payroll_sheet_description, "payroll_sheet_for_the_month_of": payroll_sheet.payroll_sheet_for_the_month_of, "payroll_sheet_for_the_year": payroll_sheet.payroll_sheet_for_the_year, "branch_name": payroll_sheet.company_branch.branch_name if payroll_sheet.company_branch is not None else "", "company_branch_id": str(payroll_sheet.company_branch.id) if payroll_sheet.company_branch is not None else "", "payroll_sheet_value": payroll_sheet.payroll_sheet_value, "payroll_sheet_total_net_pay_value": payroll_sheet.payroll_sheet_total_net_pay_value, "payroll_sheet_total_bonus_value": payroll_sheet.payroll_sheet_total_bonus_value, "payroll_sheet_total_deduction_value": payroll_sheet.payroll_sheet_total_deduction_value, "payroll_sheet_total_commission_value": payroll_sheet.payroll_sheet_total_commission_value, "payroll_sheet_approved_by_system_administrator": ( "true" if payroll_sheet.payroll_sheet_approved_by_system_administrator else "false" ), "is_terminal_dues": "true" if payroll_sheet.is_terminal_dues else "false", "currency": company_profile.company_preferred_currency, "payroll_sheet_approved_by_finance": "true" if payroll_sheet.payroll_sheet_approved_by_finance else "false", "payroll_sheet_payment_settled": "true" if payroll_sheet.payroll_sheet_payment_settled else "false", "created_by": _staff_ref_map(payroll_sheet.created_by), "last_updated_by": _staff_ref_map(payroll_sheet.last_updated_by), "created_on": ( datetime.strftime(payroll_sheet.created_on.astimezone(target_timezone), date_format) if payroll_sheet.created_on is not None else "" ), "last_updated_on": ( datetime.strftime(payroll_sheet.last_updated_on.astimezone(target_timezone), date_format) if payroll_sheet.last_updated_on is not None else "" ), "payroll_instances_list": [], } def _leave_overlaps_month(leave, month_start, month_end): if leave.leave_start_date is None or leave.leave_end_date is None: return False return leave.leave_start_date <= month_end and leave.leave_end_date >= month_start def _build_staff_leave_summary_map(staff_leave, target_timezone, date_format): staff_leave_map = {} staff_profile = staff_leave.staff_profile staff_leave_map["staff_id"] = str(staff_profile.id) if staff_profile is not None else "" staff_leave_map["staff_number"] = staff_profile.staff_number if staff_profile is not None else "" staff_leave_map["first_name"] = staff_profile.first_name if staff_profile is not None else "" staff_leave_map["last_name"] = staff_profile.last_name if staff_profile is not None else "" staff_leave_map["staff_position"] = staff_profile.staff_position.position_title if (staff_profile is not None and staff_profile.staff_position is not None) else "" staff_leave_map["company_department"] = staff_profile.company_department.department_name if (staff_profile is not None and staff_profile.company_department is not None) else "" staff_leave_map["company_department_id"] = str(staff_profile.company_department.id) if (staff_profile is not None and staff_profile.company_department is not None) else "" staff_leave_map["company_branch_name"] = staff_profile.company_branch.branch_name if (staff_profile is not None and staff_profile.company_branch is not None) else "" staff_leave_map["leave_id"] = str(staff_leave.id) staff_leave_map["leave_type"] = staff_leave.leave_type staff_leave_map["leave_description"] = staff_leave.leave_description staff_leave_map["leave_start_date"] = datetime.strftime( staff_leave.leave_start_date, '%d/%m/%Y') if staff_leave.leave_start_date is not None else "" staff_leave_map["leave_end_date"] = datetime.strftime( staff_leave.leave_end_date, '%d/%m/%Y') if staff_leave.leave_end_date is not None else "" staff_leave_map["number_of_leave_days"] = staff_leave.number_of_leave_days staff_leave_map["leave_department_approval"] = staff_leave.leave_department_approval staff_leave_map["leave_hr_approval"] = staff_leave.leave_hr_approval staff_leave_map["leave_status"] = staff_leave.leave_status creator_map = {} creator_map["staff_id"] = str( staff_leave.staff_reliever.id) if staff_leave.staff_reliever is not None else "" creator_map["staff_name"] = f'{staff_leave.staff_reliever.first_name} {staff_leave.staff_reliever.last_name}' if staff_leave.staff_reliever is not None else "" creator_map["staff_position"] = staff_leave.staff_reliever.staff_position.position_title if (staff_leave.staff_reliever is not None and staff_leave.staff_reliever.staff_position is not None) else "" staff_leave_map["staff_reliever"] = creator_map staff_leave_map["created_on"] = datetime.strftime( staff_leave.created_on.astimezone(target_timezone), date_format) if staff_leave.created_on is not None else "" staff_leave_map["last_updated_on"] = datetime.strftime( staff_leave.last_updated_on.astimezone(target_timezone), date_format) if staff_leave.last_updated_on is not None else "" if staff_profile is not None: available_leave_days = get_available_leave_days(staff_profile.id) staff_leave_map["available_leave_days"] = str(available_leave_days) if available_leave_days > 0 else "0" else: staff_leave_map["available_leave_days"] = "0" return staff_leave_map @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def overall_human_resource_dashboard(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] active_user = request.user payload = {} company_profile_map = {} company_departments_list = [] # deductions_list = [] # bonuses_list = [] work_shifts_list = [] company_departments_list = [] company_branches_list = [] company_profile_map = {} staff_positions_list = [] current_date = datetime.now().date() try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.dashboard.view"): company_profile_map["company_name"] = company_profile.company_name company_profile_map["company_postal_address"] = company_profile.company_postal_address company_profile_map["company_country_location"] = company_profile.company_country_location company_profile_map["company_phone"] = company_profile.company_phone active_staff_profile_data = get_staff_profile_data(active_user) company_departments = company_profile.company_departments.all().order_by('-id') for department in company_departments: if department.recycle_bin == False: department_map = {} department_map["department_id"] = str(department.id) department_map["department_name"] = department.department_name department_map["department_description"] = department.department_description department_map["created_on"] = datetime.strftime( department.created_on.astimezone(target_timezone), date_format) department_map["last_updated_on"] = datetime.strftime( department.last_updated_on.astimezone(target_timezone), date_format) company_departments_list.append(department_map) company_branches = company_profile.company_branches.all().order_by('-id') for branch in company_branches: if branch.recycle_bin == False and branch.branch_active == True: branch_map = {} branch_map["branch_id"] = str(branch.id) branch_map["branch_name"] = branch.branch_name branch_map["branch_description"] = branch.branch_description branch_map["branch_county_location"] = branch.branch_county_location branch_map["branch_phone"] = branch.branch_phone branch_map["main_branch"] = "true" if branch.main_branch == True else "false" branch_map["created_on"] = datetime.strftime( branch.created_on.astimezone(target_timezone), date_format) branch_map["last_updated_on"] = datetime.strftime( branch.last_updated_on.astimezone(target_timezone), date_format) company_branches_list.append(branch_map) staff_deduction_schemes = Deduction.objects.filter( recycle_bin=False).count() staff_bonus_schemes = Bonus.objects.filter( recycle_bin=False).count() staff_positions = company_profile.company_staff_positions.filter( recycle_bin=False).order_by('-id') for position in staff_positions: staff_position_map = {} staff_position_map["staff_position_id"] = str(position.id) staff_position_map["position_title"] = position.position_title staff_position_map["position_description"] = position.position_description staff_position_map["number_of_staff"] = str(position.company_positions_staff.filter( recycle_bin=False, is_profile_active=True).count()) staff_positions_list.append(staff_position_map) work_shifts = company_profile.company_work_shifts.filter( recycle_bin=False).order_by('-id') for work_shift in work_shifts: work_shift_map = {} work_shift_map["work_shift_id"] = str(work_shift.id) work_shift_map["shift_name"] = work_shift.shift_name work_shift_map["shift_hours_start"] = work_shift.shift_hours_start work_shift_map["shift_hours_end"] = work_shift.shift_hours_end work_shift_map["shift_description"] = work_shift.shift_description working_days = work_shift.workday_shifts.all().order_by('-id') working_days_list = [] for working_day in working_days: if working_day.recycle_bin == False: working_day_map = {} working_day_map["working_day_id"] = str( working_day.id) working_day_map["day_of_week_identifier"] = working_day.day_of_week_identifier # working_day_map["working_day_description"] = working_day.working_day_description working_day_map["created_on"] = datetime.strftime( working_day.created_on.astimezone(target_timezone), date_format) if working_day.created_on is not None else "" working_day_map["last_updated_on"] = datetime.strftime( working_day.last_updated_on.astimezone(target_timezone), date_format) if working_day.last_updated_on is not None else "" working_days_list.append(working_day_map) work_shift_map["working_days"] = working_days_list work_shifts_list.append(work_shift_map) payload["work_shifts_list"] = work_shifts_list payload["staff_positions_list"] = staff_positions_list payload["staff_deduction_schemes"] = str(staff_deduction_schemes) payload["staff_bonus_schemes"] = str(staff_bonus_schemes) payload["active_staff_profile_data"] = active_staff_profile_data payload["company_branches_list"] = company_branches_list payload["company_departments_list"] = company_departments_list payload["company_profile_map"] = company_profile_map return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def human_resource_staff_profiles(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] list_only = str(request.data.get("list_only", "false")).lower() == "true" active_user = request.user payload = {} staff_positions_list = [] company_branches_list = [] company_departments_list = [] current_date = datetime.now().date() try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.staff_profile.view"): active_staff_profile_data = get_staff_profile_data(active_user) staff_positions = company_profile.company_staff_positions.all().order_by('-id') for position in staff_positions: if position.recycle_bin == False: staff_position_map = {} staff_position_map["staff_position_id"] = str(position.id) staff_position_map["position_title"] = position.position_title staff_position_map["position_description"] = position.position_description staff_position_map["salary"] = position.salary staff_position_map["created_on"] = datetime.strftime( position.created_on.astimezone(target_timezone), date_format) if position.created_on is not None else "" staff_position_map["last_updated_on"] = datetime.strftime( position.last_updated_on.astimezone(target_timezone), date_format) if position.last_updated_on is not None else "" staff_positions_list.append(staff_position_map) company_branches = company_profile.company_branches.filter( recycle_bin=False).order_by('-id') company_departments = company_profile.company_departments.all().order_by('-id') for department in company_departments: if department.recycle_bin == False: department_map = {} department_map["department_id"] = str(department.id) department_map["department_name"] = department.department_name department_map["department_description"] = department.department_description department_map["created_on"] = datetime.strftime( department.created_on.astimezone(target_timezone), date_format) department_map["last_updated_on"] = datetime.strftime( department.last_updated_on.astimezone(target_timezone), date_format) company_departments_list.append(department_map) for branch in company_branches: branch_map = {} branch_map["branch_id"] = str(branch.id) branch_map["branch_name"] = branch.branch_name company_branch_staffs_list = [] branch_staffs = branch.company_branch_staffs.filter( recycle_bin=False).select_related( 'staff_position', 'company_department', 'user').order_by('-id') for staff in branch_staffs: if not list_only and staff.is_profile_active == True: if staff.type_of_employment == "contract" and staff.employment_end_date is not None: list_of_staff_ids = [] notification_exist = False all_contract_notifications = MyNotification.objects.all().filter( notification_title=f"{staff.first_name} {staff.last_name}'s Contract Expiry") for notification in all_contract_notifications: if notification.created_on.date() == current_date: notification_exist = True break subject = "" message = "" if staff.employment_end_date <= current_date: staff.is_profile_active = False staff.save() # send mail to staff requesting leave subject = f"{staff.first_name} {staff.last_name}'s Contract Expiry" message = f"The contract for {staff.first_name} {staff.last_name} of staff number {staff.staff_number} expired on {staff.employment_end_date}." # recipient_list = [f'{staff_requesting_leave.personal_email}'] # try: # send_general_mail(request, subject, # message, recipient_list) # except: # pass try: list_of_staff_ids.append( staff_profile.id) notification_title = subject notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass else: difference = relativedelta( staff.employment_end_date, current_date) # print(difference) if difference.years == 0 and difference.months < 2: if difference.weeks > 3: # send mail to staff requesting leave subject = f"{staff.first_name} {staff.last_name}'s Contract Expiry" message = f"The contract for {staff.first_name} {staff.last_name} of staff number {staff.staff_number} will expire in less than two moths time on {staff.employment_end_date}." else: subject = f"{staff.first_name} {staff.last_name}'s Contract Expiry" message = f"The contract for {staff.first_name} {staff.last_name} of staff number {staff.staff_number} will expire soon on {staff.employment_end_date}." # recipient_list = [f'{staff_requesting_leave.personal_email}'] # try: # send_general_mail(request, subject, # message, recipient_list) # except: # pass try: if notification_exist != True: list_of_staff_ids.append( staff_profile.id) notification_title = subject notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass staff_map = {} staff_map["staff_id"] = str(staff.id) staff_map["email_address"] = staff.user.username staff_map["staff_position_id"] = str( staff.staff_position.id) if staff.staff_position is not None else "" staff_map["staff_position"] = staff.staff_position.position_title if staff.staff_position is not None else "" staff_map["staff_number"] = staff.staff_number staff_map["first_name"] = staff.first_name staff_map["last_name"] = staff.last_name staff_map["date_of_birth"] = staff.date_of_birth if staff.date_of_birth is not None else '' staff_map["country_name"] = staff.country_name staff_map["identification_number"] = staff.identification_number staff_map["phone_number"] = staff.phone_number staff_map["staff_title"] = staff.staff_title staff_map["employment_start_date"] = staff.employment_start_date if staff.employment_start_date is not None else '' staff_map["employment_end_date"] = staff.employment_end_date if staff.employment_end_date is not None else '' staff_map["emergency_contact_phone"] = staff.emergency_contact_phone staff_map["company_branch_name"] = branch.branch_name staff_map["company_branch_id"] = str(branch.id) staff_map["company_department"] = staff.company_department.department_name if staff.company_department is not None else "" staff_map["company_department_id"] = str(staff.company_department.id) if staff.company_department is not None else "" staff_map["is_profile_set"] = "true" if staff.is_profile_set else "false" staff_map["is_head_of_department"] = "true" if staff.is_head_of_department else "false" staff_map["has_read_write_priviledges"] = "true" if staff.has_read_write_priviledges else "false" staff_map["is_super_admin"] = "true" if staff.is_super_admin else "false" staff_map["is_on_leave"] = "true" if staff.is_on_leave else "false" # additional staff_map["kra_pin"] = staff.kra_pin staff_map["type_of_employment"] = staff.type_of_employment staff_map["banking_institution_name"] = staff.banking_institution_name staff_map["bank_account_name"] = staff.bank_account_name staff_map["bank_account_number"] = staff.bank_account_number # added bank details staff_map["bank_branch_name"] = staff.bank_branch_name staff_map["bank_branch_code"] = staff.bank_branch_code staff_map["bank_swift_code"] = staff.bank_swift_code # end staff_map["nhif_number"] = staff.nhif_number staff_map["nhif_additional_info"] = staff.nhif_additional_info staff_map["nssf_number"] = staff.nssf_number staff_map["nssf_additional_info"] = staff.nssf_additional_info staff_map["staff_additional_info"] = staff.staff_additional_info staff_map["personal_email"] = staff.personal_email staff_map["is_profile_active"] = "true" if staff.is_profile_active == True else "false" staff_map["basic_salary"] = staff.basic_salary staff_map["currency"] = company_profile.company_preferred_currency staff_map["created_on"] = datetime.strftime( staff.created_on.astimezone(target_timezone), date_format) if staff.created_on is not None else "" staff_map["last_updated_on"] = datetime.strftime( staff.last_updated_on.astimezone(target_timezone), date_format) if staff.last_updated_on is not None else "" # end if list_only: staff_map["education_qualifications_list"] = [] staff_map["staff_training_records_list"] = [] staff_map["staff_disciplinary_records_list"] = [] else: education_qualifications_list = [] staff_training_records_list = [] staff_disciplinary_records_list = [] education_qualifications = staff.staff_education_qualifications.all().order_by('-id') for education_qualification in education_qualifications: education_qualification_map = {} education_qualification_map["education_qualification_id"] = str( education_qualification.id) education_qualification_map["qualification_title"] = education_qualification.qualification_title education_qualification_map["accredition_category"] = education_qualification.accredition_category education_qualification_map["accrediting_institution"] = education_qualification.accrediting_institution education_qualification_map["year_of_accredition"] = education_qualification.year_of_accredition education_qualification_map["created_on"] = datetime.strftime( education_qualification.created_on.astimezone(target_timezone), date_format) if education_qualification.created_on is not None else "" education_qualification_map["last_updated_on"] = datetime.strftime( education_qualification.last_updated_on.astimezone(target_timezone), date_format) if education_qualification.last_updated_on is not None else "" education_qualifications_list.append( education_qualification_map) staff_map["education_qualifications_list"] = education_qualifications_list staff_training_records = staff.staff_training_records.all().order_by('-id') for staff_training_record in staff_training_records: staff_training_record_map = {} staff_training_record_map["training_record_id"] = str( staff_training_record.id) staff_training_record_map["training_title"] = staff_training_record.training_title staff_training_record_map["training_description"] = staff_training_record.training_description staff_training_record_map["training_effective_from"] = datetime.strftime( staff_training_record.training_effective_from, date_format) if staff_training_record.training_effective_from is not None else "" staff_training_record_map["training_effective_to"] = datetime.strftime( staff_training_record.training_effective_to, date_format) if staff_training_record.training_effective_to is not None else "" staff_training_record_map["created_on"] = datetime.strftime( staff_training_record.created_on.astimezone(target_timezone), date_format) if staff_training_record.created_on is not None else "" staff_training_record_map["last_updated_on"] = datetime.strftime( staff_training_record.last_updated_on.astimezone(target_timezone), date_format) if staff_training_record.last_updated_on is not None else "" staff_training_records_list.append( staff_training_record_map) staff_map["staff_training_records_list"] = staff_training_records_list staff_disciplinary_records = staff.staff_disciplinary_records.all().order_by('-id') for staff_disciplinary_record in staff_disciplinary_records: staff_disciplinary_record_map = {} staff_disciplinary_record_map["staff_disciplinary_record_id"] = str( staff_disciplinary_record.id) staff_disciplinary_record_map[ "disciplinary_incidence_title"] = staff_disciplinary_record.disciplinary_incidence_title staff_disciplinary_record_map[ "disciplinary_incidence_description"] = staff_disciplinary_record.disciplinary_incidence_description staff_disciplinary_record_map[ "disciplinary_verdict"] = staff_disciplinary_record.disciplinary_verdict staff_disciplinary_record_map[ "disciplinary_verdict_description"] = staff_disciplinary_record.disciplinary_verdict_description staff_disciplinary_record_map[ "created_on"] = datetime.strftime( staff_disciplinary_record.created_on.astimezone(target_timezone), date_format) if staff_disciplinary_record.created_on is not None else "" staff_disciplinary_record_map[ "last_updated_on"] = datetime.strftime( staff_disciplinary_record.last_updated_on.astimezone(target_timezone), date_format) if staff_disciplinary_record.last_updated_on is not None else "" staff_disciplinary_records_list.append( staff_disciplinary_record_map) staff_map["staff_disciplinary_records_list"] = staff_disciplinary_records_list company_branch_staffs_list.append(staff_map) branch_map["branch_staff_profiles"] = company_branch_staffs_list company_branches_list.append(branch_map) payload["active_staff_profile_data"] = active_staff_profile_data payload["staff_positions_list"] = staff_positions_list payload["company_branches_list"] = company_branches_list payload["company_departments_list"] = company_departments_list # print(payload) return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def human_resource_payroll_sheets(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] page = max(int(request.data.get("page", 1) or 1), 1) page_size = min(max(int(request.data.get("page_size", 15) or 15), 1), 100) payload = {} try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.view"): payroll_sheets_qs = PayrollSheet.objects.filter( recycle_bin=False, company_branch__company_profile=company_profile, ).select_related( 'company_branch', 'created_by', 'created_by__staff_position', 'last_updated_by', 'last_updated_by__staff_position', ).order_by('-id') total_count = payroll_sheets_qs.count() offset = (page - 1) * page_size payroll_sheets_list = [ _serialize_payroll_sheet_list_item(sheet, company_profile, target_timezone, date_format) for sheet in payroll_sheets_qs[offset:offset + page_size] ] payload["payroll_sheets_list"] = payroll_sheets_list payload["payroll_sheets_total"] = total_count payload["page"] = page payload["page_size"] = page_size return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def view_single_payroll_sheet(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] active_user = request.user payload = {} current_date = datetime.now().date() try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.view"): payroll_sheet_id = int(request.data["payroll_sheet_id"]) payroll_sheet_instance = PayrollSheet.objects.get( id=payroll_sheet_id) payroll_sheet_map = {} payroll_sheet_map["payroll_sheet_id"] = str( payroll_sheet_instance.id) payroll_sheet_map["payroll_sheet_title"] = payroll_sheet_instance.payroll_sheet_title payroll_sheet_map["payroll_sheet_number"] = payroll_sheet_instance.payroll_sheet_number payroll_sheet_map["payroll_sheet_description"] = payroll_sheet_instance.payroll_sheet_description payroll_sheet_map["payroll_sheet_for_the_month_of"] = payroll_sheet_instance.payroll_sheet_for_the_month_of payroll_sheet_map["payroll_sheet_for_the_year"] = payroll_sheet_instance.payroll_sheet_for_the_year payroll_sheet_map["payroll_sheet_value"] = payroll_sheet_instance.payroll_sheet_value # added payroll_sheet_map["payroll_sheet_total_net_pay_value"] = payroll_sheet_instance.payroll_sheet_total_net_pay_value payroll_sheet_map["payroll_sheet_total_bonus_value"] = payroll_sheet_instance.payroll_sheet_total_bonus_value payroll_sheet_map["payroll_sheet_total_deduction_value"] = payroll_sheet_instance.payroll_sheet_total_deduction_value payroll_sheet_map["payroll_sheet_total_commission_value"] = payroll_sheet_instance.payroll_sheet_total_commission_value payroll_sheet_map["payroll_sheet_approved_by_system_administrator"] = "true" if payroll_sheet_instance.payroll_sheet_approved_by_system_administrator == True else "false" payroll_sheet_map["is_terminal_dues"] = "true" if payroll_sheet_instance.is_terminal_dues == True else "false" payroll_sheet_map["currency"] = company_profile.company_preferred_currency # end added payroll_sheet_map["payroll_sheet_approved_by_finance"] = "true" if payroll_sheet_instance.payroll_sheet_approved_by_finance == True else "false" payroll_sheet_map["payroll_sheet_payment_settled"] = "true" if payroll_sheet_instance.payroll_sheet_payment_settled == True else "false" staff_map = {} staff_map["staff_id"] = str( payroll_sheet_instance.created_by.id) if payroll_sheet_instance.created_by is not None else "" staff_map["staff_name"] = f'{payroll_sheet_instance.created_by.first_name} {payroll_sheet_instance.created_by.last_name}' if payroll_sheet_instance.created_by is not None else "" staff_map["staff_position"] = payroll_sheet_instance.created_by.staff_position.position_title if payroll_sheet_instance.created_by.staff_position is not None else "" payroll_sheet_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( payroll_sheet_instance.last_updated_by.id) if payroll_sheet_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{payroll_sheet_instance.last_updated_by.first_name} {payroll_sheet_instance.last_updated_by.last_name}' if payroll_sheet_instance.last_updated_by is not None else "" staff_map["staff_position"] = payroll_sheet_instance.last_updated_by.staff_position.position_title if payroll_sheet_instance.last_updated_by.staff_position is not None else "" payroll_sheet_map["last_updated_by"] = staff_map payroll_sheet_map["created_on"] = datetime.strftime( payroll_sheet_instance.created_on.astimezone(target_timezone), date_format) if payroll_sheet_instance.created_on is not None else "" payroll_sheet_map["last_updated_on"] = datetime.strftime( payroll_sheet_instance.last_updated_on.astimezone(target_timezone), date_format) if payroll_sheet_instance.last_updated_on is not None else "" payroll_instances_list = [] payroll_instances = payroll_sheet_instance.staff_payroll_items.all().order_by('-id') for payroll_instance in payroll_instances: if payroll_instance.recycle_bin == False: payroll_instance_map = {} payroll_instance_map["payroll_instance_id"] = str( payroll_instance.id) # payroll_instance_map["staff_id"] = str( # payroll_instance.staff_profile.id) payroll_instance_map["staff_financial_data"] = get_staff_finance_data( payroll_instance.staff_profile.id) deduction_instances = payroll_instance.deduction_instance.all().order_by('-id') deduction_instances_list = [] for deduction_instance in deduction_instances: if deduction_instance.recycle_bin == False: deduction_instance_map = {} deduction_instance_map["deduction_instance_id"] = str( deduction_instance.id) deduction_instance_map["deduction_id"] = str( deduction_instance.deduction.id) if deduction_instance.deduction is not None else "" deduction_instance_map["deduction_title"] = deduction_instance.deduction.deduction_title deduction_instance_map["deduction_instance_value"] = deduction_instance.deduction_instance_value staff_map = {} staff_map["staff_id"] = str( deduction_instance.created_by.id) if deduction_instance.created_by is not None else "" staff_map["staff_name"] = f'{deduction_instance.created_by.first_name} {deduction_instance.created_by.last_name}' if deduction_instance.created_by is not None else "" staff_map["staff_position"] = deduction_instance.created_by.staff_position.position_title if deduction_instance.created_by.staff_position is not None else "" deduction_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( deduction_instance.last_updated_by.id) if deduction_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{deduction_instance.last_updated_by.first_name} {deduction_instance.last_updated_by.last_name}' if deduction_instance.last_updated_by is not None else "" staff_map["staff_position"] = deduction_instance.last_updated_by.staff_position.position_title if deduction_instance.last_updated_by.staff_position is not None else "" deduction_instance_map["last_updated_by"] = staff_map deduction_instance_map["created_on"] = datetime.strftime( deduction_instance.created_on.astimezone(target_timezone), date_format) if deduction_instance.created_on is not None else "" deduction_instance_map["last_updated_on"] = datetime.strftime( deduction_instance.last_updated_on.astimezone(target_timezone), date_format) if deduction_instance.last_updated_on is not None else "" deduction_instances_list.append( deduction_instance_map) payroll_instance_map["deduction_instances_list"] = deduction_instances_list payroll_instance_map["gross_salary"] = payroll_instance.gross_salary bonus_instances = payroll_instance.bonus_instance.all().order_by('-id') bonus_instances_list = [] for bonus_instance in bonus_instances: if bonus_instance.recycle_bin == False: bonus_instance_map = {} bonus_instance_map["bonus_instance_id"] = str( bonus_instance.id) bonus_instance_map["bonus_id"] = str( bonus_instance.bonus.id) if bonus_instance.bonus is not None else "" bonus_instance_map["bonus_instance_value"] = bonus_instance.bonus_instance_value bonus_instance_map["bonus_title"] = bonus_instance.bonus.bonus_title staff_map = {} staff_map["staff_id"] = str( bonus_instance.created_by.id) if bonus_instance.created_by is not None else "" staff_map["staff_name"] = f'{bonus_instance.created_by.first_name} {bonus_instance.created_by.last_name}' if bonus_instance.created_by is not None else "" staff_map["staff_position"] = bonus_instance.created_by.staff_position.position_title if bonus_instance.created_by.staff_position is not None else "" bonus_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( bonus_instance.last_updated_by.id) if bonus_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{bonus_instance.last_updated_by.first_name} {bonus_instance.last_updated_by.last_name}' if bonus_instance.last_updated_by is not None else "" staff_map["staff_position"] = bonus_instance.last_updated_by.staff_position.position_title if bonus_instance.last_updated_by.staff_position is not None else "" bonus_instance_map["last_updated_by"] = staff_map bonus_instance_map["created_on"] = datetime.strftime( bonus_instance.created_on.astimezone(target_timezone), date_format) if deduction_instance.created_on is not None else "" bonus_instance_map["last_updated_on"] = datetime.strftime( bonus_instance.last_updated_on.astimezone(target_timezone), date_format) if bonus_instance.last_updated_on is not None else "" bonus_instances_list.append( bonus_instance_map) payroll_instance_map["bonus_instances_list"] = bonus_instances_list payroll_instance_map["net_salary"] = payroll_instance.net_salary # payroll_instance_commissions = payroll_instance.staff_payroll_instance_commissions.all() payroll_instance_commissions_list = [] for commission_sheet_instance in payroll_instance_commissions: if commission_sheet_instance.recycle_bin == False: commission_sheet_instance_map = {} commission_sheet_instance_map["commission_sheet_instance_id"] = str( commission_sheet_instance.id) commission_sheet_instance_map["customer_order_id"] = str( commission_sheet_instance.customer_order.id) if commission_sheet_instance.customer_order is not None else "" commission_sheet_instance_map[ "commission_value"] = commission_sheet_instance.commission_value commission_sheet_instance_map[ "commission_sheet_id"] = str(commission_sheet_instance.commission_sheet.id) if commission_sheet_instance.commission_sheet is not None else "" commission_sheet_instance_map[ "commission_sheet_number"] = commission_sheet_instance.commission_sheet.commission_sheet_number if commission_sheet_instance.commission_sheet is not None else "" commission_sheet_instance_map[ "commission_sheet_title"] = commission_sheet_instance.commission_sheet.commission_sheet_title if commission_sheet_instance.commission_sheet is not None else "" payroll_instance_commissions_list.append( commission_sheet_instance_map) payroll_instance_map["payroll_instance_commissions_list"] = payroll_instance_commissions_list payroll_instance_map["commissions_total"] = payroll_instance.commissions_total payroll_instance_map["is_prorated"] = "true" if payroll_instance.is_prorated == True else "false" payroll_instance_map["pro_rate_factor"] = payroll_instance.pro_rate_factor # staff_map = {} staff_map["staff_id"] = str( payroll_instance.created_by.id) if payroll_instance.created_by is not None else "" staff_map["staff_name"] = f'{payroll_instance.created_by.first_name} {payroll_instance.created_by.last_name}' if payroll_instance.created_by is not None else "" staff_map["staff_position"] = payroll_instance.created_by.staff_position.position_title if payroll_instance.created_by.staff_position is not None else "" payroll_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( payroll_instance.last_updated_by.id) if payroll_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{payroll_instance.last_updated_by.first_name} {payroll_instance.last_updated_by.last_name}' if payroll_instance.last_updated_by is not None else "" staff_map["staff_position"] = payroll_instance.last_updated_by.staff_position.position_title if payroll_instance.last_updated_by.staff_position is not None else "" payroll_instance_map["last_updated_by"] = staff_map payroll_instance_map["created_on"] = datetime.strftime( payroll_instance.created_on.astimezone(target_timezone), date_format) if payroll_instance.created_on is not None else "" payroll_instance_map["last_updated_on"] = datetime.strftime( payroll_instance.last_updated_on.astimezone(target_timezone), date_format) if payroll_instance.last_updated_on is not None else "" payroll_instances_list.append( payroll_instance_map) # print(payroll_instances_list) payroll_sheet_map["payroll_instances_list"] = payroll_instances_list payload["payroll_sheet_map"] = payroll_sheet_map return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_payroll_sheet_to_edit(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] payroll_sheet_id = int(request.data["payroll_sheet_id"]) active_user = request.user payload = {} current_date = datetime.now().date() company_departments_list = [] company_branches_list = [] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.edit"): company_departments = company_profile.company_departments.all().order_by('-id') for department in company_departments: if department.recycle_bin == False: department_map = {} department_map["department_id"] = str(department.id) department_map["department_name"] = department.department_name department_map["department_description"] = department.department_description department_map["created_on"] = datetime.strftime( department.created_on.astimezone(target_timezone), date_format) department_map["last_updated_on"] = datetime.strftime( department.last_updated_on.astimezone(target_timezone), date_format) company_departments_list.append(department_map) company_branches = company_profile.company_branches.all().order_by('-id') for branch in company_branches: if branch.recycle_bin == False and branch.branch_active == True: branch_map = {} branch_map["branch_id"] = str(branch.id) branch_map["branch_name"] = branch.branch_name company_branch_staffs_list = [] branch_staffs = branch.company_branch_staffs.all().order_by('-id') for staff in branch_staffs: staff_map = {} staff_map["staff_financial_data"] = get_staff_finance_data( staff.id) company_branch_staffs_list.append(staff_map) branch_map["company_branch_staffs_list"] = company_branch_staffs_list company_branches_list.append(branch_map) payroll_sheet_id = int(request.data["payroll_sheet_id"]) payroll_sheet_instance = PayrollSheet.objects.get( id=payroll_sheet_id) payroll_sheet_map = {} payroll_sheet_map["payroll_sheet_id"] = str( payroll_sheet_instance.id) payroll_sheet_map["payroll_sheet_title"] = payroll_sheet_instance.payroll_sheet_title payroll_sheet_map["payroll_sheet_number"] = payroll_sheet_instance.payroll_sheet_number payroll_sheet_map["payroll_sheet_description"] = payroll_sheet_instance.payroll_sheet_description payroll_sheet_map["payroll_sheet_for_the_month_of"] = payroll_sheet_instance.payroll_sheet_for_the_month_of payroll_sheet_map["payroll_sheet_for_the_year"] = payroll_sheet_instance.payroll_sheet_for_the_year payroll_sheet_map["payroll_sheet_value"] = payroll_sheet_instance.payroll_sheet_value # added payroll_sheet_map["payroll_sheet_total_net_pay_value"] = payroll_sheet_instance.payroll_sheet_total_net_pay_value payroll_sheet_map["payroll_sheet_total_bonus_value"] = payroll_sheet_instance.payroll_sheet_total_bonus_value payroll_sheet_map["payroll_sheet_total_deduction_value"] = payroll_sheet_instance.payroll_sheet_total_deduction_value payroll_sheet_map["payroll_sheet_total_commission_value"] = payroll_sheet_instance.payroll_sheet_total_commission_value payroll_sheet_map["payroll_sheet_approved_by_system_administrator"] = "true" if payroll_sheet_instance.payroll_sheet_approved_by_system_administrator == True else "false" payroll_sheet_map["is_terminal_dues"] = "true" if payroll_sheet_instance.is_terminal_dues == True else "false" payroll_sheet_map["currency"] = company_profile.company_preferred_currency # end added payroll_sheet_map["payroll_sheet_approved_by_finance"] = "true" if payroll_sheet_instance.payroll_sheet_approved_by_finance == True else "false" payroll_sheet_map["payroll_sheet_payment_settled"] = "true" if payroll_sheet_instance.payroll_sheet_payment_settled == True else "false" staff_map = {} staff_map["staff_id"] = str( payroll_sheet_instance.created_by.id) if payroll_sheet_instance.created_by is not None else "" staff_map["staff_name"] = f'{payroll_sheet_instance.created_by.first_name} {payroll_sheet_instance.created_by.last_name}' if payroll_sheet_instance.created_by is not None else "" staff_map["staff_position"] = payroll_sheet_instance.created_by.staff_position.position_title if payroll_sheet_instance.created_by.staff_position is not None else "" payroll_sheet_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( payroll_sheet_instance.last_updated_by.id) if payroll_sheet_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{payroll_sheet_instance.last_updated_by.first_name} {payroll_sheet_instance.last_updated_by.last_name}' if payroll_sheet_instance.last_updated_by is not None else "" staff_map["staff_position"] = payroll_sheet_instance.last_updated_by.staff_position.position_title if payroll_sheet_instance.last_updated_by.staff_position is not None else "" payroll_sheet_map["last_updated_by"] = staff_map payroll_sheet_map["created_on"] = datetime.strftime( payroll_sheet_instance.created_on.astimezone(target_timezone), date_format) if payroll_sheet_instance.created_on is not None else "" payroll_sheet_map["last_updated_on"] = datetime.strftime( payroll_sheet_instance.last_updated_on.astimezone(target_timezone), date_format) if payroll_sheet_instance.last_updated_on is not None else "" payroll_instances_list = [] payroll_instances = payroll_sheet_instance.staff_payroll_items.all().order_by('-id') for payroll_instance in payroll_instances: if payroll_instance.recycle_bin == False: payroll_instance_map = {} payroll_instance_map["payroll_instance_id"] = str( payroll_instance.id) # payroll_instance_map["staff_id"] = str( # payroll_instance.staff_profile.id) payroll_instance_map["staff_financial_data"] = get_staff_finance_data( payroll_instance.staff_profile.id) deduction_instances = payroll_instance.deduction_instance.all().order_by('-id') deduction_instances_list = [] for deduction_instance in deduction_instances: if deduction_instance.recycle_bin == False: deduction_instance_map = {} deduction_instance_map["deduction_instance_id"] = str( deduction_instance.id) deduction_instance_map["deduction_id"] = str( deduction_instance.deduction.id) if deduction_instance.deduction is not None else "" deduction_instance_map["deduction_title"] = deduction_instance.deduction.deduction_title deduction_instance_map["deduction_instance_value"] = deduction_instance.deduction_instance_value staff_map = {} staff_map["staff_id"] = str( deduction_instance.created_by.id) if deduction_instance.created_by is not None else "" staff_map["staff_name"] = f'{deduction_instance.created_by.first_name} {deduction_instance.created_by.last_name}' if deduction_instance.created_by is not None else "" staff_map["staff_position"] = deduction_instance.created_by.staff_position.position_title if deduction_instance.created_by.staff_position is not None else "" deduction_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( deduction_instance.last_updated_by.id) if deduction_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{deduction_instance.last_updated_by.first_name} {deduction_instance.last_updated_by.last_name}' if deduction_instance.last_updated_by is not None else "" staff_map["staff_position"] = deduction_instance.last_updated_by.staff_position.position_title if deduction_instance.last_updated_by.staff_position is not None else "" deduction_instance_map["last_updated_by"] = staff_map deduction_instance_map["created_on"] = datetime.strftime( deduction_instance.created_on.astimezone(target_timezone), date_format) if deduction_instance.created_on is not None else "" deduction_instance_map["last_updated_on"] = datetime.strftime( deduction_instance.last_updated_on.astimezone(target_timezone), date_format) if deduction_instance.last_updated_on is not None else "" deduction_instances_list.append( deduction_instance_map) payroll_instance_map["deduction_instances_list"] = deduction_instances_list payroll_instance_map["gross_salary"] = payroll_instance.gross_salary bonus_instances = payroll_instance.bonus_instance.all().order_by('-id') bonus_instances_list = [] for bonus_instance in bonus_instances: if bonus_instance.recycle_bin == False: bonus_instance_map = {} bonus_instance_map["bonus_instance_id"] = str( bonus_instance.id) bonus_instance_map["bonus_id"] = str( bonus_instance.bonus.id) if bonus_instance.bonus is not None else "" bonus_instance_map["bonus_instance_value"] = bonus_instance.bonus_instance_value bonus_instance_map["bonus_title"] = bonus_instance.bonus.bonus_title staff_map = {} staff_map["staff_id"] = str( bonus_instance.created_by.id) if bonus_instance.created_by is not None else "" staff_map["staff_name"] = f'{bonus_instance.created_by.first_name} {bonus_instance.created_by.last_name}' if bonus_instance.created_by is not None else "" staff_map["staff_position"] = bonus_instance.created_by.staff_position.position_title if bonus_instance.created_by.staff_position is not None else "" bonus_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( bonus_instance.last_updated_by.id) if bonus_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{bonus_instance.last_updated_by.first_name} {bonus_instance.last_updated_by.last_name}' if bonus_instance.last_updated_by is not None else "" staff_map["staff_position"] = bonus_instance.last_updated_by.staff_position.position_title if bonus_instance.last_updated_by.staff_position is not None else "" bonus_instance_map["last_updated_by"] = staff_map bonus_instance_map["created_on"] = datetime.strftime( bonus_instance.created_on.astimezone(target_timezone), date_format) if deduction_instance.created_on is not None else "" bonus_instance_map["last_updated_on"] = datetime.strftime( bonus_instance.last_updated_on.astimezone(target_timezone), date_format) if bonus_instance.last_updated_on is not None else "" bonus_instances_list.append( bonus_instance_map) payroll_instance_map["bonus_instances_list"] = bonus_instances_list payroll_instance_map["net_salary"] = payroll_instance.net_salary # payroll_instance_commissions = payroll_instance.staff_payroll_instance_commissions.all() payroll_instance_commissions_list = [] for commission_sheet_instance in payroll_instance_commissions: if commission_sheet_instance.recycle_bin == False: commission_sheet_instance_map = {} commission_sheet_instance_map["commission_sheet_instance_id"] = str( commission_sheet_instance.id) commission_sheet_instance_map["customer_order_id"] = str( commission_sheet_instance.customer_order.id) if commission_sheet_instance.customer_order is not None else "" commission_sheet_instance_map[ "commission_value"] = commission_sheet_instance.commission_value commission_sheet_instance_map[ "commission_sheet_id"] = str(commission_sheet_instance.commission_sheet.id) if commission_sheet_instance.commission_sheet is not None else "" commission_sheet_instance_map[ "commission_sheet_number"] = commission_sheet_instance.commission_sheet.commission_sheet_number if commission_sheet_instance.commission_sheet is not None else "" commission_sheet_instance_map[ "commission_sheet_title"] = commission_sheet_instance.commission_sheet.commission_sheet_title if commission_sheet_instance.commission_sheet is not None else "" payroll_instance_commissions_list.append( commission_sheet_instance_map) payroll_instance_map["payroll_instance_commissions_list"] = payroll_instance_commissions_list payroll_instance_map["commissions_total"] = payroll_instance.commissions_total payroll_instance_map["is_prorated"] = "true" if payroll_instance.is_prorated == True else "false" payroll_instance_map["pro_rate_factor"] = payroll_instance.pro_rate_factor # staff_map = {} staff_map["staff_id"] = str( payroll_instance.created_by.id) if payroll_instance.created_by is not None else "" staff_map["staff_name"] = f'{payroll_instance.created_by.first_name} {payroll_instance.created_by.last_name}' if payroll_instance.created_by is not None else "" staff_map["staff_position"] = payroll_instance.created_by.staff_position.position_title if payroll_instance.created_by.staff_position is not None else "" payroll_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( payroll_instance.last_updated_by.id) if payroll_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{payroll_instance.last_updated_by.first_name} {payroll_instance.last_updated_by.last_name}' if payroll_instance.last_updated_by is not None else "" staff_map["staff_position"] = payroll_instance.last_updated_by.staff_position.position_title if payroll_instance.last_updated_by.staff_position is not None else "" payroll_instance_map["last_updated_by"] = staff_map payroll_instance_map["created_on"] = datetime.strftime( payroll_instance.created_on.astimezone(target_timezone), date_format) if payroll_instance.created_on is not None else "" payroll_instance_map["last_updated_on"] = datetime.strftime( payroll_instance.last_updated_on.astimezone(target_timezone), date_format) if payroll_instance.last_updated_on is not None else "" payroll_instances_list.append( payroll_instance_map) payroll_sheet_map["payroll_instances_list"] = payroll_instances_list # print(payroll_instances_list) payload["payroll_sheet_map"] = payroll_sheet_map payload["company_branches_list"] = company_branches_list payload["company_departments_list"] = company_departments_list payload["currency"] = company_profile.company_preferred_currency return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_payroll_creation_items(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] active_user = request.user payload = {} current_date = datetime.now().date() company_departments_list = [] company_branches_list = [] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.add"): company_departments = company_profile.company_departments.all().order_by('-id') for department in company_departments: if department.recycle_bin == False: department_map = {} department_map["department_id"] = str(department.id) department_map["department_name"] = department.department_name department_map["department_description"] = department.department_description department_map["created_on"] = datetime.strftime( department.created_on.astimezone(target_timezone), date_format) department_map["last_updated_on"] = datetime.strftime( department.last_updated_on.astimezone(target_timezone), date_format) company_departments_list.append(department_map) company_branches = company_profile.company_branches.all().order_by('-id') for branch in company_branches: if branch.recycle_bin == False and branch.branch_active == True: branch_map = {} branch_map["branch_id"] = str(branch.id) branch_map["branch_name"] = branch.branch_name company_branch_staffs_list = [] branch_staffs = branch.company_branch_staffs.filter( recycle_bin=False).order_by('-id') for staff in branch_staffs: staff_map = {} staff_map["staff_financial_data"] = get_staff_finance_data( staff.id) company_branch_staffs_list.append(staff_map) branch_map["company_branch_staffs_list"] = company_branch_staffs_list company_branches_list.append(branch_map) payload["company_branches_list"] = company_branches_list payload["company_departments_list"] = company_departments_list payload["currency"] = company_profile.company_preferred_currency from human_resource.payroll_calculations import company_deduction_catalog_list from human_resource.payroll_settings import resolve_payroll_settings payload["payroll_settings"] = resolve_payroll_settings(company_profile) payload["deductions_list"] = company_deduction_catalog_list(company_profile) return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def preview_payroll_instance(request): """Recalculate statutory deductions for one staff member (Flutter formula parity).""" from human_resource.payroll_calculations import build_payroll_preview_from_staff_data company_serial_number = request.data.get("serial_number", "") staff_id = request.data.get("staff_id", "") prorate_factor = str(request.data.get("prorate_factor", "1") or "1") commissions_total = str(request.data.get("commissions_total", "0.00") or "0.00") is_prorated = str(request.data.get("is_prorated", "false")).lower() == "true" payload = {} try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not user_has_permission(staff_profile, "human_resource.payroll_sheet.add"): return Response({"message": "false", "payload": payload}, status=401) staff = StaffProfile.objects.get(id=staff_id) if staff.company_branch.company_profile_id != company_profile.id: return Response({"message": "false", "payload": payload}, status=403) staff_map = get_staff_finance_data(staff_id) preview = build_payroll_preview_from_staff_data( staff_map, prorate_factor=prorate_factor, commissions_total=commissions_total, is_prorated=is_prorated, company_profile=company_profile, ) payload["staff_financial_data"] = {**staff_map, **preview} return Response({"message": "true", "payload": payload}, status=200) except StaffProfile.DoesNotExist: return Response({"message": "false", "payload": payload}, status=404) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def human_resource_time_sheets(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] active_user = request.user payload = {} current_date = datetime.now().date() work_shifts_list = [] staff_leaves_list = [] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.dashboard.view"): active_staff_profile_data = get_staff_profile_data(active_user) work_shifts = company_profile.company_work_shifts.all().order_by('-id') for work_shift in work_shifts: if work_shift.recycle_bin == False: work_shift_map = {} work_shift_map["work_shift_id"] = str(work_shift.id) work_shift_map["shift_name"] = work_shift.shift_name work_shift_map["shift_hours_start"] = work_shift.shift_hours_start work_shift_map["shift_hours_end"] = work_shift.shift_hours_end work_shift_map["shift_description"] = work_shift.shift_description working_days = work_shift.workday_shifts.all().order_by('-id') working_days_list = [] for working_day in working_days: if working_day.recycle_bin == False: working_day_map = {} working_day_map["working_day_id"] = str( working_day.id) working_day_map["day_of_week_identifier"] = working_day.day_of_week_identifier # working_day_map["working_day_description"] = working_day.working_day_description working_day_map["created_on"] = datetime.strftime( working_day.created_on.astimezone(target_timezone), date_format) if working_day.created_on is not None else "" working_day_map["last_updated_on"] = datetime.strftime( working_day.last_updated_on.astimezone(target_timezone), date_format) if working_day.last_updated_on is not None else "" working_days_list.append(working_day_map) work_shift_map["working_days"] = working_days_list work_shift_map["created_on"] = datetime.strftime( work_shift.created_on, date_format) if work_shift.created_on is not None else "" work_shift_map["last_updated_on"] = datetime.strftime( work_shift.last_updated_on, date_format) if work_shift.last_updated_on is not None else "" work_shifts_list.append(work_shift_map) staff_leaves = StaffLeave.objects.all().order_by('-id') current_month_number = current_date.month current_year = current_date.year valid_leave_days = (current_month_number-1)*1.75 # for staff_leave in staff_leaves: if staff_leave.recycle_bin == False and staff_leave.staff_profile is not None: # compute available leave days if staff_leave.leave_hr_approval == True and staff_leave.leave_start_date.year == current_year and staff_leave.leave_department_approval == True and (staff_leave.leave_type != "sick_leave" or staff_leave.leave_type != "maternity_leave" or staff_leave.leave_type != "paternity_leave" or staff_leave.leave_type != "medical_leave" or staff_leave.leave_type != "bereavement_leave"): valid_leave_days -= int( staff_leave.number_of_leave_days) # staff_leave_map = {} staff_leave_map["staff_id"] = str( staff_leave.staff_profile.id) staff_leave_map["staff_number"] = staff_leave.staff_profile.staff_number staff_leave_map["first_name"] = staff_leave.staff_profile.first_name staff_leave_map["last_name"] = staff_leave.staff_profile.last_name staff_leave_map["staff_position"] = staff_leave.staff_profile.staff_position.position_title if staff_leave.staff_profile.staff_position is not None else "" staff_leave_map["company_department"] = staff_leave.staff_profile.company_department.department_name staff_leave_map["company_branch_name"] = staff_leave.staff_profile.company_branch.branch_name staff_leave_map["leave_id"] = str( staff_leave.id) staff_leave_map["leave_type"] = staff_leave.leave_type staff_leave_map["leave_description"] = staff_leave.leave_description staff_leave_map["leave_start_date"] = datetime.strftime( staff_leave.leave_start_date, '%d/%m/%Y') if staff_leave.leave_start_date is not None else "" staff_leave_map["leave_end_date"] = datetime.strftime( staff_leave.leave_end_date, '%d/%m/%Y') if staff_leave.leave_end_date is not None else "" # only count working days staff_leave_map["number_of_leave_days"] = staff_leave.number_of_leave_days staff_leave_map["leave_department_approval"] = staff_leave.leave_department_approval staff_leave_map["leave_hr_approval"] = staff_leave.leave_hr_approval staff_leave_map["leave_status"] = staff_leave.leave_status creator_map = {} creator_map["staff_id"] = str( staff_leave.created_by.id) if staff_leave.created_by is not None else "" creator_map["staff_name"] = f'{staff_leave.created_by.first_name} {staff_leave.created_by.last_name}' if staff_leave.created_by is not None else "" creator_map["staff_position"] = staff_leave.created_by.staff_position.position_title if staff_leave.created_by is not None else "" staff_leave_map["created_by"] = creator_map creator_map = {} creator_map["staff_id"] = str( staff_leave.last_updated_by.id) if staff_leave.last_updated_by is not None else "" creator_map["staff_name"] = f'{staff_leave.last_updated_by.first_name} {staff_leave.last_updated_by.last_name}' if staff_leave.last_updated_by is not None else "" creator_map["staff_position"] = staff_leave.last_updated_by.staff_position.position_title if staff_leave.last_updated_by is not None else "" staff_leave_map["last_updated_by"] = creator_map creator_map = {} # creator_map["staff_id"] = str( staff_leave.staff_reliever.id) if staff_leave.staff_reliever is not None else "" creator_map["staff_name"] = f'{staff_leave.staff_reliever.first_name} {staff_leave.staff_reliever.last_name}' if staff_leave.staff_reliever is not None else "" creator_map["staff_position"] = staff_leave.staff_reliever.staff_position.position_title if staff_leave.staff_reliever is not None else "" staff_leave_map["staff_reliever"] = creator_map # staff_leave_map["created_on"] = datetime.strftime( staff_leave.created_on.astimezone(target_timezone), date_format) if staff_leave.created_on is not None else "" staff_leave_map["last_updated_on"] = datetime.strftime( staff_leave.last_updated_on.astimezone(target_timezone), date_format) if staff_leave.last_updated_on is not None else "" # print(staff_leave_map) staff_leave_map["available_leave_days"] = str( valid_leave_days) if valid_leave_days > 0 else "0" staff_leaves_list.append(staff_leave_map) payload["staff_leaves_list"] = staff_leaves_list payload["active_staff_profile_data"] = active_staff_profile_data payload["work_shifts_list"] = work_shifts_list return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_staff_relievers(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] active_user = request.user payload = {} current_date = datetime.now().date() staff_reliever_list = [] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) # staff_profile = StaffProfile.objects.get(user=request.user) staff_applying_leave = StaffProfile.objects.get( id=int(staff_id)) if len(staff_id) > 0 else None all_staff_relievers = StaffProfile.objects.filter( recycle_bin=False, company_branch=staff_applying_leave.company_branch).order_by("-id") if staff_applying_leave is not None else StaffProfile.objects.filter( recycle_bin=False,).order_by("-id") for staff in all_staff_relievers: staff_map = {} staff_map["staff_id"] = str(staff.id) staff_map["staff_position_id"] = str( staff.staff_position.id) if staff.staff_position is not None else "" staff_map["staff_position"] = staff.staff_position.position_title if staff.staff_position is not None else "" staff_map["staff_number"] = staff.staff_number staff_map["first_name"] = staff.first_name staff_map["last_name"] = staff.last_name staff_map["company_department"] = staff.company_department.department_name staff_map["company_branch_name"] = staff.company_branch.branch_name staff_map["is_on_leave"] = "true" if staff.is_on_leave == True else "false" valid_leave_days = get_available_leave_days(staff.id) staff_map["available_leave_days"] = str( valid_leave_days) if valid_leave_days > 0 else "0" staff_reliever_list.append(staff_map) payload["staff_reliever_list"] = staff_reliever_list return Response({"message": "true", "payload": payload}, status=200) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def human_resource_dashboard(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data.get("serial_number") if not company_serial_number: return Response( {"detail": "serial_number is required in the POST body."}, status=400, ) active_user = request.user current_date = datetime.now().date() payload = {} staff_positions_list = [] # staff_profiles = [] deductions_list = [] bonuses_list = [] work_shifts_list = [] company_departments_list = [] company_branches_list = [] company_profile_map = {} try: try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number ) except CompanyProfile.DoesNotExist: # Dev/bootstrap fallback: if only one company exists, use it to avoid blocking # UI when frontend env serial is stale and the backend is single-tenant. company_profile = CompanyProfile.objects.filter(recycle_bin=False).order_by("-id").first() staff_profile = StaffProfile.objects.get(user=active_user) if user_has_permission(staff_profile, "human_resource.dashboard.view"): # company_profile_map["company_name"] = company_profile.company_name company_profile_map["company_postal_address"] = company_profile.company_postal_address company_profile_map["company_country_location"] = company_profile.company_country_location company_profile_map["company_phone"] = company_profile.company_phone active_staff_profile_data = get_staff_profile_data(active_user) company_departments = company_profile.company_departments.all().order_by('-id') for department in company_departments: if department.recycle_bin == False: department_map = {} department_map["department_id"] = str(department.id) department_map["department_name"] = department.department_name department_map["department_description"] = department.department_description department_map["created_on"] = datetime.strftime( department.created_on.astimezone(target_timezone), date_format) department_map["last_updated_on"] = datetime.strftime( department.last_updated_on.astimezone(target_timezone), date_format) company_departments_list.append(department_map) company_branches = company_profile.company_branches.all().order_by('-id') # criteria = {"company_profile": company_profile} if staff_profile.is_super_admin != True or check_if_system_administrator(staff_profile) != True: # criteria = {"main_branch": False} branch_cont_list = [] branch_cont_list.append(staff_profile.company_branch) company_branches = branch_cont_list # print("running") for branch in company_branches: if branch.recycle_bin == False and branch.branch_active == True: branch_map = {} branch_map["branch_id"] = str(branch.id) branch_map["branch_name"] = branch.branch_name branch_map["branch_description"] = branch.branch_description branch_map["branch_county_location"] = branch.branch_county_location branch_map["branch_phone"] = branch.branch_phone branch_map["main_branch"] = "true" if branch.main_branch == True else "false" branch_map["created_on"] = datetime.strftime( branch.created_on.astimezone(target_timezone), date_format) branch_map["last_updated_on"] = datetime.strftime( branch.last_updated_on.astimezone(target_timezone), date_format) # company_branches_list.append(branch_map) # branch users company_branch_staffs_list = [] branch_staffs = branch.company_branch_staffs.all().order_by('-id') for staff in branch_staffs: if staff.recycle_bin == False: # send contract expiry notifications if staff.is_profile_active == True: if staff.type_of_employment == "contract" and staff.employment_end_date is not None: list_of_staff_ids = [] notification_exist = False all_contract_notifications = MyNotification.objects.all().filter( notification_title=f"{staff.first_name} {staff.last_name}'s Contract Expiry") for notification in all_contract_notifications: if notification.created_on.date() == current_date: notification_exist = True break subject = "" message = "" if staff.employment_end_date <= current_date: staff.is_profile_active = False staff.save() # send mail to staff requesting leave subject = f"{staff.first_name} {staff.last_name}'s Contract Expiry" message = f"The contract for {staff.first_name} {staff.last_name} of staff number {staff.staff_number} expired on {staff.employment_end_date}." # recipient_list = [f'{staff_requesting_leave.personal_email}'] # try: # send_general_mail(request, subject, # message, recipient_list) # except: # pass try: list_of_staff_ids.append( staff_profile.id) notification_title = subject notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass else: difference = relativedelta( staff.employment_end_date, current_date) # print(difference) if difference.years == 0 and difference.months < 2: if difference.weeks > 3: # send mail to staff requesting leave subject = f"{staff.first_name} {staff.last_name}'s Contract Expiry" message = f"The contract for {staff.first_name} {staff.last_name} of staff number {staff.staff_number} will expire in less than two moths time on {staff.employment_end_date}." else: subject = f"{staff.first_name} {staff.last_name}'s Contract Expiry" message = f"The contract for {staff.first_name} {staff.last_name} of staff number {staff.staff_number} will expire soon on {staff.employment_end_date}." # recipient_list = [f'{staff_requesting_leave.personal_email}'] # try: # send_general_mail(request, subject, # message, recipient_list) # except: # pass try: if notification_exist != True: list_of_staff_ids.append( staff_profile.id) notification_title = subject notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # end time_sheets_list = [] staff_leaves_list = [] education_qualifications_list = [] staff_training_records_list = [] staff_disciplinary_records_list = [] staff_deduction_schemes_list = [] staff_bonus_schemes_list = [] staff_map = {} staff_map["staff_id"] = str(staff.id) staff_map["email_address"] = staff.user.username staff_map["staff_position"] = staff.staff_position.position_title if staff.staff_position is not None else "" staff_map["staff_position_id"] = str(staff.staff_position.id) if staff.staff_position is not None else "" staff_map["company_department_id"] = str(staff.company_department.id) if staff.company_department is not None else "" staff_map["company_branch_id"] = str(branch.id) staff_map["staff_number"] = staff.staff_number staff_map["first_name"] = staff.first_name staff_map["last_name"] = staff.last_name staff_map["date_of_birth"] = staff.date_of_birth if staff.date_of_birth is not None else '' staff_map["country_name"] = staff.country_name staff_map["identification_number"] = staff.identification_number staff_map["phone_number"] = staff.phone_number staff_map["staff_title"] = staff.staff_title staff_map["employment_start_date"] = staff.employment_start_date if staff.employment_start_date is not None else '' staff_map["employment_end_date"] = staff.employment_end_date if staff.employment_end_date is not None else '' staff_map["emergency_contact_phone"] = staff.emergency_contact_phone staff_map["company_branch_name"] = branch.branch_name staff_map["company_department"] = staff.company_department.department_name if staff.company_department is not None else "" staff_map["is_profile_set"] = "true" if staff.is_profile_set else "false" staff_map["is_head_of_department"] = "true" if staff.is_head_of_department else "false" staff_map["has_read_write_priviledges"] = "true" if staff.has_read_write_priviledges else "false" staff_map["is_super_admin"] = "true" if staff.is_super_admin else "false" staff_map["is_on_leave"] = "true" if staff.is_on_leave else "false" # additional staff_map["kra_pin"] = staff.kra_pin staff_map["type_of_employment"] = staff.type_of_employment staff_map["banking_institution_name"] = staff.banking_institution_name staff_map["bank_account_name"] = staff.bank_account_name staff_map["bank_account_number"] = staff.bank_account_number # added bank details staff_map["bank_branch_name"] = staff.bank_branch_name staff_map["bank_branch_code"] = staff.bank_branch_code staff_map["bank_swift_code"] = staff.bank_swift_code # end staff_map["nhif_number"] = staff.nhif_number staff_map["nhif_additional_info"] = staff.nhif_additional_info staff_map["nssf_number"] = staff.nssf_number staff_map["nssf_additional_info"] = staff.nssf_additional_info staff_map["staff_additional_info"] = staff.staff_additional_info staff_map["personal_email"] = staff.personal_email staff_map["is_profile_active"] = "true" if staff.is_profile_active == True else "false" staff_map["basic_salary"] = staff.basic_salary staff_map["currency"] = company_profile.company_preferred_currency # end time_sheets = staff.staff_time_sheets.all().order_by('-id') for time_sheet in time_sheets: if time_sheet.recycle_bin == False: time_sheet_map = {} time_sheet_map["time_sheet_id"] = str( time_sheet.id) time_sheet_map["check_in_time"] = datetime.strftime( time_sheet.check_in_time, date_format) if time_sheet.check_in_time is not None else "" time_sheet_map["check_out_time"] = datetime.strftime( time_sheet.check_out_time, date_format) if time_sheet.check_out_time is not None else "" time_sheet_map["created_on"] = datetime.strftime( time_sheet.created_on.astimezone(target_timezone), date_format) if time_sheet.created_on is not None else "" time_sheet_map["last_updated_on"] = datetime.strftime( time_sheet.last_updated_on.astimezone(target_timezone), date_format) if time_sheet.last_updated_on is not None else "" time_sheets_list.append(time_sheet_map) staff_map["time_sheets_list"] = time_sheets_list # adding extra data points education_qualifications = staff.staff_education_qualifications.all().order_by('-id') for education_qualification in education_qualifications: education_qualification_map = {} education_qualification_map["education_qualification_id"] = str( education_qualification.id) education_qualification_map["qualification_title"] = education_qualification.qualification_title education_qualification_map["accredition_category"] = education_qualification.accredition_category education_qualification_map["accrediting_institution"] = education_qualification.accrediting_institution education_qualification_map["year_of_accredition"] = education_qualification.year_of_accredition education_qualification_map["created_on"] = datetime.strftime( education_qualification.created_on.astimezone(target_timezone), date_format) if education_qualification.created_on is not None else "" education_qualification_map["last_updated_on"] = datetime.strftime( education_qualification.last_updated_on.astimezone(target_timezone), date_format) if education_qualification.last_updated_on is not None else "" education_qualifications_list.append( education_qualification_map) staff_map["education_qualifications_list"] = education_qualifications_list staff_training_records = staff.staff_training_records.all().order_by('-id') for staff_training_record in staff_training_records: staff_training_record_map = {} staff_training_record_map["training_record_id"] = str( staff_training_record.id) staff_training_record_map["training_title"] = staff_training_record.training_title staff_training_record_map["training_description"] = staff_training_record.training_description staff_training_record_map["training_effective_from"] = datetime.strftime( staff_training_record.training_effective_from, date_format) if staff_training_record.training_effective_from is not None else "" staff_training_record_map["training_effective_to"] = datetime.strftime( staff_training_record.training_effective_to, date_format) if staff_training_record.training_effective_to is not None else "" staff_training_record_map["created_on"] = datetime.strftime( staff_training_record.created_on.astimezone(target_timezone), date_format) if staff_training_record.created_on is not None else "" staff_training_record_map["last_updated_on"] = datetime.strftime( staff_training_record.last_updated_on.astimezone(target_timezone), date_format) if staff_training_record.last_updated_on is not None else "" staff_training_records_list.append( staff_training_record_map) staff_map["staff_training_records_list"] = staff_training_records_list staff_disciplinary_records = staff.staff_disciplinary_records.all().order_by('-id') for staff_disciplinary_record in staff_disciplinary_records: staff_disciplinary_record_map = {} staff_disciplinary_record_map["staff_disciplinary_record_id"] = str( staff_disciplinary_record.id) staff_disciplinary_record_map[ "disciplinary_incidence_title"] = staff_disciplinary_record.disciplinary_incidence_title staff_disciplinary_record_map[ "disciplinary_incidence_description"] = staff_disciplinary_record.disciplinary_incidence_description staff_disciplinary_record_map[ "disciplinary_verdict"] = staff_disciplinary_record.disciplinary_verdict staff_disciplinary_record_map[ "disciplinary_verdict_description"] = staff_disciplinary_record.disciplinary_verdict_description staff_disciplinary_record_map[ "created_on"] = datetime.strftime( staff_disciplinary_record.created_on.astimezone(target_timezone), date_format) if staff_disciplinary_record.created_on is not None else "" staff_disciplinary_record_map[ "last_updated_on"] = datetime.strftime( staff_disciplinary_record.last_updated_on.astimezone(target_timezone), date_format) if staff_disciplinary_record.last_updated_on is not None else "" staff_disciplinary_records_list.append( staff_disciplinary_record_map) staff_map["staff_disciplinary_records_list"] = staff_disciplinary_records_list # end of extra data points staff_leaves = staff.staff_leave_instances.all().order_by('-id') # # current_date = datetime.now().date() # print(current_date) current_month_number = current_date.month current_year = current_date.year valid_leave_days = (current_month_number-1)*1.75 # for staff_leave in staff_leaves: if staff_leave.recycle_bin == False: # compute available leave days if staff_leave.leave_hr_approval == True and staff_leave.leave_department_approval == True and (staff_leave.leave_type != "sick_leave" or staff_leave.leave_type != "maternity_leave" or staff_leave.leave_type != "paternity_leave" or staff_leave.leave_type != "medical_leave" or staff_leave.leave_type != "bereavement_leave"): valid_leave_days -= int( staff_leave.number_of_leave_days) # staff_leave_map = {} staff_leave_map["leave_id"] = str( staff_leave.id) staff_leave_map["leave_type"] = staff_leave.leave_type staff_leave_map["leave_description"] = staff_leave.leave_description staff_leave_map["leave_start_date"] = datetime.strftime( staff_leave.leave_start_date, '%d/%m/%Y') if staff_leave.leave_start_date is not None else "" staff_leave_map["leave_end_date"] = datetime.strftime( staff_leave.leave_end_date, '%d/%m/%Y') if staff_leave.leave_end_date is not None else "" # only count working days staff_leave_map["number_of_leave_days"] = staff_leave.number_of_leave_days staff_leave_map["leave_department_approval"] = staff_leave.leave_department_approval staff_leave_map["leave_hr_approval"] = staff_leave.leave_hr_approval staff_leave_map["leave_status"] = staff_leave.leave_status creator_map = {} creator_map["staff_id"] = str( staff_leave.created_by.id) if staff_leave.created_by is not None else "" creator_map["staff_name"] = f'{staff_leave.created_by.first_name} {staff_leave.created_by.last_name}' if staff_leave.created_by is not None else "" creator_map["staff_position"] = staff_leave.created_by.staff_position.position_title if staff_leave.created_by is not None else "" staff_leave_map["created_by"] = creator_map creator_map = {} creator_map["staff_id"] = str( staff_leave.last_updated_by.id) if staff_leave.last_updated_by is not None else "" creator_map["staff_name"] = f'{staff_leave.last_updated_by.first_name} {staff_leave.last_updated_by.last_name}' if staff_leave.last_updated_by is not None else "" creator_map["staff_position"] = staff_leave.last_updated_by.staff_position.position_title if staff_leave.last_updated_by is not None else "" staff_leave_map["last_updated_by"] = creator_map creator_map = {} # creator_map["staff_id"] = str( staff_leave.staff_reliever.id) if staff_leave.staff_reliever is not None else "" creator_map["staff_name"] = f'{staff_leave.staff_reliever.first_name} {staff_leave.staff_reliever.last_name}' if staff_leave.staff_reliever is not None else "" creator_map["staff_position"] = staff_leave.staff_reliever.staff_position.position_title if staff_leave.staff_reliever is not None else "" staff_leave_map["staff_reliever"] = creator_map # staff_leave_map["created_on"] = datetime.strftime( staff_leave.created_on.astimezone(target_timezone), date_format) if staff_leave.created_on is not None else "" staff_leave_map["last_updated_on"] = datetime.strftime( staff_leave.last_updated_on.astimezone(target_timezone), date_format) if staff_leave.last_updated_on is not None else "" # print(staff_leave_map) staff_leaves_list.append(staff_leave_map) staff_map["staff_leaves_list"] = staff_leaves_list staff_map["available_leave_days"] = str( valid_leave_days) if valid_leave_days > 0 else "0" staff_deduction_schemes = staff.staff_deduction_schemes.select_related("deduction").all().order_by('-id') for staff_deduction_scheme in staff_deduction_schemes: from human_resource.staff_deduction_utils import staff_deduction_scheme_map scheme_map = staff_deduction_scheme_map(staff_deduction_scheme, current_date) if scheme_map: staff_deduction_schemes_list.append(scheme_map) staff_map["staff_deduction_schemes_list"] = staff_deduction_schemes_list staff_bonus_schemes = staff.staff_bonus_schemes.all().order_by('-id') for staff_bonus_scheme in staff_bonus_schemes: staff_bonus_scheme_map = {} staff_bonus_scheme_map["staff_bonus_scheme_id"] = str( staff_bonus_scheme.id) staff_bonus_scheme_map["bonus_id"] = str( staff_bonus_scheme.bonus.id) staff_bonus_scheme_map["bonus_title"] = staff_bonus_scheme.bonus.bonus_title staff_bonus_scheme_map["bonus_description"] = staff_bonus_scheme.bonus.bonus_description staff_bonus_scheme_map["bonus_type"] = staff_bonus_scheme.bonus.bonus_type staff_bonus_scheme_map["bonus_amount"] = staff_bonus_scheme.bonus.bonus_amount if staff_bonus_scheme.bonus.date_effective_to >= current_date: staff_bonus_schemes_list.append( staff_bonus_scheme_map) staff_map["staff_bonus_schemes_list"] = staff_bonus_schemes_list staff_map["created_on"] = datetime.strftime( staff.created_on.astimezone(target_timezone), date_format) if staff.created_on is not None else "" staff_map["last_updated_on"] = datetime.strftime( staff.last_updated_on.astimezone(target_timezone), date_format) if staff.last_updated_on is not None else "" company_branch_staffs_list.append(staff_map) branch_map["branch_staff_profiles"] = company_branch_staffs_list payroll_sheets_list = [] payroll_sheets = branch.branch_payroll_sheets.all().order_by('-id') for payroll_sheet in payroll_sheets: if payroll_sheet.recycle_bin == False: payroll_sheet_map = {} payroll_sheet_map["payroll_sheet_id"] = str( payroll_sheet.id) payroll_sheet_map["payroll_sheet_title"] = payroll_sheet.payroll_sheet_title payroll_sheet_map["payroll_sheet_number"] = payroll_sheet.payroll_sheet_number payroll_sheet_map["payroll_sheet_description"] = payroll_sheet.payroll_sheet_description payroll_sheet_map["payroll_sheet_for_the_month_of"] = payroll_sheet.payroll_sheet_for_the_month_of payroll_sheet_map["payroll_sheet_for_the_year"] = payroll_sheet.payroll_sheet_for_the_year payroll_sheet_map["payroll_sheet_value"] = payroll_sheet.payroll_sheet_value # added payroll_sheet_map["payroll_sheet_total_net_pay_value"] = payroll_sheet.payroll_sheet_total_net_pay_value payroll_sheet_map["payroll_sheet_total_bonus_value"] = payroll_sheet.payroll_sheet_total_bonus_value payroll_sheet_map["payroll_sheet_total_deduction_value"] = payroll_sheet.payroll_sheet_total_deduction_value payroll_sheet_map["payroll_sheet_total_commission_value"] = payroll_sheet.payroll_sheet_total_commission_value payroll_sheet_map["payroll_sheet_approved_by_system_administrator"] = "true" if payroll_sheet.payroll_sheet_approved_by_system_administrator == True else "false" payroll_sheet_map["is_terminal_dues"] = "true" if payroll_sheet.is_terminal_dues == True else "false" payroll_sheet_map["currency"] = company_profile.company_preferred_currency # end added payroll_sheet_map["payroll_sheet_approved_by_finance"] = "true" if payroll_sheet.payroll_sheet_approved_by_finance == True else "false" payroll_sheet_map["payroll_sheet_payment_settled"] = "true" if payroll_sheet.payroll_sheet_payment_settled == True else "false" staff_map = {} staff_map["staff_id"] = str( payroll_sheet.created_by.id) if payroll_sheet.created_by is not None else "" staff_map["staff_name"] = f'{payroll_sheet.created_by.first_name} {payroll_sheet.created_by.last_name}' if payroll_sheet.created_by is not None else "" staff_map["staff_position"] = payroll_sheet.created_by.staff_position.position_title if payroll_sheet.created_by.staff_position is not None else "" payroll_sheet_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( payroll_sheet.last_updated_by.id) if payroll_sheet.last_updated_by is not None else "" staff_map["staff_name"] = f'{payroll_sheet.last_updated_by.first_name} {payroll_sheet.last_updated_by.last_name}' if payroll_sheet.last_updated_by is not None else "" staff_map["staff_position"] = payroll_sheet.last_updated_by.staff_position.position_title if payroll_sheet.last_updated_by.staff_position is not None else "" payroll_sheet_map["last_updated_by"] = staff_map payroll_sheet_map["created_on"] = datetime.strftime( payroll_sheet.created_on.astimezone(target_timezone), date_format) if payroll_sheet.created_on is not None else "" payroll_sheet_map["last_updated_on"] = datetime.strftime( payroll_sheet.last_updated_on.astimezone(target_timezone), date_format) if payroll_sheet.last_updated_on is not None else "" payroll_instances_list = [] payroll_instances = payroll_sheet.staff_payroll_items.all().order_by('-id') for payroll_instance in payroll_instances: if payroll_instance.recycle_bin == False: payroll_instance_map = {} payroll_instance_map["payroll_instance_id"] = str( payroll_instance.id) payroll_instance_map["staff_id"] = str( payroll_instance.staff_profile.id) deduction_instances = payroll_instance.deduction_instance.all().order_by('-id') deduction_instances_list = [] for deduction_instance in deduction_instances: if deduction_instance.recycle_bin == False: deduction_instance_map = {} deduction_instance_map["deduction_instance_id"] = str( deduction_instance.id) deduction_instance_map["deduction_id"] = str( deduction_instance.deduction.id) if deduction_instance.deduction is not None else "" deduction_instance_map["deduction_title"] = deduction_instance.deduction.deduction_title deduction_instance_map["deduction_instance_value"] = deduction_instance.deduction_instance_value staff_map = {} staff_map["staff_id"] = str( deduction_instance.created_by.id) if deduction_instance.created_by is not None else "" staff_map["staff_name"] = f'{deduction_instance.created_by.first_name} {deduction_instance.created_by.last_name}' if deduction_instance.created_by is not None else "" staff_map["staff_position"] = deduction_instance.created_by.staff_position.position_title if deduction_instance.created_by.staff_position is not None else "" deduction_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( deduction_instance.last_updated_by.id) if deduction_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{deduction_instance.last_updated_by.first_name} {deduction_instance.last_updated_by.last_name}' if deduction_instance.last_updated_by is not None else "" staff_map["staff_position"] = deduction_instance.last_updated_by.staff_position.position_title if deduction_instance.last_updated_by.staff_position is not None else "" deduction_instance_map["last_updated_by"] = staff_map deduction_instance_map["created_on"] = datetime.strftime( deduction_instance.created_on.astimezone(target_timezone), date_format) if deduction_instance.created_on is not None else "" deduction_instance_map["last_updated_on"] = datetime.strftime( deduction_instance.last_updated_on.astimezone(target_timezone), date_format) if deduction_instance.last_updated_on is not None else "" deduction_instances_list.append( deduction_instance_map) payroll_instance_map["deduction_instances_list"] = deduction_instances_list payroll_instance_map["gross_salary"] = payroll_instance.gross_salary bonus_instances = payroll_instance.bonus_instance.all().order_by('-id') bonus_instances_list = [] for bonus_instance in bonus_instances: if bonus_instance.recycle_bin == False: bonus_instance_map = {} bonus_instance_map["bonus_instance_id"] = str( bonus_instance.id) bonus_instance_map["bonus_id"] = str( bonus_instance.bonus.id) if bonus_instance.bonus is not None else "" bonus_instance_map["bonus_instance_value"] = bonus_instance.bonus_instance_value bonus_instance_map["bonus_title"] = bonus_instance.bonus.bonus_title staff_map = {} staff_map["staff_id"] = str( bonus_instance.created_by.id) if bonus_instance.created_by is not None else "" staff_map["staff_name"] = f'{bonus_instance.created_by.first_name} {bonus_instance.created_by.last_name}' if bonus_instance.created_by is not None else "" staff_map["staff_position"] = bonus_instance.created_by.staff_position.position_title if bonus_instance.created_by.staff_position is not None else "" bonus_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( bonus_instance.last_updated_by.id) if bonus_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{bonus_instance.last_updated_by.first_name} {bonus_instance.last_updated_by.last_name}' if bonus_instance.last_updated_by is not None else "" staff_map["staff_position"] = bonus_instance.last_updated_by.staff_position.position_title if bonus_instance.last_updated_by.staff_position is not None else "" bonus_instance_map["last_updated_by"] = staff_map bonus_instance_map["created_on"] = datetime.strftime( bonus_instance.created_on.astimezone(target_timezone), date_format) if deduction_instance.created_on is not None else "" bonus_instance_map["last_updated_on"] = datetime.strftime( bonus_instance.last_updated_on.astimezone(target_timezone), date_format) if bonus_instance.last_updated_on is not None else "" bonus_instances_list.append( bonus_instance_map) payroll_instance_map["bonus_instances_list"] = bonus_instances_list payroll_instance_map["net_salary"] = payroll_instance.net_salary # payroll_instance_commissions = payroll_instance.staff_payroll_instance_commissions.all() payroll_instance_commissions_list = [] for commission_sheet_instance in payroll_instance_commissions: if commission_sheet_instance.recycle_bin == False: commission_sheet_instance_map = {} commission_sheet_instance_map["commission_sheet_instance_id"] = str( commission_sheet_instance.id) commission_sheet_instance_map["customer_order_id"] = str( commission_sheet_instance.customer_order.id) if commission_sheet_instance.customer_order is not None else "" commission_sheet_instance_map[ "commission_value"] = commission_sheet_instance.commission_value commission_sheet_instance_map[ "commission_sheet_id"] = str(commission_sheet_instance.commission_sheet.id) if commission_sheet_instance.commission_sheet is not None else "" commission_sheet_instance_map[ "commission_sheet_number"] = commission_sheet_instance.commission_sheet.commission_sheet_number if commission_sheet_instance.commission_sheet is not None else "" commission_sheet_instance_map[ "commission_sheet_title"] = commission_sheet_instance.commission_sheet.commission_sheet_title if commission_sheet_instance.commission_sheet is not None else "" payroll_instance_commissions_list.append( commission_sheet_instance_map) payroll_instance_map["payroll_instance_commissions_list"] = payroll_instance_commissions_list payroll_instance_map["commissions_total"] = payroll_instance.commissions_total payroll_instance_map["is_prorated"] = "true" if payroll_instance.is_prorated == True else "false" payroll_instance_map["pro_rate_factor"] = payroll_instance.pro_rate_factor # staff_map = {} staff_map["staff_id"] = str( payroll_instance.created_by.id) if payroll_instance.created_by is not None else "" staff_map["staff_name"] = f'{payroll_instance.created_by.first_name} {payroll_instance.created_by.last_name}' if payroll_instance.created_by is not None else "" staff_map["staff_position"] = payroll_instance.created_by.staff_position.position_title if payroll_instance.created_by.staff_position is not None else "" payroll_instance_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( payroll_instance.last_updated_by.id) if payroll_instance.last_updated_by is not None else "" staff_map["staff_name"] = f'{payroll_instance.last_updated_by.first_name} {payroll_instance.last_updated_by.last_name}' if payroll_instance.last_updated_by is not None else "" staff_map["staff_position"] = payroll_instance.last_updated_by.staff_position.position_title if payroll_instance.last_updated_by.staff_position is not None else "" payroll_instance_map["last_updated_by"] = staff_map payroll_instance_map["created_on"] = datetime.strftime( payroll_instance.created_on.astimezone(target_timezone), date_format) if payroll_instance.created_on is not None else "" payroll_instance_map["last_updated_on"] = datetime.strftime( payroll_instance.last_updated_on.astimezone(target_timezone), date_format) if payroll_instance.last_updated_on is not None else "" payroll_instances_list.append( payroll_instance_map) # print(payroll_instances_list) payroll_sheet_map["payroll_instances_list"] = payroll_instances_list payroll_sheets_list.append(payroll_sheet_map) branch_map["payroll_sheets_list"] = payroll_sheets_list company_branches_list.append(branch_map) staff_positions = company_profile.company_staff_positions.all().order_by('-id') for position in staff_positions: if position.recycle_bin == False: staff_position_map = {} staff_position_map["staff_position_id"] = str(position.id) staff_position_map["position_title"] = position.position_title staff_position_map["position_description"] = position.position_description staff_position_map["salary"] = position.salary staff_position_map["created_on"] = datetime.strftime( position.created_on.astimezone(target_timezone), date_format) if position.created_on is not None else "" staff_position_map["last_updated_on"] = datetime.strftime( position.last_updated_on.astimezone(target_timezone), date_format) if position.last_updated_on is not None else "" staff_positions_list.append(staff_position_map) deductions = company_profile.company_deductions.all().order_by('-id') for deduction in deductions: if deduction.recycle_bin == False: deduction_map = {} deduction_map["deduction_id"] = str(deduction.id) deduction_map["deduction_title"] = deduction.deduction_title deduction_map["deduction_description"] = deduction.deduction_description deduction_map["deduction_type"] = deduction.deduction_type deduction_map["deduction_module"] = deduction.deduction_module deduction_map["deduction_value"] = deduction.deduction_value deduction_map["statutory_formula"] = deduction.statutory_formula or "" deduction_map["formula_config"] = deduction.formula_config or "" deduction_map["date_effective_from"] = deduction.date_effective_from deduction_map["date_effective_to"] = deduction.date_effective_to staff_map = {} staff_map["staff_id"] = str( deduction.created_by.id) if deduction.created_by is not None else "" staff_map["staff_name"] = f'{deduction.created_by.first_name} {deduction.created_by.last_name}' if deduction.created_by is not None else "" staff_map["staff_position"] = deduction.created_by.staff_position.position_title if deduction.created_by.staff_position is not None else "" deduction_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( deduction.last_updated_by.id) if deduction.last_updated_by is not None else "" staff_map["staff_name"] = f'{deduction.last_updated_by.first_name} {deduction.last_updated_by.last_name}' if deduction.last_updated_by is not None else "" staff_map["staff_position"] = deduction.last_updated_by.staff_position.position_title if deduction.last_updated_by.staff_position is not None else "" deduction_map["last_updated_by"] = staff_map deduction_map["created_on"] = datetime.strftime( deduction.created_on.astimezone(target_timezone), date_format) if deduction.created_on is not None else "" deduction_map["last_updated_on"] = datetime.strftime( deduction.last_updated_on.astimezone(target_timezone), date_format) if deduction.last_updated_on is not None else "" deductions_list.append(deduction_map) # bonuses bonuses = company_profile.company_bonuses.all().order_by('-id') for bonus in bonuses: if bonus.recycle_bin == False: bonus_map = {} bonus_map["bonus_id"] = str(bonus.id) bonus_map["bonus_title"] = bonus.bonus_title bonus_map["bonus_description"] = bonus.bonus_description bonus_map["bonus_type"] = bonus.bonus_type bonus_map["bonus_amount"] = bonus.bonus_amount bonus_map["date_effective_from"] = datetime.strftime( bonus.date_effective_from, '%d/%m/%Y') bonus_map["date_effective_to"] = datetime.strftime( bonus.date_effective_to, '%d/%m/%Y') staff_map = {} staff_map["staff_id"] = str( bonus.created_by.id) if bonus.created_by is not None else "" staff_map["staff_name"] = f'{bonus.created_by.first_name} {bonus.created_by.last_name}' if bonus.created_by is not None else "" staff_map["staff_position"] = bonus.created_by.staff_position.position_title if bonus.created_by.staff_position is not None else "" bonus_map["created_by"] = staff_map staff_map = {} staff_map["staff_id"] = str( bonus.last_updated_by.id) if bonus.last_updated_by is not None else "" staff_map["staff_name"] = f'{bonus.last_updated_by.first_name} {bonus.last_updated_by.last_name}' if bonus.last_updated_by is not None else "" staff_map["staff_position"] = bonus.last_updated_by.staff_position.position_title if bonus.last_updated_by.staff_position is not None else "" bonus_map["last_updated_by"] = staff_map bonus_map["created_on"] = datetime.strftime( bonus.created_on.astimezone(target_timezone), date_format) if bonus.created_on is not None else "" bonus_map["last_updated_on"] = datetime.strftime( bonus.last_updated_on.astimezone(target_timezone), date_format) if bonus.last_updated_on is not None else "" bonuses_list.append(bonus_map) work_shifts = company_profile.company_work_shifts.all().order_by('-id') for work_shift in work_shifts: if work_shift.recycle_bin == False: work_shift_map = {} work_shift_map["work_shift_id"] = str(work_shift.id) work_shift_map["shift_name"] = work_shift.shift_name work_shift_map["shift_hours_start"] = work_shift.shift_hours_start work_shift_map["shift_hours_end"] = work_shift.shift_hours_end work_shift_map["shift_description"] = work_shift.shift_description working_days = work_shift.workday_shifts.all().order_by('-id') working_days_list = [] for working_day in working_days: if working_day.recycle_bin == False: working_day_map = {} working_day_map["working_day_id"] = str( working_day.id) working_day_map["day_of_week_identifier"] = working_day.day_of_week_identifier # working_day_map["working_day_description"] = working_day.working_day_description working_day_map["created_on"] = datetime.strftime( working_day.created_on.astimezone(target_timezone), date_format) if working_day.created_on is not None else "" working_day_map["last_updated_on"] = datetime.strftime( working_day.last_updated_on.astimezone(target_timezone), date_format) if working_day.last_updated_on is not None else "" working_days_list.append(working_day_map) work_shift_map["working_days"] = working_days_list work_shift_map["created_on"] = datetime.strftime( work_shift.created_on, date_format) if work_shift.created_on is not None else "" work_shift_map["last_updated_on"] = datetime.strftime( work_shift.last_updated_on, date_format) if work_shift.last_updated_on is not None else "" work_shifts_list.append(work_shift_map) payload["active_staff_profile_data"] = active_staff_profile_data payload["staff_positions_list"] = staff_positions_list payload["deductions_list"] = deductions_list payload["bonuses_list"] = bonuses_list payload["work_shifts_list"] = work_shifts_list payload["company_departments_list"] = company_departments_list payload["company_branches_list"] = company_branches_list payload["company_profile_map"] = company_profile_map from human_resource.payroll_settings import resolve_payroll_settings payload["payroll_settings"] = resolve_payroll_settings(company_profile) # print(staff_positions_list) return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_staff_profile(request): date_format = '%d/%m/%Y' try: company_serial_number = request.data["serial_number"] staff_id_to_edit = request.data["staff_id_to_edit"] # print(staff_id_to_edit) # this flag checks if it is the user editing their own profile or the human resource manager that is doing the editing is_self_edit = request.data["is_self_edit"] staff_position_id = request.data["staff_position_id"] # print(staff_position_id) company_branch_id = request.data["company_branch_id"] company_department_id = request.data["company_department_id"] # print(f'{staff_id_to_edit} here') # print(f'{staff_position_id} there') # print(f'{company_branch_id} that') # print(f'{company_department_id} was') first_name = request.data["first_name"] last_name = request.data["last_name"] # email_address = request.data["email_address"] date_of_birth = parse_staff_date(request.data["date_of_birth"]) country_name = request.data["country_name"] identification_number = request.data["identification_number"] phone_number = request.data["phone_number"] personal_email = request.data["personal_email"] staff_title = request.data["staff_title"] employment_start_date = parse_staff_date(request.data["employment_start_date"]) employment_end_date = parse_staff_date(request.data["employment_end_date"]) emergency_contact_phone = request.data["emergency_contact_phone"] is_head_of_department = True if request.data["is_head_of_department"] == "true" else False has_read_write_priviledges = True if request.data[ "has_read_write_priviledges"] == "true" else False kra_pin = request.data["kra_pin"] staff_additional_info = request.data["staff_additional_info"] type_of_employment = request.data["type_of_employment"] active_user = request.user staff_profile_to_edit = StaffProfile.objects.get( user=active_user) # active_user if is_self_edit == "true": if staff_profile_to_edit.is_super_admin == True: staff_profile_serializer = StaffProfileSerializer( instance=staff_profile_to_edit, data={'staff_position': int(staff_position_id), 'company_branch': int(company_branch_id), 'company_department': int(company_department_id), 'first_name': first_name, 'last_name': last_name, 'date_of_birth': date_of_birth, 'country_name': country_name, 'identification_number': identification_number, 'phone_number': phone_number, 'staff_title': staff_title, 'employment_start_date': employment_start_date, 'employment_end_date': employment_end_date, 'emergency_contact_phone': emergency_contact_phone, 'kra_pin': kra_pin, 'staff_additional_info': staff_additional_info, 'personal_email': personal_email, 'type_of_employment': type_of_employment }) else: staff_profile_serializer = StaffProfileSerializer( instance=staff_profile_to_edit, data={'personal_email': personal_email, 'phone_number': phone_number, 'emergency_contact_phone': emergency_contact_phone, }) if staff_profile_serializer.is_valid(): staff_profile_serializer.save() if staff_profile_to_edit.is_profile_set != True: staff_profile_to_edit.is_profile_set = True staff_profile_to_edit.save() return Response({"message": "You have edited your staff profile successfully"}, status=200) else: # print(staff_profile_serializer.errors) return Response({"message": "Error editing your staff profile!"}, status=406) else: active_staff_profile = StaffProfile.objects.get(user=active_user) editPossible = checkIfStaffCanEdit( active_staff_profile, company_department_id) if editPossible == True: staff_profile_to_edit = StaffProfile.objects.get( id=int(staff_id_to_edit)) staff_profile_serializer = StaffProfileSerializer( instance=staff_profile_to_edit, data={'staff_position': int(staff_position_id), 'company_branch': int(company_branch_id), 'company_department': int(company_department_id), 'first_name': first_name, 'last_name': last_name, 'date_of_birth': date_of_birth, 'country_name': country_name, 'identification_number': identification_number, 'phone_number': phone_number, 'staff_title': staff_title, 'employment_start_date': employment_start_date, 'employment_end_date': employment_end_date, 'emergency_contact_phone': emergency_contact_phone, 'is_head_of_department': is_head_of_department, 'has_read_write_priviledges': has_read_write_priviledges, 'kra_pin': kra_pin, 'staff_additional_info': staff_additional_info, 'personal_email': personal_email, 'type_of_employment': type_of_employment }) if staff_profile_serializer.is_valid(): staff_profile_serializer.save() if staff_profile_to_edit.is_profile_set != True: staff_profile_to_edit.is_profile_set = True staff_profile_to_edit.save() return Response({"message": "Staff profile edited successfully"}, status=200) else: send_error_messages( request, f"{staff_profile_serializer.errors}") # print(staff_profile_serializer.errors) # send_error_messages(request,staff_profile_serializer.errors) return Response({"message": "Error editing staff profile!"}, status=406) else: return Response({"message": "You are unauthorised to perform this action"}, status=401) except Exception as e: message = f'{e}' send_error_messages(request, message) return Response({"message": message}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_staff_detail_mini(request): date_format = '%d/%m/%Y' active_user = request.user try: staff_id_to_edit = request.data["staff_id_to_edit"] personal_email = request.data["personal_email"] phone_number = request.data["phone_number"] emergency_contact_phone = request.data["emergency_contact_phone"] staff_profile_to_edit = StaffProfile.objects.get( user=active_user) # active_user staff_profile_serializer = StaffProfileSerializer( instance=staff_profile_to_edit, data={'personal_email': personal_email, 'phone_number': phone_number, 'emergency_contact_phone': emergency_contact_phone, }) if staff_profile_serializer.is_valid(): staff_profile_serializer.save() return Response({"message": "You have successfully edited your profile"}, status=200) else: # print(staff_profile_serializer.errors) # print(personal_email) return Response({"message": "Unable to edit your profile", }, status=406) except: # Exception as e: # print(e) return Response({"message": "Error editing your profile"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_staff_position(request): company_serial_number = request.data["serial_number"] position_title = request.data["position_title"] position_description = request.data["position_description"] salary = request.data["salary"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.staff_position.add"): serializers = staffPositionSerializer( data={'company_profile': company_profile.id, 'position_title': position_title, 'position_description': position_description, 'salary': salary}) if serializers.is_valid(): staff_position = serializers.save() try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "New Staff Position" notification_body = f"You have added a new staff position {staff_position.position_title}" create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff position created successfully", }, status=200) return Response({"message": "Unable to create staff position", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating staff position", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_staff_position(request): company_serial_number = request.data["serial_number"] staff_position_to_edit_id = request.data["staff_position_to_edit_id"] position_title = request.data["position_title"] position_description = request.data["position_description"] salary = request.data["salary"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_position_to_edit = staffPosition.objects.get( id=int(staff_position_to_edit_id)) if user_has_permission(staff_profile, "human_resource.staff_position.edit"): serializers = staffPositionSerializer(instance=staff_position_to_edit, data={ 'company_profile': company_profile.id, 'position_title': position_title, 'position_description': position_description, 'salary': salary}) if serializers.is_valid(): staff_position = serializers.save() return Response({"message": "Staff position edited successfully", }, status=200) return Response({"message": "Unable to edit staff position", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except Exception as e: print(e) return Response({"message": "Error editing staff position", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def super_hr_get_staff_profiles_to_set_up(request): payload = {} company_serial_number = request.data["serial_number"] try: # company_profile = CompanyProfile.objects.get( # company_serial_number=company_serial_number) criteria = {"is_profile_set": False} unset_staff_profiles = StaffProfile.objects.filter(**criteria) unset_staff_profiles_list = [] staff_profile = StaffProfile.objects.get(user=request.user) for staff in unset_staff_profiles: if staff.recycle_bin == False: staff_map = {} staff_map["staff_id"] = str(staff.id) staff_map["email_address"] = staff.user.username staff_map["staff_number"] = staff.staff_number # print(staff_map) unset_staff_profiles_list.append(staff_map) payload["unset_staff_profiles_list"] = unset_staff_profiles_list return Response({"message": "true", "payload": payload}, status=200) except: # Exception as e: # print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_bonus(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] bonus_title = request.data["bonus_title"] bonus_description = request.data["bonus_description"] bonus_type = request.data["bonus_type"] bonus_amount = request.data["bonus_amount"] date_effective_from = datetime.strptime( request.data["date_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_from"]) > 0 else None date_effective_to = datetime.strptime( request.data["date_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_to"]) > 0 else None try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.bonus.add"): serializers = BonusSerializer(data={'company_profile': company_profile.id, 'bonus_title': bonus_title, 'bonus_description': bonus_description, 'bonus_type': bonus_type, 'bonus_amount': bonus_amount, 'date_effective_from': date_effective_from, 'date_effective_to': date_effective_to, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if serializers.is_valid(): bonus_instance = serializers.save() try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "New Bonus Scheme Added" notification_body = f"You have added a new bonus scheme {bonus_instance.bonus_title}" create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Bonus payment item created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create bonus payment item", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating bonus payment item", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_bonus(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] bonus_id = request.data["bonus_id"] bonus_title = request.data["bonus_title"] bonus_description = request.data["bonus_description"] bonus_type = request.data["bonus_type"] bonus_amount = request.data["bonus_amount"] date_effective_from = datetime.strptime( request.data["date_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_from"]) > 0 else None date_effective_to = datetime.strptime( request.data["date_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_to"]) > 0 else None try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.bonus.edit"): bonus_instance = Bonus.objects.get(id=int(bonus_id)) serializers = BonusSerializer(instance=bonus_instance, data={'company_profile': company_profile.id, 'bonus_title': bonus_title, 'bonus_description': bonus_description, 'bonus_type': bonus_type, 'bonus_amount': bonus_amount, 'date_effective_from': date_effective_from, 'date_effective_to': date_effective_to, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if serializers.is_valid(): bonus_instance = serializers.save() return Response({"message": "Bonus payment item edited successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to edit bonus payment item", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error editing bonus payment item", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_deduction(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] deduction_title = request.data["deduction_title"] deduction_description = request.data["deduction_description"] deduction_type = request.data["deduction_type"] deduction_module = request.data["deduction_module"] deduction_value = request.data["deduction_value"] statutory_formula = str(request.data.get("statutory_formula", "") or "") formula_config = str(request.data.get("formula_config", "") or "") date_effective_from = datetime.strptime( request.data["date_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_from"]) > 0 else None date_effective_to = datetime.strptime( request.data["date_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_to"]) > 0 else None try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.deduction.add"): serializers = DeductionSerializer(data={'company_profile': company_profile.id, 'deduction_title': deduction_title, 'deduction_description': deduction_description, 'deduction_type': deduction_type, 'deduction_value': deduction_value, 'deduction_module': deduction_module, 'statutory_formula': statutory_formula, 'formula_config': formula_config, 'date_effective_from': date_effective_from, 'date_effective_to': date_effective_to, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if serializers.is_valid(): deduction_instance = serializers.save() try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "New Deduction Scheme Added" notification_body = f"You have added a new deduction scheme {deduction_instance.deduction_title}" create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Payment deduction item created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create payment deduction item", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating payment deduction item", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_deduction(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] deduction_id = request.data["deduction_id"] deduction_title = request.data["deduction_title"] deduction_description = request.data["deduction_description"] deduction_type = request.data["deduction_type"] deduction_module = request.data["deduction_module"] deduction_value = request.data["deduction_value"] statutory_formula = str(request.data.get("statutory_formula", "") or "") formula_config = str(request.data.get("formula_config", "") or "") date_effective_from = datetime.strptime( request.data["date_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_from"]) > 0 else None date_effective_to = datetime.strptime( request.data["date_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["date_effective_to"]) > 0 else None try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.deduction.edit"): deduction_instance = Deduction.objects.get(id=int(deduction_id)) serializers = DeductionSerializer(instance=deduction_instance, data={'company_profile': company_profile.id, 'deduction_title': deduction_title, 'deduction_description': deduction_description, 'deduction_type': deduction_type, 'deduction_value': deduction_value, 'deduction_module': deduction_module, 'statutory_formula': statutory_formula, 'formula_config': formula_config, 'date_effective_from': date_effective_from, 'date_effective_to': date_effective_to, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if serializers.is_valid(): deduction_instance = serializers.save() return Response({"message": "Payment deduction item edited successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to edited payment deduction item", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error editing payment deduction item", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_work_shift(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] shift_name = request.data["shift_name"] shift_hours_start = request.data["shift_hours_start"] shift_hours_end = request.data["shift_hours_end"] shift_description = request.data["shift_description"] daysList = request.data.get('daysList', []) daysList = json.loads(daysList) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.work_shift.add"): serializers = WorkShiftSerializer( data={'company_profile': company_profile.id, 'shift_name': shift_name, 'shift_hours_start': convertTo24HRFormat(shift_hours_start), 'shift_hours_end': convertTo24HRFormat(shift_hours_end), 'shift_description': shift_description}) if serializers.is_valid(): work_shift = serializers.save() for dayInstance in daysList: day_of_week_identifier = dayInstance["day_of_week_identifier"] daySerializer = WorkingDaysSerializer( data={'work_shift': work_shift.id, 'day_of_week_identifier': day_of_week_identifier}) if daySerializer.is_valid(): daySerializer.save() try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "New Work Shift Added" notification_body = f"You have added a new work shift {work_shift.shift_name}" create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # else: # print(daySerializer.errors) return Response({"message": "Work shift created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create work shift", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating work shift", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_staff_leave(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] staff_requesting_leave_id = request.data["staff_requesting_leave_id"] leave_type = request.data["leave_type"] leave_description = request.data["leave_description"] leave_start_date = datetime.strptime( request.data["leave_start_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_start_date"]) > 0 else None leave_end_date = datetime.strptime( request.data["leave_end_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_end_date"]) > 0 else None number_of_leave_days = request.data["number_of_leave_days"] reliever_id = request.data["reliever_id"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_requesting_leave = StaffProfile.objects.get( id=int(staff_requesting_leave_id)) if user_has_permission(staff_profile, "human_resource.staff_leave.add"): staff_reliever = StaffProfile.objects.get(id=int(reliever_id)) serializers = StaffLeaveSerializer( data={'company_profile': company_profile.id, 'staff_profile': staff_requesting_leave.id, 'leave_type': leave_type, 'leave_description': leave_description, 'leave_start_date': leave_start_date, 'leave_end_date': leave_end_date, 'number_of_leave_days': number_of_leave_days, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id, 'staff_reliever': staff_reliever.id}) if serializers.is_valid(): serializers.save() # send mail to staff requesting leave subject = "Leave Application" message = f"You have requested to go on leave from {leave_start_date} to {leave_end_date}. If the leave request is approved, you will be notified." recipient_list = [f'{staff_requesting_leave.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_requesting_leave.id) notification_title = "Leave Application Created" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # send mail to reliever subject = "Leave Replacement" message = f"Hello {staff_reliever.first_name} {staff_reliever.last_name}, You have been designated to temporarily fill in for {staff_requesting_leave.first_name} {staff_requesting_leave.last_name} who intends to go on leave from {leave_start_date} to {leave_end_date}. If the leave request is approved, you will be notified." recipient_list = [f'{staff_reliever.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_reliever.id) notification_title = "Leave Replacement" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # send notification to the h.o.d try: staff_department = staff_requesting_leave.company_department all_hods = StaffProfile.objects.filter( company_department=staff_department, is_head_of_department=True) list_of_staff_ids = [] for hod in all_hods: list_of_staff_ids.append(hod.id) notification_title = "Leave Application Approval Required" notification_body = f"A staff in your department with staff number {staff_requesting_leave.staff_number} has requested to go on leave. Your approval as the head of department is required as soon as possible!" create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff leave created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create staff leave", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating staff leave", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def hr_edit_staff_leave(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] leave_instance_id = request.data["leave_instance_id"] # leave_type = request.data["leave_type"] leave_edit_type = request.data["leave_edit_type"] leave_description = request.data["leave_description"] leave_start_date = datetime.strptime( request.data["leave_start_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_start_date"]) > 0 else None leave_end_date = datetime.strptime( request.data["leave_end_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_end_date"]) > 0 else None number_of_leave_days = request.data["number_of_leave_days"] # reliever_id = request.data["reliever_id"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_leave_instance = StaffLeave.objects.get( id=int(leave_instance_id)) if user_has_permission(staff_profile, "human_resource.staff_leave.edit"): # validate the leave request # get all staff leaves approved and completed for that year if leave_edit_type == "hr_approval": all_previously_approved_leaves = staff_leave_instance.staff_profile.staff_leave_instances.all( ).filter(leave_department_approval="approved", leave_hr_approval="approved") current_date = datetime.now().date() # print(current_date) current_month_number = current_date.month current_year = current_date.year valid_leave_days = (current_month_number-1)*1.75 for approved_leave in all_previously_approved_leaves: if approved_leave.leave_start_date.year == current_year and approved_leave.leave_end_date.year == current_year and (approved_leave.leave_type != "sick_leave" or approved_leave.leave_type != "maternity_leave" or approved_leave.leave_type != "paternity_leave" or approved_leave.leave_type != "medical_leave" or approved_leave.leave_type != "bereavement_leave"): valid_leave_days -= int(approved_leave.number_of_leave_days) if valid_leave_days >= int(number_of_leave_days) or (approved_leave.leave_type == "sick_leave" or approved_leave.leave_type == "maternity_leave" or approved_leave.leave_type == "paternity_leave" or approved_leave.leave_type == "medical_leave" or approved_leave.leave_type == "bereavement_leave"): # approve leave leave_serializer = StaffLeaveSerializer(instance=staff_leave_instance, data={ 'leave_start_date': leave_start_date, 'leave_end_date': leave_end_date, 'leave_description': leave_description, 'number_of_leave_days': number_of_leave_days, 'leave_hr_approval': "approved", "leave_status": "active", 'last_updated_by': staff_profile.id}) if leave_serializer.is_valid(): leave_serializer.save() # send mail to staff requesting leave subject = "Leave Application Approved" message = f"Your application to go on leave from {leave_start_date} to {leave_end_date} has been approved. You are expected to report back to work on {leave_end_date}." recipient_list = [ f'{staff_leave_instance.staff_profile.personal_email}'] try: send_general_mail( request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_profile.id) notification_title = "Leave Application Approved" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # send mail to reliever subject = "Leave Replacement Approved" message = f"Hello {staff_leave_instance.staff_reliever.first_name} {staff_leave_instance.staff_reliever.last_name}, You are hereby required to temporarily fill in for {staff_leave_instance.staff_profile.first_name} {staff_leave_instance.staff_profile.last_name} who will be on leave from {leave_start_date} to {leave_end_date}. Thank you." recipient_list = [ f'{staff_leave_instance.staff_reliever.personal_email}'] try: send_general_mail( request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_reliever.id) notification_title = "Leave Replacement Approved" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff leave approved", }, status=200) else: error_messages = [str(value[0]) for value in leave_serializer.errors.values()] return Response({"message": error_messages, }, status=406) else: return Response({"message": "The leave days that the staff is eligible for is less than the leave days applied for", }, status=406) if leave_edit_type == "hr_denial": staff_leave_instance.leave_hr_approval = "denied" staff_leave_instance.leave_status = "denied" staff_leave_instance.last_updated_by = staff_profile staff_leave_instance.save() # send mail to staff requesting leave subject = "Leave Application Denied" message = f"Your application to go on leave from {leave_start_date} to {leave_end_date} has been denied. Please contact your head of department or human resource manager." recipient_list = [ f'{staff_leave_instance.staff_profile.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_profile.id) notification_title = "Leave Application Denied" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # send mail to reliever subject = "Leave Replacement Denied" message = f"Hello {staff_leave_instance.staff_reliever.first_name} {staff_leave_instance.staff_reliever.last_name}, your designation to temporarily fill in for {staff_leave_instance.staff_profile.first_name} {staff_leave_instance.staff_profile.last_name} who requested to be on leave from {leave_start_date} to {leave_end_date} is hereby cancelled. Thank you." recipient_list = [ f'{staff_leave_instance.staff_reliever.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_reliever.id) notification_title = "Leave Replacement Denied" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff leave denied", }, status=200) if leave_edit_type == "hr_cancel": staff_leave_instance.leave_status = "cancelled" staff_leave_instance.last_updated_by = staff_profile staff_leave_instance.save() # send mail to staff requesting leave subject = "Leave Cancelled" message = f"Your application to go on leave from {leave_start_date} to {leave_end_date} has been cancelled. Please contact your head of department or human resource manager." recipient_list = [ f'{staff_leave_instance.staff_profile.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass # send mail to reliever subject = "Leave Replacement Denied" message = f"Hello {staff_leave_instance.staff_reliever.first_name} {staff_leave_instance.staff_reliever.last_name}, your designation to temporarily fill in for {staff_leave_instance.staff_profile.first_name} {staff_leave_instance.staff_profile.last_name} who requested to be on leave from {leave_start_date} to {leave_end_date} is hereby cancelled. Thank you." recipient_list = [ f'{staff_leave_instance.staff_reliever.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_reliever.id) notification_title = "Leave Replacement Denied" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff leave cancelled", }, status=200) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except Exception as e: print(e) return Response({"message": "Error updating staff leave", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def hod_edit_staff_leave(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] leave_instance_id = request.data["leave_instance_id"] leave_edit_type = request.data["leave_edit_type"] leave_start_date = datetime.strptime( request.data["leave_start_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_start_date"]) > 0 else None leave_end_date = datetime.strptime( request.data["leave_end_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_end_date"]) > 0 else None number_of_leave_days = request.data["number_of_leave_days"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_leave_instance = StaffLeave.objects.get( id=int(leave_instance_id)) # check if hod and applicant are in the same department if user_has_permission(staff_profile, "human_resource.staff_leave.approve"): if leave_edit_type == "hod_denial": staff_leave_instance.leave_department_approval = "denied" staff_leave_instance.save() try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_profile.id) notification_title = "Leave Application Denied by H.O.D" notification_body = "Sorry, your leave application has been denied by the head of your department." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # notify hr hr_department = CompanyDepartment.objects.get( department_name="human_resource_management") all_hrs = StaffProfile.objects.filter( company_department=hr_department, company_branch=staff_profile.company_branch) for hr in all_hrs: if hr.is_head_of_department == True: try: list_of_staff_ids = [] list_of_staff_ids.append( hr.id) notification_title = "Leave Application Denied by H.O.D" notification_body = f"The leave application by staff {staff_leave_instance.staff_profile.staff_number} has been denied by the respective head of department." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff leave denied", }, status=200) if leave_edit_type == "hod_approval": leave_serializer = StaffLeaveSerializer(instance=staff_leave_instance, data={ 'leave_start_date': leave_start_date, 'leave_end_date': leave_end_date, 'number_of_leave_days': number_of_leave_days, 'leave_department_approval': 'approved', 'last_updated_by': staff_profile.id}) if leave_serializer.is_valid(): leave_serializer.save() try: list_of_staff_ids = [] list_of_staff_ids.append( staff_leave_instance.staff_profile.id) notification_title = "Leave Application Approved by H.O.D" notification_body = "Your leave application has been approved by the head of your department. HR approval is now pending." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # notify hr hr_department = CompanyDepartment.objects.get( department_name="human_resource_management") all_hrs = StaffProfile.objects.filter( company_department=hr_department, company_branch=staff_profile.company_branch) for hr in all_hrs: if hr.is_head_of_department == True: try: list_of_staff_ids = [] list_of_staff_ids.append( hr.id) notification_title = "Leave Application Approved by H.O.D" notification_body = f"The leave application by staff {staff_leave_instance.staff_profile.staff_number} has been approved by the respective head of department. Human resource management approval is now pending." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Staff leave departmental approval successful", }, status=200) else: error_messages = [str(value[0]) for value in leave_serializer.errors.values()] return Response({"message": error_messages, }, status=406) except: return Response({"message": "Error updating staff leave", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_educational_qualification(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] accredited_staff_id = request.data["accredited_staff_id"] qualification_title = request.data["qualification_title"] accredition_category = request.data["accredition_category"] accrediting_institution = request.data["accrediting_institution"] year_of_accredition = request.data["year_of_accredition"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) accredited_staff = StaffProfile.objects.get( id=int(accredited_staff_id)) if user_has_permission(staff_profile, "human_resource.educational_qualification.add"): serializers = Educational_QualificationSerializer(data={'company_profile': company_profile.id, 'staff': accredited_staff.id, 'qualification_title': qualification_title, 'accredition_category': accredition_category, 'accrediting_institution': accrediting_institution, 'year_of_accredition': year_of_accredition}) if serializers.is_valid(): education_qualification = serializers.save() education_qualification_map = {} education_qualification_map["education_qualification_id"] = str( education_qualification.id) education_qualification_map["qualification_title"] = education_qualification.qualification_title education_qualification_map["accredition_category"] = education_qualification.accredition_category education_qualification_map["accrediting_institution"] = education_qualification.accrediting_institution education_qualification_map["year_of_accredition"] = education_qualification.year_of_accredition education_qualification_map["created_on"] = datetime.strftime( education_qualification.created_on, date_format) if education_qualification.created_on is not None else "" education_qualification_map["last_updated_on"] = datetime.strftime( education_qualification.last_updated_on, date_format) if education_qualification.last_updated_on is not None else "" education_qualification_map["message"] = "Education qualification created successfully" return Response(education_qualification_map, status=200) # print(serializers.errors) return Response({"message": "Unable to create education qualification", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating education qualification", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_training_record(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] trained_staff_id = request.data["trained_staff_id"] training_title = request.data["training_title"] training_description = request.data["training_description"] training_effective_from = datetime.strptime( request.data["training_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["training_effective_from"]) > 0 else None training_effective_to = datetime.strptime( request.data["training_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["training_effective_to"]) > 0 else None try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) trained_staff = StaffProfile.objects.get( id=int(trained_staff_id)) if user_has_permission(staff_profile, "human_resource.training_record.add"): serializers = Training_RecordSerializer( data={'company_profile': company_profile.id, 'staff': trained_staff.id, 'training_title': training_title, 'training_description': training_description, 'training_effective_from': training_effective_from, 'training_effective_to': training_effective_to}) if serializers.is_valid(): staff_training_record = serializers.save() staff_training_record_map = {} staff_training_record_map["training_record_id"] = str( staff_training_record.id) staff_training_record_map["training_title"] = staff_training_record.training_title staff_training_record_map["training_description"] = staff_training_record.training_description staff_training_record_map["training_effective_from"] = datetime.strftime( staff_training_record.training_effective_from, date_format) if staff_training_record.training_effective_from is not None else "" staff_training_record_map["training_effective_to"] = datetime.strftime( staff_training_record.training_effective_to, date_format) if staff_training_record.training_effective_to is not None else "" staff_training_record_map["created_on"] = datetime.strftime( staff_training_record.created_on, date_format) if staff_training_record.created_on is not None else "" staff_training_record_map["last_updated_on"] = datetime.strftime( staff_training_record.last_updated_on, date_format) if staff_training_record.last_updated_on is not None else "" staff_training_record_map["message"] = "Training record created successfully" return Response(staff_training_record_map, status=200) # print(serializers.errors) return Response({"message": "Unable to create training record", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating training record", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_disciplinary_record(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] disciplined_staff_id = request.data["disciplined_staff_id"] disciplinary_incidence_title = request.data["disciplinary_incidence_title"] disciplinary_incidence_description = request.data["disciplinary_incidence_description"] disciplinary_verdict = request.data["disciplinary_verdict"] disciplinary_verdict_description = request.data["disciplinary_verdict_description"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) disciplined_staff = StaffProfile.objects.get( id=int(disciplined_staff_id)) if user_has_permission(staff_profile, "human_resource.disciplinary_record.add"): serializers = Disciplinary_RecordSerializer( data={'company_profile': company_profile.id, 'staff': disciplined_staff.id, 'disciplinary_incidence_title': disciplinary_incidence_title, 'disciplinary_incidence_description': disciplinary_incidence_description, 'disciplinary_verdict': disciplinary_verdict, 'disciplinary_verdict_description': disciplinary_verdict_description}) if serializers.is_valid(): staff_disciplinary_record = serializers.save() staff_disciplinary_record_map = {} staff_disciplinary_record_map["staff_disciplinary_record_id"] = str( staff_disciplinary_record.id) staff_disciplinary_record_map[ "disciplinary_incidence_title"] = staff_disciplinary_record.disciplinary_incidence_title staff_disciplinary_record_map[ "disciplinary_incidence_description"] = staff_disciplinary_record.disciplinary_incidence_description staff_disciplinary_record_map[ "disciplinary_verdict"] = staff_disciplinary_record.disciplinary_verdict staff_disciplinary_record_map[ "disciplinary_verdict_description"] = staff_disciplinary_record.disciplinary_verdict_description staff_disciplinary_record_map[ "created_on"] = datetime.strftime( staff_disciplinary_record.created_on, date_format) if staff_disciplinary_record.created_on is not None else "" staff_disciplinary_record_map[ "last_updated_on"] = datetime.strftime( staff_disciplinary_record.last_updated_on, date_format) if staff_disciplinary_record.last_updated_on is not None else "" staff_disciplinary_record_map[ "message"] = "Disciplinary record created successfully" return Response(staff_disciplinary_record_map, status=200) print(serializers.errors) return Response({"message": "Unable to create disciplinary record", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating disciplinary record", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_task(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] task_title = request.data["task_title"] task_description = request.data["task_description"] task_effective_from = datetime.strptime( request.data["task_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["task_effective_from"]) > 0 else None task_effective_to = datetime.strptime( request.data["task_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["task_effective_to"]) > 0 else None staffList = request.data.get('staffList', []) staffList = json.loads(staffList) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staffIdList = [] for staffId in staffList: staffIdList.append(int(staffId["staff_id"])) if user_has_permission(staff_profile, "human_resource.task.add"): serializers = TaskSerializer( data={'company_profile': company_profile.id, 'task_title': task_title, 'task_description': task_description, 'task_effective_from': task_effective_from, 'task_effective_to': task_effective_to, 'staff': staffIdList}) if serializers.is_valid(): serializers.save() return Response({"message": "Task created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create task", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating task", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_engagement(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] engagement_title = request.data["engagement_title"] engagement_description = request.data["engagement_description"] engagement_effective_from = datetime.strptime( request.data["engagement_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["engagement_effective_from"]) > 0 else None engagement_effective_to = datetime.strptime( request.data["engagement_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["engagement_effective_to"]) > 0 else None staffList = request.data.get('staffList', []) staffList = json.loads(staffList) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staffIdList = [] for staffId in staffList: staffIdList.append(int(staffId["staff_id"])) if user_has_permission(staff_profile, "human_resource.task.add"): serializers = EngagementSerializer( data={'company_profile': company_profile.id, 'engagement_title': engagement_title, 'engagement_description': engagement_description, 'engagement_effective_from': engagement_effective_from, 'engagement_effective_to': engagement_effective_to, 'staff': staffIdList}) if serializers.is_valid(): serializers.save() return Response({"message": "Engagement created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create engagement", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating engagement", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_vacancy_record(request): date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] vacant_position_id = request.data["vacant_position_id"] vacancy_title = request.data["vacancy_title"] vacancy_description = request.data["vacancy_description"] vacancy_count = request.data["vacancy_count"] vacancy_deadline = datetime.strptime( request.data["vacancy_deadline"], date_format).strftime("%Y-%m-%d") if len(request.data["vacancy_deadline"]) > 0 else None vacant_position_requirement_list = request.data.get( 'vacant_position_requirement_list', []) vacant_position_requirement_list = json.loads( vacant_position_requirement_list) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) vacant_position = VacancyRecord.objects.get(id=int(vacant_position_id)) if user_has_permission(staff_profile, "human_resource.vacancy.add"): serializers = VacancyRecordSerializer( data={'company_profile': company_profile.id, 'vacant_position': vacant_position.id, 'vacancy_title': vacancy_title, 'vacancy_description': vacancy_description, 'vacancy_count': vacancy_count, 'vacancy_deadline': vacancy_deadline}) if serializers.is_valid(): new_vacant_position = serializers.save() for vacant_position_requirement in vacant_position_requirement_list: requirement_title = vacant_position_requirement["requirement_title"] requirement_description = vacant_position_requirement["requirement_description"] position_requirement_serializer = VacantPositionRequirementSerializer( data={'vacancy_record': new_vacant_position.id, 'requirement_title': requirement_title, 'requirement_description': requirement_description}) if position_requirement_serializer.is_valid(): position_requirement_serializer.save() return Response({"message": "Vacancy record created successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to create vacancy record", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error creating vacancy record", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_staff_financial_info(request): date_format = '%d/%m/%Y' try: company_serial_number = request.data["serial_number"] staff_id_to_edit = request.data["staff_id_to_edit"] banking_institution_name = request.data["banking_institution_name"] bank_account_name = request.data["bank_account_name"] bank_account_number = request.data["bank_account_number"] nhif_number = request.data["nhif_number"] nhif_additional_info = request.data["nhif_additional_info"] nssf_number = request.data["nssf_number"] nssf_additional_info = request.data["nssf_additional_info"] basic_salary = request.data["basic_salary"] bank_branch_name = request.data["bank_branch_name"] bank_swift_code = request.data["bank_swift_code"] bank_branch_code = request.data["bank_branch_code"] company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_profile_to_edit = StaffProfile.objects.get( id=int(staff_id_to_edit)) if user_has_permission(staff_profile, "human_resource.staff_profile.edit"): serializers = StaffProfileSerializer(instance=staff_profile_to_edit, data={'banking_institution_name': banking_institution_name, 'bank_account_name': bank_account_name, 'bank_account_number': bank_account_number, 'nhif_number': nhif_number, 'nhif_additional_info': nhif_additional_info, 'nssf_number': nssf_number, 'nssf_additional_info': nssf_additional_info, 'basic_salary': basic_salary, "bank_branch_name": bank_branch_name, "bank_swift_code": bank_swift_code, "bank_branch_code": bank_branch_code}) if serializers.is_valid(): serializers.save() return Response({"message": "Financial details updated successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to update financial details", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error updating financial details", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def save_staff_deduction_schemes(request): try: company_serial_number = request.data["serial_number"] staff_id_to_edit = request.data["staff_id_to_edit"] raw_schemes = request.data.get("staff_deduction_schemes_list", "[]") schemes = json.loads(raw_schemes) if isinstance(raw_schemes, str) else (raw_schemes or []) company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_profile_to_edit = StaffProfile.objects.get(id=int(staff_id_to_edit)) if not user_has_permission(staff_profile, "human_resource.deduction.edit"): return Response({"message": "You are unauthorised to perform this action"}, status=401) from human_resource.staff_deduction_utils import is_statutory_deduction_title kept_ids = [] for existing in staff_profile_to_edit.staff_deduction_schemes.select_related("deduction").all(): title = existing.deduction.deduction_title if existing.deduction else "" if not is_statutory_deduction_title(title): existing.delete() for item in schemes: deduction_id = str(item.get("deduction_id") or "").strip() if not deduction_id: continue deduction = Deduction.objects.get( id=int(deduction_id), company_profile=company_profile, recycle_bin=False, ) if is_statutory_deduction_title(deduction.deduction_title): continue module = str(item.get("deduction_module") or deduction.deduction_module or "fixed").strip() value = str(item.get("deduction_value") if item.get("deduction_value") is not None else deduction.deduction_value or "0").strip() scheme, _created = StaffDeductionScheme.objects.update_or_create( staff_profile=staff_profile_to_edit, deduction=deduction, defaults={ "deduction_module": module, "deduction_value": value, }, ) kept_ids.append(scheme.id) return Response({"message": "Staff deduction schemes updated successfully"}, status=200) except Deduction.DoesNotExist: return Response({"message": "One or more deductions could not be found"}, status=404) except StaffProfile.DoesNotExist: return Response({"message": "Staff profile not found"}, status=404) except Exception as e: logger.exception("save_staff_deduction_schemes failed") return Response({"message": f"Error updating staff deduction schemes: {e}"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_educational_qualification(request): date_format = '%d/%m/%Y' try: company_serial_number = request.data["serial_number"] educational_qualification_id_to_edit = request.data["educational_qualification_id_to_edit"] qualification_title = request.data["qualification_title"] accredition_category = request.data["accredition_category"] accrediting_institution = request.data["accrediting_institution"] year_of_accredition = request.data["year_of_accredition"] staff_profile = StaffProfile.objects.get(user=request.user) educational_qualification_to_edit = Educational_Qualification.objects.get( id=int(educational_qualification_id_to_edit)) company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) if user_has_permission(staff_profile, "human_resource.educational_qualification.edit"): serializers = Educational_QualificationSerializer(instance=educational_qualification_to_edit, data={'qualification_title': qualification_title, 'accredition_category': accredition_category, 'accrediting_institution': accrediting_institution, 'year_of_accredition': year_of_accredition}) if serializers.is_valid(): serializers.save() return Response({"message": "Academic record updated successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to update academic record", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error updating academic record", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_training_record(request): date_format = '%d/%m/%Y' try: company_serial_number = request.data["serial_number"] training_record_id_to_edit = request.data["staff_id_to_edit"] training_title = request.data["training_title"] training_description = request.data["training_description"] training_effective_from = datetime.strptime( request.data["training_effective_from"], date_format).strftime("%Y-%m-%d") if len(request.data["training_effective_from"]) > 0 else None training_effective_to = datetime.strptime( request.data["training_effective_to"], date_format).strftime("%Y-%m-%d") if len(request.data["training_effective_to"]) > 0 else None staff_profile = StaffProfile.objects.get(user=request.user) training_record_to_edit = Training_Record.objects.get( id=int(training_record_id_to_edit)) company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) if user_has_permission(staff_profile, "human_resource.training_record.edit"): serializers = Training_RecordSerializer(instance=training_record_to_edit, data={ 'training_title': training_title, 'training_description': training_description, 'training_effective_from': training_effective_from, 'training_effective_to': training_effective_to}) if serializers.is_valid(): serializers.save() return Response({"message": "Training record updated successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to update training record", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: return Response({"message": "Error updating training record", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_payroll_sheet(request): company_serial_number = request.data["serial_number"] company_branch_id = request.data["company_branch_id"] payroll_sheet_title = request.data["payroll_sheet_title"] payroll_sheet_description = request.data["payroll_sheet_description"] payroll_sheet_for_the_month_of = request.data["payroll_sheet_for_the_month_of"] payroll_sheet_for_the_year = request.data["payroll_sheet_for_the_year"] is_terminal_dues = request.data["is_terminal_dues"] is_terminal_dues = True if is_terminal_dues == "true" else False # calculate sheet value # grab payroll instances data payrollSheetInstancesList = request.data.get( 'payrollSheetInstancesList', []) payrollSheetInstancesList = json.loads(payrollSheetInstancesList) # print(payrollSheetInstancesList) try: # create the payroll sheet payroll_sheet_value = 0.00 payroll_sheet_total_net_pay_value = 0.00 payroll_sheet_total_bonus_value = 0.00 payroll_sheet_total_deduction_value = 0.00 payroll_sheet_total_commission_value = 0.00 company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) company_branch = CompanyBranch.objects.get(id=int(company_branch_id)) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.add"): payroll_sheet_serializer = PayrollSheetSerializer( data={'company_profile': company_profile.id, 'company_branch': company_branch.id, 'payroll_sheet_title': payroll_sheet_title, 'payroll_sheet_description': payroll_sheet_description, 'payroll_sheet_for_the_month_of': payroll_sheet_for_the_month_of, 'payroll_sheet_for_the_year': payroll_sheet_for_the_year, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id, 'is_terminal_dues': is_terminal_dues}) if payroll_sheet_serializer.is_valid(): new_payroll_sheet = payroll_sheet_serializer.save() for payrollSheetInstance in payrollSheetInstancesList: staff_on_payroll = StaffProfile.objects.get( id=int(payrollSheetInstance["staff_id"])) bonus_instance_list = payrollSheetInstance["bonus_instance_list"] payroll_sheet_bonus_instances_ids_list = [] for bonus in bonus_instance_list: bonus_id = bonus["bonus_id"] bonus_instance_value = f"{round(float(bonus['bonus_instance_value'].replace(',', '')))}" bonus_instance_serializer = BonusInstanceSerializer(data={'bonus': int( bonus_id), 'bonus_instance_value': bonus_instance_value, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if bonus_instance_serializer.is_valid(): new_bonus_instance = bonus_instance_serializer.save() payroll_sheet_bonus_instances_ids_list.append( new_bonus_instance.id) payroll_sheet_total_bonus_value += float( bonus_instance_value.replace(',', '')) # create bonus scheme if it does not exist validBonusInstance = Bonus.objects.get( id=int(bonus_id)) staff_bonus_scheme, created = StaffBonusScheme.objects.get_or_create( staff_profile=staff_on_payroll, bonus=validBonusInstance) # all bonus instances for this payroll sheet are created deduction_instance_list = payrollSheetInstance["deduction_instance_list"] # print(f'Deduction instance list: {deduction_instance_list}') payroll_sheet_deduction_instances_ids_list = [] for deduction in deduction_instance_list: deduction_id = deduction["deduction_id"] deduction_instance_value = f"{round(float(deduction['deduction_value'].replace(',','')))}" deduction_instance_serializer = DeductionInstanceSerializer( data={'deduction': int(deduction_id), 'deduction_instance_value': deduction_instance_value, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if deduction_instance_serializer.is_valid(): new_deduction_instance = deduction_instance_serializer.save() payroll_sheet_deduction_instances_ids_list.append( new_deduction_instance.id) payroll_sheet_total_deduction_value += round(float( deduction_instance_value.replace(',', ''))) # create deduction scheme if it does not exist validDeductionInstance = Deduction.objects.get( id=int(deduction_id)) staff_deduction_scheme, created = StaffDeductionScheme.objects.get_or_create( staff_profile=staff_on_payroll, deduction=validDeductionInstance) else: pass # print(deduction_instance_serializer.errors) # all deduction instances for this payroll sheet are created # creating the payroll instance pro_rate_factor = payrollSheetInstance["prorate_factor"] # print(pro_rate_factor) is_prorated = True if payrollSheetInstance["is_prorated"] == "true" else False gross_salary = payrollSheetInstance["gross_salary"] commissions_total = payrollSheetInstance["commissions_total"] payroll_sheet_total_commission_value += round( float(commissions_total.replace(',', ''))) payroll_sheet_value += round( float(gross_salary.replace(',', ''))) net_salary = f"{round(float(payrollSheetInstance['net_salary'].replace(',','')))}" payroll_sheet_total_net_pay_value += float(net_salary) created_payroll_sheet_instance_serializer = StaffPayrollInstanceSerializer(data={'staff_profile': staff_on_payroll.id, 'payroll_sheet': new_payroll_sheet.id, 'deduction_instance': payroll_sheet_deduction_instances_ids_list, 'gross_salary': gross_salary, 'bonus_instance': payroll_sheet_bonus_instances_ids_list, 'commissions_total': commissions_total, 'net_salary': net_salary, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id, 'pro_rate_factor': pro_rate_factor, 'is_prorated': is_prorated}) if created_payroll_sheet_instance_serializer.is_valid(): created_payroll_sheet_instance_serializer.save() try: from finance_and_accounting.employee_advance import schedule_advance_recoveries_for_payroll_instance schedule_advance_recoveries_for_payroll_instance( payroll_sheet=new_payroll_sheet, staff=staff_on_payroll, deduction_instance_list=deduction_instance_list, company_profile=company_profile, ) except Exception: pass else: # print(created_payroll_sheet_instance_serializer.errors) pass new_payroll_sheet.payroll_sheet_value = f'{payroll_sheet_value}' new_payroll_sheet.payroll_sheet_total_net_pay_value = f'{payroll_sheet_total_net_pay_value}' new_payroll_sheet.payroll_sheet_total_bonus_value = f'{payroll_sheet_total_bonus_value}' new_payroll_sheet.payroll_sheet_total_deduction_value = f'{payroll_sheet_total_deduction_value}' new_payroll_sheet.payroll_sheet_total_commission_value = f'{payroll_sheet_total_commission_value}' new_payroll_sheet.save() # successful creation # notify finance department try: list_of_staff_ids = [] # get finance department staff finance_dept = CompanyDepartment.objects.get( department_name="finance_and_accounting") all_finance_staff = StaffProfile.objects.filter( company_department=finance_dept) for finance_staff in all_finance_staff: list_of_staff_ids.append( finance_staff.id) notification_title = f"Staff Payroll Sheet for {new_payroll_sheet.payroll_sheet_for_the_month_of}" notification_body = f"Staff payroll sheet of number {new_payroll_sheet.payroll_sheet_number} for the month of {new_payroll_sheet.payroll_sheet_for_the_month_of} is ready. Approval by finance department is required." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": "Payroll sheet created successfully", }, status=200) else: # print(payroll_sheet_serializer.errors) return Response({"message": "Unable to create payroll sheet", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except: # Exception as e: # print(e) return Response({"message": "Error creating payroll sheet", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_payroll_sheet(request): company_serial_number = request.data["serial_number"] company_branch_id = request.data["company_branch_id"] payroll_sheet_id_to_edit = request.data["payroll_sheet_id_to_edit"] payroll_sheet_title = request.data["payroll_sheet_title"] payroll_sheet_description = request.data["payroll_sheet_description"] payroll_sheet_for_the_month_of = request.data["payroll_sheet_for_the_month_of"] payroll_sheet_for_the_year = request.data["payroll_sheet_for_the_year"] included_payroll_sheet_instances_ids = [] # new_payroll_sheet_id = None # calculate sheet value # grab payroll instances data payrollSheetInstancesList = request.data.get( 'payrollSheetInstancesList', []) payrollSheetInstancesList = json.loads(payrollSheetInstancesList) try: payroll_sheet_value = 0.00 payroll_sheet_total_net_pay_value = 0.00 payroll_sheet_total_bonus_value = 0.00 payroll_sheet_total_deduction_value = 0.00 payroll_sheet_total_commission_value = 0.00 company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) company_branch = CompanyBranch.objects.get(id=int(company_branch_id)) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.edit"): payroll_sheet_to_edit = PayrollSheet.objects.get( id=int(payroll_sheet_id_to_edit)) try: from finance_and_accounting.employee_advance import reset_scheduled_recoveries_for_payroll_sheet reset_scheduled_recoveries_for_payroll_sheet(payroll_sheet_to_edit) except Exception: pass payroll_sheet_serializer = PayrollSheetSerializer(instance=payroll_sheet_to_edit, data={'company_profile': company_profile.id, 'company_branch': company_branch.id, 'payroll_sheet_title': payroll_sheet_title, 'payroll_sheet_description': payroll_sheet_description, 'payroll_sheet_for_the_month_of': payroll_sheet_for_the_month_of, 'payroll_sheet_for_the_year': payroll_sheet_for_the_year, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if payroll_sheet_serializer.is_valid(): new_payroll_sheet = payroll_sheet_serializer.save() # print(f"{new_payroll_sheet} here") for payrollSheetInstance in payrollSheetInstancesList: # print(payrollSheetInstance) staff_on_payroll = StaffProfile.objects.get( id=int(payrollSheetInstance["staff_id"])) bonus_instance_list = payrollSheetInstance["bonus_instance_list"] payroll_instance_id = payrollSheetInstance["payroll_instance_id"] ## get_payroll_sheet_instance = None if len(payroll_instance_id) > 0: # print("running well") get_payroll_sheet_instance = StaffPayrollInstance.objects.get( id=int(payroll_instance_id)) # delete bonus instances bonus_instances_data = get_payroll_sheet_instance.bonus_instance.all() for bonus_instance_detail in bonus_instances_data: bonus_instance_detail.delete() deduction_instances_data = get_payroll_sheet_instance.deduction_instance.all() for deduction_instance_detail in deduction_instances_data: deduction_instance_detail.delete() # else: # print("not running") ## payroll_sheet_bonus_instances_ids_list = [] for bonus in bonus_instance_list: bonus_id = bonus["bonus_id"] bonus_instance_value = bonus["bonus_instance_value"] bonus_instance_serializer = BonusInstanceSerializer(data={'bonus': int( bonus_id), 'bonus_instance_value': bonus_instance_value, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if bonus_instance_serializer.is_valid(): new_bonus_instance = bonus_instance_serializer.save() payroll_sheet_bonus_instances_ids_list.append( new_bonus_instance.id) # will check to delete non_retained bonus instances payroll_sheet_total_bonus_value += float( bonus_instance_value.replace(',', '')) validBonusInstance = Bonus.objects.get( id=int(bonus_id)) staff_bonus_scheme, created = StaffBonusScheme.objects.get_or_create( staff_profile=staff_on_payroll, bonus=validBonusInstance) deduction_instance_list = payrollSheetInstance["deduction_instance_list"] payroll_sheet_deduction_instances_ids_list = [] for deduction in deduction_instance_list: deduction_id = deduction["deduction_id"] deduction_instance_value = deduction["deduction_value"] deduction_instance_serializer = DeductionInstanceSerializer( data={'deduction': int(deduction_id), 'deduction_instance_value': deduction_instance_value, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if deduction_instance_serializer.is_valid(): new_deduction_instance = deduction_instance_serializer.save() payroll_sheet_deduction_instances_ids_list.append( new_deduction_instance.id) payroll_sheet_total_deduction_value += float( deduction_instance_value.replace(',', '')) # create deduction scheme if it does not exist validDeductionInstance = Deduction.objects.get( id=int(deduction_id)) staff_deduction_scheme, created = StaffDeductionScheme.objects.get_or_create( staff_profile=staff_on_payroll, deduction=validDeductionInstance) gross_salary = payrollSheetInstance["gross_salary"] commissions_total = payrollSheetInstance["commissions_total"] payroll_sheet_total_commission_value += float( commissions_total.replace(',', '')) payroll_sheet_value += float(gross_salary.replace(',', '')) net_salary = payrollSheetInstance["net_salary"] payroll_sheet_total_net_pay_value += float( net_salary.replace(',', '')) is_prorated = True if payrollSheetInstance["is_prorated"] == "true" else False pro_rate_factor = payrollSheetInstance["pro_rate_factor"] if get_payroll_sheet_instance is not None: edited_payroll_sheet_instance_serializer = StaffPayrollInstanceSerializer(instance=get_payroll_sheet_instance, data={'deduction_instance': payroll_sheet_deduction_instances_ids_list, 'gross_salary': gross_salary, 'bonus_instance': payroll_sheet_bonus_instances_ids_list, 'commissions_total': commissions_total, 'is_prorated': is_prorated, 'pro_rate_factor': pro_rate_factor, 'net_salary': net_salary, 'last_updated_by': staff_profile.id}) if edited_payroll_sheet_instance_serializer.is_valid(): new_instance_payroll = edited_payroll_sheet_instance_serializer.save() included_payroll_sheet_instances_ids.append( new_instance_payroll.id) try: from finance_and_accounting.employee_advance import schedule_advance_recoveries_for_payroll_instance schedule_advance_recoveries_for_payroll_instance( payroll_sheet=new_payroll_sheet, staff=staff_on_payroll, deduction_instance_list=deduction_instance_list, company_profile=company_profile, ) except Exception: pass else: created_payroll_sheet_instance_serializer = StaffPayrollInstanceSerializer(data={'staff_profile': staff_on_payroll.id, 'payroll_sheet': new_payroll_sheet.id, 'deduction_instance': payroll_sheet_deduction_instances_ids_list, 'gross_salary': gross_salary, 'bonus_instance': payroll_sheet_bonus_instances_ids_list, 'commissions_total': commissions_total, 'net_salary': net_salary, 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id}) if created_payroll_sheet_instance_serializer.is_valid(): new_instance_payroll = created_payroll_sheet_instance_serializer.save() included_payroll_sheet_instances_ids.append( new_instance_payroll.id) try: from finance_and_accounting.employee_advance import schedule_advance_recoveries_for_payroll_instance schedule_advance_recoveries_for_payroll_instance( payroll_sheet=new_payroll_sheet, staff=staff_on_payroll, deduction_instance_list=deduction_instance_list, company_profile=company_profile, ) except Exception: pass # check payroll sheet instances not included new_payroll_sheet.save() # print(new_payroll_sheet) new_payroll_sheet = PayrollSheet.objects.get( id=new_payroll_sheet.id) staff_payroll_items = new_payroll_sheet.staff_payroll_items.all() for staff_payroll_item in staff_payroll_items: if staff_payroll_item.id not in included_payroll_sheet_instances_ids: staff_payroll_item.delete() new_payroll_sheet.payroll_sheet_value = f'{payroll_sheet_value}' new_payroll_sheet.payroll_sheet_total_net_pay_value = f'{payroll_sheet_total_net_pay_value}' new_payroll_sheet.payroll_sheet_total_bonus_value = f'{payroll_sheet_total_bonus_value}' new_payroll_sheet.payroll_sheet_total_deduction_value = f'{payroll_sheet_total_deduction_value}' new_payroll_sheet.payroll_sheet_total_commission_value = f'{payroll_sheet_total_commission_value}' new_payroll_sheet.save() # successful creation return Response({"message": "Payroll sheet edited successfully", }, status=200) else: # print(payroll_sheet_serializer.errors) return Response({"message": "Unable to edit payroll sheet", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except Exception as e: print(e) return Response({"message": "Error editing payroll sheet", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_my_profile_details(request): active_user = request.user try: company_serial_number = request.data["serial_number"] except KeyError: return Response( {"detail": "serial_number is required in the POST body."}, status=400, ) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number ) except CompanyProfile.DoesNotExist: # Dev/bootstrap fallback: if only one company exists, use it to avoid blocking # UI when frontend env serial is stale and the backend is single-tenant. company_profile = CompanyProfile.objects.filter(recycle_bin=False).order_by("-id").first() if company_profile is None: return Response( { "detail": "No company found for this serial number. Set VITE_COMPANY_SERIAL_NUMBER to match an existing CompanyProfile.company_serial_number in the database.", }, status=404, ) payload = {} try: active_staff_profile_data = get_staff_profile_data(active_user) # get all branch staff profiles company_branches = company_profile.company_branches.all().order_by('-id') company_branches_list = [] for branch in company_branches: if branch.recycle_bin == False and branch.branch_active == True: branch_map = {} branch_map["branch_id"] = str(branch.id) branch_map["branch_name"] = branch.branch_name company_branch_staffs_list = [] branch_staffs = branch.company_branch_staffs.all().order_by('-id') for staff in branch_staffs: if staff.recycle_bin == False: staff_map = {} staff_map["staff_id"] = str(staff.id) staff_map["email_address"] = ( staff.user.username if staff.user is not None else "" ) staff_map["staff_position"] = ( staff.staff_position.position_title if staff.staff_position is not None else "" ) staff_map["staff_number"] = staff.staff_number staff_map["first_name"] = staff.first_name staff_map["last_name"] = staff.last_name staff_map["company_branch_name"] = branch.branch_name staff_map["company_department"] = ( staff.company_department.department_name if staff.company_department is not None else "" ) company_branch_staffs_list.append(staff_map) branch_map["branch_staff_profiles"] = company_branch_staffs_list company_branches_list.append(branch_map) target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_departments_list = [] for department in company_profile.company_departments.all().order_by('-id'): if department.recycle_bin == False: department_map = {} department_map["department_id"] = str(department.id) department_map["department_name"] = department.department_name department_map["department_description"] = department.department_description department_map["created_on"] = ( datetime.strftime( department.created_on.astimezone(target_timezone), date_format ) if department.created_on is not None else "" ) department_map["last_updated_on"] = ( datetime.strftime( department.last_updated_on.astimezone(target_timezone), date_format ) if department.last_updated_on is not None else "" ) company_departments_list.append(department_map) staff_positions_list = [] for position in company_profile.company_staff_positions.all().order_by('-id'): if position.recycle_bin == False: staff_position_map = {} staff_position_map["staff_position_id"] = str(position.id) staff_position_map["position_title"] = position.position_title staff_position_map["position_description"] = position.position_description staff_position_map["salary"] = position.salary staff_position_map["created_on"] = ( datetime.strftime( position.created_on.astimezone(target_timezone), date_format ) if position.created_on is not None else "" ) staff_position_map["last_updated_on"] = ( datetime.strftime( position.last_updated_on.astimezone(target_timezone), date_format ) if position.last_updated_on is not None else "" ) staff_positions_list.append(staff_position_map) payload["company_departments_list"] = company_departments_list payload["staff_positions_list"] = staff_positions_list payload["active_staff_profile_data"] = active_staff_profile_data payload["company_branches_list"] = company_branches_list return Response({"message": "true", "payload": payload}, status=200) except Exception: logger.exception("get_my_profile_details failed") return Response( { "message": "false", "detail": "Server error while loading profile. See Django runserver log for the traceback.", }, status=500, ) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def change_password(request): date_format = '%d/%m/%Y' active_user = request.user try: company_serial_number = request.data["serial_number"] new_password = request.data["new_password"] active_user.set_password(new_password) active_user.save() return Response({"message": "Password changed successfully", }, status=200) except: return Response({"message": "Error changing password", }, status=500) # delete sections @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_bonus(request): active_user = request.user bonus_id = request.data["bonus_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.bonus.delete"): bonus_to_delete = Bonus.objects.get(id=int(bonus_id)) if bonus_to_delete.recycle_bin != True: bonus_to_delete.recycle_bin = True bonus_to_delete.save() return Response({"message": "Bonus deleted successfully", }, status=200) else: bonus_to_delete.delete() return Response({"message": "You have permanently deleted bonus item successfully", }, status=200) except: return Response({"message": "Error deleting bonus", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_deduction(request): active_user = request.user deduction_id = request.data["deduction_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.deduction.delete"): deduction_to_delete = Deduction.objects.get(id=int(deduction_id)) if deduction_to_delete.recycle_bin != True: deduction_to_delete.recycle_bin = True deduction_to_delete.save() return Response({"message": "Deduction deleted successfully", }, status=200) else: deduction_to_delete.delete() return Response({"message": "You have permanently deleted deduction item successfully", }, status=200) except: return Response({"message": "Error deleting deduction", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_payroll_sheet(request): active_user = request.user payroll_sheet_id = request.data["payroll_sheet_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payroll_sheet.delete"): payroll_sheet_to_delete = PayrollSheet.objects.get( id=int(payroll_sheet_id)) if payroll_sheet_to_delete.recycle_bin != True: payroll_sheet_to_delete.recycle_bin = True payroll_sheet_to_delete.save() return Response({"message": "Payroll sheet deleted successfully", }, status=200) else: payroll_sheet_to_delete.delete() return Response({"message": "You have permanently deleted payroll sheet successfully", }, status=200) except: return Response({"message": "Error deleting payroll_sheet", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_staff_position(request): active_user = request.user staff_position_id = request.data["staff_position_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=active_user) if user_has_permission(staff_profile, "human_resource.staff_position.delete"): staff_position_to_delete = staffPosition.objects.get( id=int(staff_position_id)) if staff_position_to_delete.recycle_bin != True: staff_position_to_delete.recycle_bin = True staff_position_to_delete.save() return Response({"message": "Staff position deleted successfully", }, status=200) else: staff_position_to_delete.delete() return Response({"message": "You have permanently deleted staff position successfully", }, status=200) except: # Exception as e: # print(e) return Response({"message": "Error deleting staff position", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_staff_profile(request): active_user = request.user staff_profile_id = request.data["staff_profile_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=active_user) if user_has_permission(staff_profile, "human_resource.staff_profile.delete"): staff_profile_to_delete = StaffProfile.objects.get( id=int(staff_profile_id)) if staff_profile_to_delete.recycle_bin != True: staff_profile_to_delete.recycle_bin = True staff_profile_to_delete.save() return Response({"message": "Staff profile deleted successfully", }, status=200) else: staff_profile_to_delete.delete() return Response({"message": "You have permanently deleted staff profile successfully", }, status=200) except: return Response({"message": "Error deleting staff profile", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_work_shift(request): active_user = request.user work_shift_id = request.data["work_shift_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=active_user) if user_has_permission(staff_profile, "human_resource.work_shift.delete"): work_shift_to_delete = WorkShift.objects.get( id=int(work_shift_id)) if work_shift_to_delete.recycle_bin != True: work_shift_to_delete.recycle_bin = True work_shift_to_delete.save() return Response({"message": "Work shift deleted successfully", }, status=200) else: work_shift_to_delete.delete() return Response({"message": "You have permanently deleted work shift successfully", }, status=200) except: return Response({"message": "Error deleting work shift", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_staff_leave(request): active_user = request.user staff_leave_id = request.data["staff_leave_id"] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=active_user) if user_has_permission(staff_profile, "human_resource.staff_leave.edit"): staff_leave_to_delete = StaffLeave.objects.get( id=int(staff_leave_id)) if staff_leave_to_delete.recycle_bin != True: staff_leave_to_delete.recycle_bin = True staff_leave_to_delete.save() return Response({"message": "Staff leave deleted successfully", }, status=200) else: staff_leave_to_delete.delete() return Response({"message": "You have permanently deleted staff leave successfully", }, status=200) except Exception as e: print(e) return Response({"message": "Error deleting staff leave", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_work_shift(request): date_format = '%d/%m/%Y' work_shift_id = request.data["work_shift_id"] company_serial_number = request.data["serial_number"] shift_name = request.data["shift_name"] shift_hours_start = request.data["shift_hours_start"] shift_hours_end = request.data["shift_hours_end"] shift_description = request.data["shift_description"] daysList = request.data.get('daysList', []) daysList = json.loads(daysList) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.work_shift.edit"): work_shift_instance = WorkShift.objects.get(id=int(work_shift_id)) serializers = WorkShiftSerializer(instance=work_shift_instance, data={'company_profile': company_profile.id, 'shift_name': shift_name, 'shift_hours_start': convertTo24HRFormat(shift_hours_start), 'shift_hours_end': convertTo24HRFormat(shift_hours_end), 'shift_description': shift_description}) if serializers.is_valid(): work_shift = serializers.save() # delete previous day instances for existingDay in work_shift_instance.workday_shifts.all(): work_shift_instance.workday_shifts.remove(existingDay) existingDay.delete() for dayInstance in daysList: day_of_week_identifier = dayInstance["day_of_week_identifier"] daySerializer = WorkingDaysSerializer( data={'work_shift': work_shift.id, 'day_of_week_identifier': day_of_week_identifier}) if daySerializer.is_valid(): daySerializer.save() # else: # print(daySerializer.errors) return Response({"message": "Work shift edited successfully", }, status=200) # print(serializers.errors) return Response({"message": "Unable to edit work shift", }, status=406) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except Exception as e: print(e) return Response({"message": "Error editing work shift", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def staff_apply_for_leave(request): payload = {} staff_leaves_list = [] try: date_format = '%d/%m/%Y' company_serial_number = request.data["serial_number"] staff_requesting_leave_id = request.data["staff_requesting_leave_id"] leave_type = request.data["leave_type"] leave_description = request.data["leave_description"] leave_start_date = datetime.strptime( request.data["leave_start_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_start_date"]) > 0 else None leave_end_date = datetime.strptime( request.data["leave_end_date"], date_format).strftime("%Y-%m-%d") if len(request.data["leave_end_date"]) > 0 else None number_of_leave_days = request.data["number_of_leave_days"] reliever_id = request.data["reliever_id"] company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) staff_requesting_leave = StaffProfile.objects.get( id=int(staff_requesting_leave_id)) if company_profile: staff_reliever = StaffProfile.objects.get(id=int(reliever_id)) serializers = StaffLeaveSerializer( data={'company_profile': company_profile.id, 'staff_profile': staff_requesting_leave.id, 'leave_type': leave_type, 'leave_description': leave_description, 'leave_start_date': leave_start_date, 'leave_end_date': leave_end_date, 'number_of_leave_days': number_of_leave_days, 'leave_department_approval': "approved" if staff_requesting_leave.is_head_of_department == True else "pending", 'created_by': staff_profile.id, 'last_updated_by': staff_profile.id, 'staff_reliever': staff_reliever.id}) if serializers.is_valid(): new_leave = serializers.save() # send mail to staff requesting leave subject = "Leave Application" message = f"You have requested to go on leave from {leave_start_date} to {leave_end_date}. If the leave request is approved, you will be notified." recipient_list = [f'{staff_requesting_leave.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_requesting_leave.id) notification_title = "Leave Application Created" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # send mail to reliever subject = "Leave Replacement" message = f"Hello {staff_reliever.first_name} {staff_reliever.last_name}, You have been designated to temporarily fill in for {staff_requesting_leave.first_name} {staff_requesting_leave.last_name} who intends to go on leave from {leave_start_date} to {leave_end_date}. If the leave request is approved, you will be notified." recipient_list = [f'{staff_reliever.personal_email}'] try: send_general_mail(request, subject, message, recipient_list) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_reliever.id) notification_title = "Leave Replacement" notification_body = message create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # send notification to the h.o.d if staff_requesting_leave.is_head_of_department != True: try: staff_department = staff_requesting_leave.company_department all_hods = StaffProfile.objects.filter( company_department=staff_department, is_head_of_department=True) list_of_staff_ids = [] for hod in all_hods: list_of_staff_ids.append(hod.id) notification_title = "Leave Application Approval Required" notification_body = f"A staff in your department with staff number {staff_requesting_leave.staff_number} has requested to go on leave. Your approval as the head of department is required as soon as possible!" create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass hr_department = CompanyDepartment.objects.get( department_name="human_resource_management") all_hrs = StaffProfile.objects.filter( company_department=hr_department, company_branch=staff_profile.company_branch) for hr in all_hrs: if hr.is_head_of_department == True: try: list_of_staff_ids = [] list_of_staff_ids.append( hr.id) notification_title = "Leave Application Created" notification_body = f"Staff {new_leave.staff_profile.staff_number} has applied to go on leave from {leave_start_date} to {leave_end_date}. Approval by the respective head of department is required before human resouce management can approve the leave." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass else: # notify hr # hr_department = CompanyDepartment.objects.get( department_name="human_resource_management") all_hrs = StaffProfile.objects.filter( company_department=hr_department, company_branch=staff_profile.company_branch) for hr in all_hrs: if hr.is_head_of_department == True: try: list_of_staff_ids = [] list_of_staff_ids.append( hr.id) notification_title = "Human Resource Leave Approval Required" notification_body = f"Staff {new_leave.staff_profile.staff_number} has applied to go on leave from {leave_start_date} to {leave_end_date}. Approval by the human resouce management is required." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # return list of staff leaves staff_leaves = staff_requesting_leave.staff_leave_instances.all().order_by("-id") for staff_leave in staff_leaves: if staff_leave.recycle_bin == False: staff_leave_map = {} staff_leave_map["leave_id"] = str( staff_leave.id) staff_leave_map["leave_type"] = staff_leave.leave_type staff_leave_map["leave_description"] = staff_leave.leave_description staff_leave_map["leave_start_date"] = datetime.strftime( staff_leave.leave_start_date, '%d/%m/%Y') if staff_leave.leave_start_date is not None else "" staff_leave_map["leave_end_date"] = datetime.strftime( staff_leave.leave_end_date, '%d/%m/%Y') if staff_leave.leave_end_date is not None else "" # only count working days staff_leave_map["number_of_leave_days"] = staff_leave.number_of_leave_days staff_leave_map["leave_department_approval"] = staff_leave.leave_department_approval staff_leave_map["leave_hr_approval"] = staff_leave.leave_hr_approval staff_leave_map["leave_status"] = staff_leave.leave_status creator_map = {} creator_map["staff_id"] = str( staff_leave.created_by.id) if staff_leave.created_by is not None else "" creator_map["staff_name"] = f'{staff_leave.created_by.first_name} {staff_leave.created_by.last_name}' if staff_leave.created_by is not None else "" creator_map["staff_position"] = staff_leave.created_by.staff_position.position_title if staff_leave.created_by is not None else "" staff_leave_map["created_by"] = creator_map creator_map = {} creator_map["staff_id"] = str( staff_leave.last_updated_by.id) if staff_leave.last_updated_by is not None else "" creator_map["staff_name"] = f'{staff_leave.last_updated_by.first_name} {staff_leave.last_updated_by.last_name}' if staff_leave.last_updated_by is not None else "" creator_map["staff_position"] = staff_leave.last_updated_by.staff_position.position_title if staff_leave.last_updated_by is not None else "" staff_leave_map["last_updated_by"] = creator_map creator_map = {} # creator_map["staff_id"] = str( staff_leave.staff_reliever.id) if staff_leave.staff_reliever is not None else "" creator_map["staff_name"] = f'{staff_leave.staff_reliever.first_name} {staff_leave.staff_reliever.last_name}' if staff_leave.staff_reliever is not None else "" creator_map["staff_position"] = staff_leave.staff_reliever.staff_position.position_title if staff_leave.staff_reliever is not None else "" staff_leave_map["staff_reliever"] = creator_map # staff_leave_map["created_on"] = datetime.strftime( staff_leave.created_on, date_format) if staff_leave.created_on is not None else "" staff_leave_map["last_updated_on"] = datetime.strftime( staff_leave.last_updated_on, date_format) if staff_leave.last_updated_on is not None else "" # print(staff_leave_map) staff_leaves_list.append(staff_leave_map) payload["staff_leaves_list"] = staff_leaves_list return Response({"message": "Leave application successful", "payload": payload}, status=200) else: # print(serializers.errors) payload["staff_leaves_list"] = [] return Response({"message": "Unable to apply for leave", "payload": payload}, status=406) else: payload["staff_leaves_list"] = [] return Response({"message": "You are unauthorised to perform this action", "payload": payload}, status=401) except: # Exception as e: # print(e) payload["staff_leaves_list"] = [] return Response({"message": "Error creating leave", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def hod_get_department_leaves(request): payload = {} staff_leaves_list = [] date_format = '%d/%m/%Y, %H:%M' try: company_serial_number = request.data["serial_number"] company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if (staff_profile.is_head_of_department == True and company_profile) or check_if_system_administrator(staff_profile): all_staff_leaves = StaffLeave.objects.all().order_by("-id") for staff_leave in all_staff_leaves: if staff_leave.recycle_bin == False and staff_leave.staff_profile.company_department == staff_profile.company_department and staff_leave.staff_profile.company_branch == staff_profile.company_branch: staff_leave_map = {} # staff_leave_map["staff_id"] = str( staff_leave.staff_profile.id) staff_leave_map["staff_number"] = staff_leave.staff_profile.staff_number staff_leave_map["first_name"] = staff_leave.staff_profile.first_name staff_leave_map["last_name"] = staff_leave.staff_profile.last_name staff_leave_map["company_department"] = staff_leave.staff_profile.company_department.department_name staff_leave_map["company_branch_name"] = staff_leave.staff_profile.company_branch.branch_name staff_leave_map["staff_position"] = staff_leave.staff_profile.staff_position.position_title staff_leave_map["email_address"] = staff_leave.staff_profile.email_address # staff_leave_map["leave_id"] = str( staff_leave.id) staff_leave_map["leave_type"] = staff_leave.leave_type staff_leave_map["leave_description"] = staff_leave.leave_description staff_leave_map["leave_start_date"] = datetime.strftime( staff_leave.leave_start_date, '%d/%m/%Y') if staff_leave.leave_start_date is not None else "" staff_leave_map["leave_end_date"] = datetime.strftime( staff_leave.leave_end_date, '%d/%m/%Y') if staff_leave.leave_end_date is not None else "" # only count working days staff_leave_map["number_of_leave_days"] = staff_leave.number_of_leave_days staff_leave_map["leave_department_approval"] = staff_leave.leave_department_approval staff_leave_map["leave_hr_approval"] = staff_leave.leave_hr_approval staff_leave_map["leave_status"] = staff_leave.leave_status creator_map = {} creator_map["staff_id"] = str( staff_leave.created_by.id) if staff_leave.created_by is not None else "" creator_map["staff_name"] = f'{staff_leave.created_by.first_name} {staff_leave.created_by.last_name}' if staff_leave.created_by is not None else "" creator_map["staff_position"] = staff_leave.created_by.staff_position.position_title if staff_leave.created_by is not None else "" staff_leave_map["created_by"] = creator_map creator_map = {} creator_map["staff_id"] = str( staff_leave.last_updated_by.id) if staff_leave.last_updated_by is not None else "" creator_map["staff_name"] = f'{staff_leave.last_updated_by.first_name} {staff_leave.last_updated_by.last_name}' if staff_leave.last_updated_by is not None else "" creator_map["staff_position"] = staff_leave.last_updated_by.staff_position.position_title if staff_leave.last_updated_by is not None else "" staff_leave_map["last_updated_by"] = creator_map creator_map = {} # creator_map["staff_id"] = str( staff_leave.staff_reliever.id) if staff_leave.staff_reliever is not None else "" creator_map["staff_name"] = f'{staff_leave.staff_reliever.first_name} {staff_leave.staff_reliever.last_name}' if staff_leave.staff_reliever is not None else "" creator_map["staff_position"] = staff_leave.staff_reliever.staff_position.position_title if staff_leave.staff_reliever is not None else "" staff_leave_map["staff_reliever"] = creator_map # staff_leave_map["created_on"] = datetime.strftime( staff_leave.created_on, date_format) if staff_leave.created_on is not None else "" staff_leave_map["last_updated_on"] = datetime.strftime( staff_leave.last_updated_on, date_format) if staff_leave.last_updated_on is not None else "" # print(staff_leave_map) staff_leaves_list.append(staff_leave_map) payload["staff_leaves_list"] = staff_leaves_list return Response({"message": "true", "payload": payload}, status=200) else: payload["staff_leaves_list"] = [] return Response({"message": "false", "payload": payload}, status=401) except: # Exception as e: # print(e) payload["staff_leaves_list"] = [] return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def send_payslip(request): active_user = request.user company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] payslip = request.data["payslip"] file_name = request.data["file_name"] subject = request.data["subject"] month_name = request.data["month_name"] try: # print("hit successful") company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payslip.view"): staff_to_send_payslip = StaffProfile.objects.get(id=int(staff_id)) print("running") try: sent = send_email_with_attachment( request, subject, f"Attached is your payslip for the month of {month_name}", staff_to_send_payslip.email_address, staff_to_send_payslip.personal_email, file_name, payslip) if sent: return Response({"message": f"Email successfully sent to {staff_to_send_payslip.staff_number}", }, status=200) return Response({ "message": "Payslip emails are disabled or SMTP is not configured. Enable attachment emails in System Admin → Email & SMTP.", }, status=406) except Exception as e: print(f"email senidng failed: {e}") return Response({"message": f"Failed to send email to {staff_to_send_payslip.staff_number}", }, status=406) else: return Response({"message": "You are unauthorised to perform this function", }, status=401) except: return Response({"message": "Error sending email to staff", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def send_p9_form(request): company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] p9_form = request.data["p9_form"] file_name = request.data["file_name"] subject = request.data["subject"] tax_year = request.data.get("tax_year") or request.data.get("payroll_year") or request.data.get("p9_year") or "" try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if user_has_permission(staff_profile, "human_resource.payslip.view"): staff_to_send = StaffProfile.objects.get(id=int(staff_id)) try: sent = send_email_with_attachment( request, subject, f"Attached is your P9 tax deduction card for the year {tax_year}", staff_to_send.email_address, staff_to_send.personal_email, file_name, p9_form, ) if sent: return Response({"message": f"Email successfully sent to {staff_to_send.staff_number}"}, status=200) return Response({ "message": "P9 emails are disabled or SMTP is not configured. Enable attachment emails in System Admin → Email & SMTP.", }, status=406) except Exception as e: print(f"P9 email sending failed: {e}") return Response({"message": f"Failed to send email to {staff_to_send.staff_number}"}, status=406) else: return Response({"message": "You are unauthorised to perform this function"}, status=401) except Exception: return Response({"message": "Error sending P9 form email to staff"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def terminate_employment(request): active_user = request.user company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] try: caller_profile = StaffProfile.objects.filter(user=active_user).first() if not user_has_permission(caller_profile, "human_resource.employment_status.edit"): return Response({"message": "You are unauthorised to perform this action", }, status=401) staff_profile = StaffProfile.objects.get(id=int(staff_id)) staff_profile.is_profile_active = False staff_profile.save() return Response({"message": f"Employment/Contract for {staff_profile.first_name} {staff_profile.last_name} of staff number {staff_profile.staff_number} terminated successfully", }, status=200) except: return Response({"message": "Error encountered while terminating employment/contract", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_staff_p9_list(request): company_serial_number = request.data.get("serial_number", "") payload = {} staff_profiles_list = [] try: if not company_serial_number: return Response({"message": "serial_number is required", "payload": payload}, status=400) company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not user_has_permission(staff_profile, "human_resource.payslip.view"): return Response({"message": "You are not authorised to view P9 staff list", "payload": payload}, status=401) staff_profiles = StaffProfile.objects.filter( recycle_bin=False, company_branch__company_profile=company_profile, ).order_by("first_name", "last_name", "-id") for staff in staff_profiles: staff_map = {} staff_map["staff_id"] = str(staff.id) staff_map["first_name"] = staff.first_name staff_map["last_name"] = staff.last_name staff_map["staff_number"] = staff.staff_number staff_map["kra_pin"] = staff.kra_pin or "" staff_map["is_profile_active"] = "true" if staff.is_profile_active == True else "false" staff_profiles_list.append(staff_map) payload["staff_profiles"] = staff_profiles_list return Response({"message": "true", "payload": payload}, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found", "payload": payload}, status=404) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_leave_summary(request): target_timezone = pytz.timezone('Africa/Nairobi') date_format = '%d/%m/%Y, %H:%M' company_serial_number = request.data.get("serial_number", "") department_id = str(request.data.get("company_department_id", "all")).strip() summary_month = request.data.get("summary_month", "") summary_year = request.data.get("summary_year", "") payload = {} leave_summary_list = [] try: if not company_serial_number: return Response({"message": "serial_number is required", "payload": payload}, status=400) month_num = _parse_summary_month(summary_month) if month_num is None: return Response({"message": "Invalid summary month", "payload": payload}, status=400) try: year_num = int(str(summary_year).strip()) except (TypeError, ValueError): return Response({"message": "Invalid summary year", "payload": payload}, status=400) month_start = date(year_num, month_num, 1) month_end = date(year_num, month_num, calendar.monthrange(year_num, month_num)[1]) month_key = list(MONTH_NAME_TO_NUMBER.keys())[month_num - 1] month_label = month_key.capitalize() company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not user_has_permission(staff_profile, "human_resource.staff_leave.view"): return Response({"message": "You are not authorised to generate leave summaries", "payload": payload}, status=401) department_name = "All Departments" if department_id and department_id.lower() != "all": department = CompanyDepartment.objects.get(id=int(department_id), recycle_bin=False) department_name = department.department_name staff_leaves = StaffLeave.objects.filter( recycle_bin=False, staff_profile__isnull=False, staff_profile__recycle_bin=False, ).select_related( 'staff_profile', 'staff_profile__company_department', 'staff_profile__company_branch', 'staff_profile__staff_position', 'staff_reliever', 'staff_reliever__staff_position', ).order_by('-leave_start_date', '-id') if department_id and department_id.lower() != "all": staff_leaves = staff_leaves.filter(staff_profile__company_department_id=int(department_id)) totals = { "leave_records_count": 0, "total_leave_days": 0, "approved_count": 0, "pending_count": 0, "denied_count": 0, "by_leave_type": {}, } for staff_leave in staff_leaves: if not _leave_overlaps_month(staff_leave, month_start, month_end): continue staff_leave_map = _build_staff_leave_summary_map(staff_leave, target_timezone, date_format) leave_summary_list.append(staff_leave_map) leave_days = 0 try: leave_days = int(staff_leave.number_of_leave_days) except (TypeError, ValueError): leave_days = 0 totals["leave_records_count"] += 1 totals["total_leave_days"] += leave_days leave_type = staff_leave.leave_type or "not_selected" totals["by_leave_type"][leave_type] = totals["by_leave_type"].get(leave_type, 0) + leave_days hr_status = str(staff_leave.leave_hr_approval or "").lower() if hr_status == "approved": totals["approved_count"] += 1 elif hr_status == "denied": totals["denied_count"] += 1 else: totals["pending_count"] += 1 payload["filters"] = { "company_department_id": department_id if department_id else "all", "department_name": department_name, "summary_month": month_key, "summary_month_label": month_label, "summary_year": str(year_num), } payload["totals"] = totals payload["leave_summary_list"] = leave_summary_list return Response({"message": "true", "payload": payload}, status=200) except CompanyDepartment.DoesNotExist: return Response({"message": "Department not found", "payload": payload}, status=404) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found", "payload": payload}, status=404) except StaffProfile.DoesNotExist: return Response({"message": "Staff profile not found", "payload": payload}, status=404) except Exception: logger.exception("generate_leave_summary failed") return Response({"message": "false", "payload": payload}, status=500) def _normalize_statutory_title(title): key = str(title or "").strip().upper() if key in ("NHIF", "SHIF"): return "SHA" return str(title or "") def _hr_can_manage_payroll(staff_profile, company_profile): # Migrated to dynamic RBAC. return bool(company_profile) and user_has_permission( staff_profile, "human_resource.payroll_sheet.view") def _filter_payroll_sheets(summary_month, summary_year, company_branch_id="all"): month_key = list(MONTH_NAME_TO_NUMBER.keys())[_parse_summary_month(summary_month) - 1] month_label = month_key.capitalize() payroll_sheets = PayrollSheet.objects.filter( recycle_bin=False, payroll_sheet_for_the_year=str(summary_year), ).select_related('company_branch') normalized_month = month_key.lower() filtered = [] for payroll_sheet in payroll_sheets: sheet_month = str(payroll_sheet.payroll_sheet_for_the_month_of or "").strip().lower() if sheet_month != normalized_month and sheet_month != month_label.lower(): continue if company_branch_id and str(company_branch_id).lower() != "all": if payroll_sheet.company_branch is None or str(payroll_sheet.company_branch.id) != str(company_branch_id): continue filtered.append(payroll_sheet) return filtered, month_key, month_label @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_payroll_summary(request): company_serial_number = request.data.get("serial_number", "") company_branch_id = str(request.data.get("company_branch_id", "all")).strip() summary_month = request.data.get("summary_month", "") summary_year = request.data.get("summary_year", "") payload = {} try: if not company_serial_number: return Response({"message": "serial_number is required", "payload": payload}, status=400) if _parse_summary_month(summary_month) is None: return Response({"message": "Invalid summary month", "payload": payload}, status=400) company_profile = CompanyProfile.objects.get(company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _hr_can_manage_payroll(staff_profile, company_profile): return Response({"message": "You are not authorised to generate payroll summaries", "payload": payload}, status=401) payroll_sheets, month_key, month_label = _filter_payroll_sheets( summary_month, summary_year, company_branch_id) payroll_summary_list = [] totals = { "payroll_sheets_count": 0, "staff_count": 0, "total_gross_value": 0.0, "total_net_pay_value": 0.0, "total_deduction_value": 0.0, "total_bonus_value": 0.0, } for payroll_sheet in payroll_sheets: staff_count = payroll_sheet.staff_payroll_items.filter(recycle_bin=False).count() gross_value = float(payroll_sheet.payroll_sheet_value or 0) net_value = float(payroll_sheet.payroll_sheet_total_net_pay_value or 0) deduction_value = float(payroll_sheet.payroll_sheet_total_deduction_value or 0) bonus_value = float(payroll_sheet.payroll_sheet_total_bonus_value or 0) payroll_summary_list.append({ "payroll_sheet_id": str(payroll_sheet.id), "payroll_sheet_title": payroll_sheet.payroll_sheet_title, "payroll_sheet_number": payroll_sheet.payroll_sheet_number, "branch_name": payroll_sheet.company_branch.branch_name if payroll_sheet.company_branch is not None else "", "payroll_sheet_value": payroll_sheet.payroll_sheet_value, "payroll_sheet_total_net_pay_value": payroll_sheet.payroll_sheet_total_net_pay_value, "payroll_sheet_total_deduction_value": payroll_sheet.payroll_sheet_total_deduction_value, "payroll_sheet_total_bonus_value": payroll_sheet.payroll_sheet_total_bonus_value, "staff_count": staff_count, }) totals["payroll_sheets_count"] += 1 totals["staff_count"] += staff_count totals["total_gross_value"] += gross_value totals["total_net_pay_value"] += net_value totals["total_deduction_value"] += deduction_value totals["total_bonus_value"] += bonus_value payload["filters"] = { "company_branch_id": company_branch_id or "all", "summary_month": month_key, "summary_month_label": month_label, "summary_year": str(summary_year), } payload["totals"] = totals payload["payroll_summary_list"] = payroll_summary_list return Response({"message": "true", "payload": payload}, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found", "payload": payload}, status=404) except StaffProfile.DoesNotExist: return Response({"message": "Staff profile not found", "payload": payload}, status=404) except Exception: logger.exception("generate_payroll_summary failed") return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_statutory_deductions_summary(request): company_serial_number = request.data.get("serial_number", "") company_branch_id = str(request.data.get("company_branch_id", "all")).strip() summary_month = request.data.get("summary_month", "") summary_year = request.data.get("summary_year", "") payload = {} try: if not company_serial_number: return Response({"message": "serial_number is required", "payload": payload}, status=400) if _parse_summary_month(summary_month) is None: return Response({"message": "Invalid summary month", "payload": payload}, status=400) company_profile = CompanyProfile.objects.get(company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _hr_can_manage_payroll(staff_profile, company_profile): return Response({"message": "You are not authorised to generate statutory summaries", "payload": payload}, status=401) payroll_sheets, month_key, month_label = _filter_payroll_sheets( summary_month, summary_year, company_branch_id) statutory_map = {} statutory_rows = [] for payroll_sheet in payroll_sheets: payroll_instances = payroll_sheet.staff_payroll_items.filter(recycle_bin=False) for payroll_instance in payroll_instances: deduction_instances = payroll_instance.deduction_instance.filter(recycle_bin=False) for deduction_instance in deduction_instances: if deduction_instance.deduction is None: continue title = _normalize_statutory_title(deduction_instance.deduction.deduction_title) title_upper = title.upper() if title_upper not in ("SHA", "NSSF", "PAYE"): continue amount = float(deduction_instance.deduction_instance_value or 0) bucket = statutory_map.setdefault(title_upper, {"total_amount": 0.0, "staff_ids": set()}) bucket["total_amount"] += amount if payroll_instance.staff_profile is not None: bucket["staff_ids"].add(payroll_instance.staff_profile.id) for statutory_type, bucket in statutory_map.items(): statutory_rows.append({ "statutory_type": statutory_type, "total_amount": f"{bucket['total_amount']:.2f}", "staff_count": len(bucket["staff_ids"]), }) statutory_rows.sort(key=lambda row: row["statutory_type"]) payload["filters"] = { "company_branch_id": company_branch_id or "all", "summary_month": month_key, "summary_month_label": month_label, "summary_year": str(summary_year), } payload["statutory_summary_list"] = statutory_rows return Response({"message": "true", "payload": payload}, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found", "payload": payload}, status=404) except StaffProfile.DoesNotExist: return Response({"message": "Staff profile not found", "payload": payload}, status=404) except Exception: logger.exception("generate_statutory_deductions_summary failed") return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_staff_p9_form(request): active_user = request.user company_serial_number = request.data.get("serial_number") if not company_serial_number: return Response( {"message": "serial_number is required", "payload": {}}, status=400, ) payroll_year = request.data.get("payroll_year") or request.data.get("payrollYear") or request.data.get("p9_year") if not payroll_year or str(payroll_year).strip().lower() in {"", "not_selected"}: return Response( {"message": "payroll_year is required", "payload": {}}, status=400, ) payroll_year = str(payroll_year).strip() payload = {} def _parse_staff_list(raw): if raw is None: return [] if isinstance(raw, str): raw = raw.strip() if not raw: return [] try: parsed = json.loads(raw) except json.JSONDecodeError: parsed = [part.strip() for part in raw.split(",") if part.strip()] raw = parsed if isinstance(raw, list): if len(raw) == 1 and isinstance(raw[0], str) and raw[0].strip().startswith("["): try: raw = json.loads(raw[0]) except json.JSONDecodeError: pass return [str(staff_id).strip() for staff_id in raw if str(staff_id).strip()] return [] staffList = _parse_staff_list( request.data.get("staffList") or request.data.get("staff_list") or request.data.get("staffIds") or request.data.get("staff_ids") ) if not staffList: single_staff_id = request.data.get("staff_id") or request.data.get("staffId") if single_staff_id and str(single_staff_id).strip().lower() not in {"", "all", "not_selected"}: staffList = [str(single_staff_id).strip()] if not staffList: return Response( {"message": "staffList or staff_id is required", "payload": {}}, status=400, ) #("january", "January"), ("february", "February"), ("march", "March"), ("april", "April"), ("may", "May"), ("june", "June"), ("july", "July"), ("august", "August"), ("september", "September"), ("october", "October"), ("november", "November"), ("december", "December"), ("not_selected", "Not Selected")] months_list = ["january","february","march","april","may","june","july","august","september","october","november","december"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not user_has_permission(staff_profile, "human_resource.payslip.view"): return Response({"message": "You are not authorised to generate P9 forms", "payload": payload}, status=401) staff_profiles = [] staff_profiles_list = [] for staff_id in staffList: staff_profile_row = StaffProfile.objects.get( id=int(staff_id), recycle_bin=False, company_branch__company_profile=company_profile, ) staff_profiles.append(staff_profile_row) for staff in staff_profiles: staff_map = {} staff_payroll_instance = staff.staff_payroll_instance.filter( recycle_bin=False, payroll_sheet__payroll_sheet_for_the_year=payroll_year,payroll_sheet__payroll_sheet_approved_by_finance=True).order_by("id") if len(staff_payroll_instance) > 0: staff_map["staff_id"] = str(staff.id) staff_map["staff_name"] = f'{staff.first_name} {staff.last_name}' staff_map["kra_pin"] = staff.kra_pin staff_map["staff_number"] = staff.staff_number payroll_instances_list = [] for month in months_list: month_payrolls = staff_payroll_instance.filter(payroll_sheet__payroll_sheet_for_the_month_of=month) if month_payrolls.count() > 0: payroll_instance_map = {} month_pay_count = 0 payroll_instance_map["payroll_sheet_for_the_month_of"] = month payroll_instance_map["payroll_instance_id"] = "" for month_pay in month_payrolls: if month_pay_count <= 0: payroll_instance_map["gross_salary"] = month_pay.gross_salary payroll_instance_map["basic_salary"] = calculate_basic_salary( month_pay) payroll_instance_map["net_salary"] = month_pay.net_salary staff_deductions = month_pay.deduction_instance.filter( recycle_bin=False).order_by("-id") staff_deductions_list = [] for deduction in staff_deductions: deduction_map = {} deduction_map["deduction_id"] = str(deduction.id) deduction_map["deduction_instance_value"] = deduction.deduction_instance_value deduction_map["deduction_title"] = deduction.deduction.deduction_title staff_deductions_list.append(deduction_map) payroll_instance_map["staff_deductions_list"] = staff_deductions_list month_pay_count += 1 else: if month_payrolls.count() > 1: payroll_instance_map["gross_salary"] = str( float(payroll_instance_map["gross_salary"])+float(month_pay.gross_salary)) payroll_instance_map["basic_salary"] = str(float(payroll_instance_map["basic_salary"])+float(calculate_basic_salary( month_pay))) payroll_instance_map["net_salary"] = str( float(payroll_instance_map["net_salary"])+float(month_pay.net_salary)) staff_deductions = month_pay.deduction_instance.filter( recycle_bin=False).order_by("-id") for deduction in staff_deductions: for saved_deduction in payroll_instance_map["staff_deductions_list"]: if saved_deduction["deduction_id"] == str(deduction.id): deduction_map["deduction_instance_value"] = str(float( deduction_map["deduction_instance_value"])+float(deduction.deduction_instance_value)) payroll_instances_list.append(payroll_instance_map) else: payroll_instance_map = {} payroll_instance_map["payroll_sheet_for_the_month_of"] = month payroll_instance_map["payroll_instance_id"] = "" payroll_instance_map["gross_salary"] = "0.00" payroll_instance_map["basic_salary"] = "0.00" payroll_instance_map["net_salary"] = "0.00" payroll_instance_map["staff_deductions_list"] = [] payroll_instances_list.append(payroll_instance_map) # for payroll_instance in staff_payroll_instance: # payroll_instance_map = {} # payroll_instance_map["payroll_instance_id"] = str( # payroll_instance.id) # payroll_instance_map["payroll_sheet_for_the_month_of"] = payroll_instance.payroll_sheet.payroll_sheet_for_the_month_of # payroll_instance_map["gross_salary"] = payroll_instance.gross_salary # payroll_instance_map["basic_salary"] = calculate_basic_salary( # payroll_instance) # payroll_instance_map["net_salary"] = payroll_instance.net_salary # staff_deductions = payroll_instance.deduction_instance.filter( # recycle_bin=False).order_by("-id") # staff_deductions_list = [] # for deduction in staff_deductions: # deduction_map = {} # deduction_map["deduction_id"] = str(deduction.id) # deduction_map["deduction_instance_value"] = deduction.deduction_instance_value # deduction_map["deduction_title"] = deduction.deduction.deduction_title # staff_deductions_list.append(deduction_map) # payroll_instance_map["staff_deductions_list"] = staff_deductions_list # # check to see if the staff has two instances of the same month # payroll_instances_list.append(payroll_instance_map) staff_map["payroll_instances_list"] = payroll_instances_list staff_profiles_list.append(staff_map) payload["staff_profiles"] = staff_profiles_list if not staff_profiles_list: return Response( { "message": f"No finance-approved payroll records found for {payroll_year}.", "payload": payload, }, status=200, ) return Response({"message": "true", "payload": payload}, status=200) except StaffProfile.DoesNotExist: return Response({"message": "Staff member not found for this company", "payload": payload}, status=404) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found", "payload": payload}, status=404) except Exception as e: print(e) return Response({"message": "false", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def activate_employment(request): active_user = request.user company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] employment_end_date = request.data["employment_end_date"] date_format = '%d/%m/%Y' try: staff_profile = StaffProfile.objects.get(id=int(staff_id)) active_hr = StaffProfile.objects.get(user=active_user) if not user_has_permission(active_hr, "human_resource.employment_status.edit"): return Response({"message": "You are unauthorised to perform this action", }, status=401) if staff_profile.type_of_employment == "contract": employment_end_date = datetime.strptime( request.data["employment_end_date"], date_format).strftime("%Y-%m-%d") if len(request.data["employment_end_date"]) > 0 else None staff_profile.employment_end_date = employment_end_date staff_profile.is_profile_active = True staff_profile.save() # print(f"Running {staff_profile.is_profile_active}") else: staff_profile.is_profile_active = True staff_profile.save() try: list_of_staff_ids = [] list_of_staff_ids.append(active_hr.id) notification_title = "Employment/Contract Renewal" notification_body = f"You have renewed employment/contract terms for {staff_profile.staff_number}." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "Employment/Contract Renewal" notification_body = f"Your employment/contract terms has been renewed successfully." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass return Response({"message": f"Employment/Contract for {staff_profile.staff_number} has been renewed", }, status=200) except Exception as e: print(e) return Response({"message": "Error encountered while renewing employment/contract", }, status=500)