diff --git a/app/services/part_catalog.py b/app/services/part_catalog.py new file mode 100644 index 0000000..f01ca01 --- /dev/null +++ b/app/services/part_catalog.py @@ -0,0 +1,237 @@ +""" +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}'