132 lines
5.1 KiB
Python
132 lines
5.1 KiB
Python
import uuid
|
|
|
|
from decimal import Decimal
|
|
|
|
from brewman.models.batch import Batch
|
|
from brewman.models.cost_centre import CostCentre
|
|
from brewman.models.inventory import Inventory
|
|
from brewman.models.journal import Journal
|
|
from brewman.models.period import Period
|
|
from brewman.models.price import Price
|
|
from brewman.models.voucher import Voucher
|
|
from brewman.models.voucher_type import VoucherType
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import distinct, func, select
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models.recipe import Recipe
|
|
from ..models.recipe_item import RecipeItem
|
|
from ..models.stock_keeping_unit import StockKeepingUnit
|
|
|
|
|
|
def calculate_prices(period_id: uuid.UUID, db: Session):
|
|
try:
|
|
item: Period = db.execute(select(Period).where(Period.id == period_id)).scalar_one()
|
|
recipes = set(
|
|
db.execute(select(distinct(StockKeepingUnit.product_id)).join(StockKeepingUnit.recipes)).scalars().all()
|
|
)
|
|
ingredients = set(db.execute(select(distinct(RecipeItem.product_id))).scalars().all())
|
|
issued_products = get_issue_prices(item, ingredients - recipes, db)
|
|
left = ingredients - recipes - issued_products.keys()
|
|
purchased_products = get_issue_prices(item, left, db)
|
|
left -= purchased_products.keys()
|
|
rest = get_rest(left, db)
|
|
prices = issued_products | purchased_products | rest
|
|
|
|
while len(recipes) > 0:
|
|
calculate_recipes(recipes, prices, db)
|
|
for pid, price in prices.items():
|
|
db.execute(
|
|
pg_insert(Price)
|
|
.values(product_id=pid, price=price, period_id=item.id)
|
|
.on_conflict_do_update(constraint="uq_prices_period_id", set_=dict(price=price))
|
|
)
|
|
except SQLAlchemyError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
|
|
|
|
def get_issue_prices(period: Period, products: set[uuid.UUID], db: Session) -> dict[uuid.UUID, Decimal]:
|
|
sum_quantity = func.sum(
|
|
Inventory.quantity * StockKeepingUnit.fraction * StockKeepingUnit.product_yield * Journal.debit
|
|
).label("quantity")
|
|
sum_net = func.sum(Inventory.rate * Inventory.quantity * Journal.debit).label("net")
|
|
|
|
d: dict[uuid.UUID, Decimal] = {}
|
|
query: list[tuple[uuid.UUID, Decimal]] = db.execute(
|
|
select(StockKeepingUnit.product_id, sum_net / sum_quantity)
|
|
.join(Inventory.batch)
|
|
.join(Batch.sku)
|
|
.join(Inventory.voucher)
|
|
.join(Voucher.journals)
|
|
.where(
|
|
Voucher.date_ >= period.valid_from,
|
|
Voucher.date_ <= period.valid_till,
|
|
Voucher.voucher_type.in_([VoucherType.ISSUE, VoucherType.CLOSING_STOCK]),
|
|
Journal.cost_centre_id != CostCentre.cost_centre_purchase(),
|
|
StockKeepingUnit.product_id.in_(products),
|
|
)
|
|
.group_by(StockKeepingUnit.product_id, Journal.debit)
|
|
).all()
|
|
for id, amount in query:
|
|
d[id] = amount
|
|
return d
|
|
|
|
|
|
def get_purchase_prices(period: Period, req: set[uuid.UUID], db: Session) -> dict[uuid.UUID, Decimal]:
|
|
sum_quantity = func.sum(
|
|
Inventory.quantity * StockKeepingUnit.fraction * StockKeepingUnit.product_yield * Journal.debit
|
|
).label("quantity")
|
|
sum_net = func.sum(Inventory.rate * Inventory.quantity * Journal.debit).label("net")
|
|
|
|
d: dict[uuid.UUID, Decimal] = {}
|
|
query: list[tuple[uuid.UUID, Decimal]] = db.execute(
|
|
select(StockKeepingUnit.product_id, sum_net / sum_quantity)
|
|
.join(Inventory.batch)
|
|
.join(Batch.sku)
|
|
.join(Inventory.voucher)
|
|
.join(Voucher.journals)
|
|
.where(
|
|
Voucher.date_ >= period.valid_from,
|
|
Voucher.date_ <= period.valid_till,
|
|
Voucher.voucher_type == VoucherType.PURCHASE,
|
|
StockKeepingUnit.product_id.in_(req),
|
|
)
|
|
.group_by(StockKeepingUnit.product_id, Journal.debit)
|
|
).all()
|
|
for id, amount in query:
|
|
d[id] = amount
|
|
return d
|
|
|
|
|
|
def get_rest(req: set[uuid.UUID], db: Session) -> dict[uuid.UUID, Decimal]:
|
|
d: dict[uuid.UUID, Decimal] = {}
|
|
query = db.execute(
|
|
select(
|
|
StockKeepingUnit.product_id,
|
|
StockKeepingUnit.cost_price / (StockKeepingUnit.fraction * StockKeepingUnit.product_yield),
|
|
).where(StockKeepingUnit.product_id.in_(req))
|
|
).all()
|
|
for id, amount in query:
|
|
d[id] = amount
|
|
return d
|
|
|
|
|
|
def calculate_recipes(recipes: set[uuid.UUID], prices: dict[uuid.UUID, Decimal], db: Session) -> None:
|
|
sq = select(RecipeItem.recipe_id).where(RecipeItem.product_id.in_(recipes))
|
|
items = (
|
|
db.execute(
|
|
select(Recipe).join(Recipe.sku).where(StockKeepingUnit.product_id.in_(recipes), Recipe.id.notin_(sq))
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
for item in items:
|
|
cost = sum(i.quantity * prices[i.product_id] for i in item.items) / (item.recipe_yield * item.sku.fraction)
|
|
prices[item.sku.product_id] = cost
|
|
recipes.remove(item.sku.product_id)
|