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

"""Audit log API routes."""

from flask import Blueprint, request
from flask_jwt_extended import jwt_required

from shopdb.core.models import AuditLog
from shopdb.utils.responses import success_response
from shopdb.utils.authz import require_permission

auditlogs_bp = Blueprint('auditlogs', __name__)


def _resolve_full_names(ssos):
    """Best-effort SSO -> full name map for audit rows. Mode-aware, guarded,
    degrades to {} so the list still renders if the directory is unreachable."""
    wanted = {s for s in ssos if s and str(s).isdigit()}
    if not wanted:
        return {}
    names = {}
    # Self-hosted directory first (no external dependency).
    try:
        from plugins.employees.models import DirectoryEmployee
        rows = DirectoryEmployee.query.filter(
            DirectoryEmployee.sso.in_(wanted)).all()
        for emp in rows:
            full = f'{(emp.firstname or "").strip()} {(emp.lastname or "").strip()}'.strip()
            if full:
                names[str(emp.sso)] = full
    except Exception:
        pass
    missing = wanted - set(names)
    if not missing:
        return names
    # External HR directory for anything still unresolved.
    try:
        from shopdb.utils.employee_db import employee_connection
        conn = employee_connection()
        placeholders = ','.join(['%s'] * len(missing))
        with conn.cursor() as cur:
            cur.execute(
                'SELECT SSO, First_Name, Last_Name FROM employees '
                f'WHERE SSO IN ({placeholders})', tuple(missing))
            for emp in cur.fetchall():
                full = f'{(emp["First_Name"] or "").strip()} {(emp["Last_Name"] or "").strip()}'.strip()
                if full:
                    names[str(emp['SSO'])] = full
        conn.close()
    except Exception:
        pass
    return names


@auditlogs_bp.route('', methods=['GET'])
@jwt_required()
@require_permission('audit.view')
def list_auditlogs():
    """
    List audit logs with filtering and pagination.

    Query params:
        page: Page number (default 1)
        perpage: Items per page (default 50, max 200)
        action: Filter by action (created, updated, deleted)
        entitytype: Filter by entity type
        userid: Filter by user ID
        search: Search in entityname or username
        from_date: Filter from date (ISO format)
        to_date: Filter to date (ISO format)
    """
    page = request.args.get('page', 1, type=int)
    perpage = min(request.args.get('perpage', 50, type=int), 200)

    query = AuditLog.query

    # Filters
    action = request.args.get('action')
    if action:
        query = query.filter(AuditLog.action == action)

    entitytype = request.args.get('entitytype')
    if entitytype:
        query = query.filter(AuditLog.entitytype == entitytype)

    userid = request.args.get('userid', type=int)
    if userid:
        query = query.filter(AuditLog.userid == userid)

    search = request.args.get('search')
    if search:
        search_term = f'%{search}%'
        query = query.filter(
            (AuditLog.entityname.ilike(search_term)) |
            (AuditLog.username.ilike(search_term))
        )

    from_date = request.args.get('from_date')
    if from_date:
        from datetime import datetime
        try:
            dt = datetime.fromisoformat(from_date.replace('Z', '+00:00'))
            query = query.filter(AuditLog.timestamp >= dt)
        except ValueError:
            pass

    to_date = request.args.get('to_date')
    if to_date:
        from datetime import datetime
        try:
            dt = datetime.fromisoformat(to_date.replace('Z', '+00:00'))
            query = query.filter(AuditLog.timestamp <= dt)
        except ValueError:
            pass

    # Order by most recent first
    query = query.order_by(AuditLog.timestamp.desc())

    # Paginate
    pagination = query.paginate(page=page, per_page=perpage, error_out=False)

    rows = [log.to_dict() for log in pagination.items]
    # Attach full names (hover tooltip on the SSO); best-effort.
    fullnames = _resolve_full_names({r.get('username') for r in rows})
    for r in rows:
        r['userfullname'] = fullnames.get(str(r.get('username')))

    return success_response(
        rows,
        meta={
            'page': page,
            'perpage': perpage,
            'total': pagination.total,
            'pages': pagination.pages
        }
    )


@auditlogs_bp.route('/entity/<entitytype>/<int:entityid>', methods=['GET'])
@jwt_required()
@require_permission('audit.view')
def get_entity_history(entitytype: str, entityid: int):
    """Get audit history for a specific entity."""
    logs = AuditLog.query.filter_by(
        entitytype=entitytype,
        entityid=entityid
    ).order_by(AuditLog.timestamp.desc()).all()

    return success_response([log.to_dict() for log in logs])


@auditlogs_bp.route('/stats', methods=['GET'])
@jwt_required()
@require_permission('audit.view')
def get_stats():
    """Get audit log statistics."""
    from sqlalchemy import func
    from datetime import datetime, timedelta, timezone

    # Actions by type
    actions = db_func_count_by(AuditLog.action)

    # Entity types
    entities = db_func_count_by(AuditLog.entitytype)

    # Recent activity (last 7 days)
    week_ago = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=7)
    recent_count = AuditLog.query.filter(AuditLog.timestamp >= week_ago).count()

    # Most active users (last 7 days)
    from shopdb.extensions import db
    active_users = db.session.query(
        AuditLog.username,
        func.count(AuditLog.auditlogid).label('count')
    ).filter(
        AuditLog.timestamp >= week_ago,
        AuditLog.username.isnot(None)
    ).group_by(AuditLog.username).order_by(func.count(AuditLog.auditlogid).desc()).limit(5).all()

    return success_response({
        'actions': actions,
        'entities': entities,
        'recentCount': recent_count,
        'activeUsers': [{'username': u[0], 'count': u[1]} for u in active_users]
    })


def db_func_count_by(column):
    """Helper to count grouped by a column."""
    from sqlalchemy import func
    from shopdb.extensions import db

    results = db.session.query(
        column,
        func.count().label('count')
    ).group_by(column).all()

    return {r[0]: r[1] for r in results}