lego-instructions-manager/app/services/logging_setup.py

"""
Logging setup and log reading.

LIM previously configured no logging at all, which left Flask's logger at
WARNING and wrote nothing to disk — so a quiet background feature like the
instruction fetcher had no trail to follow. This attaches a rotating file
handler to the root logger, so every module logger (app.services.*, and the
fetcher in particular) lands in one file, and provides the parsing the admin
log viewer needs.

Rotation is by size, not date: LOG_MAX_BYTES per file, LOG_BACKUP_COUNT old
copies, so disk use is bounded no matter how chatty things get.

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

import io
import logging
import os
import re
from logging.handlers import RotatingFileHandler

LOG_FILENAME = 'lim.log'

LEVELS = ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL')

LEVEL_BADGES = {
    'DEBUG': 'secondary',
    'INFO': 'info',
    'WARNING': 'warning',
    'ERROR': 'danger',
    'CRITICAL': 'danger',
}

# 2026-09-26 13:38:00 INFO     app.services.instruction_fetcher | message
LINE_RE = re.compile(
    r'^(?P<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+'
    r'(?P<level>[A-Z]+)\s+'
    r'(?P<logger>\S+)\s+\|\s'
    r'(?P<message>.*)$'
)

_FORMAT = '%(asctime)s %(levelname)-8s %(name)s | %(message)s'
_DATEFMT = '%Y-%m-%d %H:%M:%S'


def log_path(app):
    return os.path.join(app.config['LOG_FOLDER'], LOG_FILENAME)


def configure_logging(app):
    """
    Attach the rotating file handler. Idempotent — safe on a reloader restart.
    """
    folder = app.config['LOG_FOLDER']
    try:
        os.makedirs(folder, exist_ok=True)
    except Exception as exc:
        # A read-only or missing volume shouldn't stop the app booting.
        app.logger.warning('Could not create log folder %s: %s', folder, exc)
        return None

    level = getattr(logging, str(app.config.get('LOG_LEVEL', 'INFO')).upper(),
                    logging.INFO)
    path = log_path(app)

    root = logging.getLogger()
    for existing in root.handlers:
        if getattr(existing, '_lim_handler', False):
            existing.setLevel(level)
            root.setLevel(level)
            app.logger.setLevel(level)
            return path

    try:
        handler = RotatingFileHandler(
            path,
            maxBytes=int(app.config.get('LOG_MAX_BYTES', 2 * 1024 * 1024)),
            backupCount=int(app.config.get('LOG_BACKUP_COUNT', 5)),
            encoding='utf-8',
        )
    except Exception as exc:
        app.logger.warning('Could not open log file %s: %s', path, exc)
        return None

    handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATEFMT))
    handler.setLevel(level)
    handler._lim_handler = True

    root.addHandler(handler)
    root.setLevel(level)
    app.logger.setLevel(level)

    # Request access logs are noisy and rarely what you're looking for.
    if not app.config.get('LOG_ACCESS', False):
        logging.getLogger('werkzeug').setLevel(logging.WARNING)

    app.logger.info('Logging to %s at %s', path,
                    logging.getLevelName(level))
    return path


def get_handler():
    for handler in logging.getLogger().handlers:
        if getattr(handler, '_lim_handler', False):
            return handler
    return None


def rotate_now():
    """Force a rollover. Returns True if it happened."""
    handler = get_handler()
    if handler is None:
        return False
    handler.doRollover()
    return True


def available_files(app):
    """The live log plus its rotated backups, newest first."""
    folder = app.config['LOG_FOLDER']
    if not os.path.isdir(folder):
        return []

    found = []
    for name in os.listdir(folder):
        if name == LOG_FILENAME or name.startswith(LOG_FILENAME + '.'):
            full = os.path.join(folder, name)
            try:
                found.append({
                    'name': name,
                    'size': os.path.getsize(full),
                    'current': name == LOG_FILENAME,
                })
            except OSError:
                continue

    # lim.log first, then lim.log.1, lim.log.2, ...
    found.sort(key=lambda item: (not item['current'], item['name']))
    return found


def resolve_file(app, name):
    """Turn a user-supplied file name into a safe path inside LOG_FOLDER."""
    if not name:
        name = LOG_FILENAME
    if name != LOG_FILENAME and not name.startswith(LOG_FILENAME + '.'):
        return None
    if os.path.sep in name or '/' in name or '..' in name:
        return None
    path = os.path.join(app.config['LOG_FOLDER'], name)
    return path if os.path.isfile(path) else None


def _tail_bytes(path, limit):
    """Last `limit` bytes of a file, decoded leniently."""
    size = os.path.getsize(path)
    with io.open(path, 'rb') as handle:
        if size > limit:
            handle.seek(size - limit)
            handle.readline()  # discard the partial first line
        return handle.read().decode('utf-8', errors='replace')


def read_entries(path, max_entries=300, level=None, search=None,
                 logger_prefix=None, byte_limit=1024 * 1024):
    """
    Parse the tail of a log file into entries, newest last.

    Lines that don't match the format (traceback continuations) are attached
    to the entry above them, so a stack trace stays with its error.
    """
    if not path or not os.path.isfile(path):
        return [], 0

    text = _tail_bytes(path, byte_limit)
    entries = []

    for line in text.splitlines():
        match = LINE_RE.match(line)
        if match:
            entries.append({
                'time': match.group('time'),
                'level': match.group('level'),
                'logger': match.group('logger'),
                'message': match.group('message'),
                'extra': [],
            })
        elif entries:
            entries[-1]['extra'].append(line)

    total = len(entries)

    if level:
        wanted = str(level).upper()
        if wanted in LEVELS:
            floor = LEVELS.index(wanted)
            entries = [entry for entry in entries
                       if entry['level'] in LEVELS
                       and LEVELS.index(entry['level']) >= floor]

    if logger_prefix:
        entries = [entry for entry in entries
                   if entry['logger'].startswith(logger_prefix)]

    if search:
        needle = search.lower()
        entries = [
            entry for entry in entries
            if needle in entry['message'].lower()
            or needle in entry['logger'].lower()
            or any(needle in line.lower() for line in entry['extra'])
        ]

    for entry in entries:
        entry['badge'] = LEVEL_BADGES.get(entry['level'], 'secondary')

    return entries[-max_entries:], total