brewman/brewman/brewman/routers/reports/product_ledger.py

import uuid

from datetime import date
from decimal import Decimal
from typing import Annotated

from fastapi import APIRouter, Depends, Security
from sqlalchemy.orm import Session, contains_eager
from sqlalchemy.sql.expression import func, select

from ...core.security import get_current_active_user as get_user
from ...db.session import SessionDep
from ...models.batch import Batch
from ...models.cost_centre import CostCentre
from ...models.inventory import Inventory
from ...models.journal import Journal
from ...models.product import Product
from ...models.sku_version import SkuVersion
from ...models.stock_keeping_unit import StockKeepingUnit
from ...models.voucher import Voucher
from ...models.voucher_type import VoucherType
from ...routers.reports import report_finish_date, report_start_date
from ...schemas import product_ledger as schemas
from ...schemas.product import ProductLink
from ...schemas.user import UserToken
from .. import _sv_onclause


router = APIRouter()


@router.get("", response_model=schemas.ProductLedger)
def show_blank(
    start_date: Annotated[date, Depends(report_start_date)],
    finish_date: Annotated[date, Depends(report_finish_date)],
    user: Annotated[UserToken, Security(get_user, scopes=["product-ledger"])],
) -> schemas.ProductLedger:
    return schemas.ProductLedger(
        start_date=start_date,
        finish_date=finish_date,
        product=None,
        body=[],
    )


@router.get("/{id_}", response_model=schemas.ProductLedger)
def show_data(
    id_: uuid.UUID,
    start_date: Annotated[date, Depends(report_start_date)],
    finish_date: Annotated[date, Depends(report_finish_date)],
    user: Annotated[UserToken, Security(get_user, scopes=["product-ledger"])],
    db: SessionDep,
) -> schemas.ProductLedger:
    product = db.execute(select(Product).where(Product.id == id_)).scalar_one()
    body = build_report(
        product.id,
        start_date,
        finish_date,
        db,
    )
    return schemas.ProductLedger(
        start_date=start_date,
        finish_date=finish_date,
        product=ProductLink(id_=product.id, name=product.versions[-1].name),
        body=body,
    )


def build_report(
    product_id: uuid.UUID, start_date: date, finish_date: date, db: Session
) -> list[schemas.ProductLedgerItem]:
    running_total_q, running_total_a, opening = opening_balance(product_id, start_date, db)
    body = opening

    query: list[tuple[Voucher, Inventory, Journal, SkuVersion]] = (
        db.execute(  # type: ignore[assignment]
            select(Voucher, Inventory, Journal, SkuVersion)
            .select_from(Voucher)
            .join(Voucher.journals)
            .join(Journal.account)
            .join(Journal.cost_centre)
            .join(Voucher.inventories)
            .join(Inventory.batch)
            .join(Batch.sku)
            .join(SkuVersion, onclause=_sv_onclause(Voucher.date_))
            .join(StockKeepingUnit.product)
            .options(
                contains_eager(Voucher.journals).contains_eager(Journal.account),
                contains_eager(Voucher.journals).contains_eager(Journal.cost_centre),
            )
            .where(
                StockKeepingUnit.product_id == product_id,
                Journal.cost_centre_id != CostCentre.cost_centre_purchase(),
                Voucher.date_ >= start_date,
                Voucher.date_ <= finish_date,
            )
            .order_by(Voucher.date_, Voucher.last_edit_date)
        )
        .unique()
        .all()
    )

    for voucher, inventory, journal, stockKeepingUnit in query:
        journal_debit = journal.debit * -1
        name = (
            journal.cost_centre.name
            if voucher.voucher_type in [VoucherType.ISSUE, VoucherType.CLOSING_STOCK]
            else journal.account.name
        )
        debit_q = inventory.quantity if journal_debit == 1 else None
        debit_a = inventory.amount if journal_debit == 1 else None
        debit_u = stockKeepingUnit.units if journal_debit == 1 else None
        credit_q = inventory.quantity if journal_debit != 1 else None
        credit_a = inventory.amount if journal_debit != 1 else None
        credit_u = stockKeepingUnit.units if journal_debit != 1 else None

        running_total_q += inventory.quantity * journal_debit
        running_total_a += inventory.amount * journal_debit

        body.append(
            schemas.ProductLedgerItem(
                id_=voucher.id,
                date_=voucher.date_,
                name=name,
                url=[
                    "/",
                    voucher.voucher_type.name.replace("_", "-").lower(),
                    str(voucher.id)
                    if voucher.voucher_type != VoucherType.CLOSING_STOCK
                    else voucher.date_.strftime("%d-%b-%Y"),
                ],
                type_=voucher.voucher_type.name.replace("_", " ").title(),
                narration=voucher.narration,
                posted=voucher.posted or voucher.voucher_type in [VoucherType.ISSUE, VoucherType.CLOSING_STOCK],
                debit_quantity=debit_q,
                debit_amount=debit_a,
                debit_unit=debit_u,
                credit_quantity=credit_q,
                credit_amount=credit_a,
                credit_unit=credit_u,
                running_quantity=running_total_q,
                running_amount=running_total_a,
            )
        )

    return body


def opening_balance(
    product_id: uuid.UUID, start_date: date, db: Session
) -> tuple[Decimal, Decimal, list[schemas.ProductLedgerItem]]:
    row = db.execute(
        select(
            SkuVersion.units,
            func.sum(Inventory.quantity * Journal.debit).label("quantity"),
            func.sum(Inventory.amount * Journal.debit).label("amount"),
        )
        .select_from(Voucher)
        .join(Voucher.journals)
        .join(Voucher.inventories)
        .join(Inventory.batch)
        .join(Batch.sku)
        .join(SkuVersion, onclause=_sv_onclause(Voucher.date_))
        .where(
            Voucher.id == Inventory.voucher_id,
            Voucher.id == Journal.voucher_id,
            StockKeepingUnit.product_id == product_id,
            Journal.cost_centre_id == CostCentre.cost_centre_purchase(),
            Voucher.date_ < start_date,
        )
        .group_by(SkuVersion.units)
    ).all()

    quantity: Decimal = sum((r.quantity for r in row), start=Decimal(0))
    amount: Decimal = sum((r.amount for r in row), start=Decimal(0))

    return (
        quantity,
        amount,
        [
            schemas.ProductLedgerItem(
                id_=None,
                date_=start_date,
                name="Opening Balance",
                url=[],
                type_="Opening Balance",
                narration="",
                posted=True,
                debit_quantity=q,
                debit_amount=a,
                debit_unit=u,
                credit_quantity=Decimal(0),
                credit_amount=Decimal(0),
                credit_unit="",
                running_quantity=quantity,
                running_amount=amount,
            )
            for u, q, a in row
        ],
    )