Add parts catalogue blueprint (browse, update, import, map)
This commit is contained in:
jessikitty committed
1 parent
7d32ad8fd6
commit
41ea7cab3b
1 file changed
+232
@@ -0,0 +1,232 @@
|
||||
"""
|
||||
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'))
|
||||
Reference in new issue
Block a user