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

"""Dashboard API endpoints."""

from flask import Blueprint, current_app
from flask_jwt_extended import jwt_required

from shopdb.extensions import db
from shopdb.core.models import Asset, AssetType, AssetStatus
from shopdb.utils.responses import success_response
from shopdb.utils.authz import require_permission

dashboard_bp = Blueprint('dashboard', __name__)

# Map asset type name -> dashboard category label
_TYPE_CATEGORY = {
    'machine': 'Machine',
    'computer': 'PC',
    'printer': 'Printer',
    'network_device': 'Network',
    'measuring_tool': 'Measuring Tool',
}


def _count_by_type(assettype):
    query = db.session.query(Asset).join(AssetType).filter(
        Asset.isactive == True,
        AssetType.assettype == assettype
    )
    # Dualpath single-machine collapse: subtract the hidden secondary bays so a
    # dual-bay machine counts once. Only machines are ever Dualpath-paired, but
    # the AssetType filter above keeps this correct for any type. Default on.
    if assettype == 'machine':
        from shopdb.core.services.dualpath import (
            resolve_dualpath_pairs, dualpath_single_machine_enabled)
        if dualpath_single_machine_enabled():
            secondaryassetids = resolve_dualpath_pairs().secondaryassetids
            if secondaryassetids:
                query = query.filter(Asset.assetid.notin_(secondaryassetids))
    return query.count()


def _inactive_excluded():
    """Counted-type assets held back by isactive, i.e. soft-deleted.

    Reported so the page can account for itself. A total that silently drops
    rows is the reason somebody compares it with a list page and finds a
    difference they cannot explain.
    """
    return (db.session.query(Asset).join(AssetType)
            .filter(Asset.isactive == False,
                    AssetType.assettype.in_(COUNTEDTYPES))
            .count())


COUNTEDTYPES = ('machine', 'computer', 'network_device', 'printer',
                'measuring_tool')


def _countedassets():
    """Base query for the assets the dashboard totals describe.

    The status tiles and the type tiles MUST count the same population. They
    did not: the totals summed five specific types and subtracted dual-bay
    secondaries, while the status counts took every asset row of any type with
    no collapse. So "all assets 704" sat beside "in use 737", and both were
    correct about different things - which makes them worse than either alone.
    """
    from shopdb.core.services.dualpath import (
        resolve_dualpath_pairs, dualpath_single_machine_enabled)

    query = (db.session.query(Asset).join(AssetType)
             .filter(Asset.isactive == True,
                     AssetType.assettype.in_(COUNTEDTYPES)))
    if dualpath_single_machine_enabled():
        secondaryassetids = resolve_dualpath_pairs().secondaryassetids
        if secondaryassetids:
            query = query.filter(Asset.assetid.notin_(secondaryassetids))
    return query


@dashboard_bp.route('/summary', methods=['GET'])
@dashboard_bp.route('', methods=['GET'])
@require_permission('assets.view')
def get_dashboard():
    """Get dashboard summary data (asset-based)."""
    # Imported here, as in the helpers above: the dualpath service reaches back
    # into models, so a module-level import is circular.
    from shopdb.core.services.dualpath import dualpath_single_machine_enabled

    machine_count = _count_by_type('machine')
    pc_count = _count_by_type('computer')
    network_count = _count_by_type('network_device')
    printer_count = _count_by_type('printer')
    measuringtool_count = _count_by_type('measuring_tool')
    total = machine_count + pc_count + network_count + printer_count + measuringtool_count

    # Count by status over the SAME population as the totals above, so the
    # tiles can be read against each other.
    status_dict = {}
    for asset in _countedassets().all():
        name = asset.status.status if asset.status else 'Unknown'
        status_dict[name] = status_dict.get(name, 0) + 1

    # Recent assets
    recent = Asset.query.filter_by(isactive=True).order_by(
        Asset.createddate.desc()
    ).limit(10).all()

    return success_response({
        # Fields expected by frontend (totalmachines now means machines,
        # totalassets is the grand total - renamed with the machines plugin)
        'totalassets': total,
        'totalmachines': machine_count,
        'totalpc': pc_count,
        'totalnetwork': network_count,
        'totalprinter': printer_count,
        'totalmeasuringtool': measuringtool_count,
        'activeassets': status_dict.get('In Use', 0),
        'inrepair': status_dict.get('In Repair', 0),
        # What the totals actually cover, so the page can say so rather than
        # leaving a reader to work it out from a number that surprised them.
        # Every count here filters isactive, and the dual-path collapse is a
        # per-site toggle - asserting it unconditionally would be false at a
        # site that has it off.
        'countsactiveonly': True,
        'dualpathcollapsed': bool(dualpath_single_machine_enabled()),
        # Every status, not just the two that had callers. The counts were
        # already built for the rollup above and then thrown away, so a reader
        # could see "In Use" and had no way to learn where the rest went.
        'statuses': status_dict,
        # A DIFFERENT thing to a status, and the one that makes a total
        # disagree with a list page. isactive is the soft-delete flag: these
        # rows are excluded from every count above, so they are in no status
        # bucket and cannot be found by adding the buckets up.
        'inactiveexcluded': _inactive_excluded(),
        # Structured data
        'counts': {
            'machines': machine_count,
            'pcs': pc_count,
            'networkdevices': network_count,
            'printers': printer_count,
            'measuringtools': measuringtool_count,
            'total': total
        },
        'bystatus': status_dict,
        'recent': [
            {
                'assetid': a.assetid,
                'assetnumber': a.assetnumber,
                'assettype': a.assettype.assettype if a.assettype else None,
                'createddate': a.createddate.isoformat() + 'Z' if a.createddate else None
            }
            for a in recent
        ]
    })


