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.extensions import db
from shopdb.utils.responses import (success_response, error_response,
                                    paginated_response, ErrorCodes)
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. 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 = {}
    missing = set(wanted)
    # External HR directory. Best effort: a site with EMPLOYEE_DB_* unset, or a
    # directory that is down, gets bare SSOs rather than an error page over a
    # log it can still read.
    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'])
@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')))

    # paginated_response, like every other list in the product. This one built
    # its own meta with the fields FLAT and named 'pages', so a client written
    # against the documented meta.pagination.{page,perpage,total,totalpages}
    # found nothing where it looked - on the one list whose whole purpose is to
    # be read by somebody checking what happened.
    return paginated_response(rows, page, perpage, pagination.total)


@auditlogs_bp.route('/entity/<entitytype>/<int:entityid>', methods=['GET'])
@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'])
@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}


@auditlogs_bp.route('/retention', methods=['GET'])
@require_permission('audit.view')
def get_retention():
    """Size of the log and what the configured retention would remove.

    Drives the Site & Facility card: an admin should see the number of entries a
    purge deletes BEFORE pressing anything.
    """
    from shopdb.core.services import auditretention
    days = auditretention.configureddays()
    return success_response(auditretention.stats(days or None))


@auditlogs_bp.route('/purge', methods=['POST'])
@require_permission('audit.purge')
def purge_auditlogs():
    """Delete entries older than the retention window.

    Separate permission from audit.view: reading the log and destroying it are
    not the same act, and plenty of people need the first.

    Batch-capped, because this runs inside a request that a proxy will time out.
    An incomplete purge returns complete=false and the number still to go rather
    than reporting a finished job - the caller re-posts, or runs
    `flask audit purge --apply` on the server, which has no cap.
    """
    from flask import request as flask_request
    from shopdb.core.services import auditretention

    body = flask_request.get_json(silent=True) or {}
    days = body.get('days')
    if days is None:
        days = auditretention.configureddays()
    try:
        days = int(days)
    except (TypeError, ValueError):
        days = 0
    if days <= 0:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'No retention period is configured. Set one in Site & Facility '
            'settings before purging.')

    dryrun = bool(body.get('dryrun'))
    result = auditretention.purge(
        days, dryrun=dryrun,
        maxbatches=None if dryrun else auditretention.MAXBATCHES_REQUEST)

    if dryrun:
        message = '{:,} entry(s) would be deleted'.format(result['matched'])
    elif result['complete']:
        message = 'Deleted {:,} entry(s)'.format(result['deleted'])
    else:
        message = ('Deleted {:,} entry(s); {:,} still older than the cutoff. '
                   'Run it again to continue.').format(
                       result['deleted'], result['remaining'])

    # The purge is itself an audited act, and this row postdates the cutoff so
    # it survives the purge that created it.
    if not dryrun and result['deleted']:
        from shopdb.api import audit_log
        audit_log(action='deleted', entitytype='AuditLog',
                  entityname='purge older than {} days'.format(days),
                  changes={'deleted': result['deleted'],
                           'cutoff': result['cutoff']})

    return success_response(result, message=message)