Odoo-Modules/fusion_clock/controllers/clock_nfc_kiosk.py

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)

import logging
import re
import time
import threading
from odoo import fields, http
from odoo.http import request

_logger = logging.getLogger(__name__)
_UID_HEX_PATTERN = re.compile(r'^[0-9A-F]+$')

_DEBOUNCE_WINDOW_SECONDS = 5.0
_recent_taps = {}  # {card_uid: monotonic_ts}
_recent_taps_lock = threading.Lock()


def _is_debounced(uid):
    """Return True if this UID was tapped within the debounce window."""
    now = time.monotonic()
    with _recent_taps_lock:
        last = _recent_taps.get(uid, 0)
        if now - last < _DEBOUNCE_WINDOW_SECONDS:
            return True
        _recent_taps[uid] = now
        # Opportunistic GC: drop entries older than 60s
        stale_keys = [k for k, t in _recent_taps.items() if now - t > 60]
        for k in stale_keys:
            _recent_taps.pop(k, None)
        return False


def _strip_data_url_prefix(b64):
    """Strip 'data:image/...;base64,' prefix from a data URL, returning raw base64."""
    if not b64:
        return b''
    if isinstance(b64, str) and b64.startswith('data:'):
        comma = b64.find(',')
        if comma >= 0:
            return b64[comma + 1:].encode('ascii', errors='ignore')
    return b64.encode('ascii', errors='ignore') if isinstance(b64, str) else b64


