import io import json import openpyxl from odoo import http from odoo.exceptions import AccessError, ValidationError from odoo.http import content_disposition, request from .leave_import import HR_OFFICER_GROUP, _leave_type_code # Business-friendly Leave Balance import, ported from XENHR's own # LeaveBalancesImport.php: employee/leave-type resolved by their # human-facing names/codes, one row per employee+leave-type+year, upserted # (existing balance for that year is revised in place, same as the Laravel # version's update-vs-create branch) rather than always creating a new # allocation. carried_days from the reference file has no equivalent here -- # this deployment doesn't model a carry-forward balance (see leave_type # simplification decision) -- so the template only carries the columns that # map onto a real field: employee_name, leave_type_code, year, allocated_days. _TEMPLATE_HEADERS = ["employee_name", "leave_type_code", "year", "allocated_days"] class XenoLeaveBalanceImport(http.Controller): @http.route("/xeno_leave/leave_balance_import/template", type="http", auth="user") def leave_balance_import_template(self): if not request.env.user.has_group(HR_OFFICER_GROUP): raise AccessError("Only Time Off Officers can use the Leave Balance import.") leave_types = request.env["hr.leave.type"].search([]) workbook = openpyxl.Workbook() sheet = workbook.active sheet.title = "Leave Balances" sheet.append(_TEMPLATE_HEADERS) for cell in sheet[1]: cell.font = openpyxl.styles.Font(bold=True, color="FFFFFF") cell.fill = openpyxl.styles.PatternFill("solid", fgColor="2563EB") from datetime import date example_employee = request.env["hr.employee"].search([("active", "=", True)], limit=1) example_type = leave_types[:1] sheet.append([ example_employee.name if example_employee else "John Smith", _leave_type_code(example_type) if example_type else "AL", date.today().year, 10, ]) notes_row = 4 sheet.cell(row=notes_row, column=1, value="NOTES:").font = openpyxl.styles.Font(bold=True) notes = [ "employee_name: must match an existing employee's full Name exactly " "(case-insensitive); if two employees share a name, the row will be " "skipped and reported so you can tell them apart", "leave_type_code: the code after the dot in the leave type name (e.g. AL, SL)", "year: the calendar year this balance applies to (e.g. 2026)", "allocated_days: the total days granted for that employee/type/year", "If a balance already exists for the same employee + leave type + " "year, its allocated_days is revised instead of creating a duplicate.", ] for i, note in enumerate(notes): sheet.cell(row=notes_row + i, column=2, value=note) codes = ", ".join(f"{_leave_type_code(lt)} ({lt.name})" for lt in leave_types) sheet.cell(row=notes_row + len(notes) + 1, column=2, value=f"Available leave type codes: {codes}") for col, width in zip("ABCD", [22, 15, 8, 16]): sheet.column_dimensions[col].width = width output = io.BytesIO() workbook.save(output) output.seek(0) return request.make_response( output.read(), headers=[ ("Content-Type", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"), ("Content-Disposition", content_disposition("Leave_Balance_Import_Template.xlsx")), ], ) @http.route("/xeno_leave/leave_balance_import/upload", type="http", auth="user", methods=["POST"]) def leave_balance_import_upload(self, ufile, **kwargs): if not request.env.user.has_group(HR_OFFICER_GROUP): raise AccessError("Only Time Off Officers can use the Leave Balance import.") try: workbook = openpyxl.load_workbook(ufile, data_only=True) except Exception: return self._json_response({"errors": ["Could not read this file -- please upload a .xlsx file."], "imported": 0, "updated": 0}) sheet = workbook.active rows_iter = sheet.iter_rows(values_only=True) try: header = [str(h or "").strip() for h in next(rows_iter)] except StopIteration: return self._json_response({"errors": ["The file is empty."], "imported": 0, "updated": 0}) col_idx = {name: i for i, name in enumerate(header)} missing = [c for c in _TEMPLATE_HEADERS if c not in col_idx] if missing: return self._json_response({ "errors": [f"Missing required column(s): {', '.join(missing)}"], "imported": 0, "updated": 0}) def cell(row, name, default=""): i = col_idx.get(name) if i is None or i >= len(row) or row[i] is None: return default return row[i] # Historical/bulk backfill routinely covers employees who've since # left (now archived) -- active_test=False so they're still found by # name instead of failing as "not found" (same reasoning as # leave_import.py's own employee resolution). Employee = request.env["hr.employee"].sudo().with_context(active_test=False) LeaveType = request.env["hr.leave.type"].sudo() Allocation = request.env["hr.leave.allocation"].sudo() leave_types = LeaveType.search([]) type_by_code = {_leave_type_code(lt): lt for lt in leave_types} imported = 0 updated = 0 errors = [] for row_num, row in enumerate(rows_iter, start=2): if row is None or all(v is None for v in row): continue emp_name = str(cell(row, "employee_name")).strip() type_code = str(cell(row, "leave_type_code")).strip().upper() year_raw = cell(row, "year") days_raw = cell(row, "allocated_days") if not emp_name: errors.append(f"Row {row_num}: employee_name is required.") continue if not type_code: errors.append(f"Row {row_num}: leave_type_code is required.") continue try: year = int(year_raw) if year < 2020 or year > 2100: raise ValueError except (TypeError, ValueError): errors.append(f"Row {row_num}: year must be a valid year (e.g. 2026).") continue try: allocated_days = float(days_raw) if allocated_days < 0: raise ValueError except (TypeError, ValueError): errors.append(f"Row {row_num}: allocated_days must be a non-negative number.") continue matches = Employee.search([("name", "=ilike", emp_name)]) if not matches: errors.append(f"Row {row_num}: Employee '{emp_name}' not found.") continue if len(matches) > 1: codes = ", ".join(m.xeno_employee_code or f"id {m.id}" for m in matches) errors.append( f"Row {row_num}: '{emp_name}' matches more than one employee " f"({codes}) -- can't tell which one, skipped.") continue employee = matches leave_type = type_by_code.get(type_code) if not leave_type: errors.append(f"Row {row_num}: Leave type code '{type_code}' not found.") continue date_from = f"{year}-01-01" date_to = f"{year}-12-31" try: with request.env.cr.savepoint(): existing = Allocation.search([ ("employee_id", "=", employee.id), ("holiday_status_id", "=", leave_type.id), ("date_from", "=", date_from), ], limit=1) if existing: # Core blocks reducing number_of_days below days # already taken (raises ValidationError) -- caught # below like any other per-row failure. existing.write({"number_of_days": allocated_days}) updated += 1 else: allocation = Allocation.with_context( import_file=True, mail_create_nosubscribe=True, tracking_disable=True, ).create({ "employee_id": employee.id, "holiday_status_id": leave_type.id, "date_from": date_from, "date_to": date_to, "number_of_days": allocated_days, "name": f"{leave_type.name} ({allocated_days} day(s))", }) # create() only allows state="confirm" -- approve # separately. .sudo() makes _check_approval_update's # is_superuser() check pass regardless of who's # actually running the import, so this never depends # on the acting HR user being a configured approver # for this specific employee. Neither action_approve # nor the create() path above sends any email -- # confirmed against core source, unlike hr.leave's # own approve/refuse. allocation.sudo().action_approve() imported += 1 except ValidationError as e: errors.append(f"Row {row_num}: {e.args[0] if e.args else e}") continue except Exception as e: # noqa: BLE001 -- surfaced per-row, not fatal to the batch errors.append(f"Row {row_num}: {e}") continue return self._json_response({"imported": imported, "updated": updated, "errors": errors}) @staticmethod def _json_response(payload): return request.make_response( json.dumps(payload), headers=[("Content-Type", "application/json")])