- Plate lookup: new plate_vehicles table (v1.7 migration), plate_lookup service with Mexican plate validation, GET/POST endpoints on catalog_bp, plate search UI in catalog vehicle selector - Translations: extend PART_TRANSLATIONS from ~80 to 326 entries covering brake, engine, fuel, cooling, electrical, drivetrain, suspension, steering, exhaust, A/C, lighting, body, interior, fluids, and category translations - Bulk images: image_scraper service with download+resize+placeholder generation, bulk-images and auto-image endpoints on inventory_bp Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
188
pos/services/image_scraper.py
Normal file
188
pos/services/image_scraper.py
Normal file
@@ -0,0 +1,188 @@
|
||||
# /home/Autopartes/pos/services/image_scraper.py
|
||||
"""Bulk image downloader and processor for inventory parts.
|
||||
|
||||
Provides two capabilities:
|
||||
1. Bulk import: Accept a list of {part_number, image_url}, download each image,
|
||||
resize/optimize with Pillow, save to /pos/static/images/parts/, and update
|
||||
the inventory.image_url in the tenant DB.
|
||||
2. Auto-image: Generate a placeholder image with the part number text when no
|
||||
real image is available.
|
||||
"""
|
||||
|
||||
import io
|
||||
import os
|
||||
import logging
|
||||
import requests
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
IMAGES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'static', 'images', 'parts')
|
||||
MAX_SIZE = 800
|
||||
THUMB_SIZE = 300
|
||||
QUALITY = 85
|
||||
DOWNLOAD_TIMEOUT = 15
|
||||
|
||||
|
||||
def _ensure_dir():
|
||||
"""Ensure the parts image directory exists."""
|
||||
os.makedirs(IMAGES_DIR, exist_ok=True)
|
||||
|
||||
|
||||
def _process_and_save(image_data, tenant_id, item_id):
|
||||
"""Resize image, save full + thumbnail, return the relative URL path."""
|
||||
_ensure_dir()
|
||||
|
||||
img = Image.open(io.BytesIO(image_data))
|
||||
if img.mode not in ('RGB', 'L'):
|
||||
img = img.convert('RGB')
|
||||
|
||||
# Full size
|
||||
full = img.copy()
|
||||
full.thumbnail((MAX_SIZE, MAX_SIZE), Image.LANCZOS)
|
||||
full_path = os.path.join(IMAGES_DIR, f'{tenant_id}_{item_id}.jpg')
|
||||
full.save(full_path, format='JPEG', quality=QUALITY)
|
||||
|
||||
# Thumbnail
|
||||
thumb = img.copy()
|
||||
thumb.thumbnail((THUMB_SIZE, THUMB_SIZE), Image.LANCZOS)
|
||||
thumb_path = os.path.join(IMAGES_DIR, f'{tenant_id}_{item_id}_thumb.jpg')
|
||||
thumb.save(thumb_path, format='JPEG', quality=QUALITY)
|
||||
|
||||
return f'/pos/static/images/parts/{tenant_id}_{item_id}.jpg'
|
||||
|
||||
|
||||
def download_and_process(url, tenant_id, item_id):
|
||||
"""Download an image from a URL, process it, and save locally.
|
||||
|
||||
Returns the relative URL path on success, or raises on failure.
|
||||
"""
|
||||
resp = requests.get(url, timeout=DOWNLOAD_TIMEOUT, stream=True)
|
||||
resp.raise_for_status()
|
||||
|
||||
# Read up to 10 MB
|
||||
data = resp.content
|
||||
if len(data) > 10 * 1024 * 1024:
|
||||
raise ValueError('Image exceeds 10 MB limit')
|
||||
|
||||
return _process_and_save(data, tenant_id, item_id)
|
||||
|
||||
|
||||
def bulk_import(tenant_conn, tenant_id, items):
|
||||
"""Process a list of {part_number, image_url} items.
|
||||
|
||||
For each item:
|
||||
1. Find the inventory item by part_number
|
||||
2. Download the image
|
||||
3. Resize and save
|
||||
4. Update inventory.image_url
|
||||
|
||||
Returns: {imported: int, errors: [str]}
|
||||
"""
|
||||
imported = 0
|
||||
errors = []
|
||||
cur = tenant_conn.cursor()
|
||||
|
||||
try:
|
||||
for entry in items:
|
||||
pn = (entry.get('part_number') or '').strip()
|
||||
url = (entry.get('image_url') or '').strip()
|
||||
|
||||
if not pn or not url:
|
||||
errors.append(f'Missing part_number or image_url: {entry}')
|
||||
continue
|
||||
|
||||
# Find inventory item
|
||||
cur.execute(
|
||||
"SELECT id FROM inventory WHERE part_number = %s AND is_active = true LIMIT 1",
|
||||
(pn,)
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if not row:
|
||||
errors.append(f'Part not found: {pn}')
|
||||
continue
|
||||
|
||||
item_id = row[0]
|
||||
|
||||
try:
|
||||
rel_url = download_and_process(url, tenant_id, item_id)
|
||||
cur.execute(
|
||||
"UPDATE inventory SET image_url = %s WHERE id = %s",
|
||||
(rel_url, item_id)
|
||||
)
|
||||
tenant_conn.commit()
|
||||
imported += 1
|
||||
except Exception as e:
|
||||
tenant_conn.rollback()
|
||||
errors.append(f'{pn}: {str(e)}')
|
||||
logger.warning('Failed to import image for %s: %s', pn, e)
|
||||
|
||||
finally:
|
||||
cur.close()
|
||||
|
||||
return {'imported': imported, 'errors': errors}
|
||||
|
||||
|
||||
def generate_placeholder(tenant_id, item_id, part_number, name=''):
|
||||
"""Generate a placeholder image with the part number text.
|
||||
|
||||
Creates a 400x400 gray image with the part number centered.
|
||||
Returns the relative URL path.
|
||||
"""
|
||||
_ensure_dir()
|
||||
|
||||
width, height = 400, 400
|
||||
bg_color = (240, 240, 240)
|
||||
text_color = (80, 80, 80)
|
||||
accent_color = (245, 166, 35) # Nexus orange
|
||||
|
||||
img = Image.new('RGB', (width, height), bg_color)
|
||||
draw = ImageDraw.Draw(img)
|
||||
|
||||
# Draw accent bar at top
|
||||
draw.rectangle([0, 0, width, 6], fill=accent_color)
|
||||
|
||||
# Draw part icon placeholder (box outline)
|
||||
box_margin = 100
|
||||
draw.rectangle(
|
||||
[box_margin, box_margin, width - box_margin, height - box_margin - 40],
|
||||
outline=(200, 200, 200), width=2
|
||||
)
|
||||
|
||||
# Try to use a built-in font, fall back to default
|
||||
try:
|
||||
font_large = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 24)
|
||||
font_small = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 14)
|
||||
except (IOError, OSError):
|
||||
font_large = ImageFont.load_default()
|
||||
font_small = font_large
|
||||
|
||||
# Part number (centered)
|
||||
pn_text = part_number or 'SIN NUMERO'
|
||||
bbox = draw.textbbox((0, 0), pn_text, font=font_large)
|
||||
tw = bbox[2] - bbox[0]
|
||||
draw.text(((width - tw) / 2, height - 80), pn_text, fill=accent_color, font=font_large)
|
||||
|
||||
# Name (centered, below part number)
|
||||
if name:
|
||||
display_name = name[:30] + ('...' if len(name) > 30 else '')
|
||||
bbox2 = draw.textbbox((0, 0), display_name, font=font_small)
|
||||
tw2 = bbox2[2] - bbox2[0]
|
||||
draw.text(((width - tw2) / 2, height - 50), display_name, fill=text_color, font=font_small)
|
||||
|
||||
# "SIN IMAGEN" text centered in box
|
||||
no_img = 'SIN IMAGEN'
|
||||
bbox3 = draw.textbbox((0, 0), no_img, font=font_small)
|
||||
tw3 = bbox3[2] - bbox3[0]
|
||||
draw.text(((width - tw3) / 2, (height - 40) / 2), no_img, fill=(180, 180, 180), font=font_small)
|
||||
|
||||
# Save
|
||||
full_path = os.path.join(IMAGES_DIR, f'{tenant_id}_{item_id}.jpg')
|
||||
img.save(full_path, format='JPEG', quality=QUALITY)
|
||||
|
||||
thumb = img.copy()
|
||||
thumb.thumbnail((THUMB_SIZE, THUMB_SIZE), Image.LANCZOS)
|
||||
thumb_path = os.path.join(IMAGES_DIR, f'{tenant_id}_{item_id}_thumb.jpg')
|
||||
thumb.save(thumb_path, format='JPEG', quality=QUALITY)
|
||||
|
||||
return f'/pos/static/images/parts/{tenant_id}_{item_id}.jpg'
|
||||
Reference in New Issue
Block a user