202 lines
7.5 KiB
Python
202 lines
7.5 KiB
Python
from datetime import datetime
|
|
from typing import Dict, List
|
|
from uuid import uuid4
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
import logging
|
|
|
|
from db.models import (
|
|
Box,
|
|
File,
|
|
StagedFileProduct,
|
|
Product,
|
|
OpenBoxCard,
|
|
OpenBox,
|
|
TCGPlayerGroups,
|
|
Inventory
|
|
)
|
|
from db.utils import db_transaction
|
|
from schemas.box import CreateBoxRequest, UpdateBoxRequest, CreateOpenBoxRequest
|
|
from services.inventory import InventoryService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VALID_BOX_TYPES = {"collector", "play", "draft", "set", "commander"}
|
|
|
|
class BoxService:
|
|
def __init__(self, db: Session, inventory_service: InventoryService):
|
|
self.db = db
|
|
self.inventory_service = inventory_service
|
|
|
|
def get_staged_product_data(self, file_ids: List[str]) -> List[StagedFileProduct]:
|
|
"""Retrieve staged product data for given file IDs."""
|
|
return self.db.query(StagedFileProduct).filter(
|
|
StagedFileProduct.file_id.in_(file_ids)
|
|
).all()
|
|
|
|
def aggregate_staged_product_data(self, staged_product_data: List[StagedFileProduct]) -> Dict[Product, int]:
|
|
"""Aggregate staged product data by product and quantity."""
|
|
product_data = {}
|
|
for row in staged_product_data:
|
|
product = self.db.query(Product).filter(Product.id == row.product_id).first()
|
|
if product:
|
|
product_data[product] = product_data.get(product, 0) + row.quantity
|
|
return product_data
|
|
|
|
def add_products_to_open_box(self, open_box: OpenBox, product_data: Dict[Product, int]) -> None:
|
|
"""Add products to an open box."""
|
|
for product, quantity in product_data.items():
|
|
open_box_card = OpenBoxCard(
|
|
id=str(uuid4()),
|
|
open_box_id=open_box.id,
|
|
card_id=product.id,
|
|
quantity=quantity
|
|
)
|
|
self.db.add(open_box_card)
|
|
|
|
def validate_box_type(self, box_type: str) -> bool:
|
|
"""Validate if the box type is supported."""
|
|
return box_type in VALID_BOX_TYPES
|
|
|
|
def validate_set_code(self, set_code: str) -> bool:
|
|
"""Validate if the set code exists in TCGPlayer groups."""
|
|
return self.db.query(TCGPlayerGroups).filter(
|
|
TCGPlayerGroups.abbreviation == set_code
|
|
).first() is not None
|
|
|
|
def create_box(self, create_box_data: CreateBoxRequest) -> Box:
|
|
"""Create a new box."""
|
|
if not self.validate_box_type(create_box_data.type):
|
|
raise ValueError("Invalid box type")
|
|
if not self.validate_set_code(create_box_data.set_code):
|
|
raise ValueError("Invalid set code")
|
|
|
|
existing_box = self.db.query(Box).filter(
|
|
Box.type == create_box_data.type,
|
|
Box.set_code == create_box_data.set_code,
|
|
or_(Box.sku == create_box_data.sku, Box.sku.is_(None))
|
|
).first()
|
|
|
|
if existing_box:
|
|
return existing_box, False
|
|
else:
|
|
with db_transaction(self.db):
|
|
product = Product(
|
|
id=str(uuid4()),
|
|
type='box',
|
|
product_line='mtg'
|
|
)
|
|
box = Box(
|
|
product_id=product.id,
|
|
type=create_box_data.type,
|
|
set_code=create_box_data.set_code,
|
|
sku=create_box_data.sku,
|
|
num_cards_expected=create_box_data.num_cards_expected
|
|
)
|
|
self.db.add(product)
|
|
self.db.add(box)
|
|
|
|
return box, True
|
|
|
|
def update_box(self, box_id: str, update_box_data: UpdateBoxRequest) -> Box:
|
|
"""Update an existing box."""
|
|
box = self.db.query(Box).filter(Box.product_id == box_id).first()
|
|
if not box:
|
|
raise ValueError("Box not found")
|
|
|
|
update_data = update_box_data.dict(exclude_unset=True)
|
|
|
|
# Validate box type if it's being updated
|
|
if "type" in update_data and update_data["type"] is not None:
|
|
if not self.validate_box_type(update_data["type"]):
|
|
raise ValueError(f"Invalid box type: {update_data['type']}")
|
|
|
|
# Validate set code if it's being updated
|
|
if "set_code" in update_data and update_data["set_code"] is not None:
|
|
if not self.validate_set_code(update_data["set_code"]):
|
|
raise ValueError(f"Invalid set code: {update_data['set_code']}")
|
|
|
|
with db_transaction(self.db):
|
|
for field, value in update_data.items():
|
|
if value is not None: # Only update non-None values
|
|
setattr(box, field, value)
|
|
|
|
return box
|
|
|
|
def delete_box(self, box_id: str) -> Box:
|
|
"""Delete a box."""
|
|
box = self.db.query(Box).filter(Box.product_id == box_id).first()
|
|
product = self.db.query(Product).filter(Product.id == box_id).first()
|
|
if not box:
|
|
raise ValueError("Box not found")
|
|
|
|
with db_transaction(self.db):
|
|
self.db.delete(box)
|
|
self.db.delete(product)
|
|
return box
|
|
|
|
def open_box(self, box_id: str, box_data: CreateOpenBoxRequest) -> OpenBox:
|
|
"""Open a box and process its contents."""
|
|
box = self.db.query(Box).filter(Box.product_id == box_id).first()
|
|
if not box:
|
|
raise ValueError("Box not found")
|
|
|
|
with db_transaction(self.db):
|
|
open_box = OpenBox(
|
|
id=str(uuid4()),
|
|
product_id=box_id,
|
|
num_cards_actual=box_data.num_cards_actual,
|
|
date_opened=datetime.strptime(box_data.date_opened, "%Y-%m-%d") if box_data.date_opened else datetime.now()
|
|
)
|
|
self.db.add(open_box)
|
|
|
|
staged_product_data = self.get_staged_product_data(box_data.file_ids)
|
|
product_data = self.aggregate_staged_product_data(staged_product_data)
|
|
self.inventory_service.process_staged_products(product_data)
|
|
self.add_products_to_open_box(open_box, product_data)
|
|
|
|
# Update file box IDs
|
|
self.db.query(File).filter(File.id.in_(box_data.file_ids)).update(
|
|
{"box_id": open_box.id}, synchronize_session=False
|
|
)
|
|
|
|
return open_box
|
|
|
|
def delete_open_box(self, box_id: str) -> OpenBox:
|
|
# Fetch open box and related cards in one query
|
|
open_box = (
|
|
self.db.query(OpenBox)
|
|
.filter(OpenBox.id == box_id)
|
|
.first()
|
|
)
|
|
if not open_box:
|
|
raise ValueError("Open box not found")
|
|
|
|
# Get all open box cards and related inventory items in one query
|
|
open_box_cards = (
|
|
self.db.query(OpenBoxCard, Inventory)
|
|
.join(
|
|
Inventory,
|
|
OpenBoxCard.card_id == Inventory.product_id
|
|
)
|
|
.filter(OpenBoxCard.open_box_id == open_box.id)
|
|
.all()
|
|
)
|
|
|
|
# Process inventory adjustments
|
|
for open_box_card, inventory_item in open_box_cards:
|
|
if open_box_card.quantity > inventory_item.quantity:
|
|
raise ValueError("Open box quantity exceeds inventory quantity")
|
|
|
|
inventory_item.quantity -= open_box_card.quantity
|
|
if inventory_item.quantity == 0:
|
|
self.db.delete(inventory_item)
|
|
|
|
# Delete the open box card
|
|
self.db.delete(open_box_card)
|
|
|
|
# Execute all database operations in a single transaction
|
|
with db_transaction(self.db):
|
|
self.db.delete(open_box)
|
|
|
|
return open_box |