@dashboard_bp.route('/stats', methods=['GET'])
@require_permission('assets.view')
def get_stats():
    """Get detailed statistics by asset type."""
    type_counts = db.session.query(
        AssetType.assettype,
        db.func.count(Asset.assetid)
    ).outerjoin(
        Asset,
        db.and_(Asset.assettypeid == AssetType.assettypeid, Asset.isactive == True)
    ).group_by(AssetType.assettypeid).all()

    return success_response({
        'bytype': [
            {
                'type': t,
                'category': _TYPE_CATEGORY.get(t, t),
                'count': count
            }
            for t, count in type_counts
        ]
    })


@dashboard_bp.route('/navigation', methods=['GET'])
def get_navigation():
    """Get navigation items from all loaded plugins."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        return success_response([])

    all_items = []

    # Core navigation items (always present)
    # Every item names the permission its page needs, so the filter below can
    # hide what the caller cannot reach - a link that only leads to a 401 is
    # worse than no link.
    all_items.extend([
        {'name': 'Dashboard', 'icon': 'layout-dashboard', 'route': '/',
         'position': 0, 'permission': 'assets.view'},
        {'name': 'Map', 'icon': 'map', 'route': '/map', 'position': 4,
         'permission': 'assets.view'},
    ])

    # Collect navigation items from enabled plugins. Disabling persists to the
    # registry immediately, so a disabled plugin drops out of the menu right
    # away (its routes stay registered until the next restart - Flask cannot
    # unregister a blueprint at runtime).
    for name, plugin in pm.get_all_plugins().items():
        if not pm.registry.is_enabled(name):
            continue
        try:
            items = plugin.get_navigation_items()
            for item in items:
                item['plugin'] = name
            all_items.extend(items)
        except Exception:
            # Fail loud in dev/test; isolate a broken plugin in prod.
            if current_app.config.get('DEBUG') or current_app.config.get('TESTING'):
                raise
            current_app.logger.exception(
                'Plugin %s get_navigation_items failed', name)

    # Add core information section items
    all_items.extend([
        {'name': 'Applications', 'icon': 'app-window', 'route': '/applications',
         'position': 30, 'section': 'information',
         'permission': 'applications.view'},
        {'name': 'Reports', 'icon': 'bar-chart-3', 'route': '/reports',
         'position': 40, 'section': 'information', 'permission': 'reports.view'},
    ])

    # Sort by position
    all_items.sort(key=lambda x: x.get('position', 99))

    # A signed-out visitor is shown only what the guest role can actually
    # reach. Without this the sidebar offers them the whole app and every link
    # bounces off a 401, which reads as a broken site rather than a closed
    # door. An item with no declared permission is treated as members-only: a
    # plugin opts into being guest-visible by naming the permission its page
    # needs, which is the same string the API already gates on.
    from flask_jwt_extended import verify_jwt_in_request
    from shopdb.utils.authz import currentuser, guestholds

    verify_jwt_in_request(optional=True)
    def guestmayreach(item):
        declared = item.get('permission')
        if not declared:
            return False
        names = declared if isinstance(declared, list) else [declared]
        return any(guestholds(name) for name in names)

    if currentuser() is None:
        all_items = [item for item in all_items if guestmayreach(item)]

    return success_response(all_items)


@dashboard_bp.route('/capabilities', methods=['GET'])
def get_capabilities():
    """What THIS caller can reach, for chrome that would otherwise only 401.

    The sidebar search box was gated on being signed in, because a box that
    can only answer 401 is worse than no box. That stopped being the right
    test once the guest role could be granted assets.view: a signed-out
    visitor at a machine may now search for the asset number in front of them,
    and the box stayed hidden from exactly the person it was re-opened for.

    The client cannot answer this itself. Its hasPermission getter reads the
    signed-in user's permission list, and a guest has no user at all, so
    asking it about a guest returns false for everyone. The navigation route
    above has the same problem and solves it the same way: the server knows
    what the guest role holds, so the server decides.

    Public, and deliberately so - it is the pre-login call that decides
    whether the box is drawn. It reports only booleans about the caller, never
    what the guest role holds in general, so it says nothing to an anonymous
    reader that the rendered page does not already.
    """
    from flask_jwt_extended import verify_jwt_in_request
    from shopdb.utils.authz import caller_holds

    verify_jwt_in_request(optional=True)
    return success_response({
        # Global search is gated on assets.view, and each domain behind it now
        # gates on its own permission, so holding this reaches assets alone.
        'cansearch': caller_holds('assets.view'),
    })


@dashboard_bp.route('/widgets', methods=['GET'])
@require_permission('assets.view')
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).

    Anonymous callers get a coarse status and nothing else. The probe used to
    format str(e) straight into the body, and a pymysql OperationalError names
    the database account, the app host's address, the driver and the failing
    statement - the internal topology, handed to whoever asks, on a route
    PUBLIC_ENDPOINTS declares open because "monitoring has no account". The
    detail is logged server-side instead, where the people who need it are.

    The version goes the same way. It told an anonymous caller exactly which
    release to look up known issues for, and monitoring does not need it.
    """
    try:
        db.session.execute(db.text('SELECT 1'))
        healthy = True
    except Exception:
        current_app.logger.exception('Health check database probe failed')
        healthy = False

    # Status code stays 200 either way. A monitor keyed on the body's status
    # field is the documented contract here, and turning a degraded database
    # into a transport-level failure would change what every existing check
    # sees - a separate decision from not leaking the exception.
    return success_response({
        'status': 'ok' if healthy else 'degraded',
        'database': 'healthy' if healthy else 'unhealthy',
    })