lego-instructions-manager/app/routes/logs.py

"""
Owner-only log viewer and Brickset API call history.

These pages show everything happening inside the instance, including other
admins' activity, so they sit behind owner_required rather than
admin_required.

Created by: Jess Rogerson (yelling commands at Claude.AI)
"""

import os

from flask import (Blueprint, current_app, flash, jsonify, redirect,
                   render_template, request, send_file, url_for)
from flask_login import login_required

from app import db
from app.models.api_call import BricksetApiCall
from app.models.site_setting import SiteSetting
from app.services import logging_setup
from app.services.ownership import owner_required

logs_bp = Blueprint('logs', __name__, url_prefix='/admin/logs')

DEFAULT_LINES = 300
MAX_LINES = 2000


def _int_arg(name, default, low, high):
    try:
        value = int(request.args.get(name, default))
    except (TypeError, ValueError):
        value = default
    return max(low, min(high, value))


def _retention_days():
    raw = SiteSetting.get('api_log_retention_days', '90')
    try:
        return max(1, min(3650, int(str(raw).strip())))
    except (TypeError, ValueError):
        return 90


# ---------------------------------------------------------------------------
# application log
# ---------------------------------------------------------------------------

@logs_bp.route('/')
@login_required
@owner_required
def index():
    """Tail of the application log, filtered."""
    level = request.args.get('level', 'INFO')
    search = request.args.get('q', '').strip()
    source = request.args.get('source', '')
    lines = _int_arg('lines', DEFAULT_LINES, 25, MAX_LINES)
    filename = request.args.get('file', logging_setup.LOG_FILENAME)

    path = logging_setup.resolve_file(current_app, filename)
    entries, total = logging_setup.read_entries(
        path, max_entries=lines, level=level, search=search or None,
        logger_prefix=source or None)

    files = logging_setup.available_files(current_app)
    configured = logging_setup.get_handler() is not None

    return render_template(
        'admin/logs.html',
        entries=list(reversed(entries)),   # newest first on screen
        shown=len(entries),
        total=total,
        level=level,
        search=search,
        source=source,
        lines=lines,
        filename=filename,
        files=files,
        levels=logging_setup.LEVELS,
        configured=configured,
        log_path=logging_setup.log_path(current_app),
    )


@logs_bp.route('/tail.json')
@login_required
@owner_required
def tail_json():
    """Same data as the page, for auto-refresh."""
    level = request.args.get('level', 'INFO')
    search = request.args.get('q', '').strip()
    source = request.args.get('source', '')
    lines = _int_arg('lines', DEFAULT_LINES, 25, MAX_LINES)
    path = logging_setup.resolve_file(
        current_app, request.args.get('file', logging_setup.LOG_FILENAME))

    entries, total = logging_setup.read_entries(
        path, max_entries=lines, level=level, search=search or None,
        logger_prefix=source or None)

    return jsonify({
        'entries': list(reversed(entries)),
        'shown': len(entries),
        'total': total,
    })


@logs_bp.route('/download')
@login_required
@owner_required
def download():
    path = logging_setup.resolve_file(
        current_app, request.args.get('file', logging_setup.LOG_FILENAME))
    if not path:
        flash('No such log file.', 'warning')
        return redirect(url_for('logs.index'))
    return send_file(path, as_attachment=True,
                     download_name=os.path.basename(path))


@logs_bp.route('/rotate', methods=['POST'])
@login_required
@owner_required
def rotate():
    """Start a fresh log file, keeping the current one as a backup."""
    if logging_setup.rotate_now():
        current_app.logger.info('Log rotated by the owner')
        flash('Log rotated — the previous contents are now in lim.log.1.',
              'success')
    else:
        flash('No file handler is active, so there is nothing to rotate.',
              'warning')
    return redirect(url_for('logs.index'))


# ---------------------------------------------------------------------------
# Brickset API calls
# ---------------------------------------------------------------------------

@logs_bp.route('/brickset')
@login_required
@owner_required
def brickset():
    """Every call LIM has made to the Brickset API."""
    days = _int_arg('days', 7, 1, 90)
    page = _int_arg('page', 1, 1, 10000)
    method = request.args.get('method', '')
    only_failed = request.args.get('failed') == '1'

    query = BricksetApiCall.query
    if method:
        query = query.filter(BricksetApiCall.method == method)
    if only_failed:
        query = query.filter(BricksetApiCall.ok.is_(False))

    pagination = query.order_by(BricksetApiCall.id.desc()).paginate(
        page=page, per_page=50, error_out=False)

    methods = [row[0] for row in db.session.query(
        BricksetApiCall.method).distinct().order_by(
        BricksetApiCall.method).all()]

    return render_template(
        'admin/brickset_calls.html',
        stats=BricksetApiCall.stats(days),
        calls=pagination.items,
        pagination=pagination,
        days=days,
        method=method,
        only_failed=only_failed,
        methods=methods,
        retention_days=_retention_days(),
        quota_note=None,
    )


@logs_bp.route('/brickset/quota', methods=['POST'])
@login_required
@owner_required
def brickset_quota():
    """
    Ask Brickset what it thinks our usage has been.

    getKeyUsageStats is free — it doesn't count against the quota it reports.
    """
    from app.services.brickset_api import BricksetAPI

    if not BricksetAPI.is_configured():
        flash('Brickset credentials are not configured.', 'warning')
        return redirect(url_for('logs.brickset'))

    usage = BricksetAPI(context='quota-check').get_key_usage()
    if not usage:
        flash('Brickset returned no usage data.', 'warning')
    else:
        recent = ', '.join(
            f"{row.get('dateStamp', '?')}: {row.get('count', 0)}"
            for row in usage[:7])
        flash(f'Brickset reports (most recent first) — {recent}', 'info')
    return redirect(url_for('logs.brickset'))


@logs_bp.route('/brickset/retention', methods=['POST'])
@login_required
@owner_required
def brickset_retention():
    try:
        days = max(1, min(3650, int(request.form.get('days', 90))))
    except (TypeError, ValueError):
        days = 90
    SiteSetting.set('api_log_retention_days', str(days))
    db.session.commit()
    flash(f'API call records are now kept for {days} days.', 'success')
    return redirect(url_for('logs.brickset'))


@logs_bp.route('/brickset/prune', methods=['POST'])
@login_required
@owner_required
def brickset_prune():
    removed = BricksetApiCall.prune(_retention_days())
    flash(f'Removed {removed} old API call record(s).', 'success')
    return redirect(url_for('logs.brickset'))