shopdb-flask/shopdb/core/api/search.py

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

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 _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')

# Shipped GE defaults. Settings override these per-site; identical fallbacks
# live here so this consumer works even if the settings seed has not run.
SERVICENOW_URL_DEFAULT = (
    'https://geaerospaceqa.service-now.com/now/nav/ui/search/'
    '0f8b85d0c7922010099a308dc7c2606a/params/search-term/{ticket}/'
    'global-search-data-config-id/c861cea2c7022010099a308dc7c26041/'
)
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()]

    servicenow_enabled = settings.get('servicenow_enabled')
    if servicenow_enabled is None:
        servicenow_enabled = True

    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

    return {
        'type': asset_type_name,
        'id': result_id,
        'title': display_name,
        'subtitle': subtitle,
        'location': location_name,
        'url': url,
        'relevance': relevance
    }


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:
        logger.error(f"Application search failed: {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:
        logger.error(f"KnowledgeBase search failed: {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:
        logger.error(f"Employee search failed: {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(
            Asset.isactive == True,
            _word_match(query, Asset.assetnumber, Asset.name,
                        Asset.serialnumber, Asset.notes,
                        Asset.gaugelabreference, Asset.maintenancereference)
        ).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.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:
        logger.error(f"Asset search failed: {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:
        logger.error(f"Measuring tool search failed: {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:
        logger.error(f"USB device search failed: {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:
        logger.error(f"Printed item search failed: {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:
        logger.error(f"Custom field search failed: {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:
        logger.error(f"IP search failed: {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,
                        'url': f'/network',
                        'relevance': 70
                    })
            except ValueError:
                continue
    except ImportError:
        pass
    except Exception as e:
        logger.error(f"Subnet search failed: {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:
        logger.error(f"Computer hostname search failed: {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:
        logger.error(f"Printer hostname search failed: {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:
        logger.error(f"Network device hostname search failed: {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,
                'url': f'/notifications',
                'relevance': min(int(base_relevance), 100),
                'ticketnumber': notif.ticketnumber,
                'iscurrent': notif.is_current
            })
    except ImportError:
        pass
    except Exception as e:
        logger.error(f"Notification search failed: {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:
        logger.error(f"Machine vendor/model/type search failed: {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:
        logger.error(f"Printer vendor/model/type search failed: {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:
        logger.error(f"Network device vendor/type search failed: {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'])
@jwt_required(optional=True)
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'])
@jwt_required(optional=True)
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}%'

    # Run all search domains
    results.extend(_search_applications(query, search_term))
    results.extend(_search_knowledgebase(query, search_term))
    results.extend(_search_employees(query, search_term))
    results.extend(_search_assets(query, search_term))
    results.extend(_search_measuringtools(query, search_term))
    results.extend(_search_usbdevices(query, search_term))
    results.extend(_search_printeditems(query, search_term))
    results.extend(_search_customfields(query, search_term))
    results.extend(_search_notifications(query, search_term))
    results.extend(_search_hostnames(query, search_term))
    results.extend(_search_vendor_model_type(query, search_term))

    # IP-specific searches
    if classification['is_ip']:
        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.
    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.
        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

    redirect = _check_smart_redirect(query, classification)
    if redirect:
        response_data['redirect'] = redirect

    return success_response(response_data)