Tweaks for the new file format

payroll/services/ingest.py

@@ -1,5 +1,4 @@
 import csv
-import random
 from collections import namedtuple
 
 from django.core.files import File
@@ -57,66 +56,42 @@
 PayrollRow = namedtuple(
     "PayrollRow",
     (
-        (
-            "col_a",
-            "col_b",
-            "col_c",
-            "col_d",
-            "col_e",
-            "employee_no",
-            "col_g",
-            "col_h",
-            "col_i",
-            "col_j",
-            "col_k",
-            "col_l",
-            "col_m",
-            "col_n",
-            "col_o",
-            "col_p",
-            "col_q",
-            "pay_type",
-            "col_s",
-            "debit_amount",
-            "credit_amount",
-        )
+        "employee_no",
+        "first_name",
+        "last_name",
+        "cost_centre",
+        "programme_code",
+        "grade",
+        "assignment_status",
+        "fte",
+        "basic_pay",
+        "ernic",
+        "pension",
     ),
 )
 
 
 @transaction.atomic()
-def import_payroll(
-    hr_csv: File,
-    payroll_csv: File | None,
-    hr_csv_has_header: bool,
-    payroll_csv_has_header: bool,
-) -> str:
-    # payroll_data=[]
-    # payrol_csv_reader= csv.reader((row.decode("utf-8") for row in payroll_csv))
-    # if payroll_csv_has_header:
-    #     next(payrol_csv_reader)
+def import_payroll(payroll_csv: File) -> str:
+    csv_reader = csv.reader((row.decode("utf-8") for row in payroll_csv))
 
-    # for payroll_row in payrol_csv_reader:
-    #     payroll_data.append(map_payroll(PayrollRow(**payroll_row)))
-
-    hr_csv_reader = csv.reader((row.decode("utf-8") for row in hr_csv))
-
-    if hr_csv_has_header:
-        next(hr_csv_reader)
+    # Skip header row.
+    next(csv_reader)
 
     employees = []
     cost_centres = []
     programme_codes = []
     grades = []
-    for hr_row in hr_csv_reader:
-        employee = hr_row_to_employee(HrRow(*hr_row))
+    for hr_row in csv_reader:
+        employee = hr_row_to_employee(PayrollRow(*hr_row))
         employees.append(employee)
         cost_centres.append(employee["cost_centre"])
         programme_codes.append(employee["programme_code"])
         grades.append(employee["grade"])
-        uniq_cost_centre_codes = set(cost_centres)
-        uniq_grades = set(grades)
-        uniq_programme_codes = set(programme_codes)
+
+    uniq_cost_centre_codes = set(cost_centres)
+    uniq_grades = set(grades)
+    uniq_programme_codes = set(programme_codes)
     cost_centers = {
         centre.cost_centre_code: centre
         for centre in CostCentre.objects.filter(
@@ -163,19 +138,19 @@ def import_payroll(
     return {"failed_records": failed_records, **result}
 
 
-def hr_row_to_employee(hr_row) -> Employee:
+def hr_row_to_employee(hr_row) -> dict[str, object]:
     employee = {
         "employee_no": hr_row.employee_no,
         "first_name": hr_row.first_name,
         "last_name": hr_row.last_name,
-        "cost_centre": hr_row.cost_centre_id,
-        "programme_code": hr_row.programme_code_id,
-        "grade": hr_row.grade_id,
+        "cost_centre": hr_row.cost_centre,
+        "programme_code": hr_row.programme_code,
+        "grade": hr_row.grade,
         "assignment_status": hr_row.assignment_status,
         "fte": hr_row.fte,
-        "basic_pay": random.randint(100000, 999999),
-        "ernic": random.randint(100000, 999999),
-        "pension": random.randint(100000, 999999),
+        "basic_pay": hr_row.basic_pay,
+        "ernic": hr_row.ernic,
+        "pension": hr_row.pension,
         "has_left": False,
     }
     return employee
@@ -218,12 +193,3 @@ def bulk_update_or_create(data):
         Employee.objects.bulk_update(to_update, keys)
 
     return {"created": to_create, "updated": to_update}
-
-
-# def map_payroll(row):
-#     return {
-#         "employee_no":row.employee_no,
-#         "pay_type":row.pay_type,
-#         "debit_amount":row.debit_amount,
-#         "credit_amount":row.credit_amount
-#     }

payroll/templates/payroll/page/import_payroll.html

@@ -12,22 +12,10 @@
     <h1 class="govuk-heading-l">Import payroll</h1>
     <form method="POST" enctype="multipart/form-data">
         {% csrf_token %}
-        <div>
-            <label for="hr_csv">HR CSV file:</label>
-            <input type="file" name="hr_csv" id="hr_csv">
-        </div>
-        <div>
-            <label for="hr_csv">Has header?:</label>
-            <input type="checkbox" name="hr_csv_has_header" id="hr_csv_has_header" checked>
-        </div>
         <div>
             <label for="payroll_csv">Payroll CSV file:</label>
             <input type="file" name="payroll_csv" id="payroll_csv">
         </div>
-        <div>
-            <label for="payroll_csv">Has header?:</label>
-            <input type="checkbox" name="payroll_csv_has_header" id="payroll_csv_has_header" checked>
-        </div>
         <button type="submit">Submit</button>
     </form>
     <div>

payroll/views.py

@@ -192,19 +192,16 @@ def import_payroll_page(request: HttpRequest) -> HttpResponse:
 
     output = ""
     context = {}
+
     if request.method == "POST":
-        if "hr_csv" not in request.FILES or "payroll_csv" not in request.FILES:
-            context = {"error": "Both HR and Payroll files are required"}
+        if "payroll_csv" not in request.FILES:
+            context = {"error": "Payroll file is required"}
         else:
-            hr_csv = request.FILES["hr_csv"]
-            hr_csv_has_header = request.POST.get("hr_csv_has_header", False)
             payroll_csv = request.FILES["payroll_csv"]
-            payroll_csv_has_header = request.POST.get("hr_csv_has_header", False)
-            output = import_payroll(
-                hr_csv, payroll_csv, hr_csv_has_header, payroll_csv_has_header
-            )
-
-            context = {
-                "output": output,
-            }
+            output = import_payroll(payroll_csv)
+
+    context = {
+        "output": output,
+    }
+
     return TemplateResponse(request, "payroll/page/import_payroll.html", context)