684 lines
29 KiB
Python
684 lines
29 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2026 Nexa Systems Inc.
|
|
# License OPL-1 (Odoo Proprietary License v1.0)
|
|
|
|
import base64
|
|
import math
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from odoo import http, fields, _
|
|
from odoo.http import request
|
|
from odoo.addons.fusion_clock.models.tz_utils import get_local_today, get_local_day_boundaries
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
STREAK_MILESTONES = [5, 10, 20, 50, 100]
|
|
|
|
|
|
def haversine_distance(lat1, lon1, lat2, lon2):
|
|
"""Calculate the great-circle distance between two points on Earth (in meters)."""
|
|
R = 6371000
|
|
phi1 = math.radians(lat1)
|
|
phi2 = math.radians(lat2)
|
|
delta_phi = math.radians(lat2 - lat1)
|
|
delta_lambda = math.radians(lon2 - lon1)
|
|
|
|
a = (math.sin(delta_phi / 2) ** 2
|
|
+ math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2) ** 2)
|
|
c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
|
|
return R * c
|
|
|
|
|
|
class FusionClockAPI(http.Controller):
|
|
"""JSON API endpoints for Fusion Clock operations."""
|
|
|
|
def _get_employee(self):
|
|
user = request.env.user
|
|
return request.env['hr.employee'].sudo().search([
|
|
('user_id', '=', user.id),
|
|
], limit=1)
|
|
|
|
def _get_locations_for_employee(self, employee):
|
|
Location = request.env['fusion.clock.location'].sudo()
|
|
locations = Location.search([
|
|
('active', '=', True),
|
|
('company_id', '=', employee.company_id.id),
|
|
])
|
|
return locations.filtered(
|
|
lambda loc: loc.all_employees or employee.id in loc.employee_ids.ids
|
|
)
|
|
|
|
def _verify_location(self, latitude, longitude, employee, client_ip=None):
|
|
"""Verify GPS coordinates or IP against allowed geofences.
|
|
Returns (location_record, distance, error_detail, method) tuple.
|
|
method is 'gps' or 'ip'.
|
|
"""
|
|
locations = self._get_locations_for_employee(employee)
|
|
|
|
if not locations:
|
|
return False, 0, 'no_locations', 'gps'
|
|
|
|
gps_available = latitude != 0 or longitude != 0
|
|
|
|
geocoded = locations.filtered(lambda l: l.latitude and l.longitude
|
|
and not (l.latitude == 0.0 and l.longitude == 0.0))
|
|
|
|
nearest_distance = float('inf')
|
|
|
|
if gps_available and geocoded:
|
|
for loc in geocoded:
|
|
dist = haversine_distance(latitude, longitude, loc.latitude, loc.longitude)
|
|
if dist <= loc.radius:
|
|
return loc, dist, None, 'gps'
|
|
if dist < nearest_distance:
|
|
nearest_distance = dist
|
|
|
|
# IP fallback -- try when GPS is unavailable OR GPS is outside all geofences
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
if client_ip:
|
|
for loc in locations:
|
|
if loc.check_ip_whitelist(client_ip):
|
|
return loc, 0, None, 'ip'
|
|
|
|
if not gps_available:
|
|
return False, 0, 'gps_unavailable', 'gps'
|
|
|
|
if not geocoded:
|
|
return False, 0, 'no_geocoded', 'gps'
|
|
|
|
return False, nearest_distance, 'outside', 'gps'
|
|
|
|
def _location_error_message(self, error_type, distance=0):
|
|
if error_type == 'no_locations':
|
|
return 'No clock locations configured. Ask your manager to set up locations in Fusion Clock > Locations.'
|
|
elif error_type == 'no_geocoded':
|
|
return 'Clock locations exist but have no GPS coordinates. Ask your manager to geocode them.'
|
|
elif error_type == 'gps_unavailable':
|
|
return 'Could not determine your location. Please enable GPS/location services in your browser and device settings, then try again.'
|
|
else:
|
|
dist_str = f"{int(distance)}m" if distance < 1000 else f"{distance/1000:.1f}km"
|
|
return f'You are {dist_str} away from the nearest clock location. Please clock in/out within the allowed area.'
|
|
|
|
def _get_scheduled_times(self, employee, date):
|
|
"""Get scheduled clock-in and clock-out datetime using employee shift."""
|
|
return employee._get_fclk_scheduled_times(date)
|
|
|
|
def _check_and_create_penalty(self, employee, attendance, penalty_type, scheduled_dt, actual_dt):
|
|
"""Check if a penalty should be created and deduct minutes."""
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
if ICP.get_param('fusion_clock.enable_penalties', 'True') != 'True':
|
|
return
|
|
|
|
grace = float(ICP.get_param('fusion_clock.penalty_grace_minutes', '5'))
|
|
deduction = float(ICP.get_param('fusion_clock.penalty_deduction_minutes', '15'))
|
|
diff_minutes = abs((actual_dt - scheduled_dt).total_seconds()) / 60.0
|
|
|
|
should_penalize = False
|
|
if penalty_type == 'late_in' and actual_dt > scheduled_dt:
|
|
should_penalize = diff_minutes > grace
|
|
elif penalty_type == 'early_out' and actual_dt < scheduled_dt:
|
|
should_penalize = diff_minutes > grace
|
|
|
|
if should_penalize:
|
|
request.env['fusion.clock.penalty'].sudo().create({
|
|
'attendance_id': attendance.id,
|
|
'employee_id': employee.id,
|
|
'penalty_type': penalty_type,
|
|
'scheduled_time': scheduled_dt,
|
|
'actual_time': actual_dt,
|
|
'penalty_minutes': deduction,
|
|
'date': actual_dt.date() if isinstance(actual_dt, datetime) else get_local_today(request.env, employee),
|
|
})
|
|
|
|
# Deduct penalty minutes from attendance (adds to break deduction)
|
|
current_break = attendance.x_fclk_break_minutes or 0.0
|
|
attendance.sudo().write({
|
|
'x_fclk_break_minutes': current_break + deduction,
|
|
})
|
|
|
|
# Log penalty
|
|
log_type = 'late_clock_in' if penalty_type == 'late_in' else 'early_clock_out'
|
|
request.env['fusion.clock.activity.log'].sudo().create({
|
|
'employee_id': employee.id,
|
|
'log_type': log_type,
|
|
'description': f"{'Late' if penalty_type == 'late_in' else 'Early'} by "
|
|
f"{diff_minutes:.0f} min. {deduction:.0f} min deducted.",
|
|
'attendance_id': attendance.id,
|
|
'source': 'system',
|
|
})
|
|
|
|
# Reset streak on late clock-in
|
|
if penalty_type == 'late_in':
|
|
employee.sudo().write({'x_fclk_ontime_streak': 0})
|
|
|
|
def _apply_break_deduction(self, attendance, employee):
|
|
"""Apply automatic break deduction if configured."""
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
if ICP.get_param('fusion_clock.auto_deduct_break', 'True') != 'True':
|
|
return
|
|
|
|
threshold = float(ICP.get_param('fusion_clock.break_threshold_hours', '4.0'))
|
|
worked = attendance.worked_hours or 0.0
|
|
|
|
if worked >= threshold:
|
|
break_min = employee._get_fclk_break_minutes()
|
|
current = attendance.x_fclk_break_minutes or 0.0
|
|
# Set to whichever is higher: configured break or existing (penalty-inflated) value
|
|
new_val = max(break_min, current)
|
|
if new_val != current:
|
|
attendance.sudo().write({'x_fclk_break_minutes': new_val})
|
|
|
|
def _log_activity(self, employee, log_type, description, attendance=None,
|
|
location=None, latitude=0, longitude=0, distance=0, source='portal'):
|
|
"""Create an activity log entry."""
|
|
try:
|
|
request.env['fusion.clock.activity.log'].sudo().create({
|
|
'employee_id': employee.id,
|
|
'log_type': log_type,
|
|
'description': description,
|
|
'attendance_id': attendance.id if attendance else False,
|
|
'location_id': location.id if location else False,
|
|
'latitude': latitude,
|
|
'longitude': longitude,
|
|
'distance': distance,
|
|
'source': source,
|
|
})
|
|
except Exception as e:
|
|
_logger.error("Fusion Clock: Failed to create activity log: %s", e)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# API Endpoints
|
|
# -------------------------------------------------------------------------
|
|
|
|
@http.route('/fusion_clock/verify_location', type='jsonrpc', auth='user', methods=['POST'])
|
|
def verify_location(self, latitude=0, longitude=0, accuracy=0, **kw):
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
client_ip = request.httprequest.environ.get('HTTP_X_FORWARDED_FOR', '').split(',')[0].strip()
|
|
if not client_ip:
|
|
client_ip = request.httprequest.remote_addr
|
|
|
|
location, distance, err, method = self._verify_location(latitude, longitude, employee, client_ip)
|
|
|
|
if location:
|
|
return {
|
|
'allowed': True,
|
|
'location_id': location.id,
|
|
'location_name': location.name,
|
|
'location_address': location.address or '',
|
|
'distance': round(distance, 1),
|
|
'radius': location.radius,
|
|
'method': method,
|
|
}
|
|
else:
|
|
msg = self._location_error_message(err, distance)
|
|
return {
|
|
'allowed': False,
|
|
'nearest_distance': round(distance, 1) if distance != float('inf') else None,
|
|
'message': msg,
|
|
'error_type': err,
|
|
}
|
|
|
|
@http.route('/fusion_clock/clock_action', type='jsonrpc', auth='user', methods=['POST'])
|
|
def clock_action(self, latitude=0, longitude=0, accuracy=0, source='portal', photo=None, **kw):
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
if not employee.x_fclk_enable_clock:
|
|
return {'error': 'Fusion Clock is not enabled for your account.'}
|
|
|
|
# Check pending reason before clock-in
|
|
is_checked_in = employee.attendance_state == 'checked_in'
|
|
if not is_checked_in and employee.x_fclk_pending_reason:
|
|
return {
|
|
'requires_reason': True,
|
|
'message': 'Please provide a reason for your missed clock-out before clocking in.',
|
|
}
|
|
|
|
# Server-side location verification
|
|
client_ip = request.httprequest.environ.get('HTTP_X_FORWARDED_FOR', '').split(',')[0].strip()
|
|
if not client_ip:
|
|
client_ip = request.httprequest.remote_addr
|
|
|
|
location, distance, err, method = self._verify_location(latitude, longitude, employee, client_ip)
|
|
if not location:
|
|
err_msg = self._location_error_message(err, distance)
|
|
err_msg += f" (Your IP: {client_ip})"
|
|
self._log_activity(
|
|
employee, 'outside_geofence',
|
|
err_msg,
|
|
latitude=latitude, longitude=longitude, distance=distance, source=source,
|
|
)
|
|
return {
|
|
'error': err_msg,
|
|
'allowed': False,
|
|
'error_type': err,
|
|
}
|
|
|
|
# Log IP fallback usage
|
|
if method == 'ip':
|
|
self._log_activity(
|
|
employee, 'ip_fallback',
|
|
f"IP-based verification used (IP: {client_ip}) at {location.name}",
|
|
location=location, source=source,
|
|
)
|
|
|
|
now = fields.Datetime.now()
|
|
today = get_local_today(request.env, employee)
|
|
|
|
geo_info = {
|
|
'latitude': latitude,
|
|
'longitude': longitude,
|
|
'browser': kw.get('browser', ''),
|
|
'ip_address': client_ip,
|
|
}
|
|
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
|
|
try:
|
|
if not is_checked_in:
|
|
# CLOCK IN
|
|
attendance = employee.sudo()._attendance_action_change(geo_info)
|
|
|
|
write_vals = {
|
|
'x_fclk_location_id': location.id,
|
|
'x_fclk_in_distance': round(distance, 1),
|
|
'x_fclk_clock_source': source,
|
|
}
|
|
|
|
# Photo verification
|
|
if photo and location.require_photo:
|
|
try:
|
|
write_vals['x_fclk_checkin_photo'] = photo
|
|
except Exception:
|
|
pass
|
|
|
|
attendance.sudo().write(write_vals)
|
|
|
|
# Log clock-in
|
|
self._log_activity(
|
|
employee, 'clock_in',
|
|
f"Clocked in at {location.name} ({method.upper()})",
|
|
attendance=attendance, location=location,
|
|
latitude=latitude, longitude=longitude, distance=distance,
|
|
source=source,
|
|
)
|
|
|
|
# Check for late clock-in penalty
|
|
scheduled_in, _ = self._get_scheduled_times(employee, today)
|
|
self._check_and_create_penalty(employee, attendance, 'late_in', scheduled_in, now)
|
|
|
|
# Update on-time streak
|
|
diff_min = (now - scheduled_in).total_seconds() / 60.0 if now > scheduled_in else 0
|
|
grace = float(ICP.get_param('fusion_clock.penalty_grace_minutes', '5'))
|
|
if diff_min <= grace:
|
|
new_streak = (employee.x_fclk_ontime_streak or 0) + 1
|
|
employee.sudo().write({'x_fclk_ontime_streak': new_streak})
|
|
if new_streak in STREAK_MILESTONES:
|
|
self._log_activity(
|
|
employee, 'streak_milestone',
|
|
f"On-time streak reached {new_streak} days!",
|
|
attendance=attendance, source='system',
|
|
)
|
|
|
|
# Very late notification
|
|
very_late = float(ICP.get_param('fusion_clock.very_late_threshold_minutes', '15'))
|
|
if diff_min > very_late:
|
|
office_user_id = int(ICP.get_param('fusion_clock.office_user_id', '0'))
|
|
if office_user_id:
|
|
request.env['hr.attendance'].sudo()._fclk_notify_office(
|
|
office_user_id,
|
|
f"Very Late Clock-In: {employee.name}",
|
|
f"{employee.name} clocked in {int(diff_min)} minutes late.",
|
|
'hr.attendance',
|
|
attendance.id,
|
|
)
|
|
|
|
return {
|
|
'success': True,
|
|
'action': 'clock_in',
|
|
'attendance_id': attendance.id,
|
|
'check_in': fields.Datetime.to_string(attendance.check_in),
|
|
'location_name': location.name,
|
|
'location_address': location.address or '',
|
|
'message': f'Clocked in at {location.name}',
|
|
'streak': employee.x_fclk_ontime_streak,
|
|
}
|
|
|
|
else:
|
|
# CLOCK OUT
|
|
attendance = employee.sudo()._attendance_action_change(geo_info)
|
|
attendance.sudo().write({
|
|
'x_fclk_out_distance': round(distance, 1),
|
|
})
|
|
|
|
# Apply break deduction
|
|
self._apply_break_deduction(attendance, employee)
|
|
|
|
# Check for early clock-out penalty
|
|
_, scheduled_out = self._get_scheduled_times(employee, today)
|
|
self._check_and_create_penalty(employee, attendance, 'early_out', scheduled_out, now)
|
|
|
|
# Log clock-out
|
|
self._log_activity(
|
|
employee, 'clock_out',
|
|
f"Clocked out from {location.name}. Net: {attendance.x_fclk_net_hours:.1f}h",
|
|
attendance=attendance, location=location,
|
|
latitude=latitude, longitude=longitude, distance=distance,
|
|
source=source,
|
|
)
|
|
|
|
# Log overtime
|
|
if attendance.x_fclk_is_overtime and attendance.x_fclk_overtime_hours > 0:
|
|
self._log_activity(
|
|
employee, 'overtime',
|
|
f"Overtime: {attendance.x_fclk_overtime_hours:.1f}h beyond scheduled shift",
|
|
attendance=attendance, source='system',
|
|
)
|
|
|
|
return {
|
|
'success': True,
|
|
'action': 'clock_out',
|
|
'attendance_id': attendance.id,
|
|
'check_in': fields.Datetime.to_string(attendance.check_in),
|
|
'check_out': fields.Datetime.to_string(attendance.check_out),
|
|
'worked_hours': round(attendance.worked_hours or 0, 2),
|
|
'net_hours': round(attendance.x_fclk_net_hours or 0, 2),
|
|
'break_minutes': attendance.x_fclk_break_minutes,
|
|
'overtime_hours': round(attendance.x_fclk_overtime_hours or 0, 2),
|
|
'location_name': location.name,
|
|
'location_address': location.address or '',
|
|
'message': f'Clocked out from {location.name}',
|
|
}
|
|
|
|
except Exception as e:
|
|
_logger.error("Fusion Clock error: %s", str(e))
|
|
return {'error': str(e)}
|
|
|
|
@http.route('/fusion_clock/submit_reason', type='jsonrpc', auth='user', methods=['POST'])
|
|
def submit_reason(self, reason='', departure_time='', **kw):
|
|
"""Submit a reason for missed clock-out."""
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
if not reason:
|
|
return {'error': 'Please provide a reason.'}
|
|
|
|
# Find the last unclosed attendance or last auto-closed
|
|
last_att = request.env['hr.attendance'].sudo().search([
|
|
('employee_id', '=', employee.id),
|
|
('check_out', '!=', False),
|
|
], order='check_out desc', limit=1)
|
|
|
|
if last_att and departure_time:
|
|
try:
|
|
dep_dt = fields.Datetime.from_string(departure_time)
|
|
if dep_dt and dep_dt > last_att.check_in:
|
|
last_att.sudo().write({'check_out': dep_dt})
|
|
except Exception:
|
|
pass
|
|
|
|
# Log the reason
|
|
self._log_activity(
|
|
employee, 'reason_provided',
|
|
f"Reason for missed clock-out: {reason}. "
|
|
f"Reported departure: {departure_time or 'not specified'}",
|
|
attendance=last_att if last_att else None,
|
|
source='portal',
|
|
)
|
|
|
|
employee.sudo().write({'x_fclk_pending_reason': False})
|
|
|
|
return {'success': True, 'message': 'Reason submitted. You may now clock in.'}
|
|
|
|
@http.route('/fusion_clock/request_leave', type='jsonrpc', auth='user', methods=['POST'])
|
|
def request_leave(self, leave_date='', reason='', **kw):
|
|
"""Submit a leave request from the portal."""
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
if not leave_date or not reason:
|
|
return {'error': 'Please provide both a date and a reason.'}
|
|
|
|
try:
|
|
date_obj = fields.Date.from_string(leave_date)
|
|
except Exception:
|
|
return {'error': 'Invalid date format. Use YYYY-MM-DD.'}
|
|
|
|
existing = request.env['fusion.clock.leave.request'].sudo().search([
|
|
('employee_id', '=', employee.id),
|
|
('leave_date', '=', date_obj),
|
|
], limit=1)
|
|
if existing:
|
|
return {'error': 'A leave request already exists for this date.'}
|
|
|
|
request.env['fusion.clock.leave.request'].sudo().create({
|
|
'employee_id': employee.id,
|
|
'leave_date': date_obj,
|
|
'reason': reason,
|
|
'created_from': 'portal',
|
|
})
|
|
|
|
return {'success': True, 'message': f'Leave request for {leave_date} submitted.'}
|
|
|
|
@http.route('/fusion_clock/request_correction', type='jsonrpc', auth='user', methods=['POST'])
|
|
def request_correction(self, attendance_id=0, check_in='', check_out='', reason='', **kw):
|
|
"""Submit a timesheet correction request from the portal."""
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
if ICP.get_param('fusion_clock.enable_correction_requests', 'True') != 'True':
|
|
return {'error': 'Correction requests are not enabled.'}
|
|
|
|
if not attendance_id or not reason:
|
|
return {'error': 'Please provide the attendance record and a reason.'}
|
|
|
|
attendance = request.env['hr.attendance'].sudo().browse(attendance_id)
|
|
if not attendance.exists() or attendance.employee_id.id != employee.id:
|
|
return {'error': 'Attendance record not found.'}
|
|
|
|
vals = {
|
|
'employee_id': employee.id,
|
|
'attendance_id': attendance_id,
|
|
'reason': reason,
|
|
}
|
|
if check_in:
|
|
try:
|
|
vals['requested_check_in'] = fields.Datetime.from_string(check_in)
|
|
except Exception:
|
|
pass
|
|
if check_out:
|
|
try:
|
|
vals['requested_check_out'] = fields.Datetime.from_string(check_out)
|
|
except Exception:
|
|
pass
|
|
|
|
request.env['fusion.clock.correction'].sudo().create(vals)
|
|
return {'success': True, 'message': 'Correction request submitted for review.'}
|
|
|
|
@http.route('/fusion_clock/get_status', type='jsonrpc', auth='user', methods=['POST'])
|
|
def get_status(self, **kw):
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
is_checked_in = employee.attendance_state == 'checked_in'
|
|
|
|
result = {
|
|
'is_checked_in': is_checked_in,
|
|
'employee_name': employee.name,
|
|
'enable_clock': employee.x_fclk_enable_clock,
|
|
'pending_reason': employee.x_fclk_pending_reason,
|
|
'ontime_streak': employee.x_fclk_ontime_streak,
|
|
}
|
|
|
|
if is_checked_in:
|
|
att = request.env['hr.attendance'].sudo().search([
|
|
('employee_id', '=', employee.id),
|
|
('check_out', '=', False),
|
|
], limit=1)
|
|
if att:
|
|
result.update({
|
|
'attendance_id': att.id,
|
|
'check_in': fields.Datetime.to_string(att.check_in),
|
|
'location_name': att.x_fclk_location_id.name or '',
|
|
'location_address': att.x_fclk_location_id.address or '',
|
|
'location_id': att.x_fclk_location_id.id or False,
|
|
})
|
|
|
|
local_today = get_local_today(request.env, employee)
|
|
today_start_utc, today_end_utc = get_local_day_boundaries(request.env, local_today, employee)
|
|
today_atts = request.env['hr.attendance'].sudo().search([
|
|
('employee_id', '=', employee.id),
|
|
('check_in', '>=', fields.Datetime.to_string(today_start_utc)),
|
|
('check_in', '<', fields.Datetime.to_string(today_end_utc)),
|
|
('check_out', '!=', False),
|
|
])
|
|
result['today_hours'] = round(sum(a.x_fclk_net_hours or 0 for a in today_atts), 2)
|
|
|
|
today = get_local_today(request.env, employee)
|
|
week_start = today - timedelta(days=today.weekday())
|
|
week_start_utc, _ = get_local_day_boundaries(request.env, week_start, employee)
|
|
week_atts = request.env['hr.attendance'].sudo().search([
|
|
('employee_id', '=', employee.id),
|
|
('check_in', '>=', fields.Datetime.to_string(week_start_utc)),
|
|
('check_in', '<', fields.Datetime.to_string(today_end_utc)),
|
|
('check_out', '!=', False),
|
|
])
|
|
result['week_hours'] = round(sum(a.x_fclk_net_hours or 0 for a in week_atts), 2)
|
|
|
|
recent = request.env['hr.attendance'].sudo().search([
|
|
('employee_id', '=', employee.id),
|
|
('check_out', '!=', False),
|
|
], order='check_in desc', limit=10)
|
|
result['recent_activity'] = [{
|
|
'id': a.id,
|
|
'check_in': fields.Datetime.to_string(a.check_in),
|
|
'check_out': fields.Datetime.to_string(a.check_out),
|
|
'worked_hours': round(a.worked_hours or 0, 2),
|
|
'net_hours': round(a.x_fclk_net_hours or 0, 2),
|
|
'location': a.x_fclk_location_id.name or '',
|
|
} for a in recent]
|
|
|
|
return result
|
|
|
|
@http.route('/fusion_clock/get_locations', type='jsonrpc', auth='user', methods=['POST'])
|
|
def get_locations(self, **kw):
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found for current user.'}
|
|
|
|
locations = self._get_locations_for_employee(employee)
|
|
default_id = employee.x_fclk_default_location_id.id if employee.x_fclk_default_location_id else False
|
|
|
|
return {
|
|
'locations': [{
|
|
'id': loc.id,
|
|
'name': loc.name,
|
|
'address': loc.address or '',
|
|
'latitude': loc.latitude,
|
|
'longitude': loc.longitude,
|
|
'radius': loc.radius,
|
|
'is_default': loc.id == default_id,
|
|
'require_photo': loc.require_photo,
|
|
} for loc in locations],
|
|
'default_location_id': default_id,
|
|
}
|
|
|
|
@http.route('/fusion_clock/get_settings', type='jsonrpc', auth='user', methods=['POST'])
|
|
def get_settings(self, **kw):
|
|
ICP = request.env['ir.config_parameter'].sudo()
|
|
return {
|
|
'enable_sounds': ICP.get_param('fusion_clock.enable_sounds', 'True') == 'True',
|
|
'google_maps_api_key': ICP.get_param('fusion_clock.google_maps_api_key', ''),
|
|
'default_clock_in': float(ICP.get_param('fusion_clock.default_clock_in_time', '9.0')),
|
|
'default_clock_out': float(ICP.get_param('fusion_clock.default_clock_out_time', '17.0')),
|
|
'target_daily_hours': 8.0,
|
|
'target_weekly_hours': 40.0,
|
|
'enable_photo': ICP.get_param('fusion_clock.enable_photo_verification', 'False') == 'True',
|
|
'enable_corrections': ICP.get_param('fusion_clock.enable_correction_requests', 'True') == 'True',
|
|
}
|
|
|
|
@http.route('/fusion_clock/dashboard_data', type='jsonrpc', auth='user', methods=['POST'])
|
|
def dashboard_data(self, **kw):
|
|
"""Return dashboard data for managers."""
|
|
user = request.env.user
|
|
is_manager = user.has_group('fusion_clock.group_fusion_clock_manager')
|
|
is_team_lead = user.has_group('fusion_clock.group_fusion_clock_team_lead')
|
|
|
|
if not is_manager and not is_team_lead:
|
|
return {'error': 'Access denied.'}
|
|
|
|
now = fields.Datetime.now()
|
|
today = get_local_today(request.env)
|
|
today_start, _ = get_local_day_boundaries(request.env, today)
|
|
|
|
Attendance = request.env['hr.attendance'].sudo()
|
|
Employee = request.env['hr.employee'].sudo()
|
|
|
|
# Filter employees by access
|
|
if is_manager:
|
|
employees = Employee.search([('x_fclk_enable_clock', '=', True)])
|
|
else:
|
|
employee = self._get_employee()
|
|
if not employee:
|
|
return {'error': 'No employee record found.'}
|
|
employees = Employee.search([
|
|
('parent_id', '=', employee.id),
|
|
('x_fclk_enable_clock', '=', True),
|
|
])
|
|
|
|
emp_ids = employees.ids
|
|
|
|
# Currently clocked in
|
|
open_atts = Attendance.search([
|
|
('employee_id', 'in', emp_ids),
|
|
('check_out', '=', False),
|
|
])
|
|
clocked_in = [{
|
|
'employee': a.employee_id.name,
|
|
'check_in': fields.Datetime.to_string(a.check_in),
|
|
'location': a.x_fclk_location_id.name or '',
|
|
} for a in open_atts]
|
|
|
|
# Today stats
|
|
today_atts = Attendance.search([
|
|
('employee_id', 'in', emp_ids),
|
|
('check_in', '>=', today_start),
|
|
])
|
|
present_ids = set(a.employee_id.id for a in today_atts)
|
|
|
|
ActivityLog = request.env['fusion.clock.activity.log'].sudo()
|
|
late_count = ActivityLog.search_count([
|
|
('employee_id', 'in', emp_ids),
|
|
('log_type', '=', 'late_clock_in'),
|
|
('log_date', '>=', today_start),
|
|
])
|
|
|
|
# Pending alerts
|
|
pending_reasons = Employee.search_count([
|
|
('id', 'in', emp_ids),
|
|
('x_fclk_pending_reason', '=', True),
|
|
])
|
|
pending_corrections = request.env['fusion.clock.correction'].sudo().search_count([
|
|
('employee_id', 'in', emp_ids),
|
|
('state', '=', 'pending'),
|
|
])
|
|
|
|
return {
|
|
'clocked_in': clocked_in,
|
|
'total_employees': len(emp_ids),
|
|
'present_count': len(present_ids),
|
|
'absent_count': len(emp_ids) - len(present_ids),
|
|
'late_count': late_count,
|
|
'pending_reasons': pending_reasons,
|
|
'pending_corrections': pending_corrections,
|
|
}
|