# -*- coding: utf-8 -*- # Copyright 2026 Nexa Systems Inc. # License OPL-1 (Odoo Proprietary License v1.0) import logging from datetime import datetime, timedelta from odoo import models, fields, api from odoo.tools import float_round _logger = logging.getLogger(__name__) 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'), ('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_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 _, 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 {clock_out_time.strftime('%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 {clock_out_time.strftime('%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 {clock_out_time.strftime('%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')) yesterday = fields.Date.today() - timedelta(days=1) # Skip weekends if yesterday.weekday() >= 5: return # Skip public holidays holidays = self.env['resource.calendar.leaves'].sudo().search([ ('resource_id', '=', False), ('date_from', '<=', datetime.combine(yesterday, datetime.max.time())), ('date_to', '>=', datetime.combine(yesterday, datetime.min.time())), ]) if holidays: return 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: # Check for attendance yesterday att_count = self.sudo().search_count([ ('employee_id', '=', emp.id), ('check_in', '>=', datetime.combine(yesterday, datetime.min.time())), ('check_in', '<', datetime.combine(yesterday + timedelta(days=1), datetime.min.time())), ]) if att_count > 0: continue # Check for approved leave leave = LeaveRequest.search([ ('employee_id', '=', emp.id), ('leave_date', '=', yesterday), ], limit=1) if leave: continue # Mark absent ActivityLog.create({ 'employee_id': emp.id, 'log_type': 'absent', 'log_date': datetime.combine(yesterday, datetime.min.time().replace(hour=9)), 'description': f"No attendance recorded for {yesterday}", 'source': 'system', }) emp.sudo().write({'x_fclk_pending_reason': True}) # Check monthly threshold month_start = yesterday.replace(day=1) absence_count = ActivityLog.search_count([ ('employee_id', '=', emp.id), ('log_type', '=', 'absent'), ('log_date', '>=', datetime.combine(month_start, datetime.min.time())), ]) 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() today = fields.Date.today() # Skip weekends if today.weekday() >= 5: return employees = self.env['hr.employee'].sudo().search([ ('x_fclk_enable_clock', '=', True), ]) for emp in employees: 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: has_attendance = self.sudo().search_count([ ('employee_id', '=', emp.id), ('check_in', '>=', datetime.combine(today, datetime.min.time())), ]) 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 {scheduled_in.strftime('%I:%M %p')}.", ) 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 {scheduled_out.strftime('%I:%M %p')}. " 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 today = fields.Date.today() if today.weekday() != 0: return week_start = today - timedelta(days=7) week_end = today - timedelta(days=1) employees = self.env['hr.employee'].sudo().search([ ('x_fclk_enable_clock', '=', True), ]) company_email = self.env.company.email or '' for emp in employees: if not emp.work_email: continue atts = self.sudo().search([ ('employee_id', '=', emp.id), ('check_in', '>=', datetime.combine(week_start, datetime.min.time())), ('check_in', '<', datetime.combine(week_end + timedelta(days=1), datetime.min.time())), ('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', '>=', datetime.combine(week_start, datetime.min.time())), ('log_date', '<', datetime.combine(week_end + timedelta(days=1), datetime.min.time())), ]) streak = emp.x_fclk_ontime_streak or 0 def _row(label, value, bg=False): bg_style = 'background:#f8f9fa;' if bg else '' return ( f'' f'{label}' f'{value}' f'' ) body = ( '
' '' '' '
' '

Fusion Clock

' '

Weekly Summary

' '
' f'

Hello {emp.name},

' f'

Here is your attendance summary for {week_start} to {week_end}:

' '' + _row('Total Hours', f'{total_net}h', True) + _row('Overtime', f'{total_ot}h', False) + _row('Penalties', str(penalty_count), True) + _row('Absences', str(absence_count), False) + _row('On-Time Streak', f'{streak} days', True) + '
' '

Log in to your portal to view details.

' '

This is an automated message from Fusion Clock.

' '
' ) try: mail = self.env['mail.mail'].sudo().create({ 'subject': f'Your Weekly Attendance Summary ({week_start} - {week_end})', 'email_from': company_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': fields.Date.today(), }) 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: mail_values = { 'subject': f"Fusion Clock: {subject}", 'body_html': f"

{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)