# from django.shortcuts import render from django.db import connection import base64 from human_resource.models import Deduction, MessageStaffReadStatus, MyMessage, MyNotification, NotificationStaffReadStatus, StaffProfile, staffPosition from human_resource.serializers import DeductionSerializer, StaffDeductionSchemeSerializer from system_administration.utils import create_folder_if_not_exists, create_messages, create_notifications, generate_random_string, generateNextStaffNumber, get_staff_profile_data, send_email_signup 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 rest_framework.authtoken.models import Token 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 import json import traceback from django.http import JsonResponse import pytz @api_view(['POST']) def get_admin_creation_status(request): serialNumber = request.data["serial_number"] # system_admin_already_exist = SystemAdminCreationStatus.objects.exists() payload = {} payload["company_serial_number"] = "" payload["company_name"] = "" payload["company_description"] = "" payload["company_postal_address"] = "" payload["company_country_location"] = "" payload["company_phone"] = "" payload["company_preferred_currency"] = "" try: company_profile = CompanyProfile.objects.get( company_serial_number=serialNumber) try: system_admin = SystemAdminCreationStatus.objects.get( company_profile=company_profile) if system_admin.systemAdminCreated == True: payload["company_serial_number"] = company_profile.company_serial_number payload["company_name"] = company_profile.company_name payload["company_description"] = company_profile.company_description payload["company_postal_address"] = company_profile.company_postal_address payload["company_country_location"] = company_profile.company_country_location payload["company_phone"] = company_profile.company_phone payload["company_preferred_currency"] = company_profile.company_preferred_currency return Response({"message": "true", "payload": payload, }, status=200) else: return Response({"message": "false", "payload": payload, }, status=200) except: # Exception as e: # print (e) # print (payload) return Response({"message": "false", "payload": payload, }, status=401) except: return Response({"message": "false", "payload": payload, }, status=406) @api_view(['POST']) def validate_serial_number(request): serialNumber = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=serialNumber) if company_profile.company_serial_number == serialNumber: return Response({"message": "true"}, status=200) except: return Response({"message": "false"}, status=200) @api_view(['POST']) def create_system_admin(request): email = request.data['username'] password = request.data['password'] company_serial_number = request.data["serial_number"] try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) criteria = {"company_profile": company_profile} system_admin_already_exist = SystemAdminCreationStatus.objects.filter( **criteria).exists() if system_admin_already_exist == False: serializers = UserSerializer( data={'username': email, 'password': password}) if serializers.is_valid(): user = serializers.save() user.is_staff = True user.save() new_user = authenticate(username=email, password=password) # create main company branch main_branch, created = CompanyBranch.objects.get_or_create( company_profile=company_profile, main_branch=True) # create staff position staff_position, created = staffPosition.objects.get_or_create(company_profile=company_profile, position_title="System Administrator") # create department department, created = CompanyDepartment.objects.get_or_create(company_profile=company_profile, department_name="system_and_administration", department_description="Tasked with overseeing the ERP system's health, security, and performance, this department ensures that the organization's technology backbone operates seamlessly to meet business needs.") # create staff profile staff, created = StaffProfile.objects.get_or_create( user=new_user, staff_position=staff_position, staff_number="MEL001", email_address=new_user.username, company_branch=main_branch, company_department=department, has_read_write_priviledges=True, is_head_of_department=True) # creates the system administrator role # create message try: list_of_staff_ids = [] list_of_staff_ids.append(staff.id) message_title = "Welcome" message_body = f"You have been registered to Megawatt Energies as the system administrator. Your staff number is {staff.staff_number}." create_messages(message_title, message_body, list_of_staff_ids) except: pass # create default deduction schemes # create default deduction schemes # paye current_year = datetime.now().year date_effective_from = datetime(current_year, 1, 1) date_effective_to = datetime(current_year+2, 1, 1) paye_deduction_serializer = DeductionSerializer(data={'company_profile': company_profile.id, 'deduction_title': "PAYE", 'deduction_description': "Pay As You Earn tax (Income Tax)", 'deduction_type': "tax", 'deduction_value': "0.0", 'deduction_module': 'other', 'date_effective_from': date_effective_from.date(), 'date_effective_to': date_effective_to.date(), 'created_by': staff.id, 'last_updated_by': staff.id}) if paye_deduction_serializer.is_valid(): paye_deduction_instance = paye_deduction_serializer.save() else: pass # print(paye_deduction_serializer.errors) # shif # current_year = datetime.now().year date_effective_from = datetime(current_year, 7, 1) date_effective_to = datetime(current_year+2, 1, 1) shif_deduction_serializer = DeductionSerializer(data={'company_profile': company_profile.id, 'deduction_title': "SHA", 'deduction_description': "Social Health Authority contribution", 'deduction_type': "insurance", 'deduction_value': "2.75", 'deduction_module': "percentage", 'date_effective_from': date_effective_from.date(), 'date_effective_to': date_effective_to.date(), 'created_by': staff.id, 'last_updated_by': staff.id}) if shif_deduction_serializer.is_valid(): shif_deduction_instance = shif_deduction_serializer.save() else: pass # print(shif_deduction_serializer.errors) # pension # current_year = datetime.now().year date_effective_from = datetime(current_year, 1, 1) date_effective_to = datetime(current_year+2, 1, 1) nssf_deduction_serializer = DeductionSerializer(data={'company_profile': company_profile.id, 'deduction_title': "NSSF", 'deduction_description': "Pension Contribution", 'deduction_type': "pension", 'deduction_value': "0.00", 'deduction_module': 'other', 'date_effective_from': date_effective_from.date(), 'date_effective_to': date_effective_to.date(), 'created_by': staff.id, 'last_updated_by': staff.id}) if nssf_deduction_serializer.is_valid(): nssf_deduction_instance = nssf_deduction_serializer.save() else: pass # print(nssf_deduction_serializer.errors) # create default deduction schemes for the super hr # paye paye_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': paye_deduction_instance.id}) if paye_deduction_scheme_serializer.is_valid(): paye_deduction_scheme_serializer.save() else: pass # print(paye_deduction_scheme_serializer.errors) # shif shif_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': shif_deduction_instance.id}) if shif_deduction_scheme_serializer.is_valid(): shif_deduction_scheme_serializer.save() else: pass # print(shif_deduction_scheme_serializer.errors) # nssf nssf_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': nssf_deduction_instance.id}) if nssf_deduction_scheme_serializer.is_valid(): nssf_deduction_scheme_serializer.save() else: pass # print(nssf_deduction_scheme_serializer.errors) # end SystemAdminCreationStatus.objects.create( user=new_user, systemAdminCreated=True, company_profile=company_profile) # add send email to super admin with their staff number message = f"You have successfully registered on Megawatt Energies ERP. Your staff number is {staff.staff_number}" recipient_list = [f'{email}'] try: send_email_signup(request, message, recipient_list) except: pass return Response({"message": "System Administrator created successfully", }, status=200) else: # print(serializers.errors) return Response({"message": "Unable to create system administrator account", }, status=406) else: # system admin already exists return Response({"message": "System Admin already exists!", }, status=401) except: # Exception as e: # print(e) return Response({"message": "Error creating system administrator account!", }, status=500) @api_view(['POST']) def staff_login(request): from system_administration.security_settings import ( build_staff_login_payload, create_login_otp_challenge, get_or_create_security_settings, is_ip_allowed_for_staff, resolve_company_from_staff, ) staff_number = request.data['staff_number'] password = request.data['password'] payload = {} payload["user_token"] = "" payload["company_department"] = "" payload["full_name"] = "" payload["user_role"] = "" try: staff_profile = StaffProfile.objects.select_related( "user", "company_branch", "company_branch__company_profile", "company_department", "staff_position", ).get(staff_number=staff_number) if staff_profile.user is None: return Response( {"message": "Staff account is not linked to a login user.", "payload": payload}, status=500, ) try: create_folder_if_not_exists() except Exception: pass username = staff_profile.user.username user = authenticate(username=username, password=password) if user is not None and staff_profile.is_profile_active is not True: return Response( { "message": "Your staff profile is inactive. Contact HR to renew your employment or contract.", "payload": payload, }, status=403, ) if user is not None and staff_profile.is_profile_active is True: company_profile = resolve_company_from_staff(staff_profile) security = get_or_create_security_settings(company_profile) if company_profile else None if security and not is_ip_allowed_for_staff(request, staff_profile, security): return Response( { "message": "Access denied. Your IP address is not on the company whitelist.", "payload": payload, }, status=403, ) if security and security.enable_login_otp: if not company_profile: return Response( { "message": "Login OTP is enabled but the staff branch is not linked to a company profile.", "payload": payload, }, status=403, ) if not (staff_profile.email_address or '').strip(): return Response( { "message": "Login OTP is enabled but your staff profile has no email address.", "payload": payload, }, status=403, ) challenge = create_login_otp_challenge(staff_profile, security, company_profile) if not challenge.get('sent'): return Response( { "message": "Unable to send login verification code. Ask your administrator to check SMTP settings.", "payload": payload, }, status=503, ) payload["otp_required"] = True payload["challenge_id"] = challenge["challenge_id"] payload["masked_email"] = challenge["masked_email"] payload["otp_ttl_minutes"] = security.otp_ttl_minutes return Response( {"message": "Verification code sent to your email.", "payload": payload}, status=200, ) # SPA uses Token auth; session login is optional when middleware provides a session. http_request = getattr(request, "_request", request) if hasattr(http_request, "session"): login(http_request, user) token, _created = Token.objects.get_or_create(user=user) payload.update(build_staff_login_payload(staff_profile, user, token)) return Response({"message": "Login successful", "payload": payload}, status=200) else: return Response({"message": "Invalid credentials", "payload": payload}, status=406) except Exception as e: print(e) return Response( {"message": "Login failed. Check staff number and password.", "payload": payload}, status=500, ) @api_view(['POST']) def verify_login_otp(request): from system_administration.security_settings import ( build_staff_login_payload, verify_login_otp_challenge, ) staff_number = request.data.get('staff_number') challenge_id = request.data.get('challenge_id') otp_code = request.data.get('otp_code') payload = { "user_token": "", "company_department": "", "full_name": "", "user_role": "", } ok, message, staff_profile = verify_login_otp_challenge(staff_number, challenge_id, otp_code) if not ok or staff_profile is None: return Response({"message": message, "payload": payload}, status=406) user = staff_profile.user if user is None or staff_profile.is_profile_active is not True: return Response({"message": "Staff account is inactive.", "payload": payload}, status=403) http_request = getattr(request, "_request", request) if hasattr(http_request, "session"): login(http_request, user) token, _created = Token.objects.get_or_create(user=user) payload.update(build_staff_login_payload(staff_profile, user, token)) return Response({"message": "Login successful", "payload": payload}, status=200) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_user(request): try: company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] staff_profile = StaffProfile.objects.get(user=request.user) company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) if staff_profile.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: staff_to_delete = StaffProfile.objects.get(id=int(staff_id)) staff_to_delete.recycle_bin = True staff_to_delete.save() staff_to_delete.user.is_active = False staff_to_delete.user.save() return Response({"message": "Staff profile deleted successfully", }, status=200) else: return Response({"message": "You are unauthorised to perform this action", }, status=401) except Exception as e: print(e) return Response({"message": "Unable to delete staff profile", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def system_admin_dashboard(request): date_format = '%d/%m/%Y, %H:%M' company_serial_number = str(request.data.get("serial_number", "")).strip() active_user = request.user payload = {} company_profile_map = {} company_departments_list = [] company_branches_list = [] company_suppliers_list = [] company_customers_list = [] try: target_timezone = pytz.timezone('Africa/Nairobi') if len(company_serial_number) == 0: return Response({"message": "serial_number is required", "payload": payload}, status=400) company_profile = CompanyProfile.objects.filter( company_serial_number=company_serial_number).first() if company_profile is None: return Response({"message": "Invalid company serial number", "payload": payload}, status=404) staff_profile = StaffProfile.objects.get(user=request.user) staff_position_title = staff_profile.staff_position.position_title if staff_profile.staff_position is not None else "" if staff_position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: active_staff_profile_data = get_staff_profile_data(active_user) # print(active_staff_profile_data) company_profile_map["company_name"] = company_profile.company_name company_profile_map["company_description"] = company_profile.company_description 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 company_profile_map["company_preferred_currency"] = company_profile.company_preferred_currency company_profile_map["company_profile_set"] = "true" if company_profile.company_profile_set == True else "false" company_profile_map["company_super_hr_created"] = "true" if company_profile.company_super_hr_created == True else "false" company_profile_map["created_on"] = datetime.strftime( company_profile.created_on.astimezone(target_timezone), date_format) company_profile_map["last_updated_on"] = datetime.strftime( company_profile.last_updated_on.astimezone(target_timezone), date_format) # company_departments company_departments = company_profile.company_departments.all().order_by("-id") # print(company_departments) 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_name 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) # users company_branch_staffs_list = [] branch_staffs = branch.company_branch_staffs.all().order_by("-id") for staff in branch_staffs: if staff.user is not None and staff.user.is_active == True: 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_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" staff_map["is_profile_active"] = "true" if staff.is_profile_active == True else "false" staff_map["created_on"] = datetime.strftime( staff.created_on.astimezone(target_timezone), date_format) staff_map["last_updated_on"] = datetime.strftime( staff.last_updated_on.astimezone(target_timezone), date_format) # added info 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 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 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["basic_salary"] = staff.basic_salary staff_map["time_sheets_list"] = [] staff_map["education_qualifications_list"] = [] staff_map["staff_training_records_list"] = [] staff_map["staff_disciplinary_records_list"] = [] staff_map["staff_leaves_list"] = [] staff_map["available_leave_days"] = "0" staff_map["staff_bonus_schemes_list"] = [] staff_map["staff_deduction_schemes_list"] = [] # added staff_map["personal_email"] = staff.personal_email # staff_map["currency"] = company_profile.company_preferred_currency # end company_branch_staffs_list.append(staff_map) branch_map["branch_staff_profiles"] = company_branch_staffs_list company_suppliers = company_profile.company_suppliers.all() for supplier in company_suppliers: supplier_map = {} supplier_map["supplier_id"] = str(supplier.id) supplier_map["supplier_name"] = supplier.supplier_name supplier_map["supplier_phone"] = supplier.supplier_phone supplier_map["supplier_email"] = supplier.supplier_email supplier_map["supplier_address"] = supplier.supplier_address supplier_map["supplier_description"] = supplier.supplier_description supplier_map["created_by"] = f'{supplier.created_by.first_name} {supplier.created_by.last_name}' if supplier.created_by is not None else "" supplier_map["last_updated_by"] = f'{supplier.last_updated_by.first_name} {supplier.last_updated_by.last_name}' if supplier.last_updated_by is not None else "" supplier_map["created_on"] = datetime.strftime( supplier.created_on, date_format) supplier_map["last_updated_on"] = datetime.strftime( supplier.last_updated_on, date_format) company_suppliers_list.append(supplier_map) company_customers = company_profile.company_customers.all().order_by("-id") for customer in company_customers: if customer.recycle_bin == False: customer_map = {} customer_map["customer_id"] = str(customer.id) customer_map["customer_name"] = f'{customer.customer_first_name} {customer.customer_last_name}' customer_map["email_address"] = customer.email_address customer_map["phone_number"] = customer.phone_number customer_map["customer_title"] = customer.customer_title customer_map["is_profile_set"] = "true" if customer.is_profile_set == True else "false" customer_map["created_on"] = datetime.strftime( customer.created_on.astimezone(target_timezone), date_format) customer_map["last_updated_on"] = datetime.strftime( customer.last_updated_on.astimezone(target_timezone), date_format) company_customers_list.append(customer_map) payload["company_profile"] = company_profile_map payload["active_staff_profile_data"] = active_staff_profile_data payload["company_departments_list"] = company_departments_list payload["company_branches_list"] = company_branches_list payload["company_suppliers_list"] = company_suppliers_list payload["company_customers_list"] = company_customers_list # print(payload) return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: # payload["company_profile"] = company_profile # payload["active_staff_profile_data"] = active_staff_profile_data # payload["company_departments_list"] = company_departments_list # payload["company_branches_list "] = company_branches_list # payload["company_suppliers_list"] = company_suppliers_list # payload["company_customers_list"] = company_customers_list traceback.print_exc() print(e) return Response({"message": "Unable to load system admin dashboard", "payload": payload}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_company_profile(request): 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 staff_profile.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: company_name = request.data['company_name'] company_description = request.data['company_description'] company_postal_address = request.data['company_postal_address'] company_country_location = request.data['company_country_location'] company_phone = request.data['company_phone'] company_preferred_currency = request.data['company_preferred_currency'] serializer = CompanyProfileSerializer(instance=company_profile, data={'company_name': company_name, 'company_description': company_description, 'company_postal_address': company_postal_address, 'company_country_location': company_country_location, 'company_phone': company_phone, 'company_preferred_currency': company_preferred_currency}) if serializer.is_valid(): company_profile = serializer.save() company_profile.company_profile_set = True company_profile.save() return Response({"message": "Company profile set up successful"}, status=200) else: return Response({"message": "Error setting up company profile"}, status=406) else: return Response({"message": "You are unauthorised to perform this function"}, status=401) except: # Exception as e: # print(e) return Response({"message": "Unable to set up company profile"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_company_branch(request): 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 staff_profile.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: branch_id = request.data['branch_id'] branchInstance = CompanyBranch.objects.get( id=int(branch_id), company_profile=company_profile) branch_name = request.data['branch_name'] branch_description = request.data['branch_description'] branch_county_location = request.data['branch_county_location'] branch_phone = request.data['branch_phone'] serializer = CompanyBranchSerializer(instance=branchInstance, data={'branch_name': branch_name, 'branch_description': branch_description, 'branch_county_location': branch_county_location, 'branch_phone': branch_phone, }) if serializer.is_valid(): company_branch = serializer.save() # automatically create the remaining departments departments_list = [{"department_name": "procurement", "department_description": "Responsible for receiving and reviewing purchase requisitions from various departments within the organization, especialy the warehouse management department. The department creates purchase orders from the purchase requisitions it receives."}, {"department_name": "human_resource_management", "department_description": "Responsible for overseeing personnel-related functions, including recruitment, employee onboarding and performance management"}, {"department_name": "warehouse_management", "department_description": "This department is tasked with efficiently overseeing and optimizing the movement and storage of goods within the organization's warehouses. It utilizes the ERP system to manage inventory levels, track shipments, and enhance overall supply chain visibility."}, {"department_name": "Sales", "department_description": "Aims to drive revenue growth, achieve sales targets, and contribute to the company's overall success and profitability."}, {"department_name": "marketing", "department_description": "The marketing department focuses on driving revenue and promoting products or services. It utilizes the ERP system to manage customer relationships, track sales activities, and analyze marketing campaigns to enhance overall business growth."}, {"department_name": "finance_and_accounting", "department_description": "Responsible for managing financial transactions, budgeting, and financial reporting."}, {"department_name": "management", "department_description": "Collaborates with various departments to ensure alignment with organizational objectives and effective implementation of policies and initiatives."}] for department in departments_list: new_department, created = CompanyDepartment.objects.get_or_create(company_profile=company_profile, department_name=department["department_name"], department_description=department["department_description"]) company_profile.company_departments_set = True company_profile.save() return Response({"message": "Company main branch set up successful"}, status=200) else: return Response({"message": "Error setting up company main branch"}, status=406) else: return Response({"message": "You are unauthorised to perform this function"}, status=401) except: # Exception as e: # print(e) return Response({"message": "Unable to set up company main branch"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_other_branch(request): 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.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: branch_name = request.data['branch_name'] branch_description = request.data['branch_description'] branch_county_location = request.data['branch_county_location'] branch_phone = request.data['branch_phone'] serializer = CompanyBranchSerializer( data={'company_profile': company_profile.id, 'branch_name': branch_name, 'branch_description': branch_description, 'branch_county_location': branch_county_location, 'branch_phone': branch_phone, }) if serializer.is_valid(): company_branch = serializer.save() return Response({"message": "Company branch created successfully"}, status=200) else: return Response({"message": "Error setting up company branch"}, status=406) except: return Response({"message": "Unable to set up company branch"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def edit_other_branch(request): 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.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: branch_name = request.data['branch_name'] branch_description = request.data['branch_description'] branch_county_location = request.data['branch_county_location'] branch_phone = request.data['branch_phone'] branch_id = request.data["branch_id"] branch_instance = CompanyBranch.objects.get(id=int(branch_id)) serializer = CompanyBranchSerializer(instance=branch_instance, data={'company_profile': company_profile.id, 'branch_name': branch_name, 'branch_description': branch_description, 'branch_county_location': branch_county_location, 'branch_phone': branch_phone, }) if serializer.is_valid(): company_branch = serializer.save() return Response({"message": "Company branch edited successfully"}, status=200) else: return Response({"message": "Error editing company branch"}, status=406) except: return Response({"message": "Unable to edit company branch"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def delete_other_branch(request): 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.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: branch_id = request.data["branch_id"] branch_instance = CompanyBranch.objects.get(id=int(branch_id)) branch_instance.recycle_bin = True branch_instance.save() return Response({"message": "Company branch deleted successfully"}, status=200) else: return Response({"message": "You are unauthozed to perform this action"}, status=401) except: return Response({"message": "Error deleting company branch"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_super_hr_user(request): company_serial_number = request.data["serial_number"] email = request.data['username'] password = generate_random_string(6) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if staff_profile.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: serializers = UserSerializer( data={'username': email, 'password': password}) if serializers.is_valid(): user = serializers.save() user.is_staff = True user.save() new_user = authenticate(username=email, password=password) staff_position, created = staffPosition.objects.get_or_create(company_profile=company_profile, position_title="Head of Human Resource") main_branch = CompanyBranch.objects.get( company_profile=company_profile, main_branch=True) department = CompanyDepartment.objects.get( department_name="human_resource_management") # print(main_branch) # print(department) # generate staffnumber for the superhr time_stamp = datetime.now() numberReg = re.sub( r'[^0-9]', '', time_stamp.strftime('%M')) staff_number = "MEL"+f'{user.id}'+numberReg staff, created = StaffProfile.objects.get_or_create( user=new_user, staff_position=staff_position, email_address=new_user.username, staff_number=staff_number, company_branch=main_branch, company_department=department, has_read_write_priviledges=True, is_head_of_department=True, is_super_admin=True) # creates the system administrator role # print("running well") company_profile.company_super_hr_created = True company_profile.save() try: list_of_staff_ids = [] list_of_staff_ids.append(staff.id) message_title = "Welcome" message_body = f"You have been registered to {company_profile.company_name} as the head of human resource. Your staff number is {staff.staff_number}." create_messages(message_title, message_body, list_of_staff_ids) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "New User Added" notification_body = f"You have added user with email address {new_user.user_name} to {company_profile.company_name} ERP as the head of human resource." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # create default deduction schemes for the super hr # retrive paye paye_scheme = Deduction.objects.get( deduction_title="PAYE", company_profile=company_profile) sha_scheme = Deduction.objects.filter( company_profile=company_profile, deduction_title__in=["SHA", "SHIF"], recycle_bin=False, ).order_by("-id").first() nssf_scheme = Deduction.objects.get( deduction_title="NSSF", company_profile=company_profile) # paye paye_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': paye_scheme.id}) if paye_deduction_scheme_serializer.is_valid(): paye_deduction_scheme_serializer.save() if sha_scheme: sha_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': sha_scheme.id}) if sha_deduction_scheme_serializer.is_valid(): sha_deduction_scheme_serializer.save() # nssf nssf_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': nssf_scheme.id}) if nssf_deduction_scheme_serializer.is_valid(): nssf_deduction_scheme_serializer.save() message = f"You have successfully registered on Megawatt Energies ERP. Your staff number is {staff.staff_number}. Your autogenerated password is {password}. Once logged in, change your password." recipient_list = [f'{email}'] try: send_email_signup(request, message, recipient_list) except: pass return Response({"message": "Head of Human Resource account created successfully", }, status=200) else: return Response({"message": "Unable to create account for Head of Human Resource", }, status=406) else: return Response({"message": "You are unauthorized to perform this action", }, status=401) except: # Exception as e: # print(e) return Response({"message": "Error creating account for Head of Human Resource", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def create_other_staff_user(request): company_serial_number = request.data["serial_number"] email = request.data['username'] password = f'MEL@{generate_random_string(6)}' try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if staff_profile.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department == True and staff_profile.has_read_write_priviledges == True and company_profile: serializers = UserSerializer( data={'username': email, 'password': password}) if serializers.is_valid(): user = serializers.save() user.is_staff = True user.is_active = True user.set_password(password) user.save() new_user = authenticate(username=email, password=password) #print(new_user.username) staff_number = generateNextStaffNumber(company_profile.id) print(staff_number) staff, created = StaffProfile.objects.get_or_create( user=new_user, email_address=new_user.username, staff_number=staff_number) try: list_of_staff_ids = [] list_of_staff_ids.append(staff.id) message_title = "Welcome" message_body = f"You have been registered to {company_profile.company_name}. Your staff number is {staff.staff_number}." create_messages(message_title, message_body, list_of_staff_ids) except: pass try: list_of_staff_ids = [] list_of_staff_ids.append(staff_profile.id) notification_title = "New User Added" notification_body = f"You have added user with email address {new_user.user_name} to {company_profile.company_name} ERP." create_notifications(notification_title, notification_body, list_of_staff_ids) except: pass # create default deduction schemes for the super hr paye_scheme = Deduction.objects.get( deduction_title="PAYE", company_profile=company_profile) sha_scheme = Deduction.objects.filter( company_profile=company_profile, deduction_title__in=["SHA", "SHIF"], recycle_bin=False, ).order_by("-id").first() nssf_scheme = Deduction.objects.get( deduction_title="NSSF", company_profile=company_profile) paye_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': paye_scheme.id}) if paye_deduction_scheme_serializer.is_valid(): paye_deduction_scheme_serializer.save() if sha_scheme: sha_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': sha_scheme.id}) if sha_deduction_scheme_serializer.is_valid(): sha_deduction_scheme_serializer.save() nssf_deduction_scheme_serializer = StaffDeductionSchemeSerializer(data={ 'staff_profile': staff.id, 'deduction': nssf_scheme.id}) if nssf_deduction_scheme_serializer.is_valid(): nssf_deduction_scheme_serializer.save() message = f"You have successfully registered on Megawatt Energies ERP. Your staff number is {staff.staff_number}. Your autogenerated password is {password}. Once logged in, change your password." recipient_list = [f'{email}'] try: send_email_signup(request, message, recipient_list) except: pass return Response({"message": "User account created successfully", }, status=200) else: return Response({"message": "Unable to create user account", }, status=406) else: print("not running") return Response({"message": "You are unauthorized to perform this action", }, status=401) except Exception as e: print(e) return Response({"message": "Error creating user account", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_messages_and_notifications(request): date_format = '%d/%m/%Y, %H:%M' company_serial_number = str(request.data.get("serial_number", "")).strip() active_user = request.user payload = {} messages_list = [] notifications_list = [] target_timezone = pytz.timezone('Africa/Nairobi') try: if len(company_serial_number) == 0: return Response({"message": "serial_number is required", "payload": payload}, status=400) company_profile = CompanyProfile.objects.filter( company_serial_number=company_serial_number).first() if company_profile is None: return Response({"message": "Invalid company serial number", "payload": payload}, status=404) staff_profile = StaffProfile.objects.get(user=active_user) # all_notifications = MyNotification.objects.filter(addressed_to=staff_profile).order_by('-id')[:50] all_notifications = NotificationStaffReadStatus.objects.filter( staff=staff_profile).order_by('-id')[:50] all_messages = MessageStaffReadStatus.objects.filter( staff=staff_profile).order_by('-id')[:50] if company_profile: for notification in all_notifications: if notification.notification is None: continue notification_map = {} notification_map["notification_id"] = str(notification.id) notification_map["notification_title"] = notification.notification.notification_title notification_map["notification_body"] = notification.notification.notification_body notification_map["created_on"] = datetime.strftime( notification.notification.created_on.astimezone(target_timezone), date_format) if notification.notification.created_on is not None else "" notification_map["notification_read"] = "true" if notification.notification_read == True else "false" notifications_list.append(notification_map) for message in all_messages: if message.message is None: continue message_map = {} message_map["message_id"] = str(message.id) message_map["message_title"] = message.message.message_title message_map["message_body"] = message.message.message_body message_map["created_on"] = datetime.strftime( message.message.created_on.astimezone(target_timezone), date_format) if message.message.created_on is not None else "" message_map["message_read"] = "true" if message.message_read == True else "false" messages_list.append(message_map) payload["notifications_list"] = notifications_list payload["messages_list"] = messages_list return Response({"message": "true", "payload": payload}, status=200) else: return Response({"message": "false", "payload": payload}, status=401) except Exception as e: traceback.print_exc() print(e) return Response({"message": "Unable to load messages and notifications", "payload": payload}, status=500) def _parse_id_list(raw): """Normalise an optional id list that may arrive as a real list or a JSON string. Returns a list of ints. An empty/missing value yields [] (meaning "all records"). """ if raw is None: return [] if isinstance(raw, str): raw = raw.strip() if len(raw) == 0: return [] try: raw = json.loads(raw) except (ValueError, TypeError): return [] if not isinstance(raw, (list, tuple)): return [] parsed = [] for value in raw: try: parsed.append(int(value)) except (ValueError, TypeError): continue return parsed @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def mark_notifications_as_read(request): try: company_serial_number = str(request.data.get("serial_number", "")).strip() if len(company_serial_number) == 0: return Response({"message": "serial_number is required", }, status=400) company_profile = CompanyProfile.objects.filter( company_serial_number=company_serial_number).first() if company_profile is None: return Response({"message": "false", }, status=401) staff_profile = StaffProfile.objects.get(user=request.user) # Only operate on records belonging to the current user. read_status_qs = NotificationStaffReadStatus.objects.filter( staff=staff_profile) id_list = _parse_id_list( request.data.get("notification_staff_read_status_id_list")) if id_list: read_status_qs = read_status_qs.filter(id__in=id_list) read_status_qs.filter(notification_read=False).update( notification_read=True) return Response({"message": "true", }, status=200) except Exception as e: traceback.print_exc() print(e) return Response({"message": "false", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def mark_messages_as_read(request): try: company_serial_number = str(request.data.get("serial_number", "")).strip() if len(company_serial_number) == 0: return Response({"message": "serial_number is required", }, status=400) company_profile = CompanyProfile.objects.filter( company_serial_number=company_serial_number).first() if company_profile is None: return Response({"message": "false", }, status=401) staff_profile = StaffProfile.objects.get(user=request.user) # Only operate on records belonging to the current user. read_status_qs = MessageStaffReadStatus.objects.filter( staff=staff_profile) id_list = _parse_id_list( request.data.get("message_staff_read_status_id_list")) if id_list: read_status_qs = read_status_qs.filter(id__in=id_list) read_status_qs.filter(message_read=False).update(message_read=True) return Response({"message": "true", }, status=200) except Exception as e: traceback.print_exc() print(e) return Response({"message": "false", }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def change_password_for_user(request): date_format = '%d/%m/%Y' # active_user = request.user try: company_serial_number = request.data["serial_number"] staff_id = request.data["staff_id"] new_password = request.data["new_password"] active_user = StaffProfile.objects.get(id=int(staff_id)).user active_user.set_password(new_password) active_user.save() return Response({"message": "Password changed successfully", }, status=200) except Exception as e: print(e) return Response({"message": "Error changing password", }, status=500) def _require_system_admin(staff_profile, company_profile): return ( staff_profile.staff_position is not None and staff_profile.staff_position.position_title == "System Administrator" and staff_profile.is_head_of_department is True and staff_profile.has_read_write_priviledges is True and company_profile is not None ) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_company_email_settings(request): from system_administration.email_settings import email_settings_to_map, get_or_create_email_settings company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) settings_obj = get_or_create_email_settings(company_profile) return Response({ "message": "true", "payload": {"email_settings": email_settings_to_map(settings_obj)}, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error loading email settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def update_company_email_settings(request): from system_administration.email_settings import email_settings_to_map, get_or_create_email_settings company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) settings_obj = get_or_create_email_settings(company_profile) data = request.data for field in ( 'smtp_host', 'smtp_username', 'default_from_email', 'signup_from_email', 'agent_signup_from_email', 'marketing_from_email', 'error_notification_recipients', ): if field in data: setattr(settings_obj, field, data.get(field) or '') if 'smtp_port' in data: try: settings_obj.smtp_port = int(data.get('smtp_port') or 587) except (TypeError, ValueError): return Response({"message": "Invalid SMTP port"}, status=406) for flag in ( 'smtp_use_tls', 'smtp_use_ssl', 'enable_error_notifications', 'enable_general_notifications', 'enable_signup_notifications', 'enable_attachment_emails', 'enable_marketing_emails', ): if flag in data: setattr(settings_obj, flag, str(data.get(flag)).lower() in ('true', '1', 'yes', 'on')) if data.get('smtp_password'): settings_obj.smtp_password = data.get('smtp_password') settings_obj.save() return Response({ "message": "Email settings updated successfully", "payload": {"email_settings": email_settings_to_map(settings_obj)}, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error updating email settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def test_company_email_settings(request): from system_administration.email_settings import get_or_create_email_settings, send_test_email company_serial_number = request.data.get("serial_number") test_recipient = (request.data.get("test_recipient") or '').strip() try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) if not test_recipient: return Response({"message": "Test recipient email is required"}, status=406) settings_obj = get_or_create_email_settings(company_profile) send_test_email(settings_obj, test_recipient) return Response({"message": f"Test email sent to {test_recipient}"}, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": f"Test email failed: {e}"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_company_security_settings(request): from system_administration.security_settings import get_or_create_security_settings, security_settings_to_map company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) settings_obj = get_or_create_security_settings(company_profile) return Response({ "message": "true", "payload": {"security_settings": security_settings_to_map(settings_obj)}, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error loading security settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def update_company_security_settings(request): from system_administration.security_settings import get_or_create_security_settings, security_settings_to_map company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) settings_obj = get_or_create_security_settings(company_profile) data = request.data if 'allowed_ips' in data: settings_obj.allowed_ips = data.get('allowed_ips') or '' for flag in ('enable_ip_whitelist', 'enable_login_otp'): if flag in data: setattr(settings_obj, flag, str(data.get(flag)).lower() in ('true', '1', 'yes', 'on')) for field in ('otp_length', 'otp_ttl_minutes', 'otp_max_attempts'): if field in data: try: setattr(settings_obj, field, int(data.get(field))) except (TypeError, ValueError): return Response({"message": f"Invalid {field}"}, status=406) settings_obj.otp_length = max(4, min(int(settings_obj.otp_length or 6), 8)) settings_obj.otp_ttl_minutes = max(1, min(int(settings_obj.otp_ttl_minutes or 10), 60)) settings_obj.otp_max_attempts = max(1, min(int(settings_obj.otp_max_attempts or 5), 10)) settings_obj.save() return Response({ "message": "Security settings updated successfully", "payload": {"security_settings": security_settings_to_map(settings_obj)}, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error updating security settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def test_ip_whitelist(request): from system_administration.security_settings import ( client_ip_from_request, get_or_create_security_settings, is_ip_allowed_for_staff, parse_allowed_ips, ) company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) settings_obj = get_or_create_security_settings(company_profile) client_ip = client_ip_from_request(request) allowed = is_ip_allowed_for_staff(request, staff_profile, settings_obj) return Response({ "message": "true", "payload": { "client_ip": client_ip, "is_allowed": allowed, "whitelist_enabled": settings_obj.enable_ip_whitelist, "allowed_ip_count": len(parse_allowed_ips(settings_obj.allowed_ips or '')), }, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error testing IP whitelist"}, status=500) # db codes # with connection.cursor() as cursor: # cursor.execute( # "ALTER TABLE sales_and_marketing_salesquotation ADD COLUMN quotation_price_change_request_active BOOLEAN DEFAULT 0;") # ─── SMS SETTINGS (system administration) ─────────────────────────────────── from .models import CompanySmsSettings as _CompanySmsSettings @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_company_sms_settings(request): """Return current SMS settings. API key is masked.""" company_serial_number = request.data.get("serial_number", "") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) settings_obj = _CompanySmsSettings.objects.filter( company_profile=company_profile).first() if settings_obj: key = settings_obj.at_api_key or '' masked = (key[:4] + ('*' * max(0, len(key) - 8)) + key[-4:]) if len(key) > 8 else ('****' if key else '') return Response({ "message": "true", "payload": { "at_username": settings_obj.at_username, "at_api_key_masked": masked, "at_sender_id": settings_obj.at_sender_id, "configured": bool(settings_obj.at_username and settings_obj.at_api_key), }, }, status=200) return Response({ "message": "true", "payload": { "at_username": "", "at_api_key_masked": "", "at_sender_id": "", "configured": False, }, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error loading SMS settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def update_company_sms_settings(request): """Save Africa's Talking credentials for the company.""" company_serial_number = request.data.get("serial_number", "") at_username = (request.data.get("at_username") or "").strip() at_api_key = (request.data.get("at_api_key") or "").strip() at_sender_id = (request.data.get("at_sender_id") or "").strip() if not at_username or not at_api_key: return Response({"message": "Username and API key are required."}, status=400) try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) settings_obj, _ = _CompanySmsSettings.objects.get_or_create( company_profile=company_profile) settings_obj.at_username = at_username settings_obj.at_api_key = at_api_key settings_obj.at_sender_id = at_sender_id settings_obj.save() return Response({ "message": "true", "payload": "SMS settings saved successfully.", }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error saving SMS settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def check_company_sms_balance(request): """SMS balance check is unavailable — marketing SMS module removed.""" try: return Response({ "message": "false", "error": "SMS balance check is not available.", }, status=200) except Exception as e: print(e) return Response({"message": "Error checking SMS balance"}, status=500) # --------------------------------------------------------------------------- # TECHNICAL INTEGRATION SETTINGS (M-Pesa Daraja, etc.) — System Admin owned # --------------------------------------------------------------------------- @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_company_integration_settings(request): from system_administration.integration_settings import ( branch_gl_account_options, crm_integration_to_map, get_or_create_crm_integration, get_or_create_mpesa_config, mpesa_config_to_map, ) company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) branch = staff_profile.company_branch cfg = get_or_create_mpesa_config(branch) crm = get_or_create_crm_integration(company_profile) return Response({ "message": "true", "payload": { "mpesa": mpesa_config_to_map(cfg), "gl_accounts": branch_gl_account_options(branch), "crm": crm_integration_to_map(crm), }, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except StaffProfile.DoesNotExist: return Response({"message": "Staff profile not found for this user"}, status=404) except Exception as e: print(e) traceback.print_exc() return Response({ "message": "Error loading integration settings", "error": str(e), }, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def update_crm_integration_settings(request): from system_administration.integration_settings import ( apply_crm_integration_update, crm_integration_to_map, get_or_create_crm_integration, ) company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) obj = get_or_create_crm_integration(company_profile) obj = apply_crm_integration_update(obj, request.data) return Response({ "message": "true", "payload": {"crm": crm_integration_to_map(obj)}, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error saving CRM integration settings"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def generate_crm_webhook_secret(request): """Generate a strong shared secret for the CRM webhooks, store it, and return it ONCE so the admin can copy it into the CRM. Subsequent reads never expose it again (only a `*_set` boolean).""" import secrets from system_administration.integration_settings import ( crm_integration_to_map, get_or_create_crm_integration, ) company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) obj = get_or_create_crm_integration(company_profile) new_secret = secrets.token_urlsafe(36) obj.crm_webhook_secret = new_secret obj.save() return Response({ "message": "true", "payload": { # Plaintext returned only on generation — copy it into the CRM now. "crm_webhook_secret": new_secret, "crm": crm_integration_to_map(obj), }, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error generating CRM webhook secret"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def update_mpesa_api_config(request): from system_administration.integration_settings import ( apply_mpesa_config_update, get_or_create_mpesa_config, mpesa_config_to_map, ) company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) cfg = get_or_create_mpesa_config(staff_profile.company_branch) cfg = apply_mpesa_config_update(cfg, request.data, staff_profile) return Response({ "message": "true", "payload": {"mpesa": mpesa_config_to_map(cfg)}, }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "Error saving M-Pesa configuration"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def test_mpesa_api_config(request): from system_administration.integration_settings import get_or_create_mpesa_config company_serial_number = request.data.get("serial_number") try: company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return Response({"message": "You are unauthorised to perform this function"}, status=401) from finance_and_accounting.mpesa_daraja import test_connection cfg = get_or_create_mpesa_config(staff_profile.company_branch) result = test_connection(cfg) return Response({"message": "true", "payload": result}, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) return Response({"message": "false", "payload": {"connected": False, "error": str(e)}}, status=200) # --------------------------------------------------------------------------- # DATABASE MIGRATIONS — System Admin only # --------------------------------------------------------------------------- def _migration_admin_response(request, company_serial_number): company_profile = CompanyProfile.objects.get( company_serial_number=company_serial_number) staff_profile = StaffProfile.objects.get(user=request.user) if not _require_system_admin(staff_profile, company_profile): return None, Response( {"message": "You are unauthorised to perform this function"}, status=401, ) return company_profile, None @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def get_database_migration_status(request): from system_administration.migration_service import get_migration_status company_serial_number = request.data.get("serial_number") try: _, denied = _migration_admin_response(request, company_serial_number) if denied is not None: return denied return Response({ "message": "true", "payload": get_migration_status(), }, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except Exception as e: print(e) traceback.print_exc() return Response({"message": "Error checking migration status"}, status=500) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def run_database_migrations(request): from system_administration.migration_service import run_pending_migrations company_serial_number = request.data.get("serial_number") try: _, denied = _migration_admin_response(request, company_serial_number) if denied is not None: return denied payload = run_pending_migrations() return Response({"message": "true", "payload": payload}, status=200) except CompanyProfile.DoesNotExist: return Response({"message": "Company not found"}, status=404) except RuntimeError as e: return Response({"message": "false", "error": str(e)}, status=500) except Exception as e: print(e) traceback.print_exc() return Response({"message": "Error running migrations", "error": str(e)}, status=500)