2025-01-28 19:18:51 +01:00
|
|
|
import os
|
|
|
|
import logging
|
|
|
|
from sqlite3 import Row
|
|
|
|
from typing import Any, TYPE_CHECKING
|
|
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
|
|
from flask import current_app, url_for
|
|
|
|
|
|
|
|
from .exceptions import ErrorException
|
|
|
|
from .rebrickable_image import RebrickableImage
|
|
|
|
from .record import BrickRecord
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
from .minifigure import BrickMinifigure
|
|
|
|
from .set import BrickSet
|
|
|
|
from .socket import BrickSocket
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
|
# A part from Rebrickable
|
|
|
|
class RebrickablePart(BrickRecord):
|
|
|
|
socket: 'BrickSocket'
|
|
|
|
brickset: 'BrickSet | None'
|
|
|
|
minifigure: 'BrickMinifigure | None'
|
|
|
|
|
|
|
|
# Queries
|
|
|
|
select_query: str = 'rebrickable/part/select'
|
|
|
|
insert_query: str = 'rebrickable/part/insert'
|
|
|
|
|
|
|
|
def __init__(
|
|
|
|
self,
|
|
|
|
/,
|
|
|
|
*,
|
|
|
|
brickset: 'BrickSet | None' = None,
|
|
|
|
minifigure: 'BrickMinifigure | None' = None,
|
|
|
|
record: Row | dict[str, Any] | None = None
|
|
|
|
):
|
|
|
|
super().__init__()
|
|
|
|
|
|
|
|
# Save the brickset
|
|
|
|
self.brickset = brickset
|
|
|
|
|
|
|
|
# Save the minifigure
|
|
|
|
self.minifigure = minifigure
|
|
|
|
|
|
|
|
# Ingest the record if it has one
|
|
|
|
if record is not None:
|
|
|
|
self.ingest(record)
|
|
|
|
|
|
|
|
# Insert the part from Rebrickable
|
2025-01-28 23:07:12 +01:00
|
|
|
def insert_rebrickable(self, /) -> None:
|
2025-01-28 19:18:51 +01:00
|
|
|
if self.brickset is None:
|
|
|
|
raise ErrorException('Importing a part from Rebrickable outside of a set is not supported') # noqa: E501
|
|
|
|
|
|
|
|
# Insert the Rebrickable part to the database
|
2025-01-28 23:07:12 +01:00
|
|
|
self.insert(
|
2025-01-28 19:18:51 +01:00
|
|
|
commit=False,
|
|
|
|
no_defer=True,
|
|
|
|
override_query=RebrickablePart.insert_query
|
|
|
|
)
|
|
|
|
|
2025-01-28 23:07:12 +01:00
|
|
|
if not current_app.config['USE_REMOTE_IMAGES']:
|
|
|
|
RebrickableImage(
|
|
|
|
self.brickset,
|
|
|
|
minifigure=self.minifigure,
|
|
|
|
part=self,
|
|
|
|
).download()
|
2025-01-28 19:18:51 +01:00
|
|
|
|
|
|
|
# Return a dict with common SQL parameters for a part
|
|
|
|
def sql_parameters(self, /) -> dict[str, Any]:
|
|
|
|
parameters = super().sql_parameters()
|
|
|
|
|
|
|
|
# Set id
|
|
|
|
if self.brickset is not None:
|
|
|
|
parameters['id'] = self.brickset.fields.id
|
|
|
|
|
|
|
|
# Use the minifigure number if present,
|
|
|
|
if self.minifigure is not None:
|
|
|
|
parameters['figure'] = self.minifigure.fields.figure
|
|
|
|
else:
|
|
|
|
parameters['figure'] = None
|
|
|
|
|
|
|
|
return parameters
|
|
|
|
|
|
|
|
# Self url
|
|
|
|
def url(self, /) -> str:
|
|
|
|
return url_for(
|
|
|
|
'part.details',
|
|
|
|
part=self.fields.part,
|
|
|
|
color=self.fields.color,
|
|
|
|
)
|
|
|
|
|
|
|
|
# Compute the url for the bricklink page
|
|
|
|
def url_for_bricklink(self, /) -> str:
|
|
|
|
if current_app.config['BRICKLINK_LINKS']:
|
|
|
|
try:
|
|
|
|
return current_app.config['BRICKLINK_LINK_PART_PATTERN'].format( # noqa: E501
|
|
|
|
part=self.fields.part,
|
|
|
|
)
|
|
|
|
except Exception:
|
|
|
|
pass
|
|
|
|
|
|
|
|
return ''
|
|
|
|
|
|
|
|
# Compute the url for the part image
|
|
|
|
def url_for_image(self, /) -> str:
|
|
|
|
if not current_app.config['USE_REMOTE_IMAGES']:
|
|
|
|
if self.fields.image is None:
|
|
|
|
file = RebrickableImage.nil_name()
|
|
|
|
else:
|
|
|
|
file = self.fields.image_id
|
|
|
|
|
|
|
|
return RebrickableImage.static_url(file, 'PARTS_FOLDER')
|
|
|
|
else:
|
|
|
|
if self.fields.image is None:
|
|
|
|
return current_app.config['REBRICKABLE_IMAGE_NIL']
|
|
|
|
else:
|
|
|
|
return self.fields.image
|
|
|
|
|
|
|
|
# Compute the url for missing part
|
|
|
|
def url_for_missing(self, /) -> str:
|
|
|
|
# Different URL for a minifigure part
|
|
|
|
if self.minifigure is not None:
|
|
|
|
figure = self.minifigure.fields.figure
|
|
|
|
else:
|
|
|
|
figure = None
|
|
|
|
|
|
|
|
return url_for(
|
|
|
|
'set.missing_part',
|
|
|
|
id=self.fields.id,
|
|
|
|
figure=figure,
|
|
|
|
part=self.fields.part,
|
|
|
|
color=self.fields.color,
|
|
|
|
spare=self.fields.spare,
|
|
|
|
)
|
|
|
|
|
|
|
|
# Compute the url for the rebrickable page
|
|
|
|
def url_for_rebrickable(self, /) -> str:
|
|
|
|
if current_app.config['REBRICKABLE_LINKS']:
|
|
|
|
try:
|
|
|
|
return current_app.config['REBRICKABLE_LINK_PART_PATTERN'].format( # noqa: E501
|
|
|
|
part=self.fields.part,
|
|
|
|
color=self.fields.color,
|
|
|
|
)
|
|
|
|
except Exception:
|
|
|
|
pass
|
|
|
|
|
|
|
|
return ''
|
|
|
|
|
|
|
|
# Normalize from Rebrickable
|
|
|
|
@staticmethod
|
|
|
|
def from_rebrickable(
|
|
|
|
data: dict[str, Any],
|
|
|
|
/,
|
|
|
|
*,
|
|
|
|
brickset: 'BrickSet | None' = None,
|
|
|
|
minifigure: 'BrickMinifigure | None' = None,
|
|
|
|
**_,
|
|
|
|
) -> dict[str, Any]:
|
|
|
|
record = {
|
|
|
|
'id': None,
|
|
|
|
'figure': None,
|
|
|
|
'part': data['part']['part_num'],
|
|
|
|
'color': data['color']['id'],
|
|
|
|
'spare': data['is_spare'],
|
|
|
|
'quantity': data['quantity'],
|
|
|
|
'rebrickable_inventory': data['id'],
|
|
|
|
'element': data['element_id'],
|
|
|
|
'color_id': data['color']['id'],
|
|
|
|
'color_name': data['color']['name'],
|
|
|
|
'color_rgb': data['color']['rgb'],
|
|
|
|
'color_transparent': data['color']['is_trans'],
|
|
|
|
'name': data['part']['name'],
|
|
|
|
'category': data['part']['part_cat_id'],
|
|
|
|
'image': data['part']['part_img_url'],
|
|
|
|
'image_id': None,
|
|
|
|
'url': data['part']['part_url'],
|
|
|
|
'print': data['part']['print_of']
|
|
|
|
}
|
|
|
|
|
|
|
|
if brickset is not None:
|
|
|
|
record['id'] = brickset.fields.id
|
|
|
|
|
|
|
|
if minifigure is not None:
|
|
|
|
record['figure'] = minifigure.fields.figure
|
|
|
|
|
|
|
|
# Extract the file name
|
|
|
|
if record['image'] is not None:
|
|
|
|
image_id, _ = os.path.splitext(
|
|
|
|
os.path.basename(
|
|
|
|
urlparse(record['image']).path
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
if image_id is not None or image_id != '':
|
|
|
|
record['image_id'] = image_id
|
|
|
|
|
|
|
|
return record
|