shopdb-flask/plugins/printedparts/services/badges.py

"""Badge resolution for the printedparts plugin.

Same input contract as the USB plugin (deliberately copied, not imported -
cross-plugin imports break the shopdb.api-only contract):

- all digits            -> an SSO typed or scanned directly
- 0<digits>BZ           -> a physical badge wrapping a PayNo (keyboard-wedge
                           scanners emit this shape)
- anything else         -> unresolvable

Name lookup honors the site's employee directory mode (the same setting the
employees/usb plugins use):

- selfhosted: the employees plugin's directoryemployees table, by SSO. The
  table has no PayNo column, so a PayNo badge only resolves when the wrapped
  digits are themselves the SSO.
- external: the HR directory via employee_connection() - SSO badges by SSO,
  PayNo badges by PayNo (which also recovers the real SSO to record).

Policy (Setting printedparts_unknown_badge): 'deny' (default) rejects a badge
with no directory match; 'allow' records the SSO with an empty name.
"""

import logging
import re

from shopdb.api import Setting, employee_connection

logger = logging.getLogger(__name__)

_PAYNO_BADGE = re.compile(r'^0(\d+)BZ$', re.IGNORECASE)


class BadgeError(ValueError):
    """Raised when a badge cannot be accepted under the site policy."""


def _parse_badge(badge):
    """Return ('sso'|'payno', digits) or raise BadgeError on unknown shape.

    The shapes themselves live in shopdb.api (normalise_badge), because the USB
    kiosk reads the same physical badges and used to disagree: it took the
    string raw, so the PayNo wrapper resolved to nobody and an operator whose
    scanner emits it could take a part and not check out a drive.

    That also closed a subtler hole. This used str.isdigit(), which is true for
    Arabic-Indic and other decimal forms, so a badge written in them was
    accepted, resolved against int() to a REAL employee, and then stored as an
    SSO string that joins to nothing.
    """
    from shopdb.api import BadgeShapeError, badge_kind, normalise_badge

    try:
        return badge_kind(badge), normalise_badge(badge)
    except BadgeShapeError as bad:
        raise BadgeError(str(bad))


def _selfhosted_lookup(digits):
    """(sso, name) from the employees plugin directory, or None."""
    try:
        from plugins.employees.models import DirectoryEmployee
        from shopdb.api import db
        employee = db.session.get(DirectoryEmployee, int(digits))
        if employee:
            return digits, f'{employee.firstname} {employee.lastname}'.strip()
    except Exception:
        logger.exception('Self-hosted directory lookup failed for %s', digits)
    return None


def _external_lookup(kind, digits):
    """(sso, name) from the HR directory, or None. PayNo badges resolve to
    the employee's real SSO."""
    try:
        conn = employee_connection()
    except Exception:
        logger.exception('HR directory connection failed')
        return None
    try:
        with conn.cursor() as cursor:
            column = 'SSO' if kind == 'sso' else 'PayNo'
            cursor.execute(
                f'SELECT SSO, First_Name, Last_Name FROM employees '
                f'WHERE {column} = %s', (digits,))
            row = cursor.fetchone()
        if row:
            sso = str(row[0] if not isinstance(row, dict) else row['SSO'])
            first = row[1] if not isinstance(row, dict) else row['First_Name']
            last = row[2] if not isinstance(row, dict) else row['Last_Name']
            return sso, f'{(first or "").strip()} {(last or "").strip()}'.strip()
    except Exception:
        logger.exception('HR directory lookup failed for %s %s', kind, digits)
    finally:
        try:
            conn.close()
        except Exception:
            pass
    return None


def resolve_badge(badge):
    """Return (sso, name) for a scanned badge, enforcing the site policy.

    Raises BadgeError with a kiosk-displayable message when the badge shape is
    unrecognized or the policy denies an unmatched badge.
    """
    kind, digits = _parse_badge(badge)

    mode = (Setting.get('employee_directory_mode') or 'selfhosted').lower()
    if mode == 'external':
        resolved = _external_lookup(kind, digits)
    else:
        resolved = _selfhosted_lookup(digits)

    if resolved is None:
        policy = (Setting.get('printedparts_unknown_badge') or 'deny').lower()
        if policy != 'allow':
            raise BadgeError('Badge not recognized - see the parts team')
        return digits, ''
    return resolved