c1a4c9eba6
The last two items from the run. The PC detail page has always rendered a Business Unit row, and the API has always accepted the field. Only the form had no control, so it could be read and never set from the UI - a value that appeared the moment somebody set it through the API and could not be reached otherwise. The form has the picker now, loaded with listAll() rather than a capped page. Global search excluded inactive assets, so a retired record existed, was reachable by URL, and could not be found by anybody who did not already know where it was - including whoever wanted to restore it, which the retire dialog promises they can. Retired assets are searchable now. That is only safe with the ordering, so it is part of the change rather than a follow-up: somebody searching a machine number almost always wants the machine still on the floor, so a retired match drops 60 points of relevance, sorts below every live one, and carries a 'retired' flag the results page renders as a badge. Indistinguishable retired results would be worse than none. Four tests, verified by planting the isactive filter back: three fail. Scope: product
1374 lines
54 KiB
Python
1374 lines
54 KiB
Python
"""Global search API endpoint with full search parity."""
|
|
|
|
import re
|
|
import ipaddress
|
|
import logging
|
|
|
|
from datetime import datetime, timezone
|
|
from flask import Blueprint, request, current_app
|
|
from flask_jwt_extended import jwt_required
|
|
from sqlalchemy.orm import joinedload
|
|
|
|
from shopdb.extensions import db, cache
|
|
from shopdb.core.models import (
|
|
Application, Setting,
|
|
Asset, AssetType, Communication, Vendor, Model,
|
|
CustomField, CustomFieldValue
|
|
)
|
|
from shopdb.core.api.settings import get_cached_settings
|
|
from shopdb.core.services.spellfix import Vocabulary
|
|
from shopdb.utils.responses import success_response
|
|
from shopdb.utils.authz import require_permission, caller_holds
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
search_bp = Blueprint('search', __name__)
|
|
|
|
|
|
# EVERY searcher below truncates with .limit(). An unordered LIMIT lets the
|
|
# database return a DIFFERENT subset of the matching rows between two identical
|
|
# requests - the same search run twice came back with different hits, which reads
|
|
# as the search being broken rather than as a missing ORDER BY. Each query
|
|
# therefore ends with a TOTAL order: a display key plus the primary key, so ties
|
|
# cannot reorder. Relevance is applied afterwards in Python, over a stable set.
|
|
|
|
|
|
def _word_match(query, *columns, extra=None):
|
|
"""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.
|
|
|
|
`extra` is an optional callable taking one word and returning a further
|
|
clause to OR in for that word. Use it where a word may be satisfied by a
|
|
RELATED row rather than a column of this table - a knowledge base article
|
|
matching on its topic's name, say. Splitting per word matters there: without
|
|
it a word can only ever be matched by the columns, so a query naming the
|
|
topic and a keyword ('CMM Community') matches neither side alone.
|
|
"""
|
|
words = [w for w in query.split() if w]
|
|
if not words:
|
|
words = ['']
|
|
|
|
def per_word(word):
|
|
clauses = [c.ilike(f'%{word}%') for c in columns]
|
|
if extra is not None:
|
|
clauses.append(extra(word))
|
|
return db.or_(*clauses)
|
|
|
|
return db.and_(*[per_word(w) for w in words])
|
|
|
|
|
|
def _notefailure(domain, exc):
|
|
"""Record that one search domain failed this request, and log it.
|
|
|
|
Every domain below catches broadly and returns [] so one broken table cannot
|
|
take the whole search box down. That is right, but on its own it makes a
|
|
DOWN domain indistinguishable from an EMPTY one: the technician sees "no
|
|
results", is told they may have misspelled it, and concludes the record does
|
|
not exist. It does; the search just could not look.
|
|
|
|
The names collected here travel back in the response so the UI can say which
|
|
part of the search did not run.
|
|
"""
|
|
from flask import g
|
|
logger.error(f"{domain} search failed: {exc}")
|
|
failures = getattr(g, '_searchfailures', None)
|
|
if failures is None:
|
|
failures = g._searchfailures = []
|
|
if domain not in failures:
|
|
failures.append(domain)
|
|
|
|
|
|
def _searchfailures():
|
|
from flask import g
|
|
return list(getattr(g, '_searchfailures', []) or [])
|
|
|
|
|
|
def _require_enabled(name):
|
|
"""Raise ImportError when the named plugin is disabled.
|
|
|
|
Each plugin-scoped search block already catches ImportError and skips the
|
|
domain, so a disabled plugin is treated exactly like an absent one: its rows
|
|
drop out of search results. Honors runtime enable/disable.
|
|
"""
|
|
pm = current_app.extensions.get('plugin_manager')
|
|
if pm and not pm.registry.is_enabled(name):
|
|
raise ImportError(f'{name} plugin disabled')
|
|
|
|
# Fallbacks for when the settings seed has not run. The ServiceNow URL is BLANK
|
|
# on purpose and must stay that way: a hardcoded tenant here would link one
|
|
# site's technicians into another site's ServiceNow, and the code below already
|
|
# treats a blank URL as "ServiceNow not configured" and renders tickets as plain
|
|
# text. There is no sensible cross-site default for a tenant URL.
|
|
SERVICENOW_URL_DEFAULT = ''
|
|
SERVICENOW_PREFIXES_DEFAULT = 'GEINC,GECHG,GERIT,GESCT'
|
|
EMPLOYEEID_PATTERN_DEFAULT = r'^\d{9}$'
|
|
|
|
|
|
def _get_search_integrations():
|
|
"""Resolve the settings-driven search integration config.
|
|
|
|
Reads employeeid_pattern, servicenow_ticket_prefixes, servicenow_enabled
|
|
and servicenow_search_url from cached settings, falling back to the shipped
|
|
GE defaults for any missing key. An invalid employeeid_pattern regex falls
|
|
back to the default rather than raising (search must never 500 on bad
|
|
config). ServiceNow is inactive when disabled, when the URL is blank, or
|
|
when no ticket prefixes are configured.
|
|
"""
|
|
settings = get_cached_settings()
|
|
|
|
# Employee-ID pattern. Bad regex falls back so search never 500s.
|
|
pattern = settings.get('employeeid_pattern') or EMPLOYEEID_PATTERN_DEFAULT
|
|
try:
|
|
employeeid_re = re.compile(pattern)
|
|
except re.error:
|
|
employeeid_re = re.compile(EMPLOYEEID_PATTERN_DEFAULT)
|
|
|
|
# Ticket prefixes -> case-insensitive alternation built at request time.
|
|
prefixes_raw = settings.get('servicenow_ticket_prefixes')
|
|
if prefixes_raw is None:
|
|
prefixes_raw = SERVICENOW_PREFIXES_DEFAULT
|
|
prefixes = [p.strip() for p in str(prefixes_raw).split(',') if p.strip()]
|
|
|
|
# Missing means unconfigured, and unconfigured means OFF - matching the
|
|
# shipped default. Defaulting to True here made a site with no ServiceNow
|
|
# settings at all behave as though the integration were on.
|
|
servicenow_enabled = settings.get('servicenow_enabled')
|
|
if servicenow_enabled is None:
|
|
servicenow_enabled = False
|
|
|
|
servicenow_url = settings.get('servicenow_search_url')
|
|
if servicenow_url is None:
|
|
servicenow_url = SERVICENOW_URL_DEFAULT
|
|
|
|
servicenow_active = bool(servicenow_enabled) and bool(servicenow_url) and bool(prefixes)
|
|
|
|
prefix_re = None
|
|
if servicenow_active:
|
|
alternation = '|'.join(re.escape(p) for p in prefixes)
|
|
prefix_re = re.compile(r'^(' + alternation + r')\d+', re.IGNORECASE)
|
|
|
|
return {
|
|
'employeeid_re': employeeid_re,
|
|
'prefix_re': prefix_re,
|
|
'servicenow_active': servicenow_active,
|
|
'servicenow_url': servicenow_url,
|
|
}
|
|
|
|
|
|
def _classify_query(query, integrations):
|
|
"""Analyze the query string to determine its nature."""
|
|
prefix_re = integrations['prefix_re']
|
|
sn_match = prefix_re.match(query) if prefix_re else None
|
|
return {
|
|
'is_ip': bool(re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', query)),
|
|
'is_sso': bool(integrations['employeeid_re'].match(query)),
|
|
'is_servicenow': bool(sn_match),
|
|
'servicenow_prefix': sn_match.group(1) if sn_match else None,
|
|
'is_fqdn': bool(re.match(r'^[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+$', query)),
|
|
}
|
|
|
|
|
|
def _asset_presentation_map():
|
|
"""assettype -> route from enabled plugins' get_asset_presentation (ADR-010).
|
|
|
|
Lets core route a plugin-owned asset type in search rows without hardcoding
|
|
the route. Cached per request (built once per search, not per result).
|
|
"""
|
|
from flask import g
|
|
cached = getattr(g, '_asset_presentation_map', None)
|
|
if cached is not None:
|
|
return cached
|
|
result = {}
|
|
pm = current_app.extensions.get('plugin_manager')
|
|
if pm:
|
|
for plugin_name, plugin in pm.get_all_plugins().items():
|
|
if not pm.registry.is_enabled(plugin_name):
|
|
continue
|
|
try:
|
|
for entry in plugin.get_asset_presentation() or []:
|
|
if entry.get('assettype') and entry.get('route'):
|
|
result[entry['assettype']] = entry['route']
|
|
except Exception:
|
|
continue
|
|
g._asset_presentation_map = result
|
|
return result
|
|
|
|
|
|
def _get_asset_result(asset, query, relevance=None):
|
|
"""Build a search result dict from an Asset object."""
|
|
asset_type_name = asset.assettype.assettype if asset.assettype else 'asset'
|
|
|
|
plugin_id = asset.assetid
|
|
if asset_type_name == 'machine' and hasattr(asset, 'machine') and asset.machine:
|
|
plugin_id = asset.machine.machineid
|
|
elif asset_type_name == 'computer' and hasattr(asset, 'computer') and asset.computer:
|
|
plugin_id = asset.computer.computerid
|
|
elif asset_type_name == 'network_device' and hasattr(asset, 'network_device') and asset.network_device:
|
|
plugin_id = asset.network_device.networkdeviceid
|
|
elif asset_type_name == 'printer' and hasattr(asset, 'printer') and asset.printer:
|
|
plugin_id = asset.printer.printerid
|
|
elif asset_type_name == 'measuring_tool' and hasattr(asset, 'measuringtool') and asset.measuringtool:
|
|
plugin_id = asset.measuringtool.measuringtoolid
|
|
|
|
# Prefer a plugin's declared get_asset_presentation route (ADR-010): core
|
|
# stops hardcoding the plugin's route as each plugin declares one. Those
|
|
# routes take the core assetid (via a by-asset resolver). Types that have
|
|
# not declared fall back to the legacy id-keyed map below.
|
|
presentation = _asset_presentation_map()
|
|
if asset_type_name in presentation:
|
|
url = presentation[asset_type_name].replace('{assetid}', str(asset.assetid))
|
|
result_id = asset.assetid
|
|
else:
|
|
url_map = {
|
|
'machine': f"/machines/{plugin_id}",
|
|
'computer': f"/pcs/{plugin_id}",
|
|
'network_device': f"/network/{plugin_id}",
|
|
'printer': f"/printers/{plugin_id}",
|
|
'measuring_tool': f"/measuringtools/{plugin_id}",
|
|
}
|
|
url = url_map.get(asset_type_name, f"/assets/{asset.assetid}")
|
|
result_id = plugin_id
|
|
|
|
display_name = asset.display_name
|
|
subtitle = None
|
|
if asset.name and asset.assetnumber != asset.name:
|
|
subtitle = asset.assetnumber
|
|
|
|
location_name = asset.location.locationname if asset.location else None
|
|
|
|
if relevance is None:
|
|
relevance = 15
|
|
|
|
# A retired asset never outranks a live one, whatever it matched on: the
|
|
# person searching almost always wants the machine that is still on the
|
|
# floor. Flagged too, so the result can say so rather than looking like any
|
|
# other hit.
|
|
# getattr, because this builder is also called with lightweight
|
|
# stand-ins that carry only the fields a result needs. An object that does
|
|
# not say is treated as live, which is the safe direction: a live asset
|
|
# ranked correctly beats a live asset buried as retired.
|
|
retired = getattr(asset, 'isactive', True) is False
|
|
if retired:
|
|
relevance = max(1, relevance - 60)
|
|
|
|
return {
|
|
'type': asset_type_name,
|
|
'id': result_id,
|
|
'title': display_name,
|
|
'subtitle': subtitle,
|
|
'location': location_name,
|
|
'url': url,
|
|
'relevance': relevance,
|
|
'retired': retired
|
|
}
|
|
|
|
|
|
def _search_applications(query, search_term):
|
|
"""Search Applications by name and description."""
|
|
results = []
|
|
try:
|
|
apps = Application.query.filter(
|
|
Application.isactive == True,
|
|
_word_match(query, Application.appname, Application.appdescription)
|
|
).order_by(Application.appname, Application.appid).limit(10).all()
|
|
|
|
for app in apps:
|
|
relevance = 20
|
|
if query.lower() == app.appname.lower():
|
|
relevance = 100
|
|
elif query.lower() in app.appname.lower():
|
|
relevance = 50
|
|
|
|
results.append({
|
|
'type': 'application',
|
|
'id': app.appid,
|
|
'title': app.appname,
|
|
'subtitle': app.appdescription[:100] if app.appdescription else None,
|
|
'url': f"/applications/{app.appid}",
|
|
'relevance': relevance
|
|
})
|
|
except Exception as e:
|
|
_notefailure("Application", e)
|
|
return results
|
|
|
|
|
|
def _search_guides(query, search_term):
|
|
"""Search guides by title, summary, tags and BODY.
|
|
|
|
The body matters more here than in most domains. People look for a procedure
|
|
by something written halfway down it - "the one about the drive letter" -
|
|
not by a title they never read. The body is a JSON blob of blocks, so this
|
|
is a LIKE against it: crude, but it finds the guide, and a guide corpus is
|
|
tens of documents rather than millions of rows.
|
|
|
|
PUBLISHED ONLY, and published TEXT only. A draft is deliberately invisible
|
|
to readers, and search is the one place that rule would otherwise leak -
|
|
somebody would find a half-written procedure through the search box and
|
|
follow it.
|
|
|
|
Restricting to published GUIDES is not enough on its own, which is how this
|
|
first shipped: title/summary/tags/blocksjson on the Guide row are the
|
|
WORKING copy. A published guide with unpublished edits would match on a word
|
|
that exists only in the draft, and hand the reader a hit that is not in the
|
|
page when they open it. So the match, the title and the summary all come
|
|
from the published revision.
|
|
"""
|
|
results = []
|
|
try:
|
|
_require_enabled('guides')
|
|
from plugins.guides.models import Guide, GuideRevision
|
|
|
|
published = db.aliased(GuideRevision)
|
|
rows = db.session.query(Guide, published).join(
|
|
published, Guide.publishedrevisionid == published.guiderevisionid
|
|
).filter(
|
|
Guide.isactive == True,
|
|
_word_match(query, published.title, published.summary,
|
|
published.tags, published.blocksjson)
|
|
).order_by(Guide.modifieddate.desc()).limit(20).all()
|
|
|
|
for guide, revision in rows:
|
|
# A title match outranks a body match: someone searching "marker"
|
|
# wants the guide called that, not every guide mentioning one.
|
|
relevance = 10
|
|
if query.lower() in (revision.title or '').lower():
|
|
relevance += 20
|
|
if revision.tags and query.lower() in revision.tags.lower():
|
|
relevance += 15
|
|
|
|
results.append({
|
|
'type': 'guide',
|
|
'id': guide.guideid,
|
|
'title': revision.title,
|
|
'subtitle': revision.summary,
|
|
'url': f"/guides/{guide.slug}",
|
|
'relevance': relevance
|
|
})
|
|
except ImportError:
|
|
pass # guides plugin absent or disabled
|
|
except Exception as e:
|
|
_notefailure("Guides", e)
|
|
return results
|
|
|
|
|
|
def _search_knowledgebase(query, search_term):
|
|
"""Search Knowledge Base by description and keywords.
|
|
|
|
An article whose topic is a RETIRED application is excluded, matching
|
|
GET /api/knowledgebase. Filtering it out of the plugin's own listing while
|
|
global search still returned it is not a rule at all: the article was two
|
|
keystrokes away, and the result printed the retired application's name as its
|
|
subject, which reads as though it were still in service.
|
|
|
|
A null topic still matches. Not every article is about an application.
|
|
|
|
The topic's NAME is searched too, per word. An article tagged 'community'
|
|
under the CMM topic has neither word in both of its own columns, so a search
|
|
for 'CMM Community' found it in neither the plugin's listing nor here.
|
|
"""
|
|
results = []
|
|
try:
|
|
_require_enabled('knowledgebase')
|
|
from plugins.knowledgebase.models import KnowledgeBase
|
|
retired = db.session.query(Application.appid).filter(
|
|
Application.isactive.is_(False))
|
|
|
|
def topic_named(word):
|
|
"""Articles whose topic is an ACTIVE application named like word.
|
|
|
|
Active only, for the same reason the retired filter above exists: a
|
|
decommissioned application is not a topic anyone should be offered.
|
|
"""
|
|
return KnowledgeBase.appid.in_(
|
|
db.session.query(Application.appid).filter(
|
|
Application.appname.ilike(f'%{word}%'),
|
|
Application.isactive.is_(True)))
|
|
|
|
kb_articles = KnowledgeBase.query.filter(
|
|
KnowledgeBase.isactive == True,
|
|
db.or_(KnowledgeBase.appid.is_(None),
|
|
KnowledgeBase.appid.notin_(retired)),
|
|
_word_match(query, KnowledgeBase.shortdescription,
|
|
KnowledgeBase.keywords, extra=topic_named)
|
|
).order_by(KnowledgeBase.clicks.desc(),
|
|
KnowledgeBase.linkid).limit(20).all()
|
|
|
|
for kb in kb_articles:
|
|
relevance = 10 + (kb.clicks or 0) * 0.1
|
|
if kb.keywords and query.lower() in kb.keywords.lower():
|
|
relevance += 15
|
|
|
|
results.append({
|
|
'type': 'knowledgebase',
|
|
'id': kb.linkid,
|
|
'title': kb.shortdescription,
|
|
'subtitle': kb.application.appname if kb.application else None,
|
|
'url': f"/knowledgebase/{kb.linkid}",
|
|
'linkurl': kb.linkurl,
|
|
'relevance': relevance
|
|
})
|
|
except ImportError:
|
|
pass # knowledgebase plugin absent or disabled
|
|
except Exception as e:
|
|
_notefailure("KnowledgeBase", e)
|
|
return results
|
|
|
|
|
|
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)."""
|
|
settings = get_cached_settings()
|
|
mode = (settings.get('employee_directory_mode') or 'external').lower()
|
|
if mode == 'selfhosted':
|
|
try:
|
|
from plugins.employees.models import DirectoryEmployee
|
|
except ImportError:
|
|
return []
|
|
rows = DirectoryEmployee.query.filter(
|
|
_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 '
|
|
f'WHERE {where} ORDER BY Last_Name, First_Name LIMIT 10', params)
|
|
rows = cur.fetchall()
|
|
emp_conn.close()
|
|
return rows
|
|
|
|
|
|
def _search_employees(query, search_term):
|
|
"""Search employees (selfhosted directory or external HR DB, per mode)."""
|
|
results = []
|
|
try:
|
|
for emp in _employee_rows(query):
|
|
full_name = f"{emp['First_Name'].strip()} {emp['Last_Name'].strip()}"
|
|
sso_str = str(emp['SSO'])
|
|
|
|
relevance = 20
|
|
if query == sso_str:
|
|
relevance = 100
|
|
elif query.lower() == full_name.lower():
|
|
relevance = 95
|
|
elif query.lower() in full_name.lower():
|
|
relevance = 60
|
|
|
|
results.append({
|
|
'type': 'employee',
|
|
'id': emp['SSO'],
|
|
'title': full_name,
|
|
'subtitle': emp.get('Team') or emp.get('Role') or f"SSO: {sso_str}",
|
|
'url': f"/employees/{emp['SSO']}",
|
|
'relevance': relevance
|
|
})
|
|
except Exception as e:
|
|
_notefailure("Employee", e)
|
|
return results
|
|
|
|
|
|
def _search_assets(query, search_term):
|
|
"""Search unified Assets table by number, name, serial, notes and the two
|
|
optional identifiers.
|
|
|
|
gaugelabreference and maintenancereference are searched for EVERY asset type
|
|
(ADR-001). Settings lets a site enable either identifier on machines, PCs,
|
|
printers and network devices, but only the measuring-tools searcher looked at
|
|
gaugelabreference and nothing looked at maintenancereference at all - so a
|
|
tag an operator was told to record was one nobody could search by. An
|
|
identifier that can be entered has to be findable, or it is a write-only
|
|
field.
|
|
|
|
The per-type `identifier_<name>_<assettype>_enabled` toggles are NOT applied
|
|
here. They govern whether the field is SHOWN on that type; a value already in
|
|
the row is still the tag written on the physical machine, and matching it is
|
|
strictly better than returning nothing to someone reading it off a label.
|
|
"""
|
|
results = []
|
|
try:
|
|
assets = Asset.query.join(AssetType).options(
|
|
joinedload(Asset.assettype),
|
|
joinedload(Asset.location),
|
|
).filter(
|
|
# Retired assets are INCLUDED, and ranked below live ones. Retire
|
|
# is a soft delete whose confirm dialog promises the record can be
|
|
# restored - and the one place somebody looks for a machine they
|
|
# retired last month is the search box. Excluding them meant the
|
|
# record existed, was reachable by URL, and could not be found by
|
|
# anybody who did not already know where it was.
|
|
_word_match(query, Asset.assetnumber, Asset.name,
|
|
Asset.serialnumber, Asset.notes,
|
|
Asset.gaugelabreference, Asset.maintenancereference)
|
|
).order_by(Asset.isactive.desc(), Asset.assetnumber,
|
|
Asset.assetid).limit(15).all()
|
|
|
|
for asset in assets:
|
|
relevance = 15
|
|
if asset.assetnumber and query.lower() == asset.assetnumber.lower():
|
|
relevance = 100
|
|
elif asset.name and query.lower() == asset.name.lower():
|
|
relevance = 90
|
|
elif asset.gaugelabreference and query.lower() == asset.gaugelabreference.lower():
|
|
relevance = 88
|
|
elif asset.maintenancereference and query.lower() == asset.maintenancereference.lower():
|
|
relevance = 86
|
|
elif asset.serialnumber and query.lower() == asset.serialnumber.lower():
|
|
relevance = 85
|
|
elif asset.name and query.lower() in asset.name.lower():
|
|
relevance = 50
|
|
|
|
results.append(_get_asset_result(asset, query, relevance))
|
|
except Exception as e:
|
|
_notefailure("Asset", e)
|
|
return results
|
|
|
|
|
|
def _search_measuringtools(query, search_term):
|
|
"""Search measuring tools, including the gaugelabreference identifier.
|
|
|
|
The generic asset search already matches number/name/serial across all asset
|
|
types; this gated searcher adds gaugelabreference (the gage tag the gage lab
|
|
searches by) and drops out when the measuringtools plugin is disabled.
|
|
"""
|
|
results = []
|
|
try:
|
|
_require_enabled('measuringtools')
|
|
from plugins.measuringtools.models import MeasuringTool
|
|
assets = db.session.query(Asset).join(
|
|
MeasuringTool, MeasuringTool.assetid == Asset.assetid
|
|
).options(
|
|
joinedload(Asset.assettype),
|
|
joinedload(Asset.location),
|
|
).filter(
|
|
Asset.isactive == True,
|
|
_word_match(query, Asset.assetnumber, Asset.name,
|
|
Asset.serialnumber, Asset.gaugelabreference)
|
|
).order_by(Asset.assetnumber, Asset.assetid).limit(15).all()
|
|
|
|
for asset in assets:
|
|
relevance = 15
|
|
if asset.assetnumber and query.lower() == asset.assetnumber.lower():
|
|
relevance = 100
|
|
elif asset.gaugelabreference and query.lower() == asset.gaugelabreference.lower():
|
|
relevance = 90
|
|
elif asset.serialnumber and query.lower() == asset.serialnumber.lower():
|
|
relevance = 85
|
|
elif asset.name and query.lower() in asset.name.lower():
|
|
relevance = 50
|
|
results.append(_get_asset_result(asset, query, relevance))
|
|
except ImportError:
|
|
pass # measuringtools plugin absent or disabled
|
|
except Exception as e:
|
|
_notefailure("Measuring tool", e)
|
|
return results
|
|
|
|
|
|
def _search_usbdevices(query, search_term):
|
|
"""Search USB devices by serial, asset tag and product name.
|
|
|
|
A USB device is NOT an asset - it lives in the usb plugin's own table - so
|
|
the generic asset search cannot see it and these records were unreachable
|
|
from search entirely. Serial number is the field people actually have in
|
|
hand: it is what is printed on the stick they are holding.
|
|
|
|
currentusername is deliberately NOT searched. It records who holds the
|
|
device, and making search a way to list what a named person has checked out
|
|
is a different feature from finding a device.
|
|
"""
|
|
results = []
|
|
try:
|
|
_require_enabled('usb')
|
|
from plugins.usb.models import USBDevice
|
|
devices = USBDevice.query.filter(
|
|
USBDevice.isactive == True,
|
|
_word_match(query, USBDevice.serialnumber, USBDevice.assetnumber,
|
|
USBDevice.label, USBDevice.productname)
|
|
).order_by(USBDevice.serialnumber,
|
|
USBDevice.usbdeviceid).limit(10).all()
|
|
|
|
for device in devices:
|
|
relevance = 20
|
|
if query.lower() == (device.serialnumber or '').lower():
|
|
relevance = 100
|
|
elif query.lower() == (device.assetnumber or '').lower():
|
|
relevance = 90
|
|
elif query.lower() == (device.label or '').lower():
|
|
relevance = 85
|
|
elif query.lower() in (device.label or '').lower():
|
|
relevance = 50
|
|
elif query.lower() in (device.productname or '').lower():
|
|
relevance = 40
|
|
|
|
results.append({
|
|
'type': 'usb_device',
|
|
'id': device.usbdeviceid,
|
|
'title': device.label or device.productname or device.serialnumber,
|
|
'subtitle': device.assetnumber or device.serialnumber,
|
|
'url': f'/usb/{device.usbdeviceid}',
|
|
'relevance': relevance,
|
|
})
|
|
except ImportError:
|
|
pass # usb plugin absent or disabled
|
|
except Exception as e:
|
|
_notefailure("USB device", e)
|
|
return results
|
|
|
|
|
|
def _search_printeditems(query, search_term):
|
|
"""Search printed items by bin code, gage-lab tag, name and description.
|
|
|
|
Printed items are their own records, not assets, so the generic asset search
|
|
never covered them. itemcode (the bin label, e.g. 3DP-0042) and gagelabtag
|
|
are both unique and both printed on physical labels, which makes them the
|
|
likeliest thing anyone types into search.
|
|
"""
|
|
results = []
|
|
try:
|
|
_require_enabled('printedparts')
|
|
from plugins.printedparts.models import PrintedItem
|
|
items = PrintedItem.query.filter(
|
|
PrintedItem.isactive == True,
|
|
_word_match(query, PrintedItem.itemcode, PrintedItem.gagelabtag,
|
|
PrintedItem.itemname, PrintedItem.itemdescription)
|
|
).order_by(PrintedItem.itemcode,
|
|
PrintedItem.printeditemid).limit(10).all()
|
|
|
|
for item in items:
|
|
relevance = 20
|
|
if query.lower() == (item.itemcode or '').lower():
|
|
relevance = 100
|
|
elif query.lower() == (item.gagelabtag or '').lower():
|
|
relevance = 95
|
|
elif query.lower() == (item.itemname or '').lower():
|
|
relevance = 90
|
|
elif query.lower() in (item.itemname or '').lower():
|
|
relevance = 50
|
|
|
|
subtitle = item.itemcode or item.gagelabtag
|
|
if item.binlocation:
|
|
subtitle = f'{subtitle} - {item.binlocation}' if subtitle else item.binlocation
|
|
|
|
results.append({
|
|
'type': 'printed_item',
|
|
'id': item.printeditemid,
|
|
'title': item.itemname,
|
|
'subtitle': subtitle,
|
|
'url': f'/printedparts/{item.printeditemid}',
|
|
'relevance': relevance,
|
|
})
|
|
except ImportError:
|
|
pass # printedparts plugin absent or disabled
|
|
except Exception as e:
|
|
_notefailure("Printed item", e)
|
|
return results
|
|
|
|
|
|
def _search_customfields(query, search_term):
|
|
"""Search custom-field VALUES for fields flagged searchable.
|
|
|
|
Joins CustomFieldValue -> CustomField (searchable + active) -> Asset (active)
|
|
and emits a normal asset result routed to that asset's detail page. Results
|
|
are deduped against built-in-field asset matches by the shared (type, id) key
|
|
in global_search, and the asset's search_<type>_enabled domain toggle is
|
|
applied by the same end-of-request domain filter. The matched field label is
|
|
put in the subtitle so a value hit reads sensibly.
|
|
"""
|
|
results = []
|
|
try:
|
|
rows = db.session.query(Asset, CustomField).join(
|
|
CustomFieldValue, CustomFieldValue.assetid == Asset.assetid
|
|
).join(
|
|
CustomField, CustomField.fieldid == CustomFieldValue.fieldid
|
|
).options(
|
|
joinedload(Asset.assettype),
|
|
joinedload(Asset.location),
|
|
).filter(
|
|
Asset.isactive == True,
|
|
CustomField.searchable == True,
|
|
CustomField.isactive == True,
|
|
_word_match(query, CustomFieldValue.value),
|
|
).order_by(Asset.assetnumber, CustomField.fieldid).limit(15).all()
|
|
|
|
for asset, field in rows:
|
|
result = _get_asset_result(asset, query, relevance=40)
|
|
result['subtitle'] = field.label
|
|
results.append(result)
|
|
except Exception as e:
|
|
_notefailure("Custom field", e)
|
|
return results
|
|
|
|
|
|
def _search_by_ip(query, search_term):
|
|
"""Search Communications table for IP address matches."""
|
|
results = []
|
|
try:
|
|
comms = Communication.query.filter(
|
|
_word_match(query, Communication.ipaddress)
|
|
).options(
|
|
joinedload(Communication.asset).joinedload(Asset.assettype),
|
|
joinedload(Communication.asset).joinedload(Asset.location),
|
|
).order_by(Communication.ipaddress,
|
|
Communication.communicationid).limit(10).all()
|
|
|
|
seen_assets = set()
|
|
for comm in comms:
|
|
asset = comm.asset
|
|
if not asset or not asset.isactive or asset.assetid in seen_assets:
|
|
continue
|
|
seen_assets.add(asset.assetid)
|
|
|
|
relevance = 80 if query == comm.ipaddress else 40
|
|
result = _get_asset_result(asset, query, relevance)
|
|
result['subtitle'] = comm.ipaddress
|
|
results.append(result)
|
|
except Exception as e:
|
|
_notefailure("IP", e)
|
|
return results
|
|
|
|
|
|
def _search_subnets(query):
|
|
"""Find which subnet an IP address belongs to."""
|
|
results = []
|
|
try:
|
|
_require_enabled('network')
|
|
from plugins.network.models import Subnet
|
|
ip_obj = ipaddress.ip_address(query)
|
|
subnets = Subnet.query.filter(Subnet.isactive == True).all()
|
|
for subnet in subnets:
|
|
try:
|
|
network = ipaddress.ip_network(subnet.cidr, strict=False)
|
|
if ip_obj in network:
|
|
results.append({
|
|
'type': 'subnet',
|
|
'id': subnet.subnetid,
|
|
'title': f'{subnet.name} ({subnet.cidr})',
|
|
'subtitle': subnet.description or subnet.subnettype,
|
|
# the matched subnet, not the hub: searching an IP
|
|
# already identified WHICH subnet, so landing on the
|
|
# hub makes the reader find it again by hand
|
|
'url': f'/networks/{subnet.subnetid}',
|
|
'relevance': 70
|
|
})
|
|
except ValueError:
|
|
continue
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Subnet", e)
|
|
return results
|
|
|
|
|
|
def _search_hostnames(query, search_term):
|
|
"""Search hostname fields across Computer, Printer, NetworkDevice."""
|
|
results = []
|
|
|
|
# Search Computers
|
|
try:
|
|
_require_enabled('computers')
|
|
from plugins.computers.models import Computer
|
|
computers = Computer.query.filter(
|
|
_word_match(query, Computer.hostname)
|
|
).options(
|
|
joinedload(Computer.asset).joinedload(Asset.assettype),
|
|
joinedload(Computer.asset).joinedload(Asset.location),
|
|
).order_by(Computer.hostname, Computer.computerid).limit(10).all()
|
|
|
|
for comp in computers:
|
|
if comp.asset and comp.asset.isactive:
|
|
relevance = 85 if query.lower() == (comp.hostname or '').lower() else 40
|
|
result = _get_asset_result(comp.asset, query, relevance)
|
|
result['subtitle'] = comp.hostname
|
|
results.append(result)
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Computer hostname", e)
|
|
|
|
# Search Printers
|
|
try:
|
|
_require_enabled('printers')
|
|
from plugins.printers.models import Printer
|
|
printers = Printer.query.filter(
|
|
_word_match(query, Printer.hostname, Printer.sharename,
|
|
Printer.windowsname)
|
|
).options(
|
|
joinedload(Printer.asset).joinedload(Asset.assettype),
|
|
joinedload(Printer.asset).joinedload(Asset.location),
|
|
).order_by(Printer.hostname, Printer.printerid).limit(10).all()
|
|
|
|
for printer in printers:
|
|
if printer.asset and printer.asset.isactive:
|
|
match_field = printer.hostname or printer.sharename or ''
|
|
relevance = 85 if query.lower() == match_field.lower() else 40
|
|
result = _get_asset_result(printer.asset, query, relevance)
|
|
result['subtitle'] = printer.hostname or printer.sharename
|
|
results.append(result)
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Printer hostname", e)
|
|
|
|
# Search Network Devices
|
|
try:
|
|
_require_enabled('network')
|
|
from plugins.network.models import NetworkDevice
|
|
devices = NetworkDevice.query.filter(
|
|
_word_match(query, NetworkDevice.hostname)
|
|
).options(
|
|
joinedload(NetworkDevice.asset).joinedload(Asset.assettype),
|
|
joinedload(NetworkDevice.asset).joinedload(Asset.location),
|
|
).order_by(NetworkDevice.hostname,
|
|
NetworkDevice.networkdeviceid).limit(10).all()
|
|
|
|
for device in devices:
|
|
if device.asset and device.asset.isactive:
|
|
relevance = 85 if query.lower() == (device.hostname or '').lower() else 40
|
|
result = _get_asset_result(device.asset, query, relevance)
|
|
result['subtitle'] = device.hostname
|
|
results.append(result)
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Network device hostname", e)
|
|
|
|
return results
|
|
|
|
|
|
def _search_notifications(query, search_term):
|
|
"""Search notifications with time-weighted relevance."""
|
|
results = []
|
|
try:
|
|
_require_enabled('notifications')
|
|
from plugins.notifications.models import Notification
|
|
|
|
notifications = Notification.query.options(
|
|
joinedload(Notification.notificationtype)
|
|
).filter(
|
|
_word_match(query, Notification.notification,
|
|
Notification.ticketnumber)
|
|
).order_by(Notification.starttime.desc()).limit(15).all()
|
|
|
|
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
for notif in notifications:
|
|
base_relevance = 20
|
|
if notif.ticketnumber and query.lower() == notif.ticketnumber.lower():
|
|
base_relevance = 85
|
|
|
|
# Time-weighted relevance
|
|
if notif.is_current:
|
|
base_relevance *= 3
|
|
elif notif.starttime and notif.starttime > now:
|
|
base_relevance *= 2
|
|
elif notif.endtime and (now - notif.endtime).days < 7:
|
|
base_relevance = int(base_relevance * 1.5)
|
|
|
|
results.append({
|
|
'type': 'notification',
|
|
'id': notif.notificationid,
|
|
'title': notif.title,
|
|
'subtitle': notif.notificationtype.typename if notif.notificationtype else None,
|
|
# the notification itself, not the list it sits in: the
|
|
# `notification-detail` route renders one by id
|
|
'url': f'/notifications/{notif.notificationid}',
|
|
'relevance': min(int(base_relevance), 100),
|
|
'ticketnumber': notif.ticketnumber,
|
|
'iscurrent': notif.is_current
|
|
})
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Notification", e)
|
|
return results
|
|
|
|
|
|
def _search_vendor_model_type(query, search_term):
|
|
"""Search assets by vendor name, model name, or machine/device type name."""
|
|
results = []
|
|
|
|
# Machines: vendor, model, machinetype
|
|
try:
|
|
_require_enabled('machines')
|
|
from plugins.machines.models import Machine, MachineType
|
|
machine_assets = db.session.query(Asset).join(
|
|
Machine, Machine.assetid == Asset.assetid
|
|
).outerjoin(
|
|
Vendor, Machine.vendorid == Vendor.vendorid
|
|
).outerjoin(
|
|
Model, Machine.modelnumberid == Model.modelnumberid
|
|
).outerjoin(
|
|
MachineType, Machine.machinetypeid == MachineType.machinetypeid
|
|
).options(
|
|
joinedload(Asset.assettype),
|
|
joinedload(Asset.location),
|
|
).filter(
|
|
Asset.isactive == True,
|
|
_word_match(query, Vendor.vendor, Model.modelnumber,
|
|
MachineType.machinetype)
|
|
).order_by(Asset.assetnumber, Asset.assetid).limit(10).all()
|
|
|
|
for asset in machine_assets:
|
|
results.append(_get_asset_result(asset, query, 30))
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Machine vendor/model/type", e)
|
|
|
|
# Printers: vendor, model, printertype
|
|
try:
|
|
_require_enabled('printers')
|
|
from plugins.printers.models import Printer, PrinterType
|
|
printer_assets = db.session.query(Asset).join(
|
|
Printer, Printer.assetid == Asset.assetid
|
|
).outerjoin(
|
|
Vendor, Printer.vendorid == Vendor.vendorid
|
|
).outerjoin(
|
|
Model, Printer.modelnumberid == Model.modelnumberid
|
|
).outerjoin(
|
|
PrinterType, Printer.printertypeid == PrinterType.printertypeid
|
|
).options(
|
|
joinedload(Asset.assettype),
|
|
joinedload(Asset.location),
|
|
).filter(
|
|
Asset.isactive == True,
|
|
_word_match(query, Vendor.vendor, Model.modelnumber,
|
|
PrinterType.printertype)
|
|
).order_by(Asset.assetnumber, Asset.assetid).limit(10).all()
|
|
|
|
for asset in printer_assets:
|
|
results.append(_get_asset_result(asset, query, 30))
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Printer vendor/model/type", e)
|
|
|
|
# Network Devices: vendor, networkdevicetype
|
|
try:
|
|
_require_enabled('network')
|
|
from plugins.network.models import NetworkDevice, NetworkDeviceType
|
|
netdev_assets = db.session.query(Asset).join(
|
|
NetworkDevice, NetworkDevice.assetid == Asset.assetid
|
|
).outerjoin(
|
|
Vendor, NetworkDevice.vendorid == Vendor.vendorid
|
|
).outerjoin(
|
|
NetworkDeviceType, NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid
|
|
).options(
|
|
joinedload(Asset.assettype),
|
|
joinedload(Asset.location),
|
|
).filter(
|
|
Asset.isactive == True,
|
|
_word_match(query, Vendor.vendor,
|
|
NetworkDeviceType.networkdevicetype)
|
|
).order_by(Asset.assetnumber, Asset.assetid).limit(10).all()
|
|
|
|
for asset in netdev_assets:
|
|
results.append(_get_asset_result(asset, query, 30))
|
|
except ImportError:
|
|
pass
|
|
except Exception as e:
|
|
_notefailure("Network device vendor/type", e)
|
|
|
|
return results
|
|
|
|
|
|
def _check_smart_redirect(query, classification):
|
|
"""Check if query exactly matches a single entity for smart redirect."""
|
|
# Exact SSO match
|
|
if classification['is_sso']:
|
|
try:
|
|
# Shared env-backed connection helper; never hardcode creds.
|
|
from shopdb.utils.employee_db import employee_connection
|
|
emp_conn = employee_connection()
|
|
with emp_conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT SSO, First_Name, Last_Name FROM employees WHERE SSO = %s LIMIT 1',
|
|
(query,)
|
|
)
|
|
emp = cur.fetchone()
|
|
emp_conn.close()
|
|
if emp:
|
|
name = f"{emp['First_Name'].strip()} {emp['Last_Name'].strip()}"
|
|
return {
|
|
'type': 'employee',
|
|
'url': f"/employees/{emp['SSO']}",
|
|
'label': name
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# Exact asset number match
|
|
try:
|
|
asset = Asset.query.options(
|
|
joinedload(Asset.assettype),
|
|
).filter(
|
|
Asset.assetnumber == query,
|
|
Asset.isactive == True
|
|
).first()
|
|
if asset:
|
|
result = _get_asset_result(asset, query)
|
|
return {
|
|
'type': result['type'],
|
|
'url': result['url'],
|
|
'label': asset.display_name
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# Exact printer CSF/share name
|
|
try:
|
|
_require_enabled('printers')
|
|
from plugins.printers.models import Printer
|
|
printer = Printer.query.options(
|
|
joinedload(Printer.asset).joinedload(Asset.assettype),
|
|
).filter(
|
|
db.or_(
|
|
Printer.sharename == query,
|
|
Printer.windowsname == query
|
|
)
|
|
).first()
|
|
if printer and printer.asset and printer.asset.isactive:
|
|
return {
|
|
'type': 'printer',
|
|
'url': f"/printers/{printer.printerid}",
|
|
'label': printer.sharename or printer.asset.display_name
|
|
}
|
|
except ImportError:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
# Exact hostname match (FQDN or bare hostname)
|
|
hostname_plugins = []
|
|
try:
|
|
_require_enabled('computers')
|
|
from plugins.computers.models import Computer
|
|
hostname_plugins.append(('computer', Computer, 'computerid', '/pcs'))
|
|
except ImportError:
|
|
pass
|
|
try:
|
|
_require_enabled('printers')
|
|
from plugins.printers.models import Printer
|
|
hostname_plugins.append(('printer', Printer, 'printerid', '/printers'))
|
|
except ImportError:
|
|
pass
|
|
try:
|
|
_require_enabled('network')
|
|
from plugins.network.models import NetworkDevice
|
|
hostname_plugins.append(('network_device', NetworkDevice, 'networkdeviceid', '/network'))
|
|
except ImportError:
|
|
pass
|
|
|
|
for type_name, PluginModel, id_field, url_prefix in hostname_plugins:
|
|
try:
|
|
device = PluginModel.query.options(
|
|
joinedload(PluginModel.asset)
|
|
).filter(
|
|
PluginModel.hostname == query
|
|
).first()
|
|
if device and device.asset and device.asset.isactive:
|
|
return {
|
|
'type': type_name,
|
|
'url': f"{url_prefix}/{getattr(device, id_field)}",
|
|
'label': device.hostname
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
# Exact IP match
|
|
if classification['is_ip']:
|
|
try:
|
|
comm = Communication.query.options(
|
|
joinedload(Communication.asset).joinedload(Asset.assettype)
|
|
).filter(
|
|
Communication.ipaddress == query
|
|
).first()
|
|
if comm and comm.asset and comm.asset.isactive:
|
|
result = _get_asset_result(comm.asset, query)
|
|
return {
|
|
'type': result['type'],
|
|
'url': result['url'],
|
|
'label': f"{comm.asset.display_name} ({comm.ipaddress})"
|
|
}
|
|
except Exception:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
# How long a harvested vocabulary is reused. Long enough that a burst of
|
|
# searching does not rebuild it, short enough that something entered this
|
|
# morning is correctable before lunch.
|
|
_VOCABULARY_TTL = 600
|
|
|
|
# Per-column ceiling on harvested terms. A correction is only offered when a
|
|
# search found NOTHING, so the vocabulary exists to catch a typo against the
|
|
# names people actually use - not to mirror the database. Without a bound this
|
|
# would load every asset number on a site into memory to answer one query.
|
|
_VOCABULARY_LIMIT = 5000
|
|
|
|
|
|
def _search_vocabulary():
|
|
"""Words worth correcting a search against, harvested from the data.
|
|
|
|
From the DATA and not a dictionary, because no dictionary holds Genspect,
|
|
Telesis, Keyence or wax-trace - and because a vendor added this morning is
|
|
then correctable this morning.
|
|
|
|
Cached: this runs only on a search that found nothing, but a person who
|
|
mistypes once usually mistypes twice.
|
|
"""
|
|
cached = cache.get('search_vocabulary')
|
|
if cached is not None:
|
|
return cached
|
|
|
|
vocabulary = Vocabulary()
|
|
columns = (
|
|
(Asset, (Asset.assetnumber, Asset.name)),
|
|
(Vendor, (Vendor.vendor,)),
|
|
(Model, (Model.modelnumber,)),
|
|
(Application, (Application.appname,)),
|
|
(AssetType, (AssetType.assettype,)),
|
|
)
|
|
for model, fields in columns:
|
|
for field in fields:
|
|
try:
|
|
rows = (db.session.query(field)
|
|
.filter(field.isnot(None))
|
|
.distinct()
|
|
.limit(_VOCABULARY_LIMIT)
|
|
.all())
|
|
except Exception:
|
|
# A plugin table missing on a lean install must not break
|
|
# search. No vocabulary is worse than the alternative, not
|
|
# fatal.
|
|
logger.debug('vocabulary: skipped %s', field, exc_info=True)
|
|
continue
|
|
for (value,) in rows:
|
|
vocabulary.add(value)
|
|
|
|
cache.set('search_vocabulary', vocabulary, timeout=_VOCABULARY_TTL)
|
|
return vocabulary
|
|
|
|
|
|
def _suggestion_for(query):
|
|
"""A better spelling of `query`, or None. Never raises into a search."""
|
|
try:
|
|
return _search_vocabulary().suggest_query(query)
|
|
except Exception:
|
|
logger.warning('vocabulary lookup failed', exc_info=True)
|
|
return None
|
|
|
|
|
|
@search_bp.route('/suggest', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def search_suggest():
|
|
"""A better spelling for a query that found nothing, or null.
|
|
|
|
ITS OWN ENDPOINT, deliberately, rather than a `suggestion` key added to
|
|
every list response. Thirty routes take a search parameter across a dozen
|
|
files, so folding it into each one is a large change today and a thing every
|
|
future plugin has to remember. A page that finds no rows calls this instead;
|
|
a page that has not adopted it simply shows no suggestion, which is what it
|
|
showed before.
|
|
|
|
Returns null rather than a poor guess. The caller offers "did you mean" and
|
|
a person decides - nothing searches on their behalf.
|
|
"""
|
|
query = request.args.get('q', '').strip()
|
|
if not query or len(query) < 2 or len(query) > 200:
|
|
return success_response({'query': query, 'suggestion': None})
|
|
return success_response({
|
|
'query': query,
|
|
'suggestion': _suggestion_for(query),
|
|
})
|
|
|
|
|
|
@search_bp.route('', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def global_search():
|
|
"""
|
|
Global search across multiple entity types.
|
|
|
|
Returns combined results from assets, applications, knowledge base,
|
|
employees, notifications, IP addresses, hostnames, and vendor/model/type.
|
|
Supports smart redirects and ServiceNOW ticket detection.
|
|
"""
|
|
query = request.args.get('q', '').strip()
|
|
|
|
if not query or len(query) < 2:
|
|
return success_response({
|
|
'results': [],
|
|
'query': query,
|
|
'message': 'Search query must be at least 2 characters'
|
|
})
|
|
|
|
if len(query) > 200:
|
|
return success_response({
|
|
'results': [],
|
|
'query': query[:200],
|
|
'message': 'Search query too long'
|
|
})
|
|
|
|
integrations = _get_search_integrations()
|
|
classification = _classify_query(query, integrations)
|
|
|
|
# ServiceNOW prefix detection - return redirect immediately
|
|
if classification['is_servicenow']:
|
|
from urllib.parse import quote
|
|
servicenow_url = integrations['servicenow_url'].format(ticket=quote(query))
|
|
return success_response({
|
|
'results': [],
|
|
'query': query,
|
|
'total': 0,
|
|
'counts': {},
|
|
'redirect': {
|
|
'type': 'servicenow',
|
|
'url': servicenow_url,
|
|
'label': f'Open {query} in ServiceNOW'
|
|
}
|
|
})
|
|
|
|
results = []
|
|
search_term = f'%{query}%'
|
|
|
|
# Every domain answers only to a caller who holds ITS permission.
|
|
#
|
|
# The route is gated on assets.view alone, and that used to be the whole
|
|
# check: one permission then queried twelve domains, so it read out of the
|
|
# staff directory, the USB checkout history, the notification log and the
|
|
# network inventory to anybody who could look up an asset number. The
|
|
# SEARCH_DOMAINS settings are feature toggles a site sets, not authorization,
|
|
# and were never doing this job.
|
|
#
|
|
# caller_holds rather than require_permission: a search that refuses the
|
|
# whole request because one domain is off-limits is useless. Each domain
|
|
# simply contributes nothing.
|
|
domains = [
|
|
('assets.view', _search_assets),
|
|
# Emit asset results routed to an asset page, so they belong to the
|
|
# same grant as the assets themselves.
|
|
('assets.view', _search_customfields),
|
|
('assets.view', _search_vendor_model_type),
|
|
('applications.view', _search_applications),
|
|
('kb.view', _search_knowledgebase),
|
|
('guides.view', _search_guides),
|
|
('employees.view', _search_employees),
|
|
('measuringtools.view', _search_measuringtools),
|
|
('usb.view', _search_usbdevices),
|
|
('printedparts.view', _search_printeditems),
|
|
('notifications.view', _search_notifications),
|
|
('computers.view', _search_hostnames),
|
|
]
|
|
for permission, search in domains:
|
|
if caller_holds(permission):
|
|
results.extend(search(query, search_term))
|
|
|
|
# IP-specific searches. An address and the subnet it sits in are network
|
|
# inventory, and knowing what answers on 10.20.x is a different question
|
|
# from looking up an asset number.
|
|
if classification['is_ip'] and caller_holds('network.view'):
|
|
results.extend(_search_by_ip(query, search_term))
|
|
results.extend(_search_subnets(query))
|
|
|
|
# Sort by relevance (highest first)
|
|
results.sort(key=lambda x: x['relevance'], reverse=True)
|
|
|
|
# Remove duplicates (prefer higher relevance)
|
|
seen_ids = {}
|
|
unique_results = []
|
|
for r in results:
|
|
key = (r['type'], r['id'])
|
|
if key not in seen_ids:
|
|
seen_ids[key] = True
|
|
unique_results.append(r)
|
|
|
|
# Drop result types disabled in Settings > Search (search_<type>_enabled).
|
|
# Missing key = enabled. One query, default-on.
|
|
disabled_types = {
|
|
s.key[len('search_'):-len('_enabled')]
|
|
for s in Setting.query.filter_by(category='search').all()
|
|
if s.get_typed_value() is False
|
|
}
|
|
if disabled_types:
|
|
unique_results = [r for r in unique_results if r['type'] not in disabled_types]
|
|
|
|
# Compute type counts before truncation
|
|
type_counts = {}
|
|
for r in unique_results:
|
|
t = r['type']
|
|
type_counts[t] = type_counts.get(t, 0) + 1
|
|
|
|
total_all = len(unique_results)
|
|
|
|
# A search that found NOTHING is the only evidence of what people cannot
|
|
# find, and it is not otherwise recorded anywhere. Logged at INFO with a
|
|
# stable prefix so a week of it can be grepped into a list, which is what
|
|
# should decide whether spelling correction is worth wiring in - the recent
|
|
# multi-word and cross-field matching changes may already have fixed a good
|
|
# share of what used to fail. Guessing at that would be building for a
|
|
# problem nobody has measured.
|
|
degraded = _searchfailures()
|
|
|
|
suggestion = None
|
|
if not unique_results:
|
|
logger.info('search-no-results query=%r', query)
|
|
# Offered, never applied. The response carries the better spelling and
|
|
# the caller decides: silently searching for something else is how
|
|
# somebody orders the wrong cartridge.
|
|
#
|
|
# Still offered on a degraded run, because the response also carries
|
|
# `degraded` and the UI states it ABOVE the suggestion - the reader is
|
|
# told the search was incomplete before being offered a spelling.
|
|
#
|
|
# Suppressing it instead was tried and is worse: a domain that is
|
|
# chronically unreachable (an external HR directory, say) would silently
|
|
# disable spelling help for every search, for ever, and nothing would
|
|
# say why. Naming the failure is the honest fix; withholding the
|
|
# suggestion just removes a second useful signal.
|
|
suggestion = _suggestion_for(query)
|
|
|
|
# Limit total results
|
|
unique_results = unique_results[:50]
|
|
|
|
# Check for smart redirect
|
|
response_data = {
|
|
'results': unique_results,
|
|
'query': query,
|
|
'total': len(unique_results),
|
|
'total_all': total_all,
|
|
'counts': type_counts,
|
|
}
|
|
if suggestion:
|
|
response_data['suggestion'] = suggestion
|
|
if degraded:
|
|
# The UI says which part of the search did not run. Without this a
|
|
# broken domain reads as "that record does not exist".
|
|
response_data['degraded'] = degraded
|
|
|
|
redirect = _check_smart_redirect(query, classification)
|
|
if redirect:
|
|
response_data['redirect'] = redirect
|
|
|
|
return success_response(response_data)
|