class FusionClockNfcKiosk(http.Controller):
    """NFC tap-to-clock kiosk controller. Reuses FusionClockAPI helpers."""

    @staticmethod
    def _normalize_uid(uid):
        """Normalize an NFC card UID to canonical hex (uppercase, colon-separated).

        Returns None if the input is empty or not valid hex.
        """
        if not uid:
            return None
        cleaned = uid.strip().upper().replace('-', '').replace(':', '').replace(' ', '')
        if not cleaned or not _UID_HEX_PATTERN.match(cleaned):
            return None
        if len(cleaned) % 2 != 0:
            return None
        return ':'.join(cleaned[i:i+2] for i in range(0, len(cleaned), 2))

    @http.route('/fusion_clock/kiosk/nfc', type='http', auth='user', website=True)
    def nfc_kiosk_page(self, **kw):
        """Render the NFC kiosk page for a wall-mounted tablet."""
        user = request.env.user
        if not user.has_group('fusion_clock.group_fusion_clock_manager'):
            return request.redirect('/my')

        ICP = request.env['ir.config_parameter'].sudo()
        if ICP.get_param('fusion_clock.enable_nfc_kiosk', 'False') != 'True':
            return request.redirect('/my')

        company = request.env.company
        location = company.x_fclk_nfc_kiosk_location_id
        values = {
            'page_name': 'nfc_kiosk',
            'company_name': company.name,
            'location_name': location.name if location else 'No location configured',
            'location_configured': bool(location),
            'photo_required': ICP.get_param('fusion_clock.nfc_photo_required', 'True') == 'True',
            'debug_enabled': ICP.get_param('fusion_clock.nfc_kiosk_debug', 'False') == 'True',
        }
        return request.render('fusion_clock.nfc_kiosk_page', values)

    @staticmethod
    def _check_enroll_password(env, supplied):
        """Verify the enroll-mode password. Empty config = always-allow for managers."""
        configured = env['ir.config_parameter'].sudo().get_param('fusion_clock.nfc_enroll_password', '')
        if not configured:
            return True
        return (supplied or '') == configured

    @http.route('/fusion_clock/kiosk/nfc/enroll', type='jsonrpc', auth='user', methods=['POST'])
    def nfc_enroll(self, employee_id=0, card_uid='', enroll_password='', **kw):
        """Bind an NFC card UID to an employee. Manager-gated, password-gated."""
        user = request.env.user
        if not user.has_group('fusion_clock.group_fusion_clock_manager'):
            return {'error': 'access_denied'}

        if not self._check_enroll_password(request.env, enroll_password):
            return {'error': 'invalid_password'}

        normalized = self._normalize_uid(card_uid)
        if not normalized:
            return {'error': 'invalid_uid'}

        Employee = request.env['hr.employee'].sudo()
        target = Employee.browse(int(employee_id or 0))
        if not target.exists():
            return {'error': 'employee_not_found'}

        existing = Employee.search([
            ('x_fclk_nfc_card_uid', '=', normalized),
            ('id', '!=', target.id),
        ], limit=1)
        if existing:
            return {
                'error': 'card_already_assigned',
                'existing_employee': existing.name,
            }

        target.x_fclk_nfc_card_uid = normalized

        # Activity log (uses 'card_enrollment' + 'nfc_kiosk' selections added in Task 2)
        request.env['fusion.clock.activity.log'].sudo().create({
            'employee_id': target.id,
            'log_type': 'card_enrollment',
            'description': f"NFC card {normalized} enrolled by {user.name}",
            'source': 'nfc_kiosk',
        })

        return {
            'success': True,
            'employee_name': target.name,
            'card_uid': normalized,
        }

    @http.route('/fusion_clock/kiosk/nfc/tap', type='jsonrpc', auth='user', methods=['POST'])
    def nfc_tap(self, card_uid='', photo_b64='', **kw):
        """Toggle attendance state for the employee owning this card UID."""
        user = request.env.user
        if not user.has_group('fusion_clock.group_fusion_clock_manager'):
            return {'error': 'access_denied'}

        ICP = request.env['ir.config_parameter'].sudo()
        if ICP.get_param('fusion_clock.enable_nfc_kiosk', 'False') != 'True':
            return {'error': 'kiosk_disabled'}

        normalized = self._normalize_uid(card_uid)
        if not normalized:
            return {'error': 'invalid_uid'}

        if _is_debounced(normalized):
            return {'error': 'debounce'}

        photo_required = ICP.get_param('fusion_clock.nfc_photo_required', 'True') == 'True'
        if photo_required and not photo_b64:
            return {'error': 'photo_required', 'message': 'Camera unavailable. Ask IT to check the kiosk.'}
        photo_bytes = _strip_data_url_prefix(photo_b64) if photo_b64 else b''

        company = request.env.company
        location = company.x_fclk_nfc_kiosk_location_id
        if not location:
            return {'error': 'no_location_configured'}

        Employee = request.env['hr.employee'].sudo()
        employee = Employee.search([('x_fclk_nfc_card_uid', '=', normalized)], limit=1)
        if not employee:
            _logger.warning("[nfc-kiosk] Unknown NFC card tapped: %s", normalized)
            return {'error': 'card_unknown', 'message': 'Card not enrolled. See your manager.'}

        if not employee.x_fclk_enable_clock:
            return {'error': 'clock_disabled', 'message': 'Clock disabled for this account.'}

        from .clock_api import FusionClockAPI
        api = FusionClockAPI()

        is_checked_in = employee.attendance_state == 'checked_in'
        now = fields.Datetime.now()
        today = now.date()

        geo_info = {
            'latitude': 0,
            'longitude': 0,
            'browser': 'nfc_kiosk',
            'ip_address': request.httprequest.remote_addr or '',
        }

        attendance = employee.sudo()._attendance_action_change(geo_info)

        if not is_checked_in:
            attendance.sudo().write({
                'x_fclk_location_id': location.id,
                'x_fclk_in_distance': 0.0,
                'x_fclk_clock_source': 'nfc_kiosk',
                'x_fclk_check_in_photo': photo_bytes if photo_bytes else False,
            })
            api._log_activity(
                employee, 'clock_in',
                f"NFC kiosk clock-in at {location.name}",
                attendance=attendance, location=location,
                latitude=0, longitude=0, distance=0,
                source='nfc_kiosk',
            )
            scheduled_in, _ = api._get_scheduled_times(employee, today)
            api._check_and_create_penalty(employee, attendance, 'late_in', scheduled_in, now)
            return {
                'success': True,
                'action': 'clock_in',
                'employee_name': employee.name,
                'employee_avatar_url': f'/web/image/hr.employee/{employee.id}/avatar_128',
                'message': f'{employee.name} clocked in at {location.name}',
                'net_hours_today': 0.0,
            }
        else:
            attendance.sudo().write({
                'x_fclk_out_distance': 0.0,
                'x_fclk_check_out_photo': photo_bytes if photo_bytes else False,
            })
            api._apply_break_deduction(attendance, employee)
            _, scheduled_out = api._get_scheduled_times(employee, today)
            api._check_and_create_penalty(employee, attendance, 'early_out', scheduled_out, now)
            api._log_activity(
                employee, 'clock_out',
                f"NFC kiosk clock-out from {location.name}. Net: {attendance.x_fclk_net_hours:.1f}h",
                attendance=attendance, location=location,
                latitude=0, longitude=0, distance=0,
                source='nfc_kiosk',
            )
            return {
                'success': True,
                'action': 'clock_out',
                'employee_name': employee.name,
                'employee_avatar_url': f'/web/image/hr.employee/{employee.id}/avatar_128',
                'message': f'{employee.name} clocked out',
                'net_hours_today': round(attendance.x_fclk_net_hours or 0, 2),
            }

    @http.route('/fusion_clock/kiosk/nfc/employee_search', type='jsonrpc', auth='user', methods=['POST'])
    def nfc_employee_search(self, query='', **kw):
        """Delegate to the existing kiosk search to avoid duplication."""
        from .clock_kiosk import FusionClockKiosk
        return FusionClockKiosk().kiosk_search(query=query)