# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
import logging
import re
from datetime import timedelta
from odoo import api, fields, models, _
from odoo.exceptions import ValidationError
_logger = logging.getLogger(__name__)
class FusionClockSchedule(models.Model):
_name = 'fusion.clock.schedule'
_description = 'Clock Shift Schedule Entry'
_order = 'schedule_date, employee_id'
_rec_name = 'display_name'
employee_id = fields.Many2one(
'hr.employee',
string='Employee',
required=False, # open (unassigned) shifts have no employee until claimed
index=True,
ondelete='cascade',
)
is_open = fields.Boolean(
string='Open Shift',
default=False,
index=True,
help="An unassigned shift any eligible employee can claim from the portal.",
)
schedule_date = fields.Date(
string='Date',
required=True,
index=True,
)
shift_id = fields.Many2one(
'fusion.clock.shift',
string='Shift Template',
ondelete='set null',
)
is_off = fields.Boolean(
string='Off',
default=False,
index=True,
)
start_time = fields.Float(
string='Start Time',
default=9.0,
)
end_time = fields.Float(
string='End Time',
default=17.0,
)
break_minutes = fields.Float(
string='Break (min)',
default=30.0,
)
planned_hours = fields.Float(
string='Hours',
compute='_compute_planned_hours',
store=True,
)
crosses_midnight = fields.Boolean(
string='Overnight',
compute='_compute_planned_hours',
store=True,
help="Set automatically when the shift ends on the next day "
"(end time on or before start time).",
)
note = fields.Char(string='Note')
role_id = fields.Many2one(
'fusion.clock.role',
string='Role',
help="Shift role — drives the colour/label shown on the employee's "
"portal schedule. Defaults from the shift template or the "
"employee's Default Shift Role.",
)
recurrence_id = fields.Many2one(
'fusion.clock.schedule.recurrence',
string='Recurrence',
ondelete='set null',
index=True,
help="Set when this entry was generated by a recurring rule.",
)
company_id = fields.Many2one(
'res.company',
string='Company',
compute='_compute_fclk_company',
store=True,
readonly=False,
index=True,
)
department_id = fields.Many2one(
'hr.department',
string='Department',
related='employee_id.department_id',
store=True,
readonly=True,
)
display_name = fields.Char(
compute='_compute_display_name',
store=True,
)
state = fields.Selection(
[('draft', 'Draft'), ('posted', 'Posted')],
string='Status',
default='draft',
index=True,
help="Only POSTED entries drive reminders, absence checks and penalties. "
"Draft entries are ignored by automation until the team lead posts them.",
)
posted_date = fields.Datetime(string='Posted On', readonly=True)
# No hard UNIQUE(employee, date): the per-day model now allows split shifts
# and open (unassigned) shifts. The shift planner still manages one cell per
# day in place; the attendance contract (_get_fclk_day_plan) resolves
# multiple posted rows into a single work-window.
@api.depends('employee_id')
def _compute_fclk_company(self):
for rec in self:
if rec.employee_id:
rec.company_id = rec.employee_id.company_id
elif not rec.company_id:
rec.company_id = self.env.company
@api.constrains('employee_id', 'is_open')
def _check_employee_or_open(self):
for rec in self:
if not rec.employee_id and not rec.is_open:
raise ValidationError(
_("A shift must have an employee unless it is an open shift."))
@api.depends('is_off', 'start_time', 'end_time', 'break_minutes')
def _compute_planned_hours(self):
for rec in self:
rec.crosses_midnight = False
if rec.is_off:
rec.planned_hours = 0.0
continue
start = rec.start_time or 0.0
end = rec.end_time or 0.0
if end <= start:
# Overnight: the shift ends on the following day.
rec.crosses_midnight = True
raw_hours = (24.0 - start) + end
else:
raw_hours = end - start
rec.planned_hours = round(max(raw_hours - ((rec.break_minutes or 0.0) / 60.0), 0.0), 2)
@api.depends('employee_id', 'schedule_date', 'is_off', 'start_time', 'end_time')
def _compute_display_name(self):
for rec in self:
emp = rec.employee_id.name or ''
date_str = str(rec.schedule_date) if rec.schedule_date else ''
rec.display_name = f"{emp} - {date_str} - {rec.fclk_display_value()}"
@api.constrains('is_off', 'start_time', 'end_time', 'break_minutes')
def _check_schedule_times(self):
for rec in self:
if rec.break_minutes < 0:
raise ValidationError(_("Break minutes cannot be negative."))
if rec.is_off:
continue
if rec.start_time < 0 or rec.start_time >= 24:
raise ValidationError(_("Start time must be between 00:00 and 23:59."))
if rec.end_time < 0 or rec.end_time > 24:
raise ValidationError(_("End time must be between 00:00 and 24:00."))
# Overnight shifts (end on/before start) are allowed and span midnight.
if rec.end_time <= rec.start_time:
shift_minutes = ((24.0 - rec.start_time) + rec.end_time) * 60.0
else:
shift_minutes = (rec.end_time - rec.start_time) * 60.0
if rec.break_minutes >= shift_minutes:
raise ValidationError(_("Break duration must be shorter than the scheduled shift."))
@api.onchange('shift_id')
def _onchange_shift_id(self):
for rec in self:
if rec.shift_id:
rec.is_off = False
rec.start_time = rec.shift_id.start_time
rec.end_time = rec.shift_id.end_time
rec.break_minutes = rec.shift_id.break_minutes
@api.model
def fclk_float_to_display(self, value):
value = float(value or 0.0)
hour = int(value)
minute = int(round((value - hour) * 60))
if minute == 60:
hour += 1
minute = 0
suffix = 'am' if hour < 12 or hour == 24 else 'pm'
display_hour = hour % 12
if display_hour == 0:
display_hour = 12
return f"{display_hour}:{minute:02d} {suffix}"
def fclk_display_value(self):
self.ensure_one()
if self.is_off:
return 'OFF'
return (
f"{self.env['fusion.clock.schedule'].fclk_float_to_display(self.start_time)} - "
f"{self.env['fusion.clock.schedule'].fclk_float_to_display(self.end_time)}"
)
@api.model
def fclk_hours_display(self, hours):
hours = float(hours or 0.0)
whole = int(hours)
minutes = int(round((hours - whole) * 60))
if minutes == 60:
whole += 1
minutes = 0
return f"{whole}:{minutes:02d}"
@api.model
def _fclk_parse_time_part(self, raw):
text = (raw or '').strip().lower().replace('.', '')
match = re.match(r'^(\d{1,2})(?::(\d{1,2}))?\s*(am|pm)?$', text)
if not match:
raise ValidationError(_("Could not understand time '%s'.") % raw)
hour = int(match.group(1))
minute = int(match.group(2) or 0)
meridiem = match.group(3)
if minute < 0 or minute > 59:
raise ValidationError(_("Minutes must be between 00 and 59."))
if meridiem:
if hour < 1 or hour > 12:
raise ValidationError(_("12-hour times must use hours from 1 to 12."))
if meridiem == 'am':
hour = 0 if hour == 12 else hour
else:
hour = 12 if hour == 12 else hour + 12
elif hour > 24:
raise ValidationError(_("Hours must be between 0 and 24."))
return hour + (minute / 60.0)
@api.model
def fclk_parse_planner_input(self, input_value, default_break_minutes=30.0):
text = (input_value or '').strip()
if not text:
return {'clear': True}
if text.upper() == 'OFF':
return {
'clear': False,
'is_off': True,
'shift_id': False,
'start_time': 0.0,
'end_time': 0.0,
'break_minutes': 0.0,
}
normalized = (
text.replace('–', '-')
.replace('—', '-')
.replace(' to ', '-')
.replace(' TO ', '-')
)
parts = [p.strip() for p in normalized.split('-', 1)]
if len(parts) != 2 or not parts[0] or not parts[1]:
raise ValidationError(_("Enter a shift as '9-5', '9:00-5:30', '9:00 am - 5:30 pm', or OFF."))
start = self._fclk_parse_time_part(parts[0])
end = self._fclk_parse_time_part(parts[1])
if end <= start and end + 12 <= 24:
end += 12
if end <= start:
raise ValidationError(_("End time must be after start time. Overnight shifts are not supported yet."))
return {
'clear': False,
'is_off': False,
'shift_id': False,
'start_time': start,
'end_time': end,
'break_minutes': float(default_break_minutes or 0.0),
}
@api.model
def fclk_values_from_planner_payload(self, payload, employee):
payload = payload or {}
if 'start_time' in payload and 'end_time' in payload and not payload.get('shift_id'):
if payload.get('is_off'):
return {
'clear': False,
'is_off': True,
'shift_id': False,
'start_time': 0.0,
'end_time': 0.0,
'break_minutes': 0.0,
}
return {
'clear': False,
'is_off': False,
'shift_id': False,
'start_time': float(payload.get('start_time') or 0.0),
'end_time': float(payload.get('end_time') or 0.0),
'break_minutes': float(payload.get('break_minutes') or 0.0),
}
shift_id = int(payload.get('shift_id') or 0)
if shift_id:
shift = self.env['fusion.clock.shift'].sudo().browse(shift_id)
if not shift.exists():
raise ValidationError(_("Selected shift template no longer exists."))
return {
'clear': False,
'shift_id': shift.id,
'is_off': False,
'start_time': shift.start_time,
'end_time': shift.end_time,
'break_minutes': shift.break_minutes,
}
default_break = employee._get_fclk_break_minutes() if employee else 30.0
return self.fclk_parse_planner_input(payload.get('input', ''), default_break)
@api.model
def fclk_snapshot(self, schedule):
if not schedule:
return ''
return schedule.fclk_display_value()
@api.model
def fclk_apply_planner_cell(self, employee, schedule_date, payload, user=None):
self = self.sudo()
employee = employee.sudo()
date_obj = fields.Date.to_date(schedule_date)
if not employee.exists() or not date_obj:
raise ValidationError(_("Invalid employee or schedule date."))
existing = self.search([
('employee_id', '=', employee.id),
('schedule_date', '=', date_obj),
], limit=1)
old_value = self.fclk_snapshot(existing)
parsed = self.fclk_values_from_planner_payload(payload, employee)
if parsed.get('clear'):
if existing:
existing.unlink()
new_schedule = self.browse()
new_value = ''
else:
# Resolve the role: explicit payload role wins, then the shift
# template's role, then the employee's default role.
role_id = payload.get('role_id')
if not role_id:
shift_id = parsed.get('shift_id')
shift = self.env['fusion.clock.shift'].browse(shift_id) if shift_id else None
if shift and shift.role_id:
role_id = shift.role_id.id
elif employee.x_fclk_default_role_id:
role_id = employee.x_fclk_default_role_id.id
vals = {
'employee_id': employee.id,
'schedule_date': date_obj,
'shift_id': parsed.get('shift_id') or False,
'role_id': int(role_id) if role_id else False,
'is_off': bool(parsed.get('is_off')),
'start_time': parsed.get('start_time') or 0.0,
'end_time': parsed.get('end_time') or 0.0,
'break_minutes': parsed.get('break_minutes') or 0.0,
'note': payload.get('note') or False,
# Any planner edit returns the cell to draft; it must be re-posted
# before automation acts on it.
'state': 'draft',
'posted_date': False,
}
if existing:
existing.write(vals)
new_schedule = existing
else:
new_schedule = self.create(vals)
new_value = new_schedule.fclk_display_value()
if old_value != new_value:
self.env['fusion.clock.schedule.audit'].sudo().create({
'schedule_id': new_schedule.id if new_schedule else False,
'employee_id': employee.id,
'schedule_date': date_obj,
'old_value': old_value,
'new_value': new_value,
'changed_by_id': (user or self.env.user).id,
'changed_at': fields.Datetime.now(),
'company_id': employee.company_id.id,
'department_id': employee.department_id.id,
})
return new_schedule
@api.model
def fclk_cell_payload(self, employee, date_obj, schedule=None):
schedule = schedule or self.search([
('employee_id', '=', employee.id),
('schedule_date', '=', date_obj),
], limit=1)
Schedule = self.env['fusion.clock.schedule']
if schedule:
return {
'schedule_id': schedule.id,
'source': 'schedule',
'state': schedule.state,
'input': schedule.fclk_display_value(),
'label': schedule.fclk_display_value(),
'is_off': schedule.is_off,
'shift_id': schedule.shift_id.id or False,
'start_time': schedule.start_time,
'end_time': schedule.end_time,
'break_minutes': schedule.break_minutes,
'hours': schedule.planned_hours,
'hours_display': Schedule.fclk_hours_display(schedule.planned_hours),
'note': schedule.note or '',
'role_id': schedule.role_id.id or False,
'role_name': schedule.role_id.name or '',
'role_color': schedule.role_id._get_color_from_code() if schedule.role_id else '',
'recurring': bool(schedule.recurrence_id),
}
plan = employee._get_fclk_day_plan(date_obj)
return {
'schedule_id': False,
'source': plan.get('source') or 'none',
'state': False,
'input': plan.get('label') or '',
'label': plan.get('label') or '',
'is_off': plan.get('is_off', False),
'shift_id': False,
'start_time': plan.get('start_time') or 0.0,
'end_time': plan.get('end_time') or 0.0,
'break_minutes': plan.get('break_minutes') or 0.0,
'hours': plan.get('hours') or 0.0,
'hours_display': Schedule.fclk_hours_display(plan.get('hours') or 0.0),
'note': '',
'role_id': False,
'role_name': '',
'role_color': '',
}
@api.model
def fclk_attach_recurrence(self, schedule, repeat_vals):
"""Attach a recurrence rule to a seed schedule cell and generate it
forward. ``repeat_vals`` mirrors the recurrence fields."""
schedule = schedule.sudo()
if not schedule:
raise ValidationError(_("Pick a shift to repeat first."))
rule = self.env['fusion.clock.schedule.recurrence'].sudo().create({
'repeat_interval': int(repeat_vals.get('repeat_interval') or 1),
'repeat_unit': repeat_vals.get('repeat_unit') or 'week',
'repeat_type': repeat_vals.get('repeat_type') or 'forever',
'repeat_until': repeat_vals.get('repeat_until') or False,
'repeat_number': int(repeat_vals.get('repeat_number') or 1),
'company_id': schedule.company_id.id or self.env.company.id,
})
schedule.recurrence_id = rule.id
rule._generate()
return rule
@api.model
def fclk_clear_recurrence(self, schedule):
"""Detach + stop the recurrence on a seed cell (keeps posted rows)."""
schedule = schedule.sudo()
rule = schedule.recurrence_id
if rule:
rule._stop(fields.Date.today())
schedule.recurrence_id = False
if not rule.schedule_ids:
rule.unlink()
return True
# ----- Open shifts + bulk apply (native "Apply Also To" / self-assign) -----
@api.model
def fclk_create_open_shifts(self, company, date_obj, start, end,
role_id=False, count=1, break_minutes=0.0, note=None):
"""Create N open (unassigned) shifts for a day, available to claim."""
date_obj = fields.Date.to_date(date_obj)
if not date_obj:
raise ValidationError(_("Pick a date for the open shift."))
company_id = (company.id if company else False) or self.env.company.id
vals_list = [{
'is_open': True,
'schedule_date': date_obj,
'start_time': float(start or 0.0),
'end_time': float(end or 0.0),
'break_minutes': float(break_minutes or 0.0),
'role_id': int(role_id) if role_id else False,
'company_id': company_id,
'note': note or False,
'state': 'posted',
} for _i in range(max(1, int(count or 1)))]
return self.sudo().create(vals_list)
@api.model
def fclk_claim_open_shift(self, schedule, employee):
"""Assign an open shift to an employee (portal self-assign)."""
schedule = schedule.sudo()
employee = employee.sudo()
if not schedule or not schedule.is_open:
raise ValidationError(_("This shift is no longer available."))
if not employee:
raise ValidationError(_("No employee to assign this shift to."))
# If the shift carries a role and the employee has an explicit allowed
# list, enforce eligibility (no list = eligible for anything).
if schedule.role_id and employee.x_fclk_role_ids \
and schedule.role_id not in employee.x_fclk_role_ids:
raise ValidationError(_("You are not eligible for this shift's role."))
schedule.write({'employee_id': employee.id, 'is_open': False})
return schedule
@api.model
def fclk_release_shift(self, schedule, employee):
"""Release a claimed shift back to the open pool (portal self-unassign),
respecting the company's days-before cutoff."""
schedule = schedule.sudo()
if not schedule or schedule.employee_id != employee.sudo():
raise ValidationError(_("You can only release your own shift."))
cutoff = schedule.company_id.fclk_self_unassign_days_before or 0
if (schedule.schedule_date - fields.Date.today()).days < cutoff:
raise ValidationError(_("It is too late to release this shift."))
schedule.write({'employee_id': False, 'is_open': True})
return schedule
@api.model
def fclk_bulk_apply(self, employees, date_obj, payload, user=None):
"""Apply the same shift payload to several employees in one go
(native replacement for Planning's 'Apply Also To')."""
results = self.browse()
for employee in employees:
results |= self.fclk_apply_planner_cell(employee, date_obj, dict(payload or {}), user)
return results
@api.model
def fclk_email_posted_range(self, employee, start, end, message=None):
"""Email one employee a summary of their POSTED shifts between two
dates (inclusive). Optional ``message`` is shown above the schedule."""
employee = employee.sudo()
if not employee.work_email:
return False
from .hr_attendance import _fclk_email_wrap
entries = self.sudo().search([
('employee_id', '=', employee.id),
('schedule_date', '>=', start),
('schedule_date', '<=', end),
('state', '=', 'posted'),
])
by_date = {entry.schedule_date: entry for entry in entries}
rows = []
day = start
while day <= end:
entry = by_date.get(day)
rows.append((
day.strftime('%a %b %d'),
entry.fclk_display_value() if entry else 'Not scheduled',
))
day += timedelta(days=1)
company = employee.company_id or self.env.company
summary = (
f'Hello {employee.name}, your shifts for '
f'{start.strftime("%b %d")} - {end.strftime("%b %d, %Y")} '
f'have been posted.'
)
if message:
summary += f'
{message}'
body = _fclk_email_wrap(
company_name=company.name or '',
title='Your Posted Schedule',
summary=summary,
sections=[('Schedule', rows)],
note='Log in to your portal for details.',
)
try:
mail = self.env['mail.mail'].sudo().create({
'subject': f'Your schedule: {start.strftime("%b %d")} - {end.strftime("%b %d")}',
'email_from': company.email or '',
'email_to': employee.work_email,
'body_html': body,
'auto_delete': True,
})
mail.send()
return True
except Exception as exc:
_logger.error(
"Fusion Clock: failed to email posted schedule to %s: %s", employee.name, exc
)
return False
@api.model
def fclk_email_posted_week(self, employee, week_start, week_end):
"""Back-compat wrapper — email one employee their posted week."""
return self.fclk_email_posted_range(employee, week_start, week_end)
@api.model
def fclk_publish_range(self, employees, start, end, message=None):
"""Post every draft shift in [start, end] for the given employees and
email each affected employee. Returns (posted_count, notified_count)."""
Schedule = self.sudo()
domain = [
('employee_id', 'in', employees.ids),
('schedule_date', '>=', start),
('schedule_date', '<=', end),
('state', '!=', 'posted'),
]
# Never auto-post open (unassigned) shifts (Phase B field).
if 'is_open' in Schedule._fields:
domain.append(('is_open', '=', False))
drafts = Schedule.search(domain)
posted = len(drafts)
affected = drafts.mapped('employee_id')
if drafts:
drafts.write({'state': 'posted', 'posted_date': fields.Datetime.now()})
notified = 0
for employee in affected:
if Schedule.fclk_email_posted_range(employee, start, end, message=message):
notified += 1
return posted, notified
@api.model
def _fclk_port_planning_data(self):
"""Port Odoo Planning data (roles, employee roles, slots) into the
native models. Safe no-op when planning is not installed. Returns a
dict of counts. Called by the 19.0.5.0.0 migration and by tests."""
import pytz
counts = {'roles': 0, 'employees': 0, 'slots': 0, 'skipped': 0}
env = self.env
has_roles = 'planning.role' in env
has_slots = 'planning.slot' in env
if not has_roles and not has_slots:
return counts
Role = env['fusion.clock.role'].sudo()
role_map = {}
if has_roles:
for prole in env['planning.role'].sudo().with_context(active_test=False).search([]):
target = Role.with_context(active_test=False).search(
[('name', '=ilike', prole.name)], limit=1) or Role.create({
'name': prole.name, 'color': prole.color or 1, 'active': prole.active})
role_map[prole.id] = target.id
counts['roles'] = len(role_map)
Employee = env['hr.employee'].sudo().with_context(active_test=False)
for emp in Employee.search([]):
vals = {}
if emp._fields.get('default_planning_role_id') and emp.default_planning_role_id:
mapped = role_map.get(emp.default_planning_role_id.id)
if mapped:
vals['x_fclk_default_role_id'] = mapped
if emp._fields.get('planning_role_ids') and emp.planning_role_ids:
mapped_ids = [role_map[r.id] for r in emp.planning_role_ids if r.id in role_map]
if mapped_ids:
vals['x_fclk_role_ids'] = [(6, 0, mapped_ids)]
if vals:
emp.write(vals)
counts['employees'] += 1
if has_slots:
Schedule = self.sudo()
for slot in env['planning.slot'].sudo().search([], order='start_datetime'):
if not slot.start_datetime or not slot.end_datetime:
counts['skipped'] += 1
continue
employee = slot.employee_id if 'employee_id' in slot._fields else False
tz_name = ((employee.tz if employee else False)
or (slot.resource_id.tz if slot.resource_id else False)
or env.company.partner_id.tz or 'UTC')
try:
tz = pytz.timezone(tz_name)
except Exception:
tz = pytz.UTC
local_start = pytz.utc.localize(slot.start_datetime).astimezone(tz)
local_end = pytz.utc.localize(slot.end_datetime).astimezone(tz)
span_hours = (slot.end_datetime - slot.start_datetime).total_seconds() / 3600.0
allocated = slot.allocated_hours if 'allocated_hours' in slot._fields else span_hours
vals = {
'employee_id': employee.id if employee else False,
'is_open': not bool(employee),
'schedule_date': local_start.date(),
'start_time': round(local_start.hour + local_start.minute / 60.0, 2),
'end_time': round(local_end.hour + local_end.minute / 60.0, 2),
'break_minutes': round(max(0.0, span_hours - (allocated or span_hours)) * 60.0, 0),
'role_id': role_map.get(slot.role_id.id) if slot.role_id else False,
'note': slot.name or False,
'state': 'posted' if slot.state == 'published' else 'draft',
}
with env.cr.savepoint():
try:
Schedule.create(vals)
counts['slots'] += 1
except Exception as exc:
counts['skipped'] += 1
_logger.warning("Fusion Clock: skip planning.slot %s (%s).", slot.id, exc)
return counts
class FusionClockScheduleAudit(models.Model):
_name = 'fusion.clock.schedule.audit'
_description = 'Clock Schedule Change Audit'
_order = 'changed_at desc, id desc'
_rec_name = 'display_name'
schedule_id = fields.Many2one(
'fusion.clock.schedule',
string='Schedule',
ondelete='set null',
index=True,
)
employee_id = fields.Many2one(
'hr.employee',
string='Employee',
required=True,
index=True,
ondelete='cascade',
)
schedule_date = fields.Date(
string='Schedule Date',
required=True,
index=True,
)
old_value = fields.Char(string='Old Value')
new_value = fields.Char(string='New Value')
changed_by_id = fields.Many2one(
'res.users',
string='Changed By',
required=True,
ondelete='restrict',
)
changed_at = fields.Datetime(
string='Changed At',
default=fields.Datetime.now,
required=True,
index=True,
)
company_id = fields.Many2one(
'res.company',
string='Company',
index=True,
)
department_id = fields.Many2one(
'hr.department',
string='Department',
index=True,
)
display_name = fields.Char(
compute='_compute_display_name',
store=True,
)
@api.depends('employee_id', 'schedule_date', 'old_value', 'new_value')
def _compute_display_name(self):
for rec in self:
rec.display_name = "%s - %s: %s -> %s" % (
rec.employee_id.name or '',
rec.schedule_date or '',
rec.old_value or 'blank',
rec.new_value or 'blank',
)