import datetime import io import json import openpyxl from markupsafe import escape from odoo import http from odoo.exceptions import AccessError, ValidationError from odoo.http import content_disposition, request HR_OFFICER_GROUP = "hr_holidays.group_hr_holidays_user" # Business-friendly Leave Request import, ported from XENHR's own # LeaveApplicationsImport.php: employee/leave-type resolved by their # human-facing codes (not internal database ids), row-by-row validation # that skips-and-reports a bad row instead of failing the whole file, and # the ability to backfill historical leaves already in a decided state # (approved/rejected/cancelled), not just pending ones. Column names and # validation rules mirror the Laravel version's import/template classes # 1:1 so this deployment's HR staff (already trained on that file format) # can reuse the same spreadsheets without learning a new one. _STATUS_TO_ODOO_STATE = { "pending": "confirm", "approved": "validate", "rejected": "refuse", "cancelled": "cancel", } _VALID_REQUEST_UNITS = {"full_day", "part_time", "first_half", "second_half"} _TEMPLATE_HEADERS = [ "employee_name", "leave_type_code", "leave_base_type", "start_date", "end_date", "part_time_start", "part_time_end", "reason", "status", "reviewer_note", ] def _leave_type_code(leave_type): """This deployment names leave types "Full Name . CODE" (e.g. "Annual Leave . AL") -- the trailing segment after the last '.' is the code, same convention leave_type_colors.js's leaveTypeAbbr() uses client-side for display; this is the server-side equivalent, used for lookup.""" name = leave_type.name or "" if "." not in name: return name.strip().upper() return name.rsplit(".", 1)[1].strip().upper() class XenoLeaveImport(http.Controller): @http.route("/xeno_leave/leave_import/template", type="http", auth="user") def leave_import_template(self): if not request.env.user.has_group(HR_OFFICER_GROUP): raise AccessError("Only Time Off Officers can use the Leave Request import.") leave_types = request.env["hr.leave.type"].search([]) workbook = openpyxl.Workbook() sheet = workbook.active sheet.title = "Leave Requests" 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") 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", "full_day", "6/1/2026", "8/1/2026", "", "", "Sample leave reason", "pending", "", ]) notes_row = 5 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)", "leave_base_type: full_day | part_time | first_half | second_half", "part_time_start / part_time_end: HH:MM, required only when leave_base_type is part_time", "start_date / end_date: format D/M/YYYY, e.g. 5/1/2026 = 5 January 2026", "status: pending | approved | rejected | cancelled", "reviewer_note: optional", "A row overlapping an employee's existing leave will be skipped with an error.", ] 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("ABCDEFGHIJ", [14, 15, 14, 12, 12, 15, 13, 30, 10, 20]): 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_Request_Import_Template.xlsx")), ], ) @http.route("/xeno_leave/leave_import/upload", type="http", auth="user", methods=["POST"]) def leave_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 Request 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}) 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}) col_idx = {name: i for i, name in enumerate(header)} missing = [c for c in ("employee_name", "leave_type_code", "start_date", "end_date") if c not in col_idx] if missing: return self._json_response({ "errors": [f"Missing required column(s): {', '.join(missing)}"], "imported": 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 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". Employee = request.env["hr.employee"].sudo().with_context(active_test=False) LeaveType = request.env["hr.leave.type"].sudo() Leave = request.env["hr.leave"].sudo() leave_types = LeaveType.search([]) type_by_code = {_leave_type_code(lt): lt for lt in leave_types} imported = 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() base_type = str(cell(row, "leave_base_type") or "full_day").strip().lower() start_date = self._to_date_str(cell(row, "start_date")) end_date = self._to_date_str(cell(row, "end_date")) reason = str(cell(row, "reason")).strip() status = str(cell(row, "status") or "pending").strip().lower() reviewer_note = str(cell(row, "reviewer_note")).strip() pt_start = str(cell(row, "part_time_start")).strip() pt_end = str(cell(row, "part_time_end")).strip() 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 if not start_date or not end_date: errors.append(f"Row {row_num}: start_date and end_date must be valid dates (D/M/YYYY).") continue if end_date < start_date: errors.append(f"Row {row_num}: end_date must be on or after start_date.") continue if status not in _STATUS_TO_ODOO_STATE: errors.append(f"Row {row_num}: status must be one of pending/approved/rejected/cancelled.") continue if base_type not in _VALID_REQUEST_UNITS: errors.append(f"Row {row_num}: leave_base_type '{base_type}' is not recognized.") 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 if base_type != "full_day" and not leave_type.xeno_flexible_duration: errors.append( f"Row {row_num}: '{leave_type.name}' doesn't support Half-Day/Part-time " f"(enable Flexible Duration on the leave type first).") continue vals = { "employee_id": employee.id, "holiday_status_id": leave_type.id, "xeno_request_unit": base_type, "request_date_from": start_date, "request_date_to": end_date, "name": reason or False, } if base_type == "part_time": if not pt_start or not pt_end: errors.append(f"Row {row_num}: part_time_start and part_time_end are " f"required for part_time leave.") continue vals["request_hour_from"] = self._to_hour_float(pt_start) vals["request_hour_to"] = self._to_hour_float(pt_end) # A constraint raised mid-row (overlap, part-time-hours-out-of- # range, etc.) leaves the surrounding transaction unusable for # any further query until rolled back -- a savepoint per row # means only THIS row's work is undone, not the whole batch # processed so far, matching how core's own base_import module # isolates per-row failures. try: with request.env.cr.savepoint(): # xeno_hr_direct_create already skips building/emailing # an approval chain and the approve/reject bypass # helpers below skip their own final-notice email -- # tracking_disable is the extra belt-and-suspenders # layer covering everything else mail.thread could # still notify on (the reviewer_note message_post # below, any chatter tracking messages), so a bulk # historical import can never email anyone regardless # of a follower's own notification preference. leave = Leave.with_context( xeno_hr_direct_create=True, tracking_disable=True, ).create(vals) odoo_state = _STATUS_TO_ODOO_STATE[status] if odoo_state == "validate": leave._xeno_hr_created_approve() elif odoo_state == "refuse": leave._xeno_hr_created_reject() elif odoo_state == "cancel": # action_cancel() is NOT a state-transition method -- # it only returns an ir.actions.act_window dict that # opens the "Cancel Time Off" wizard for a human to # confirm in the UI; calling it directly does # nothing (confirmed live: state stayed unchanged). # The wizard's own confirm button calls # _action_user_cancel(), which itself requires # can_cancel (computed for the ACTING user, not # appropriate here since HR is cancelling on # someone else's behalf) before delegating to # _force_cancel() -- so call _force_cancel() # directly, same "skip the human-facing gate, keep # the real side effects" pattern as # _xeno_hr_created_approve() above. No reason means # its own notify-responsibles block is skipped too. leave._force_cancel(notify_responsibles=False) if reviewer_note: leave.message_post(body=escape(reviewer_note)) 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 imported += 1 return self._json_response({"imported": imported, "errors": errors}) @staticmethod def _to_date_str(value): # Template columns are documented/shown as D/M/YYYY (e.g. "5/1/2026" # = 5 January 2026); %d/%m/%Y also happily accepts zero-padded # input ("05/01/2026"). ISO YYYY-MM-DD is still accepted as a # fallback for robustness (e.g. a re-exported file, or a value # someone typed the old way) -- returned/stored internally as ISO # either way, since that's what hr.leave's own Date fields expect. if isinstance(value, (datetime.date, datetime.datetime)): return value.strftime("%Y-%m-%d") value = str(value).strip() if not value: return None for fmt in ("%d/%m/%Y", "%Y-%m-%d"): try: return datetime.datetime.strptime(value, fmt).strftime("%Y-%m-%d") except ValueError: continue return None @staticmethod def _to_hour_float(value): if isinstance(value, datetime.time): return value.hour + value.minute / 60.0 parts = str(value).strip().split(":") h = int(parts[0]) if parts and parts[0] else 0 m = int(parts[1]) if len(parts) > 1 and parts[1] else 0 return h + m / 60.0 @staticmethod def _json_response(payload): return request.make_response( json.dumps(payload), headers=[("Content-Type", "application/json")])