238 lines
8.2 KiB
Python
238 lines
8.2 KiB
Python
"""
|
|
Parts catalogue support: seeding the category tree and looking parts up.
|
|
|
|
Brickset's API has no parts endpoint - v3 covers sets, minifigs and
|
|
instructions only - so part numbers, names and artwork come from Rebrickable
|
|
instead. That is optional: without a key you can still type labels in by hand
|
|
and upload your own images.
|
|
|
|
Get a free key at https://rebrickable.com/api/ and set it either in .env as
|
|
REBRICKABLE_API_KEY or from Admin > Parts Labels > Settings.
|
|
|
|
Created by: Jess Rogerson (yelling commands at Claude.AI)
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Dict, List, Optional
|
|
|
|
import requests
|
|
from flask import current_app
|
|
|
|
from app import db
|
|
from app.models.part_label import PartCategory, slugify
|
|
from app.models.site_setting import SiteSetting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TAXONOMY_FILE = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
'data', 'part_taxonomy.json')
|
|
|
|
|
|
# ---------------------------------------------------------------- taxonomy
|
|
|
|
def load_taxonomy() -> List[Dict]:
|
|
"""Read the seed file. Returns [] and logs if it is missing or broken."""
|
|
try:
|
|
with open(TAXONOMY_FILE, 'r', encoding='utf-8') as handle:
|
|
return json.load(handle).get('categories', [])
|
|
except Exception as exc:
|
|
logger.error('Could not read part taxonomy from %s: %s',
|
|
TAXONOMY_FILE, exc)
|
|
return []
|
|
|
|
|
|
def seed_categories(force: bool = False) -> int:
|
|
"""
|
|
Create any seed category that is missing, matched on slug.
|
|
|
|
Idempotent: existing rows are left exactly as they are, so a category you
|
|
renamed stays renamed. Returns the number of rows created. With
|
|
force=False (the default) it does nothing once any category exists,
|
|
which keeps it cheap to call on every page load.
|
|
"""
|
|
if not force and PartCategory.query.first() is not None:
|
|
return 0
|
|
|
|
created = 0
|
|
for order, parent in enumerate(load_taxonomy()):
|
|
parent_slug = slugify(parent['name'])
|
|
row = PartCategory.query.filter_by(slug=parent_slug).first()
|
|
if row is None:
|
|
row = PartCategory(code=parent.get('code'), name=parent['name'],
|
|
slug=parent_slug, sort_order=order,
|
|
colour=parent.get('colour'))
|
|
db.session.add(row)
|
|
db.session.flush()
|
|
created += 1
|
|
|
|
for child_order, child in enumerate(parent.get('children', [])):
|
|
child_slug = slugify(f"{parent['name']}-{child['name']}")
|
|
if PartCategory.query.filter_by(slug=child_slug).first():
|
|
continue
|
|
db.session.add(PartCategory(
|
|
code=child.get('code'), name=child['name'], slug=child_slug,
|
|
parent_id=row.id, sort_order=child_order,
|
|
colour=child.get('colour')))
|
|
created += 1
|
|
|
|
if created:
|
|
db.session.commit()
|
|
logger.info('Seeded %s part categories', created)
|
|
return created
|
|
|
|
|
|
def ensure_seeded():
|
|
"""Called from the labels index so a fresh install is never empty."""
|
|
try:
|
|
return seed_categories(force=False)
|
|
except Exception as exc:
|
|
db.session.rollback()
|
|
logger.error('Category seeding failed: %s', exc)
|
|
return 0
|
|
|
|
|
|
# ------------------------------------------------------------- rebrickable
|
|
|
|
class RebrickableAPI:
|
|
"""Thin read-only client for the Rebrickable parts endpoints."""
|
|
|
|
BASE_URL = 'https://rebrickable.com/api/v3/lego'
|
|
TIMEOUT = 15
|
|
|
|
def __init__(self):
|
|
self.api_key = self.get_key()
|
|
|
|
@staticmethod
|
|
def get_key() -> Optional[str]:
|
|
"""Site setting first, then config/.env."""
|
|
key = (SiteSetting.get('rebrickable_api_key', '') or '').strip()
|
|
if key:
|
|
return key
|
|
return (current_app.config.get('REBRICKABLE_API_KEY')
|
|
or os.environ.get('REBRICKABLE_API_KEY'))
|
|
|
|
@classmethod
|
|
def is_configured(cls) -> bool:
|
|
try:
|
|
return bool(cls.get_key())
|
|
except Exception:
|
|
return False
|
|
|
|
def _get(self, path: str, params: Optional[Dict] = None) -> Optional[Dict]:
|
|
if not self.api_key:
|
|
return None
|
|
headers = {
|
|
'Authorization': f'key {self.api_key}',
|
|
'Accept': 'application/json',
|
|
}
|
|
url = f'{self.BASE_URL}/{path.lstrip("/")}'
|
|
try:
|
|
response = requests.get(url, headers=headers, params=params or {},
|
|
timeout=self.TIMEOUT)
|
|
except requests.RequestException as exc:
|
|
logger.error('Rebrickable request failed (%s): %s', url, exc)
|
|
return None
|
|
|
|
if response.status_code == 401:
|
|
logger.error('Rebrickable rejected the API key')
|
|
return None
|
|
if response.status_code == 404:
|
|
return None
|
|
if response.status_code != 200:
|
|
logger.error('Rebrickable returned %s for %s',
|
|
response.status_code, url)
|
|
return None
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
logger.error('Rebrickable returned non-JSON for %s', url)
|
|
return None
|
|
|
|
def search_parts(self, query: str, limit: int = 20) -> List[Dict]:
|
|
"""Free-text search. Returns [] when unconfigured or on any error."""
|
|
query = (query or '').strip()
|
|
if not query:
|
|
return []
|
|
data = self._get('parts/', {'search': query, 'page_size': limit})
|
|
if not data:
|
|
return []
|
|
return [self._normalise(item) for item in data.get('results', [])]
|
|
|
|
def get_part(self, part_number: str) -> Optional[Dict]:
|
|
"""Exact part number lookup, e.g. '3001'."""
|
|
part_number = (part_number or '').strip()
|
|
if not part_number:
|
|
return None
|
|
data = self._get(f'parts/{part_number}/')
|
|
return self._normalise(data) if data else None
|
|
|
|
@staticmethod
|
|
def _normalise(item: Dict) -> Dict:
|
|
name = item.get('name') or ''
|
|
return {
|
|
'part_number': item.get('part_num'),
|
|
'name': name,
|
|
'dimensions': _guess_dimensions(name),
|
|
'image_url': item.get('part_img_url'),
|
|
'part_url': item.get('part_url'),
|
|
'external_ids': item.get('external_ids') or {},
|
|
}
|
|
|
|
|
|
def _guess_dimensions(name: str) -> Optional[str]:
|
|
"""
|
|
Pull '2 x 4' out of 'Brick 2 x 4'. Rebrickable part names put the
|
|
geometry in the name, so this is just a tidy-up for the label layout,
|
|
and it is fine for it to come back empty.
|
|
"""
|
|
import re
|
|
match = re.search(r'(\d+(?:\.\d+)?\s*x\s*\d+(?:\.\d+)?(?:\s*x\s*\d+(?:\.\d+)?)?)',
|
|
name or '', re.IGNORECASE)
|
|
return match.group(1).replace(' ', ' ').strip() if match else None
|
|
|
|
|
|
def cache_part_image(image_url: str, part_number: str) -> Optional[str]:
|
|
"""
|
|
Download a remote part image into UPLOAD_FOLDER/parts so labels print
|
|
without needing the network. Returns the stored relative path, or None
|
|
on any failure - the caller falls back to the remote URL.
|
|
"""
|
|
if not image_url:
|
|
return None
|
|
|
|
folder = os.path.join(current_app.config['UPLOAD_FOLDER'], 'parts')
|
|
os.makedirs(folder, exist_ok=True)
|
|
|
|
safe = ''.join(c for c in (part_number or 'part')
|
|
if c.isalnum() or c in ('-', '_'))[:40] or 'part'
|
|
extension = os.path.splitext(image_url.split('?')[0])[1].lower()
|
|
if extension not in ('.png', '.jpg', '.jpeg', '.gif', '.webp'):
|
|
extension = '.png'
|
|
filename = f'{safe}{extension}'
|
|
destination = os.path.join(folder, filename)
|
|
|
|
try:
|
|
response = requests.get(image_url, timeout=20)
|
|
response.raise_for_status()
|
|
with open(destination, 'wb') as handle:
|
|
handle.write(response.content)
|
|
except Exception as exc:
|
|
logger.warning('Could not cache part image %s: %s', image_url, exc)
|
|
return None
|
|
|
|
# Pillow is already a dependency; shrink anything daft before it ends up
|
|
# on a 20mm label.
|
|
try:
|
|
from PIL import Image
|
|
with Image.open(destination) as img:
|
|
if max(img.size) > 600:
|
|
img.thumbnail((600, 600))
|
|
img.save(destination)
|
|
except Exception:
|
|
pass
|
|
|
|
return f'parts/{filename}'
|