""" The local parts catalogue - admin only. Browse the Rebrickable catalogue after the Update button has pulled it in, and turn any row into a drawer label without typing it out. Routes (all under /admin/labels/catalog): / browse the catalogue /update start a sync (background thread) /status JSON progress, polled by the Update button /import create labels from one or more catalogue rows /map map a Rebrickable category to one of your drawer groups Created by: Jess Rogerson (yelling commands at Claude.AI) """ import logging from flask import (Blueprint, current_app, flash, jsonify, redirect, render_template, request, url_for) from flask_login import current_user, login_required from sqlalchemy import or_ from app import db from app.models.part_catalog_item import (CatalogSync, PartCatalogCategory, PartCatalogItem) from app.models.part_label import PartCategory, PartLabel from app.routes.admin import admin_required from app.services import catalog_sync from app.services.part_catalog import RebrickableAPI, cache_part_image logger = logging.getLogger(__name__) catalog_bp = Blueprint('catalog', __name__, url_prefix='/admin/labels/catalog') PER_PAGE = 60 @catalog_bp.route('/') @login_required @admin_required def index(): search = (request.args.get('q') or '').strip() category_id = request.args.get('category', type=int) only = request.args.get('only') or '' page = request.args.get('page', 1, type=int) query = PartCatalogItem.query if category_id: query = query.filter(PartCatalogItem.category_id == category_id) if search: pattern = f'%{search}%' query = query.filter(or_( PartCatalogItem.name.ilike(pattern), PartCatalogItem.part_num.ilike(pattern), PartCatalogItem.dimensions.ilike(pattern), )) if only == 'labelled': query = query.filter(PartCatalogItem.has_label.is_(True)) elif only == 'unlabelled': query = query.filter(PartCatalogItem.has_label.is_(False)) pagination = (query.order_by(PartCatalogItem.name, PartCatalogItem.part_num) .paginate(page=page, per_page=PER_PAGE, error_out=False)) categories = (PartCatalogCategory.query .order_by(PartCatalogCategory.name).all()) return render_template( 'admin/labels/catalog.html', items=pagination.items, pagination=pagination, categories=categories, active_category=(PartCatalogCategory.query.get(category_id) if category_id else None), search=search, only=only, stats=catalog_sync.catalog_stats(), drawer_choices=PartCategory.choices(), syncing=catalog_sync.is_running(), ) @catalog_bp.route('/update', methods=['POST']) @login_required @admin_required def update(): """Start a catalogue sync. Returns JSON for the button, or redirects.""" run = catalog_sync.start_sync(current_app._get_current_object(), user_id=current_user.id) if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify(run.to_dict()) if run.is_running: flash('Catalogue update started. It runs in the background.', 'info') return redirect(url_for('catalog.index')) @catalog_bp.route('/status') @login_required @admin_required def status(): run = CatalogSync.latest() if run is None: return jsonify({'status': 'none', 'finished': True, 'progress': 0}) payload = run.to_dict() payload.update(catalog_counts()) return jsonify(payload) def catalog_counts(): return { 'total_parts': PartCatalogItem.query.count(), 'total_categories': PartCatalogCategory.query.count(), } @catalog_bp.route('/import', methods=['POST']) @login_required @admin_required def import_items(): """ Turn one or more catalogue rows into drawer labels. The drawer category comes from the form if given, otherwise from the mapping on the part's Rebrickable category, otherwise nothing - a label with no category still prints fine. """ part_nums = request.form.getlist('part_nums') if len(part_nums) == 1 and ',' in part_nums[0]: part_nums = [p.strip() for p in part_nums[0].split(',') if p.strip()] if not part_nums: flash('Nothing selected to import.', 'warning') return redirect(request.referrer or url_for('catalog.index')) forced_category = request.form.get('category_id', type=int) or None location = (request.form.get('location') or '').strip() or None fetch_images = bool(request.form.get('fetch_images')) items = (PartCatalogItem.query .filter(PartCatalogItem.part_num.in_(part_nums)).all()) existing = { label.part_number for label in PartLabel.query.filter(PartLabel.part_number.in_(part_nums)).all() if label.part_number } api = RebrickableAPI() if (fetch_images and RebrickableAPI.is_configured()) else None created = 0 skipped = 0 for item in items: if item.part_num in existing: item.has_label = True skipped += 1 continue category_id = forced_category if category_id is None and item.catalog_category: category_id = item.catalog_category.drawer_category_id label = PartLabel( part_number=item.part_num, name=_strip_dimensions(item.name, item.dimensions), dimensions=item.dimensions, category_id=category_id, location=location, source='rebrickable', created_by_id=current_user.id, ) if api is not None: data = api.get_part(item.part_num) if data and data.get('image_url'): label.image_url = data['image_url'] item.image_url = data['image_url'] cached = cache_part_image(data['image_url'], item.part_num) if cached: label.image = cached db.session.add(label) item.has_label = True created += 1 db.session.commit() if created: message = f'Created {created} label{"" if created == 1 else "s"}.' if skipped: message += f' {skipped} already had one.' flash(message, 'success') else: flash('Those parts already have labels.', 'info') return redirect(request.referrer or url_for('catalog.index')) def _strip_dimensions(name, dimensions): """ 'Brick 2 x 4' + '2 x 4' -> 'Brick'. The label stores name and dimensions separately so the print options can show or hide the geometry; leaving it in both places double-prints it. """ if not dimensions or not name: return (name or '')[:160] trimmed = name for variant in (dimensions, dimensions.replace(' ', '')): if variant and variant.lower() in trimmed.lower(): start = trimmed.lower().index(variant.lower()) trimmed = (trimmed[:start] + trimmed[start + len(variant):]) trimmed = ' '.join(trimmed.split()).strip(' ,-') return (trimmed or name)[:160] @catalog_bp.route('/map', methods=['POST']) @login_required @admin_required def map_category(): """Point a Rebrickable category at one of your drawer categories.""" catalog_category = PartCatalogCategory.query.get_or_404( request.form.get('catalog_category_id', type=int)) catalog_category.drawer_category_id = ( request.form.get('drawer_category_id', type=int) or None) db.session.commit() target = catalog_category.drawer_category if target: flash(f'"{catalog_category.name}" now imports into ' f'{target.full_name}.', 'success') else: flash(f'Mapping cleared for "{catalog_category.name}".', 'info') return redirect(request.referrer or url_for('catalog.index'))