diff --git a/.gitignore b/.gitignore index 9592a83e..ba374815 100644 --- a/.gitignore +++ b/.gitignore @@ -76,3 +76,6 @@ secrets.yml *_secrets credentials.json scripts/site_imports/wjf/idmap.json + +# work-PC publication clone: bundle drop folder for the transfer pipeline +_transfer/ diff --git a/plugins/printedparts/services/badges.py b/plugins/printedparts/services/badges.py index 81c7f430..13ea6462 100644 --- a/plugins/printedparts/services/badges.py +++ b/plugins/printedparts/services/badges.py @@ -8,18 +8,25 @@ cross-plugin imports break the shopdb.api-only contract): scanners emit this shape) - anything else -> unresolvable -Names come from the employees plugin's self-hosted directory, looked up by -SSO. The directory carries no PayNo column, so PayNo badges resolve only when -the wrapped digits are themselves the SSO (true at sites whose badges encode -the SSO); otherwise they fall to the unknown-badge policy. +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 +from shopdb.api import Setting, employee_connection + +logger = logging.getLogger(__name__) _PAYNO_BADGE = re.compile(r'^0(\d+)BZ$', re.IGNORECASE) @@ -28,17 +35,59 @@ class BadgeError(ValueError): """Raised when a badge cannot be accepted under the site policy.""" -def _directory_name(sso): - """Best-effort display name from the employees plugin directory.""" +def _parse_badge(badge): + """Return ('sso'|'payno', digits) or raise BadgeError on unknown shape.""" + badge = (badge or '').strip() + if not badge: + raise BadgeError('Scan or enter a badge') + if badge.isdigit(): + return 'sso', badge + match = _PAYNO_BADGE.match(badge) + if match: + return 'payno', match.group(1) + raise BadgeError('Unrecognized badge format') + + +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 - if sso and str(sso).isdigit(): - employee = db.session.get(DirectoryEmployee, int(sso)) - if employee: - return f'{employee.firstname} {employee.lastname}'.strip() + employee = db.session.get(DirectoryEmployee, int(digits)) + if employee: + return digits, f'{employee.firstname} {employee.lastname}'.strip() except Exception: - pass + 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 @@ -48,22 +97,17 @@ def resolve_badge(badge): Raises BadgeError with a kiosk-displayable message when the badge shape is unrecognized or the policy denies an unmatched badge. """ - badge = (badge or '').strip() - if not badge: - raise BadgeError('Scan or enter a badge') + kind, digits = _parse_badge(badge) - if badge.isdigit(): - sso = badge + mode = (Setting.get('employee_directory_mode') or 'selfhosted').lower() + if mode == 'external': + resolved = _external_lookup(kind, digits) else: - match = _PAYNO_BADGE.match(badge) - if not match: - raise BadgeError('Unrecognized badge format') - sso = match.group(1) + resolved = _selfhosted_lookup(digits) - name = _directory_name(sso) - if name is None: + 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 sso, '' - return sso, name + return digits, '' + return resolved