From a515c28e3b8351800b79f8a354f91e7054fe24b4 Mon Sep 17 00:00:00 2001 From: cproudlock Date: Fri, 26 Jun 2026 20:00:10 -0400 Subject: [PATCH] Decouple core from plugins; wire widgets hook; drop dead search hook Architectural pass from the skill review ("plugin is the product" boundary). Core no longer imports plugin models at module load (was a hard import-time dependency that broke core if the computers plugin was absent/disabled): - collector.py, applications.py, reports.py: lazy + guarded imports of the computers plugin models. Endpoints that need install-tracking now return 503 when the plugin is absent instead of failing at import. Search honors runtime enable/disable: - search.py: _require_enabled(name) raises ImportError for a disabled plugin, so each plugin-scoped block skips it (a disabled plugin's rows leave search). - Replace hardcoded root/rootpassword employee-DB connection in _search_employees with the shared env-backed employee_connection helper. Plugin hooks (integrating-plugin-hooks: every hook needs a consumer): - get_dashboard_widgets: add the consumer GET /api/dashboard/widgets (5 plugins already implemented the hook; it had none). Skips disabled, isolates in prod. - get_searchable_fields: REMOVED. Zero plugins implemented it and there was no consumer; global search is a core concern over the asset model. Contract reduction, __contract_version__ 0.3.0 -> 0.4.0. Docs/contract: PLUGIN-HOOKS.md (widgets consumer note, searchable-fields removal, 0.4.0), PLUGIN-QUICKSTART.md, ADR-001 hook list. Tests: widgets endpoint aggregate + disabled-skip; contract tests for the removed/added hooks. 151 tests pass, naming/style green, app boots all 6 plugins. Co-Authored-By: Claude Opus 4.8 --- docs/PLUGIN-HOOKS.md | 28 +- docs/PLUGIN-QUICKSTART.md | 3 +- .../adr/ADR-001-asset-as-platform-contract.md | 2 +- shopdb/__init__.py | 5 +- shopdb/core/api/applications.py | 69 +- shopdb/core/api/collector.py | 29 +- shopdb/core/api/dashboard.py | 31 + shopdb/core/api/reports.py | 11 +- shopdb/core/api/search.py | 1473 +++++++++-------- shopdb/plugins/base.py | 28 - tests/test_core/test_dashboard_widgets.py | 38 + tests/test_plugin_contract.py | 15 +- 12 files changed, 940 insertions(+), 792 deletions(-) create mode 100644 tests/test_core/test_dashboard_widgets.py diff --git a/docs/PLUGIN-HOOKS.md b/docs/PLUGIN-HOOKS.md index 422789a..5eb16d0 100644 --- a/docs/PLUGIN-HOOKS.md +++ b/docs/PLUGIN-HOOKS.md @@ -9,7 +9,7 @@ The contract is locked in [ADR-001](../docs/adr/ADR-001-asset-as-platform-contra The framework declares its contract version in `shopdb/__init__.py`: ```python -__contract_version__ = '0.3.0' +__contract_version__ = '0.4.0' ``` Each plugin's `manifest.json` declares the range of contract versions it supports: @@ -166,6 +166,10 @@ class NotificationsPlugin(BasePlugin): }] ``` +Consumed by `GET /api/dashboard/widgets`, which merges widgets from all enabled +plugins sorted by `position` (disabled plugins are skipped; a broken plugin is +isolated in prod, re-raised in dev/test). + ### `get_navigation_items() -> List[Dict]` Returns navigation menu items. @@ -181,24 +185,10 @@ class ComputersPlugin(BasePlugin): }] ``` -### `get_searchable_fields() -> List[Dict]` - -Declares fields the plugin contributes to global search. - -```python -from .models import Computer - -class ComputersPlugin(BasePlugin): - def get_searchable_fields(self): - return [{ - 'model': Computer, - 'search_fields': ['hostname', 'serialnumber', 'currentuser'], - 'result_type': 'computer', - 'url_template': '/computers/{id}', - 'title_field': 'hostname', - 'subtitle_field': 'currentuser', - }] -``` +> Removed in contract 0.4.0: `get_searchable_fields`. Global search +> (`/api/search`) is a core concern that queries the asset model directly and +> already covers every bundled asset type; no plugin ever implemented the hook. +> Search honors runtime plugin enable/disable. ### `get_collector_schema() -> Optional[Dict]` diff --git a/docs/PLUGIN-QUICKSTART.md b/docs/PLUGIN-QUICKSTART.md index 1c38c6b..04b0ca6 100644 --- a/docs/PLUGIN-QUICKSTART.md +++ b/docs/PLUGIN-QUICKSTART.md @@ -119,10 +119,9 @@ Override hooks on the plugin class as needed. See [PLUGIN-HOOKS.md](PLUGIN-HOOKS | Hook | Adds | |------|------| -| `get_searchable_fields` | Plugin contributes to the global search endpoint | | `get_navigation_items` | Plugin shows up in the sidebar nav | | `get_dashboard_widgets` | Plugin's dashboard widget appears on the home page | -| `get_collector_schema` | Plugin accepts external pushes at `/api/collector/` | +| `get_collector_schema` + `apply_collector_payload` | Plugin accepts external pushes at `/api/collector/` | Each hook has a default that does nothing. Override only what your plugin needs. diff --git a/docs/adr/ADR-001-asset-as-platform-contract.md b/docs/adr/ADR-001-asset-as-platform-contract.md index ffa5d27..bb47159 100644 --- a/docs/adr/ADR-001-asset-as-platform-contract.md +++ b/docs/adr/ADR-001-asset-as-platform-contract.md @@ -58,7 +58,7 @@ Adding a name here is a minor (additive) contract change; removing one is major. #### Plugin contract -- `BasePlugin` ABC and its hooks (search, navigation, dashboard, relationships, collector schema + `apply_collector_payload`) +- `BasePlugin` ABC and its hooks (navigation, dashboard widgets, collector schema + `apply_collector_payload`). Note: `get_searchable_fields` was removed in contract 0.4.0 - global search is a core concern over the asset model, not a per-plugin hook. #### Excluded from the contract for v1 diff --git a/shopdb/__init__.py b/shopdb/__init__.py index 7dab0d3..71b1a77 100644 --- a/shopdb/__init__.py +++ b/shopdb/__init__.py @@ -15,7 +15,10 @@ from .plugins import plugin_manager # 0.3.0: shopdb.api expanded to the full plugin import surface (db, cache, # model bases, core models, response + pagination helpers, employee_connection) # so plugins no longer import internal core paths. Additive, hence minor bump. -__contract_version__ = '0.3.0' +# 0.4.0: removed the never-implemented get_searchable_fields hook (search is a +# core concern over the asset model) and wired the get_dashboard_widgets hook to +# a real consumer (/api/dashboard/widgets). Pre-1.0 contract reduction. +__contract_version__ = '0.4.0' def create_app(config_name: str = None) -> Flask: diff --git a/shopdb/core/api/applications.py b/shopdb/core/api/applications.py index 8017121..682c2a2 100644 --- a/shopdb/core/api/applications.py +++ b/shopdb/core/api/applications.py @@ -7,13 +7,49 @@ from shopdb.extensions import db from shopdb.core.models import ( Application, AppVersion, AppOwner, SupportTeam, AuditLog ) -from plugins.computers.models import Computer, ComputerInstalledApp from shopdb.utils.responses import ( success_response, error_response, paginated_response, ErrorCodes ) + + +def _computer_models(): + """Lazily import the computers plugin models, or None if unavailable. + + Application install-tracking is a join over the computers plugin's tables. + Importing lazily keeps the applications API importable when the computers + plugin is absent or disabled. + """ + try: + from plugins.computers.models import Computer, ComputerInstalledApp + return Computer, ComputerInstalledApp + except ImportError: + return None + + +def _installed_count(appid): + """Count active installs of an app, 0 when the computers plugin is absent.""" + models = _computer_models() + if not models: + return 0 + _, ComputerInstalledApp = models + return ComputerInstalledApp.query.filter_by(appid=appid, isactive=True).count() + + +def _require_computer_models(): + """Resolve (Computer, ComputerInstalledApp) or a 503 response tuple. + + Usage: `models, err = _require_computer_models(); if err: return err`. + """ + models = _computer_models() + if not models: + return None, error_response( + ErrorCodes.INTERNAL_ERROR, + 'Install tracking requires the computers plugin', + http_code=503) + return models, None from shopdb.utils.pagination import get_pagination_params, paginate_query applications_bp = Blueprint('applications', __name__) @@ -66,8 +102,7 @@ def list_applications(): } else: app_dict['supportteam'] = None - app_dict['installedcount'] = ComputerInstalledApp.query.filter_by( - appid=app.appid, isactive=True).count() + app_dict['installedcount'] = _installed_count(app.appid) data.append(app_dict) return paginated_response(data, page, per_page, total) @@ -97,8 +132,7 @@ def get_application(app_id: int): else: data['supportteam'] = None data['versions'] = [v.to_dict() for v in app.versions.filter_by(isactive=True).order_by(AppVersion.version.desc()).all()] - data['installedcount'] = ComputerInstalledApp.query.filter_by( - appid=app.appid, isactive=True).count() + data['installedcount'] = _installed_count(app.appid) return success_response(data) @@ -263,6 +297,11 @@ def list_installed_machines(app_id: int): if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) + models, err = _require_computer_models() + if err: + return err + Computer, ComputerInstalledApp = models + installed = ComputerInstalledApp.query.filter_by( appid=app_id, isactive=True).all() data = [] @@ -293,6 +332,11 @@ def list_installed_machines(app_id: int): @jwt_required(optional=True) def list_machine_applications(machine_id: int): """List all applications installed on a computer.""" + models, err = _require_computer_models() + if err: + return err + Computer, ComputerInstalledApp = models + comp = Computer.query.get(machine_id) if not comp: return error_response(ErrorCodes.NOT_FOUND, 'Computer not found', http_code=404) @@ -305,6 +349,11 @@ def list_machine_applications(machine_id: int): @jwt_required() def install_application(machine_id: int): """Install an application on a computer.""" + models, err = _require_computer_models() + if err: + return err + Computer, ComputerInstalledApp = models + comp = Computer.query.get(machine_id) if not comp: return error_response(ErrorCodes.NOT_FOUND, 'Computer not found', http_code=404) @@ -351,6 +400,11 @@ def install_application(machine_id: int): @jwt_required() def uninstall_application(machine_id: int, app_id: int): """Uninstall an application from a computer.""" + models, err = _require_computer_models() + if err: + return err + Computer, ComputerInstalledApp = models + installed = ComputerInstalledApp.query.filter_by( computerid=machine_id, appid=app_id, @@ -370,6 +424,11 @@ def uninstall_application(machine_id: int, app_id: int): @jwt_required() def update_installed_app(machine_id: int, app_id: int): """Update installed application (e.g., change version).""" + models, err = _require_computer_models() + if err: + return err + Computer, ComputerInstalledApp = models + installed = ComputerInstalledApp.query.filter_by( computerid=machine_id, appid=app_id, diff --git a/shopdb/core/api/collector.py b/shopdb/core/api/collector.py index d98f87f..39a8528 100644 --- a/shopdb/core/api/collector.py +++ b/shopdb/core/api/collector.py @@ -12,13 +12,27 @@ from flask import Blueprint, request, current_app from shopdb.extensions import db from shopdb.core.models import Asset, Application -from plugins.computers.models import Computer, ComputerInstalledApp from shopdb.utils.responses import success_response, error_response, ErrorCodes collector_bp = Blueprint('collector', __name__) +def _computer_models(): + """Lazily import the computers plugin models. + + The legacy /pc, /apps, /heartbeat, /bulk endpoints predate the generic + collector contract and are computers-specific. Importing the plugin lazily + (instead of at module load) keeps core importable when the computers plugin + is absent or disabled. Returns (Computer, ComputerInstalledApp) or None. + """ + try: + from plugins.computers.models import Computer, ComputerInstalledApp + return Computer, ComputerInstalledApp + except ImportError: + return None + + def require_api_key(f): """Require API key authentication.""" @wraps(f) @@ -40,7 +54,15 @@ def require_api_key(f): def _find_pc(hostname): - """Find a computer by hostname, falling back to its asset number.""" + """Find a computer by hostname, falling back to its asset number. + + Returns None if not found or if the computers plugin is unavailable; + callers already treat None as a 404. + """ + models = _computer_models() + if not models: + return None + Computer, _ = models comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first() if comp: return comp @@ -237,6 +259,9 @@ def update_installed_apps(): f'PC with hostname {hostname} not found', http_code=404) + # comp existing implies the computers plugin is loaded. + _, ComputerInstalledApp = _computer_models() + updated_count = created_count = skipped_count = 0 for app_data in apps: diff --git a/shopdb/core/api/dashboard.py b/shopdb/core/api/dashboard.py index a4371ea..3dab775 100644 --- a/shopdb/core/api/dashboard.py +++ b/shopdb/core/api/dashboard.py @@ -152,6 +152,37 @@ def get_navigation(): return success_response(all_items) +@dashboard_bp.route('/widgets', methods=['GET']) +@jwt_required(optional=True) +def get_widgets(): + """Aggregate dashboard widget definitions from all enabled plugins. + + Consumer for the BasePlugin.get_dashboard_widgets hook. Skips disabled + plugins, isolates a broken plugin in prod (re-raises in dev/test), and + returns the merged list sorted by position. + """ + pm = current_app.extensions.get('plugin_manager') + if not pm: + return success_response([]) + + widgets = [] + for name, plugin in pm.get_all_plugins().items(): + if not pm.registry.is_enabled(name): + continue + try: + for widget in plugin.get_dashboard_widgets() or []: + widget['plugin'] = name + widgets.append(widget) + except Exception: + if current_app.config.get('DEBUG') or current_app.config.get('TESTING'): + raise + current_app.logger.exception( + 'Plugin %s get_dashboard_widgets failed', name) + + widgets.sort(key=lambda w: w.get('position', 99)) + return success_response(widgets) + + @dashboard_bp.route('/health', methods=['GET']) def health_check(): """Health check endpoint (no auth required).""" diff --git a/shopdb/core/api/reports.py b/shopdb/core/api/reports.py index d7a0002..5d64a54 100644 --- a/shopdb/core/api/reports.py +++ b/shopdb/core/api/reports.py @@ -11,7 +11,6 @@ from shopdb.core.models import ( Asset, AssetType, AssetStatus, Application, KnowledgeBase ) -from plugins.computers.models import Computer, ComputerInstalledApp from shopdb.utils.responses import success_response, error_response, ErrorCodes reports_bp = Blueprint('reports', __name__) @@ -317,6 +316,16 @@ def software_compliance(): - appid: Filter to specific application - format: 'json' (default) or 'csv' """ + # Install tracking lives in the computers plugin; degrade gracefully if + # it is not installed. + try: + from plugins.computers.models import Computer, ComputerInstalledApp + except ImportError: + return error_response( + ErrorCodes.INTERNAL_ERROR, + 'Software compliance requires the computers plugin', + http_code=503) + # Get required applications required_apps = Application.query.filter( Application.isactive == True, diff --git a/shopdb/core/api/search.py b/shopdb/core/api/search.py index 27bf054..4e8e8a4 100644 --- a/shopdb/core/api/search.py +++ b/shopdb/core/api/search.py @@ -1,727 +1,746 @@ -"""Global search API endpoint with full search parity.""" - -import re -import ipaddress -import logging - -from datetime import datetime -from flask import Blueprint, request -from flask_jwt_extended import jwt_required -from sqlalchemy.orm import joinedload - -from shopdb.extensions import db -from shopdb.core.models import ( - Application, KnowledgeBase, - Asset, AssetType, Communication, Vendor, Model -) -from shopdb.utils.responses import success_response - -logger = logging.getLogger(__name__) - -search_bp = Blueprint('search', __name__) - -# ServiceNOW URL template -SERVICENOW_URL = ( - 'https://geit.service-now.com/now/nav/ui/search/' - '0f8b85d0c7922010099a308dc7c2606a/params/search-term/{ticket}/' - 'global-search-data-config-id/c861cea2c7022010099a308dc7c26041/' - 'back-button-label/IT4IT%20Homepage/search-context/now%2Fnav%2Fui' -) - - -def _classify_query(query): - """Analyze the query string to determine its nature.""" - return { - 'is_ip': bool(re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', query)), - 'is_sso': bool(re.match(r'^\d{9}$', query)), - 'is_servicenow': bool(re.match(r'^(GEINC|GECHG|GERIT|GESCT)\d+', query, re.IGNORECASE)), - 'servicenow_prefix': re.match(r'^(GEINC|GECHG|GERIT|GESCT)', query, re.IGNORECASE).group(1) if re.match(r'^(GEINC|GECHG|GERIT|GESCT)', query, re.IGNORECASE) else None, - 'is_fqdn': bool(re.match(r'^[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+$', query)), - } - - -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 == 'equipment' and hasattr(asset, 'equipment') and asset.equipment: - plugin_id = asset.equipment.equipmentid - 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 - - url_map = { - 'equipment': f"/machines/{plugin_id}", - 'computer': f"/pcs/{plugin_id}", - 'network_device': f"/network/{plugin_id}", - 'printer': f"/printers/{plugin_id}", - } - url = url_map.get(asset_type_name, f"/assets/{asset.assetid}") - - 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': plugin_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, - db.or_( - Application.appname.ilike(search_term), - Application.appdescription.ilike(search_term) - ) - ).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.""" - results = [] - try: - kb_articles = KnowledgeBase.query.filter( - KnowledgeBase.isactive == True, - db.or_( - KnowledgeBase.shortdescription.ilike(search_term), - KnowledgeBase.keywords.ilike(search_term) - ) - ).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 Exception as e: - logger.error(f"KnowledgeBase search failed: {e}") - return results - - -def _search_employees(query, search_term): - """Search Employees in separate wjf_employees database.""" - results = [] - try: - import pymysql - emp_conn = pymysql.connect( - host='localhost', - user='root', - password='rootpassword', - database='wjf_employees', - cursorclass=pymysql.cursors.DictCursor - ) - - with emp_conn.cursor() as cur: - cur.execute(''' - SELECT SSO, First_Name, Last_Name, Team, Role - FROM employees - WHERE First_Name LIKE %s - OR Last_Name LIKE %s - OR CAST(SSO AS CHAR) LIKE %s - ORDER BY Last_Name, First_Name - LIMIT 10 - ''', (search_term, search_term, search_term)) - employees = cur.fetchall() - - emp_conn.close() - - for emp in employees: - 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.""" - results = [] - try: - assets = Asset.query.join(AssetType).options( - joinedload(Asset.assettype), - joinedload(Asset.location), - ).filter( - Asset.isactive == True, - db.or_( - Asset.assetnumber.ilike(search_term), - Asset.name.ilike(search_term), - Asset.serialnumber.ilike(search_term), - Asset.notes.ilike(search_term) - ) - ).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.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_by_ip(query, search_term): - """Search Communications table for IP address matches.""" - results = [] - try: - comms = Communication.query.filter( - Communication.ipaddress.ilike(search_term) - ).options( - joinedload(Communication.asset).joinedload(Asset.assettype), - joinedload(Communication.asset).joinedload(Asset.location), - ).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: - 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: - from plugins.computers.models import Computer - computers = Computer.query.filter( - Computer.hostname.ilike(search_term) - ).options( - joinedload(Computer.asset).joinedload(Asset.assettype), - joinedload(Computer.asset).joinedload(Asset.location), - ).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: - from plugins.printers.models import Printer - printers = Printer.query.filter( - db.or_( - Printer.hostname.ilike(search_term), - Printer.sharename.ilike(search_term), - Printer.windowsname.ilike(search_term), - ) - ).options( - joinedload(Printer.asset).joinedload(Asset.assettype), - joinedload(Printer.asset).joinedload(Asset.location), - ).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: - from plugins.network.models import NetworkDevice - devices = NetworkDevice.query.filter( - NetworkDevice.hostname.ilike(search_term) - ).options( - joinedload(NetworkDevice.asset).joinedload(Asset.assettype), - joinedload(NetworkDevice.asset).joinedload(Asset.location), - ).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: - from plugins.notifications.models import Notification - - notifications = Notification.query.options( - joinedload(Notification.notificationtype) - ).filter( - db.or_( - Notification.notification.ilike(search_term), - Notification.ticketnumber.ilike(search_term) - ) - ).order_by(Notification.starttime.desc()).limit(15).all() - - now = datetime.utcnow() - 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 equipment/device type name.""" - results = [] - - # Equipment: vendor, model, equipmenttype - try: - from plugins.equipment.models import Equipment, EquipmentType - equipment_assets = db.session.query(Asset).join( - Equipment, Equipment.assetid == Asset.assetid - ).outerjoin( - Vendor, Equipment.vendorid == Vendor.vendorid - ).outerjoin( - Model, Equipment.modelnumberid == Model.modelnumberid - ).outerjoin( - EquipmentType, Equipment.equipmenttypeid == EquipmentType.equipmenttypeid - ).options( - joinedload(Asset.assettype), - joinedload(Asset.location), - ).filter( - Asset.isactive == True, - db.or_( - Vendor.vendor.ilike(search_term), - Model.modelnumber.ilike(search_term), - EquipmentType.equipmenttype.ilike(search_term) - ) - ).limit(10).all() - - for asset in equipment_assets: - results.append(_get_asset_result(asset, query, 30)) - except ImportError: - pass - except Exception as e: - logger.error(f"Equipment vendor/model/type search failed: {e}") - - # Printers: vendor, model, printertype - try: - 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, - db.or_( - Vendor.vendor.ilike(search_term), - Model.modelnumber.ilike(search_term), - PrinterType.printertype.ilike(search_term) - ) - ).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: - 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, - db.or_( - Vendor.vendor.ilike(search_term), - NetworkDeviceType.networkdevicetype.ilike(search_term) - ) - ).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: - import pymysql - emp_conn = pymysql.connect( - host='localhost', - user='root', - password='rootpassword', - database='wjf_employees', - cursorclass=pymysql.cursors.DictCursor - ) - 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: - 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: - from plugins.computers.models import Computer - hostname_plugins.append(('computer', Computer, 'computerid', '/pcs')) - except ImportError: - pass - try: - from plugins.printers.models import Printer - hostname_plugins.append(('printer', Printer, 'printerid', '/printers')) - except ImportError: - pass - try: - 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 - - -@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' - }) - - classification = _classify_query(query) - - # ServiceNOW prefix detection - return redirect immediately - if classification['is_servicenow']: - from urllib.parse import quote - servicenow_url = 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_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) - - # 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) - - # 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, - } - - redirect = _check_smart_redirect(query, classification) - if redirect: - response_data['redirect'] = redirect - - return success_response(response_data) +"""Global search API endpoint with full search parity.""" + +import re +import ipaddress +import logging + +from datetime import datetime +from flask import Blueprint, request, current_app +from flask_jwt_extended import jwt_required +from sqlalchemy.orm import joinedload + +from shopdb.extensions import db +from shopdb.core.models import ( + Application, KnowledgeBase, + Asset, AssetType, Communication, Vendor, Model +) +from shopdb.utils.responses import success_response + +logger = logging.getLogger(__name__) + +search_bp = Blueprint('search', __name__) + + +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') + +# ServiceNOW URL template +SERVICENOW_URL = ( + 'https://geit.service-now.com/now/nav/ui/search/' + '0f8b85d0c7922010099a308dc7c2606a/params/search-term/{ticket}/' + 'global-search-data-config-id/c861cea2c7022010099a308dc7c26041/' + 'back-button-label/IT4IT%20Homepage/search-context/now%2Fnav%2Fui' +) + + +def _classify_query(query): + """Analyze the query string to determine its nature.""" + return { + 'is_ip': bool(re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', query)), + 'is_sso': bool(re.match(r'^\d{9}$', query)), + 'is_servicenow': bool(re.match(r'^(GEINC|GECHG|GERIT|GESCT)\d+', query, re.IGNORECASE)), + 'servicenow_prefix': re.match(r'^(GEINC|GECHG|GERIT|GESCT)', query, re.IGNORECASE).group(1) if re.match(r'^(GEINC|GECHG|GERIT|GESCT)', query, re.IGNORECASE) else None, + 'is_fqdn': bool(re.match(r'^[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)+$', query)), + } + + +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 == 'equipment' and hasattr(asset, 'equipment') and asset.equipment: + plugin_id = asset.equipment.equipmentid + 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 + + url_map = { + 'equipment': f"/machines/{plugin_id}", + 'computer': f"/pcs/{plugin_id}", + 'network_device': f"/network/{plugin_id}", + 'printer': f"/printers/{plugin_id}", + } + url = url_map.get(asset_type_name, f"/assets/{asset.assetid}") + + 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': plugin_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, + db.or_( + Application.appname.ilike(search_term), + Application.appdescription.ilike(search_term) + ) + ).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.""" + results = [] + try: + kb_articles = KnowledgeBase.query.filter( + KnowledgeBase.isactive == True, + db.or_( + KnowledgeBase.shortdescription.ilike(search_term), + KnowledgeBase.keywords.ilike(search_term) + ) + ).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 Exception as e: + logger.error(f"KnowledgeBase search failed: {e}") + return results + + +def _search_employees(query, search_term): + """Search Employees in separate wjf_employees database.""" + results = [] + try: + # Use the 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, Team, Role + FROM employees + WHERE First_Name LIKE %s + OR Last_Name LIKE %s + OR CAST(SSO AS CHAR) LIKE %s + ORDER BY Last_Name, First_Name + LIMIT 10 + ''', (search_term, search_term, search_term)) + employees = cur.fetchall() + + emp_conn.close() + + for emp in employees: + 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.""" + results = [] + try: + assets = Asset.query.join(AssetType).options( + joinedload(Asset.assettype), + joinedload(Asset.location), + ).filter( + Asset.isactive == True, + db.or_( + Asset.assetnumber.ilike(search_term), + Asset.name.ilike(search_term), + Asset.serialnumber.ilike(search_term), + Asset.notes.ilike(search_term) + ) + ).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.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_by_ip(query, search_term): + """Search Communications table for IP address matches.""" + results = [] + try: + comms = Communication.query.filter( + Communication.ipaddress.ilike(search_term) + ).options( + joinedload(Communication.asset).joinedload(Asset.assettype), + joinedload(Communication.asset).joinedload(Asset.location), + ).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( + Computer.hostname.ilike(search_term) + ).options( + joinedload(Computer.asset).joinedload(Asset.assettype), + joinedload(Computer.asset).joinedload(Asset.location), + ).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( + db.or_( + Printer.hostname.ilike(search_term), + Printer.sharename.ilike(search_term), + Printer.windowsname.ilike(search_term), + ) + ).options( + joinedload(Printer.asset).joinedload(Asset.assettype), + joinedload(Printer.asset).joinedload(Asset.location), + ).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( + NetworkDevice.hostname.ilike(search_term) + ).options( + joinedload(NetworkDevice.asset).joinedload(Asset.assettype), + joinedload(NetworkDevice.asset).joinedload(Asset.location), + ).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( + db.or_( + Notification.notification.ilike(search_term), + Notification.ticketnumber.ilike(search_term) + ) + ).order_by(Notification.starttime.desc()).limit(15).all() + + now = datetime.utcnow() + 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 equipment/device type name.""" + results = [] + + # Equipment: vendor, model, equipmenttype + try: + _require_enabled('equipment') + from plugins.equipment.models import Equipment, EquipmentType + equipment_assets = db.session.query(Asset).join( + Equipment, Equipment.assetid == Asset.assetid + ).outerjoin( + Vendor, Equipment.vendorid == Vendor.vendorid + ).outerjoin( + Model, Equipment.modelnumberid == Model.modelnumberid + ).outerjoin( + EquipmentType, Equipment.equipmenttypeid == EquipmentType.equipmenttypeid + ).options( + joinedload(Asset.assettype), + joinedload(Asset.location), + ).filter( + Asset.isactive == True, + db.or_( + Vendor.vendor.ilike(search_term), + Model.modelnumber.ilike(search_term), + EquipmentType.equipmenttype.ilike(search_term) + ) + ).limit(10).all() + + for asset in equipment_assets: + results.append(_get_asset_result(asset, query, 30)) + except ImportError: + pass + except Exception as e: + logger.error(f"Equipment 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, + db.or_( + Vendor.vendor.ilike(search_term), + Model.modelnumber.ilike(search_term), + PrinterType.printertype.ilike(search_term) + ) + ).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, + db.or_( + Vendor.vendor.ilike(search_term), + NetworkDeviceType.networkdevicetype.ilike(search_term) + ) + ).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: + import pymysql + emp_conn = pymysql.connect( + host='localhost', + user='root', + password='rootpassword', + database='wjf_employees', + cursorclass=pymysql.cursors.DictCursor + ) + 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 + + +@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' + }) + + classification = _classify_query(query) + + # ServiceNOW prefix detection - return redirect immediately + if classification['is_servicenow']: + from urllib.parse import quote + servicenow_url = 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_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) + + # 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) + + # 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, + } + + redirect = _check_smart_redirect(query, classification) + if redirect: + response_data['redirect'] = redirect + + return success_response(response_data) diff --git a/shopdb/plugins/base.py b/shopdb/plugins/base.py index 7882e1b..d1d215d 100644 --- a/shopdb/plugins/base.py +++ b/shopdb/plugins/base.py @@ -178,31 +178,3 @@ class BasePlugin(ABC): } """ return [] - - def get_searchable_fields(self) -> List[Dict]: - """ - Return fields this plugin contributes to global search. - - Each field: { - 'model': Type, # SQLAlchemy model class - 'field': str, # Column name to search - 'result_type': str, # Type identifier for search results - 'url_template': str, # URL template with {id} placeholder - 'title_field': str, # Field to use for result title - 'subtitle_field': str, # Optional field for subtitle - 'relevance_boost': int # Optional relevance score multiplier - } - - Example for equipment plugin: - return [{ - 'model': Equipment, - 'join_model': Asset, - 'join_condition': Equipment.assetid == Asset.assetid, - 'search_fields': ['assetnumber', 'name', 'serialnumber'], - 'result_type': 'equipment', - 'url_template': '/equipment/{id}', - 'title_field': 'assetnumber', - 'subtitle_field': 'name', - }] - """ - return [] diff --git a/tests/test_core/test_dashboard_widgets.py b/tests/test_core/test_dashboard_widgets.py new file mode 100644 index 0000000..3bcdff4 --- /dev/null +++ b/tests/test_core/test_dashboard_widgets.py @@ -0,0 +1,38 @@ +"""Tests for the dashboard-widgets hook consumer (/api/dashboard/widgets). + +Pins the wiring added for the BasePlugin.get_dashboard_widgets hook: the +endpoint aggregates enabled plugins' widgets and skips disabled ones. + +Plugin enabled-state is monkeypatched (not persisted) so these tests do not +mutate the shared instance/plugins.json registry file. +""" + + +def _widget_plugins(client, headers): + response = client.get('/api/dashboard/widgets', headers=headers) + assert response.status_code == 200, response.get_json() + widgets = response.get_json()['data'] + assert isinstance(widgets, list) + return widgets, {w.get('plugin') for w in widgets} + + +def test_widgets_endpoint_aggregates_enabled_plugins(app, client, auth_headers, + monkeypatch): + """An enabled plugin that implements the hook contributes a widget.""" + pm = app.extensions['plugin_manager'] + monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True) + + widgets, plugins = _widget_plugins(client, auth_headers) + assert 'computers' in plugins # computers implements get_dashboard_widgets + positions = [w.get('position', 99) for w in widgets] + assert positions == sorted(positions) + + +def test_widgets_endpoint_skips_disabled_plugin(app, client, auth_headers, + monkeypatch): + """A disabled plugin's widgets drop out of the aggregate.""" + pm = app.extensions['plugin_manager'] + monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'computers') + + _, plugins = _widget_plugins(client, auth_headers) + assert 'computers' not in plugins diff --git a/tests/test_plugin_contract.py b/tests/test_plugin_contract.py index 05e5ba8..621a9a2 100644 --- a/tests/test_plugin_contract.py +++ b/tests/test_plugin_contract.py @@ -126,12 +126,15 @@ def test_plugin_get_navigation_items_is_iterable(plugin_instances, name): assert isinstance(items, list) -@pytest.mark.parametrize('name', BUNDLED_PLUGINS) -def test_plugin_get_searchable_fields_is_iterable(plugin_instances, name): - """get_searchable_fields returns a list (default empty).""" - plugin = plugin_instances[name] - fields = plugin.get_searchable_fields() - assert isinstance(fields, list) +def test_baseplugin_has_no_searchable_fields_hook(): + """get_searchable_fields was removed: global search is a core concern over + the asset model, and no plugin ever implemented the hook (contract 0.4.0).""" + assert not hasattr(BasePlugin, 'get_searchable_fields') + + +def test_baseplugin_has_dashboard_widgets_hook(): + """The dashboard widgets hook is on the contract surface and consumed.""" + assert hasattr(BasePlugin, 'get_dashboard_widgets') def test_baseplugin_does_not_have_event_handlers_hook():