printedparts: badge resolution honors the employee directory mode

At sites running the external HR directory the self-hosted table is
empty, so every kiosk badge fell to the deny policy. The resolver now
branches on employee_directory_mode: selfhosted reads the directory
table by SSO; external queries the HR directory, resolving PayNo
badges by their real PayNo column.
This commit is contained in:
2026-07-17 10:39:12 -04:00
parent 11c8d79ba4
commit 503c0cfcf4
2 changed files with 72 additions and 25 deletions

3
.gitignore vendored
View File

@@ -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/

View File

@@ -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