125 lines
3.3 KiB
Python
125 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.cart import available_stock
|
|
from app.models import Order, OrderItem, Product, ProductKey
|
|
|
|
|
|
class OrderError(Exception):
|
|
pass
|
|
|
|
|
|
def make_public_id() -> str:
|
|
return secrets.token_hex(4).upper()
|
|
|
|
|
|
def create_order_from_cart(
|
|
db: Session,
|
|
*,
|
|
email: str,
|
|
cart: dict[str, int],
|
|
customer_note: str = "",
|
|
) -> Order:
|
|
if not cart:
|
|
raise OrderError("Корзина пуста")
|
|
|
|
email = email.strip().lower()
|
|
if "@" not in email or "." not in email.split("@")[-1]:
|
|
raise OrderError("Укажите корректный email")
|
|
|
|
product_ids = [int(pid) for pid in cart.keys()]
|
|
products = {
|
|
p.id: p
|
|
for p in db.scalars(
|
|
select(Product).where(Product.id.in_(product_ids), Product.is_active.is_(True))
|
|
).all()
|
|
}
|
|
|
|
order = Order(
|
|
public_id=make_public_id(),
|
|
email=email,
|
|
status="pending",
|
|
total=Decimal("0.00"),
|
|
customer_note=customer_note.strip()[:500],
|
|
)
|
|
db.add(order)
|
|
db.flush()
|
|
|
|
total = Decimal("0.00")
|
|
for pid_str, qty in cart.items():
|
|
product = products.get(int(pid_str))
|
|
if not product:
|
|
raise OrderError("Товар недоступен")
|
|
qty = int(qty)
|
|
if qty <= 0:
|
|
continue
|
|
stock = available_stock(db, product.id)
|
|
if stock < qty:
|
|
raise OrderError(
|
|
f"Недостаточно ключей для «{product.title}» (в наличии {stock})"
|
|
)
|
|
line = product.price * qty
|
|
total += line
|
|
db.add(
|
|
OrderItem(
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
title=product.title,
|
|
unit_price=product.price,
|
|
quantity=qty,
|
|
)
|
|
)
|
|
|
|
if total <= 0:
|
|
raise OrderError("Нечего оформлять")
|
|
|
|
order.total = total
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order
|
|
|
|
|
|
def pay_and_deliver(db: Session, order: Order) -> Order:
|
|
"""Demo payment: mark paid and assign unused digital keys."""
|
|
if order.status == "delivered":
|
|
return order
|
|
if order.status == "cancelled":
|
|
raise OrderError("Заказ отменён")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
items = db.scalars(select(OrderItem).where(OrderItem.order_id == order.id)).all()
|
|
for item in items:
|
|
keys = db.scalars(
|
|
select(ProductKey)
|
|
.where(
|
|
ProductKey.product_id == item.product_id,
|
|
ProductKey.is_sold.is_(False),
|
|
)
|
|
.order_by(ProductKey.id)
|
|
.limit(item.quantity)
|
|
.with_for_update()
|
|
).all()
|
|
if len(keys) < item.quantity:
|
|
raise OrderError(f"Не хватает ключей для «{item.title}»")
|
|
|
|
codes: list[str] = []
|
|
for key in keys:
|
|
key.is_sold = True
|
|
key.order_id = order.id
|
|
key.sold_at = now
|
|
codes.append(key.code)
|
|
item.delivered_codes = "\n".join(codes)
|
|
|
|
order.status = "delivered"
|
|
order.paid_at = now
|
|
order.delivered_at = now
|
|
db.commit()
|
|
db.refresh(order)
|
|
return order
|