Files
Odoo-Modules/fusion_clock/models/hr_attendance.py
gsinghpal 50c209b8d3 feat(fusion_clock): NFC kiosk attendance fields + activity-log selections
- Add 'nfc_kiosk' to x_fclk_clock_source selection on hr.attendance
- Add x_fclk_check_in_photo and x_fclk_check_out_photo Binary fields (attachment=True)
- Add 'card_enrollment' and 'unknown_card_tap' to activity log log_type selection
- Add 'nfc_kiosk' to activity log source selection
- Add TestNfcAttendanceFields test class (3 tests); all 6 fusion_clock tests pass

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 00:24:35 -04:00

604 lines
23 KiB
Python

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
import logging
import pytz
from datetime import datetime, timedelta
from odoo import models, fields, api
from odoo.tools import float_round
from .tz_utils import get_local_today, get_local_day_boundaries
_logger = logging.getLogger(__name__)
def _fclk_utc_to_local_str(dt, employee, fmt='%I:%M %p'):
"""Convert a naive UTC datetime to a formatted string in the employee's timezone."""
import pytz
tz_name = (
employee.resource_id.tz
or (employee.user_id.partner_id.tz if employee.user_id else False)
or employee.company_id.partner_id.tz
or 'UTC'
)
utc_dt = pytz.UTC.localize(dt)
local_dt = utc_dt.astimezone(pytz.timezone(tz_name))
return local_dt.strftime(fmt)
_FCLK_ACCENT = '#10B981'
_FCLK_FONT = ("-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,Arial,sans-serif")
def _fclk_email_section(heading, rows):
"""Build a details table matching the Fusion email design system."""
if not rows:
return ''
html = (
'<table style="width:100%;border-collapse:collapse;margin:0 0 24px 0;">'
f'<tr><td colspan="2" style="padding:10px 14px;font-size:12px;font-weight:600;'
f'opacity:0.55;text-transform:uppercase;letter-spacing:0.5px;'
f'border-bottom:2px solid rgba(128,128,128,0.25);">{heading}</td></tr>'
)
for label, value in rows:
if value is None or value == '' or value is False:
continue
html += (
f'<tr>'
f'<td style="padding:10px 14px;opacity:0.6;font-size:14px;'
f'border-bottom:1px solid rgba(128,128,128,0.15);width:35%;">{label}</td>'
f'<td style="padding:10px 14px;font-size:14px;'
f'border-bottom:1px solid rgba(128,128,128,0.15);">{value}</td>'
f'</tr>'
)
html += '</table>'
return html
def _fclk_email_wrap(
company_name,
title,
summary,
sections=None,
note=None,
attachments_note=None,
extra_html='',
):
"""Build a complete Fusion Clock email matching the Fusion design system.
No user signatures are appended.
"""
parts = [
f'<div style="font-family:{_FCLK_FONT};max-width:600px;margin:0 auto;">',
f'<div style="height:4px;background-color:{_FCLK_ACCENT};"></div>',
'<div style="padding:32px 28px;">',
f'<p style="color:{_FCLK_ACCENT};font-size:13px;font-weight:600;'
f'letter-spacing:0.5px;text-transform:uppercase;margin:0 0 24px 0;">'
f'{company_name}</p>',
f'<h2 style="font-size:22px;font-weight:700;margin:0 0 6px 0;'
f'line-height:1.3;">{title}</h2>',
f'<p style="opacity:0.65;font-size:15px;line-height:1.5;'
f'margin:0 0 24px 0;">{summary}</p>',
]
if sections:
for heading, rows in sections:
parts.append(_fclk_email_section(heading, rows))
if note:
parts.append(
f'<div style="border-left:3px solid {_FCLK_ACCENT};padding:12px 16px;'
f'margin:0 0 24px 0;">'
f'<p style="margin:0;font-size:14px;line-height:1.5;">{note}</p></div>'
)
if extra_html:
parts.append(extra_html)
if attachments_note:
parts.append(
'<div style="padding:10px 14px;border:1px dashed rgba(128,128,128,0.35);'
'border-radius:6px;margin:0 0 24px 0;">'
'<p style="margin:0;font-size:13px;opacity:0.65;">'
f'<strong style="opacity:1;">Attached:</strong> {attachments_note}</p></div>'
)
parts.append('</div>')
parts.append(
'<div style="padding:16px 28px;text-align:center;">'
'<p style="opacity:0.5;font-size:11px;line-height:1.5;margin:0;">'
f'{company_name}<br/>'
'This is an automated notification from Fusion Clock.</p></div>'
)
parts.append('</div>')
return ''.join(parts)
class HrAttendance(models.Model):
_inherit = 'hr.attendance'
x_fclk_location_id = fields.Many2one(
'fusion.clock.location',
string='Clock Location',
help="The geofenced location where employee clocked in.",
)
x_fclk_clock_source = fields.Selection(
[
('portal', 'Portal'),
('portal_fab', 'Portal FAB'),
('systray', 'Systray'),
('backend_fab', 'Backend FAB'),
('kiosk', 'Kiosk'),
('nfc_kiosk', 'NFC Kiosk'),
('manual', 'Manual'),
('auto', 'Auto Clock-Out'),
],
string='Clock Source',
tracking=True,
help="How this attendance was recorded.",
)
x_fclk_in_distance = fields.Float(
string='Check-In Distance (m)',
digits=(10, 2),
help="Distance from location center at clock-in, in meters.",
)
x_fclk_out_distance = fields.Float(
string='Check-Out Distance (m)',
digits=(10, 2),
help="Distance from location center at clock-out, in meters.",
)
x_fclk_check_in_photo = fields.Binary(
string='Check-In Photo',
attachment=True,
help="Front-camera photo captured at NFC kiosk clock-in.",
)
x_fclk_check_out_photo = fields.Binary(
string='Check-Out Photo',
attachment=True,
help="Front-camera photo captured at NFC kiosk clock-out.",
)
x_fclk_break_minutes = fields.Float(
string='Break (min)',
default=0.0,
tracking=True,
help="Break duration in minutes to deduct from worked hours.",
)
x_fclk_net_hours = fields.Float(
string='Net Hours',
compute='_compute_net_hours',
store=True,
tracking=True,
help="Worked hours minus break deduction.",
)
x_fclk_penalty_ids = fields.One2many(
'fusion.clock.penalty',
'attendance_id',
string='Penalties',
)
x_fclk_auto_clocked_out = fields.Boolean(
string='Auto Clocked Out',
default=False,
help="Set to true if this attendance was automatically closed.",
)
x_fclk_grace_used = fields.Boolean(
string='Grace Period Used',
default=False,
help="Whether the grace period was consumed before auto clock-out.",
)
# Overtime
x_fclk_overtime_hours = fields.Float(
string='Overtime (h)',
compute='_compute_overtime_hours',
store=True,
help="Hours beyond the scheduled shift for this day.",
)
x_fclk_is_overtime = fields.Boolean(
string='Has Overtime',
compute='_compute_overtime_hours',
store=True,
)
# Photo verification
x_fclk_checkin_photo = fields.Binary(
string='Check-In Photo',
attachment=True,
help="Selfie captured at clock-in for verification.",
)
@api.depends('worked_hours', 'x_fclk_break_minutes')
def _compute_net_hours(self):
for att in self:
break_hours = (att.x_fclk_break_minutes or 0.0) / 60.0
raw = att.worked_hours or 0.0
att.x_fclk_net_hours = max(raw - break_hours, 0.0)
@api.depends('x_fclk_net_hours')
def _compute_overtime_hours(self):
ICP = self.env['ir.config_parameter'].sudo()
enable_ot = ICP.get_param('fusion_clock.enable_overtime', 'True') == 'True'
daily_threshold = float(ICP.get_param('fusion_clock.daily_overtime_threshold', '8.0'))
for att in self:
if not enable_ot or not att.check_out:
att.x_fclk_overtime_hours = 0.0
att.x_fclk_is_overtime = False
continue
employee = att.employee_id
scheduled_hours = employee._get_fclk_scheduled_hours() if employee else daily_threshold
net = att.x_fclk_net_hours or 0.0
if net > scheduled_hours:
att.x_fclk_overtime_hours = round(net - scheduled_hours, 2)
att.x_fclk_is_overtime = True
else:
att.x_fclk_overtime_hours = 0.0
att.x_fclk_is_overtime = False
@api.model
def _cron_fusion_auto_clock_out(self):
"""Cron job: auto clock-out employees after shift + grace period."""
ICP = self.env['ir.config_parameter'].sudo()
if ICP.get_param('fusion_clock.enable_auto_clockout', 'True') != 'True':
return
max_shift = float(ICP.get_param('fusion_clock.max_shift_hours', '12.0'))
grace_min = float(ICP.get_param('fusion_clock.grace_period_minutes', '15'))
office_user_id = int(ICP.get_param('fusion_clock.office_user_id', '0'))
now = fields.Datetime.now()
open_attendances = self.sudo().search([
('check_out', '=', False),
])
ActivityLog = self.env['fusion.clock.activity.log'].sudo()
for att in open_attendances:
check_in = att.check_in
if not check_in:
continue
employee = att.employee_id
emp_tz = pytz.timezone(employee.tz or self.env.company.tz or 'UTC')
check_in_date = pytz.UTC.localize(check_in).astimezone(emp_tz).date()
_, scheduled_out = employee._get_fclk_scheduled_times(check_in_date)
deadline = scheduled_out + timedelta(minutes=grace_min)
max_deadline = check_in + timedelta(hours=max_shift)
effective_deadline = min(deadline, max_deadline)
if now > effective_deadline:
clock_out_time = min(effective_deadline, now)
try:
att.sudo().write({
'check_out': clock_out_time,
'x_fclk_auto_clocked_out': True,
'x_fclk_grace_used': True,
'x_fclk_clock_source': 'auto',
})
# Apply break deduction
threshold = float(ICP.get_param('fusion_clock.break_threshold_hours', '4.0'))
if (att.worked_hours or 0) >= threshold:
break_min = employee._get_fclk_break_minutes()
att.sudo().write({'x_fclk_break_minutes': break_min})
att.sudo().message_post(
body=f"Auto clocked out at {_fclk_utc_to_local_str(clock_out_time, employee, '%H:%M')} "
f"(grace period expired). Net hours: {att.x_fclk_net_hours:.1f}h",
message_type='comment',
subtype_xmlid='mail.mt_note',
)
# Log to activity log
ActivityLog.create({
'employee_id': employee.id,
'log_type': 'auto_clock_out',
'description': f"Auto clocked out at {_fclk_utc_to_local_str(clock_out_time, employee, '%H:%M')}. "
f"Net hours: {att.x_fclk_net_hours:.1f}h",
'attendance_id': att.id,
'location_id': att.x_fclk_location_id.id if att.x_fclk_location_id else False,
'source': 'system',
})
# Set pending reason
employee.sudo().write({'x_fclk_pending_reason': True})
# Notify office user
self._fclk_notify_office(
office_user_id,
f"Auto Clock-Out: {employee.name}",
f"{employee.name} was auto-clocked out at {_fclk_utc_to_local_str(clock_out_time, employee, '%H:%M')}. "
f"Please review and correct if needed.",
'hr.attendance',
att.id,
)
_logger.info(
"Fusion Clock: Auto clocked out %s (attendance %s)",
employee.name, att.id,
)
except Exception as e:
_logger.error(
"Fusion Clock: Failed to auto clock-out attendance %s: %s",
att.id, str(e),
)
@api.model
def _cron_fusion_check_absences(self):
"""Cron job: check for absent employees (no attendance on workday)."""
ICP = self.env['ir.config_parameter'].sudo()
office_user_id = int(ICP.get_param('fusion_clock.office_user_id', '0'))
max_absences = int(ICP.get_param('fusion_clock.max_monthly_absences', '3'))
employees = self.env['hr.employee'].sudo().search([
('x_fclk_enable_clock', '=', True),
])
ActivityLog = self.env['fusion.clock.activity.log'].sudo()
LeaveRequest = self.env['fusion.clock.leave.request'].sudo()
for emp in employees:
yesterday = get_local_today(self.env, emp) - timedelta(days=1)
if yesterday.weekday() >= 5:
continue
day_start, day_end = get_local_day_boundaries(self.env, yesterday, emp)
holidays = self.env['resource.calendar.leaves'].sudo().search([
('resource_id', '=', False),
('date_from', '<=', day_end),
('date_to', '>=', day_start),
])
if holidays:
continue
att_count = self.sudo().search_count([
('employee_id', '=', emp.id),
('check_in', '>=', day_start),
('check_in', '<', day_end),
])
if att_count > 0:
continue
leave = LeaveRequest.search([
('employee_id', '=', emp.id),
('leave_date', '=', yesterday),
], limit=1)
if leave:
continue
ActivityLog.create({
'employee_id': emp.id,
'log_type': 'absent',
'log_date': day_start,
'description': f"No attendance recorded for {yesterday}",
'source': 'system',
})
emp.sudo().write({'x_fclk_pending_reason': True})
month_start = yesterday.replace(day=1)
month_boundary_start, _ = get_local_day_boundaries(self.env, month_start, emp)
absence_count = ActivityLog.search_count([
('employee_id', '=', emp.id),
('log_type', '=', 'absent'),
('log_date', '>=', month_boundary_start),
])
if absence_count >= max_absences:
self._fclk_notify_office(
office_user_id,
f"Excessive Absences: {emp.name}",
f"{emp.name} has {absence_count} absences this month "
f"(threshold: {max_absences}). Please review.",
'hr.employee',
emp.id,
)
_logger.info("Fusion Clock: Marked %s as absent for %s", emp.name, yesterday)
@api.model
def _cron_fusion_employee_reminders(self):
"""Cron job: send clock-in/out reminders to employees."""
ICP = self.env['ir.config_parameter'].sudo()
if ICP.get_param('fusion_clock.enable_employee_notifications', 'True') != 'True':
return
reminder_in_min = float(ICP.get_param('fusion_clock.reminder_before_shift_minutes', '30'))
reminder_out_min = float(ICP.get_param('fusion_clock.reminder_before_end_minutes', '15'))
now = fields.Datetime.now()
employees = self.env['hr.employee'].sudo().search([
('x_fclk_enable_clock', '=', True),
])
for emp in employees:
today = get_local_today(self.env, emp)
if today.weekday() >= 5:
continue
if emp.x_fclk_last_reminder_date == today:
continue
scheduled_in, scheduled_out = emp._get_fclk_scheduled_times(today)
is_checked_in = emp.attendance_state == 'checked_in'
# Missed clock-in reminder
reminder_deadline = scheduled_in + timedelta(minutes=reminder_in_min)
if not is_checked_in and now > reminder_deadline:
today_start, _ = get_local_day_boundaries(self.env, today, emp)
has_attendance = self.sudo().search_count([
('employee_id', '=', emp.id),
('check_in', '>=', today_start),
])
if has_attendance == 0:
self._fclk_send_employee_reminder(
emp,
"Clock-In Reminder",
f"Hi {emp.name}, you haven't clocked in yet today. "
f"Your shift started at {_fclk_utc_to_local_str(scheduled_in, emp)}.",
)
emp.sudo().write({'x_fclk_last_reminder_date': today})
# Clock-out reminder
reminder_before_end = scheduled_out - timedelta(minutes=reminder_out_min)
if is_checked_in and now > reminder_before_end and now < scheduled_out:
self._fclk_send_employee_reminder(
emp,
"Clock-Out Reminder",
f"Hi {emp.name}, your shift ends at {_fclk_utc_to_local_str(scheduled_out, emp)}. "
f"Don't forget to clock out.",
)
emp.sudo().write({'x_fclk_last_reminder_date': today})
@api.model
def _cron_fusion_weekly_summary(self):
"""Cron job: send weekly summary email to employees (Monday 8 AM)."""
ICP = self.env['ir.config_parameter'].sudo()
if ICP.get_param('fusion_clock.send_weekly_summary', 'True') != 'True':
return
employees = self.env['hr.employee'].sudo().search([
('x_fclk_enable_clock', '=', True),
])
company = self.env.company
company_email = company.email or ''
company_name = company.name or ''
for emp in employees:
if not emp.work_email:
continue
today = get_local_today(self.env, emp)
if today.weekday() != 0:
continue
week_start = today - timedelta(days=7)
week_end = today - timedelta(days=1)
ws_fmt = week_start.strftime('%b %d, %Y')
we_fmt = week_end.strftime('%b %d, %Y')
ws_start, _ = get_local_day_boundaries(self.env, week_start, emp)
_, we_end = get_local_day_boundaries(self.env, week_end, emp)
atts = self.sudo().search([
('employee_id', '=', emp.id),
('check_in', '>=', ws_start),
('check_in', '<', we_end),
('check_out', '!=', False),
])
total_net = round(sum(a.x_fclk_net_hours or 0 for a in atts), 1)
total_ot = round(sum(a.x_fclk_overtime_hours or 0 for a in atts), 1)
penalty_count = self.env['fusion.clock.penalty'].sudo().search_count([
('employee_id', '=', emp.id),
('date', '>=', week_start),
('date', '<=', week_end),
])
ActivityLog = self.env['fusion.clock.activity.log'].sudo()
absence_count = ActivityLog.search_count([
('employee_id', '=', emp.id),
('log_type', '=', 'absent'),
('log_date', '>=', ws_start),
('log_date', '<', we_end),
])
streak = emp.x_fclk_ontime_streak or 0
emp_company = emp.company_id or company
body = _fclk_email_wrap(
company_name=emp_company.name or company_name,
title='Weekly Summary',
summary=(
f'Hello <strong>{emp.name}</strong>, here is your attendance '
f'summary for <strong>{ws_fmt}</strong> to <strong>{we_fmt}</strong>.'
),
sections=[('Summary', [
('Total Hours', f'{total_net}h'),
('Overtime', f'{total_ot}h'),
('Penalties', str(penalty_count)),
('Absences', str(absence_count)),
('On-Time Streak', f'{streak} days'),
])],
note='Log in to <a href="/my/clock" style="color:#10B981;">'
'your portal</a> to view full details.',
)
try:
from_email = emp_company.email or company_email
mail = self.env['mail.mail'].sudo().create({
'subject': f'Your Weekly Attendance Summary ({ws_fmt} - {we_fmt})',
'email_from': from_email,
'email_to': emp.work_email,
'body_html': body,
'auto_delete': True,
})
mail.send()
except Exception as e:
_logger.error("Fusion Clock: Failed to send weekly summary to %s: %s", emp.name, e)
@api.model
def _fclk_notify_office(self, office_user_id, summary, note, res_model, res_id):
"""Create a mail.activity for the office user."""
if not office_user_id:
return
office_user = self.env['res.users'].sudo().browse(office_user_id)
if not office_user.exists():
return
try:
self.env['mail.activity'].sudo().create({
'activity_type_id': self.env.ref('mail.mail_activity_data_todo').id,
'summary': summary,
'note': note,
'user_id': office_user_id,
'res_model_id': self.env['ir.model']._get_id(res_model),
'res_id': res_id,
'date_deadline': get_local_today(self.env),
})
except Exception as e:
_logger.error("Fusion Clock: Failed to create office activity: %s", e)
@api.model
def _fclk_send_employee_reminder(self, employee, subject, body):
"""Send a notification to an employee via internal note."""
try:
if employee.user_id:
employee.user_id.sudo().notify_info(
message=body,
title=subject,
sticky=False,
)
except Exception:
pass
try:
if employee.work_email:
company = employee.company_id or self.env.company
company_email = company.email or ''
company_name = company.name or ''
html_body = _fclk_email_wrap(
company_name=company_name,
title=subject,
summary=body,
note='Log in to <a href="/my/clock" style="color:#10B981;">'
'your portal</a> to view your attendance details.',
)
mail_values = {
'subject': f"Fusion Clock: {subject}",
'email_from': company_email,
'body_html': html_body,
'email_to': employee.work_email,
'auto_delete': True,
}
self.env['mail.mail'].sudo().create(mail_values).send()
except Exception as e:
_logger.error("Fusion Clock: Failed to send reminder to %s: %s", employee.name, e)