""" Admin pages for the automatic instruction fetcher. Everything here is admin-only and reuses admin.admin_required so there is one definition of "is this person allowed in". Created by: Jess Rogerson (yelling commands at Claude.AI) """ from flask import (Blueprint, current_app, flash, jsonify, redirect, render_template, request, url_for) from flask_login import current_user, login_required from app import db from app.models.instruction_fetch import InstructionFetchJob from app.models.set import Set from app.models.site_setting import SiteSetting from app.routes.admin import admin_required from app.services import fetch_worker from app.services import instruction_fetcher as fetcher instruction_fetch_bp = Blueprint('instruction_fetch', __name__, url_prefix='/admin/instruction-fetch') def _int_form(name, default, low, high): try: value = int(request.form.get(name, default)) except (TypeError, ValueError): value = default return max(low, min(high, value)) @instruction_fetch_bp.route('/') @login_required @admin_required def index(): """Control panel for the fetcher.""" summary = fetcher.status_summary() jobs = InstructionFetchJob.query.order_by( InstructionFetchJob.id.desc()).limit(50).all() return render_template('admin/instruction_fetch.html', summary=summary, jobs=jobs, worker_running=fetch_worker.is_running()) @instruction_fetch_bp.route('/status') @login_required @admin_required def status(): """JSON for the page's auto-refresh.""" summary = fetcher.status_summary() summary['worker_running'] = fetch_worker.is_running() return jsonify(summary) @instruction_fetch_bp.route('/settings', methods=['POST']) @login_required @admin_required def save_settings(): SiteSetting.set('fetch_enabled', bool(request.form.get('fetch_enabled'))) SiteSetting.set('fetch_interval_minutes', str(_int_form('fetch_interval_minutes', 10, fetcher.MIN_INTERVAL_MINUTES, fetcher.MAX_INTERVAL_MINUTES))) SiteSetting.set('fetch_file_delay_seconds', str(_int_form('fetch_file_delay_seconds', 5, 0, 300))) SiteSetting.set('fetch_max_attempts', str(_int_form('fetch_max_attempts', 3, 1, 10))) SiteSetting.set('fetch_skip_sets_with_pdfs', bool(request.form.get('fetch_skip_sets_with_pdfs'))) SiteSetting.set('fetch_allow_setid_lookup', bool(request.form.get('fetch_allow_setid_lookup'))) db.session.commit() flash('Fetcher settings saved.', 'success') return redirect(url_for('instruction_fetch.index')) @instruction_fetch_bp.route('/queue', methods=['POST']) @login_required @admin_required def queue(): """Add official sets to the queue.""" scope = request.form.get('scope', 'missing') limit = request.form.get('limit', '').strip() limit = int(limit) if limit.isdigit() and int(limit) > 0 else None if scope == 'all': added = fetcher.queue_all_official(current_user.id, limit) label = 'official set(s)' elif scope == 'retry-failed': added = fetcher.requeue_failed(current_user.id) label = 'failed job(s) requeued' else: added = fetcher.queue_missing(current_user.id, limit) label = 'set(s) with no PDF yet' if added: minutes = added * fetcher.interval_minutes() hours = round(minutes / 60, 1) flash(f'Queued {added} {label}. At one every ' f'{fetcher.interval_minutes()} minutes that is about ' f'{hours} hours of background work.', 'success') if not fetcher.is_enabled(): flash('The fetcher is currently paused — switch it on below to ' 'start working through the queue.', 'warning') else: flash('Nothing new to queue.', 'info') return redirect(url_for('instruction_fetch.index')) @instruction_fetch_bp.route('/queue-set/', methods=['POST']) @login_required @admin_required def queue_set(set_id): """Queue one specific set (used from the set page).""" lego_set = Set.query.get_or_404(set_id) if lego_set.is_moc: flash('MOCs have no Brickset entry to fetch from.', 'warning') elif fetcher.queue_sets([lego_set], current_user.id): flash(f'{lego_set.set_number} queued for an instruction fetch.', 'success') else: flash(f'{lego_set.set_number} is already in the queue.', 'info') return redirect(request.referrer or url_for('instruction_fetch.index')) @instruction_fetch_bp.route('/run-now', methods=['POST']) @login_required @admin_required def run_now(): """Process one queued set immediately, in the background.""" force = bool(request.form.get('force')) if not fetcher.is_due() and not force: wait = fetcher.seconds_until_due() flash(f'Still cooling down — {wait // 60}m {wait % 60}s until the ' f'next allowed fetch. Tick "ignore the timer" to override.', 'warning') return redirect(url_for('instruction_fetch.index')) counts = InstructionFetchJob.counts() if not counts['queued']: flash('The queue is empty.', 'info') return redirect(url_for('instruction_fetch.index')) fetch_worker.run_once_async(current_app._get_current_object(), force=force) flash('Fetching one set now — the table below will update shortly.', 'info') return redirect(url_for('instruction_fetch.index')) @instruction_fetch_bp.route('/jobs//cancel', methods=['POST']) @login_required @admin_required def cancel_job(job_id): job = InstructionFetchJob.query.get_or_404(job_id) if job.is_open: job.status = 'cancelled' job.message = 'cancelled by admin' db.session.commit() flash('Job cancelled.', 'success') else: flash('That job has already finished.', 'info') return redirect(url_for('instruction_fetch.index')) @instruction_fetch_bp.route('/jobs//retry', methods=['POST']) @login_required @admin_required def retry_job(job_id): job = InstructionFetchJob.query.get_or_404(job_id) if InstructionFetchJob.open_for_set(job.set_id): flash('That set is already in the queue.', 'info') else: job.status = 'queued' job.attempts = 0 job.message = None job.not_before = None db.session.commit() flash('Job requeued.', 'success') return redirect(url_for('instruction_fetch.index')) @instruction_fetch_bp.route('/clear', methods=['POST']) @login_required @admin_required def clear(): scope = request.form.get('scope', 'finished') removed = fetcher.clear_jobs(scope) flash(f'Removed {removed} job record(s).', 'success') return redirect(url_for('instruction_fetch.index'))