Add part catalogue service (taxonomy seeding, Rebrickable lookup)

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}'