233 lines
7.9 KiB
Python
233 lines
7.9 KiB
Python
"""
|
|
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'))
|