ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

display FQDN keying + kiosk dropdown + dispatcher + multi-word search

Browse Source
This commit is contained in:
Cameron Proudlock
2026-07-29 07:45:33 -04:00
parent 14c0d4be1e
commit 02147827f0
12 changed files with 408 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
shopdb/core/api/dashboarddefaults.py
View File

@@ -32,6 +32,18 @@ def _serialize(default):
return data
def _resolve_default(fqdn, ipaddress):
"""Find the active mapping for a display: FQDN first (stable), then IP."""
default = None
if fqdn:
default = DashboardDefault.query.filter_by(
fqdn=fqdn, isactive=True).first()
if not default and ipaddress:
default = DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first()
return default
@dashboarddefaults_bp.route('/visitor-location', methods=['GET'])
def visitor_location():
"""Resolve the business unit for the calling display by its IP.
@@ -39,16 +51,18 @@ def visitor_location():
Unauthenticated: kiosks/lobby displays hit this. Returns the mapped
business unit, or a null businessunitid when the IP is not mapped.
"""
fqdn = (request.args.get('fqdn') or '').strip().lower() or None
ipaddress = request.args.get('ipaddress') or _request_ip()
default = DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first()
default = _resolve_default(fqdn, ipaddress)
if not default:
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'businessunitid': None,
'businessunit': None,
})
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'businessunitid': default.businessunitid,
'businessunit': default.businessunit.businessunit
@@ -64,11 +78,12 @@ def display_role():
(dashboard / lobby / partskiosk), the frontend path it maps to, and the
business unit for the dashboard role. Unmapped IP -> null role.
"""
fqdn = (request.args.get('fqdn') or '').strip().lower() or None
ipaddress = request.args.get('ipaddress') or _request_ip()
default = DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first()
default = _resolve_default(fqdn, ipaddress)
if not default:
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'role': None,
'path': None,
@@ -76,6 +91,7 @@ def display_role():
'businessunit': None,
})
return success_response({
'fqdn': fqdn,
'ipaddress': ipaddress,
'role': default.displayrole,
'path': default.displaypath,
@@ -100,36 +116,42 @@ def list_defaults():
def create_default():
"""Create a visitor-IP -> business-unit mapping."""
data = request.get_json() or {}
ipaddress = (data.get('ipaddress') or '').strip()
fqdn = (data.get('fqdn') or '').strip().lower() or None
ipaddress = (data.get('ipaddress') or '').strip() or None
role = (data.get('displayrole') or 'dashboard').strip()
businessunitid = data.get('businessunitid')
if not ipaddress:
return error_response(ErrorCodes.VALIDATION_ERROR, 'ipaddress is required')
if not fqdn and not ipaddress:
return error_response(ErrorCodes.VALIDATION_ERROR,
'fqdn or ipaddress is required')
if role not in DISPLAY_ROLES:
return error_response(ErrorCodes.VALIDATION_ERROR,
f'displayrole must be one of {", ".join(DISPLAY_ROLES)}')
# Only the dashboard role needs a business unit.
# Only the dashboard role needs a business unit (location).
if role == 'dashboard':
if not businessunitid:
return error_response(ErrorCodes.VALIDATION_ERROR,
'businessunitid is required for the dashboard role')
'a location is required for the dashboard role')
if not db.session.get(BusinessUnit, businessunitid):
return error_response(ErrorCodes.NOT_FOUND, 'Business unit not found',
return error_response(ErrorCodes.NOT_FOUND, 'Location not found',
http_code=404)
else:
businessunitid = None
if DashboardDefault.query.filter_by(ipaddress=ipaddress, isactive=True).first():
if fqdn and DashboardDefault.query.filter_by(fqdn=fqdn, isactive=True).first():
return error_response(ErrorCodes.CONFLICT,
f"{fqdn} is already mapped", http_code=409)
if ipaddress and DashboardDefault.query.filter_by(
ipaddress=ipaddress, isactive=True).first():
return error_response(ErrorCodes.CONFLICT,
f"IP {ipaddress} is already mapped", http_code=409)
default = DashboardDefault(ipaddress=ipaddress, displayrole=role,
default = DashboardDefault(fqdn=fqdn, ipaddress=ipaddress, displayrole=role,
businessunitid=businessunitid,
description=data.get('description'))
db.session.add(default)
db.session.flush()
AuditLog.log('created', 'DashboardDefault', entityid=default.dashboarddefaultid,
entityname=ipaddress)
entityname=(fqdn or ipaddress))
db.session.commit()
return success_response(_serialize(default), message='Mapping created',
http_code=201)
@@ -156,17 +178,22 @@ def update_default(default_id):
return error_response(ErrorCodes.NOT_FOUND, 'Business unit not found',
http_code=404)
default.businessunitid = data['businessunitid']
if 'ipaddress' in data and data['ipaddress']:
default.ipaddress = data['ipaddress'].strip()
if 'fqdn' in data:
default.fqdn = (data.get('fqdn') or '').strip().lower() or None
if 'ipaddress' in data:
default.ipaddress = (data.get('ipaddress') or '').strip() or None
if 'description' in data:
default.description = data['description']
# Non-dashboard roles carry no business unit; dashboard needs one.
if not default.fqdn and not default.ipaddress:
return error_response(ErrorCodes.VALIDATION_ERROR,
'fqdn or ipaddress is required')
# Non-dashboard roles carry no location; dashboard needs one.
if default.displayrole != 'dashboard':
default.businessunitid = None
elif not default.businessunitid:
return error_response(ErrorCodes.VALIDATION_ERROR,
'businessunitid is required for the dashboard role')
'a location is required for the dashboard role')
db.session.commit()
return success_response(_serialize(default), message='Mapping updated')

117
shopdb/core/api/search.py
View File

@@ -23,6 +23,24 @@ logger = logging.getLogger(__name__)
search_bp = Blueprint('search', __name__)
def _word_match(query, *columns):
"""SQL clause matching rows that contain EVERY word of the query, each word
in any of the columns, in any order.
A plain `col.ilike('%CSF Roles%')` needs the exact contiguous phrase, so a
multi-word search returned nothing. This splits the query into words and ANDs
them (OR across the columns per word): 'CSF Roles' matches a row with 'CSF'
and 'Roles' anywhere in the searched fields.
"""
words = [w for w in query.split() if w]
if not words:
words = ['']
return db.and_(*[
db.or_(*[c.ilike(f'%{w}%') for c in columns])
for w in words
])
def _require_enabled(name):
"""Raise ImportError when the named plugin is disabled.
@@ -194,10 +212,7 @@ def _search_applications(query, search_term):
try:
apps = Application.query.filter(
Application.isactive == True,
db.or_(
Application.appname.ilike(search_term),
Application.appdescription.ilike(search_term)
)
_word_match(query, Application.appname, Application.appdescription)
).limit(10).all()
for app in apps:
@@ -228,10 +243,8 @@ def _search_knowledgebase(query, search_term):
from plugins.knowledgebase.models import KnowledgeBase
kb_articles = KnowledgeBase.query.filter(
KnowledgeBase.isactive == True,
db.or_(
KnowledgeBase.shortdescription.ilike(search_term),
KnowledgeBase.keywords.ilike(search_term)
)
_word_match(query, KnowledgeBase.shortdescription,
KnowledgeBase.keywords)
).limit(20).all()
for kb in kb_articles:
@@ -255,7 +268,7 @@ def _search_knowledgebase(query, search_term):
return results
def _employee_rows(search_term):
def _employee_rows(query):
"""Employee rows matching search_term, from the SELFHOSTED directory table
or the external HR DB per the employee_directory_mode setting. Returns dicts
shaped like the external query (SSO/First_Name/Last_Name/Team/Role)."""
@@ -267,25 +280,30 @@ def _employee_rows(search_term):
except ImportError:
return []
rows = DirectoryEmployee.query.filter(
db.or_(
DirectoryEmployee.firstname.ilike(search_term),
DirectoryEmployee.lastname.ilike(search_term),
db.cast(DirectoryEmployee.sso, db.String).ilike(search_term),
)
_word_match(query, DirectoryEmployee.firstname,
DirectoryEmployee.lastname,
db.cast(DirectoryEmployee.sso, db.String))
).order_by(DirectoryEmployee.lastname, DirectoryEmployee.firstname).limit(10).all()
return [{'SSO': e.sso, 'First_Name': e.firstname or '',
'Last_Name': e.lastname or '', 'Team': e.team, 'Role': e.role}
for e in rows]
# External HR DB: shared env-backed connection helper; never hardcode creds.
# Word-AND so "First Last" matches (each word in any name/SSO field), not just
# the exact contiguous phrase. Parameterized - one LIKE triple per word.
from shopdb.utils.employee_db import employee_connection
words = [w for w in query.split() if w] or ['']
clauses, params = [], []
for w in words:
clauses.append('(First_Name LIKE %s OR Last_Name LIKE %s '
'OR CAST(SSO AS CHAR) LIKE %s)')
like = f'%{w}%'
params += [like, like, like]
where = ' AND '.join(clauses)
emp_conn = employee_connection()
with emp_conn.cursor() as cur:
cur.execute('''
SELECT SSO, First_Name, Last_Name, Team, Role
FROM employees
WHERE First_Name LIKE %s OR Last_Name LIKE %s OR CAST(SSO AS CHAR) LIKE %s
ORDER BY Last_Name, First_Name LIMIT 10
''', (search_term, search_term, search_term))
cur.execute(
'SELECT SSO, First_Name, Last_Name, Team, Role FROM employees '
f'WHERE {where} ORDER BY Last_Name, First_Name LIMIT 10', params)
rows = cur.fetchall()
emp_conn.close()
return rows
@@ -295,7 +313,7 @@ def _search_employees(query, search_term):
"""Search employees (selfhosted directory or external HR DB, per mode)."""
results = []
try:
for emp in _employee_rows(search_term):
for emp in _employee_rows(query):
full_name = f"{emp['First_Name'].strip()} {emp['Last_Name'].strip()}"
sso_str = str(emp['SSO'])
@@ -329,12 +347,8 @@ def _search_assets(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Asset.assetnumber.ilike(search_term),
Asset.name.ilike(search_term),
Asset.serialnumber.ilike(search_term),
Asset.notes.ilike(search_term)
)
_word_match(query, Asset.assetnumber, Asset.name,
Asset.serialnumber, Asset.notes)
).limit(15).all()
for asset in assets:
@@ -372,12 +386,8 @@ def _search_measuringtools(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Asset.assetnumber.ilike(search_term),
Asset.name.ilike(search_term),
Asset.serialnumber.ilike(search_term),
Asset.gaugelabreference.ilike(search_term),
)
_word_match(query, Asset.assetnumber, Asset.name,
Asset.serialnumber, Asset.gaugelabreference)
).limit(15).all()
for asset in assets:
@@ -421,7 +431,7 @@ def _search_customfields(query, search_term):
Asset.isactive == True,
CustomField.searchable == True,
CustomField.isactive == True,
CustomFieldValue.value.ilike(search_term),
_word_match(query, CustomFieldValue.value),
).limit(15).all()
for asset, field in rows:
@@ -438,7 +448,7 @@ def _search_by_ip(query, search_term):
results = []
try:
comms = Communication.query.filter(
Communication.ipaddress.ilike(search_term)
_word_match(query, Communication.ipaddress)
).options(
joinedload(Communication.asset).joinedload(Asset.assettype),
joinedload(Communication.asset).joinedload(Asset.location),
@@ -498,7 +508,7 @@ def _search_hostnames(query, search_term):
_require_enabled('computers')
from plugins.computers.models import Computer
computers = Computer.query.filter(
Computer.hostname.ilike(search_term)
_word_match(query, Computer.hostname)
).options(
joinedload(Computer.asset).joinedload(Asset.assettype),
joinedload(Computer.asset).joinedload(Asset.location),
@@ -520,11 +530,8 @@ def _search_hostnames(query, search_term):
_require_enabled('printers')
from plugins.printers.models import Printer
printers = Printer.query.filter(
db.or_(
Printer.hostname.ilike(search_term),
Printer.sharename.ilike(search_term),
Printer.windowsname.ilike(search_term),
)
_word_match(query, Printer.hostname, Printer.sharename,
Printer.windowsname)
).options(
joinedload(Printer.asset).joinedload(Asset.assettype),
joinedload(Printer.asset).joinedload(Asset.location),
@@ -547,7 +554,7 @@ def _search_hostnames(query, search_term):
_require_enabled('network')
from plugins.network.models import NetworkDevice
devices = NetworkDevice.query.filter(
NetworkDevice.hostname.ilike(search_term)
_word_match(query, NetworkDevice.hostname)
).options(
joinedload(NetworkDevice.asset).joinedload(Asset.assettype),
joinedload(NetworkDevice.asset).joinedload(Asset.location),
@@ -577,10 +584,8 @@ def _search_notifications(query, search_term):
notifications = Notification.query.options(
joinedload(Notification.notificationtype)
).filter(
db.or_(
Notification.notification.ilike(search_term),
Notification.ticketnumber.ilike(search_term)
)
_word_match(query, Notification.notification,
Notification.ticketnumber)
).order_by(Notification.starttime.desc()).limit(15).all()
now = datetime.now(timezone.utc).replace(tzinfo=None)
@@ -635,11 +640,8 @@ def _search_vendor_model_type(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Vendor.vendor.ilike(search_term),
Model.modelnumber.ilike(search_term),
MachineType.machinetype.ilike(search_term)
)
_word_match(query, Vendor.vendor, Model.modelnumber,
MachineType.machinetype)
).limit(10).all()
for asset in machine_assets:
@@ -666,11 +668,8 @@ def _search_vendor_model_type(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Vendor.vendor.ilike(search_term),
Model.modelnumber.ilike(search_term),
PrinterType.printertype.ilike(search_term)
)
_word_match(query, Vendor.vendor, Model.modelnumber,
PrinterType.printertype)
).limit(10).all()
for asset in printer_assets:
@@ -695,10 +694,8 @@ def _search_vendor_model_type(query, search_term):
joinedload(Asset.location),
).filter(
Asset.isactive == True,
db.or_(
Vendor.vendor.ilike(search_term),
NetworkDeviceType.networkdevicetype.ilike(search_term)
)
_word_match(query, Vendor.vendor,
NetworkDeviceType.networkdevicetype)
).limit(10).all()
for asset in netdev_assets: