barker/barker/barker/routers/reports/discount_report.py

from datetime import date, datetime, timedelta

from fastapi import APIRouter, Depends, HTTPException, Security, status
from sqlalchemy import func
from sqlalchemy.orm import Session

from ...core.config import settings
from ...core.security import get_current_active_user as get_user
from ...db.session import SessionLocal
from ...models import (
    Inventory,
    Kot,
    Product,
    Reprint,
    SaleCategory,
    Settlement,
    SettleOption,
    Voucher,
    VoucherType,
)
from ...schemas.auth import UserToken


router = APIRouter()


# Dependency
def get_db() -> Session:
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@router.get("")
def discount_report_view(
    s: str = None,
    f: str = None,
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["discount-report"]),
):
    start_date = date.today() if s is None else datetime.strptime(s, "%d-%b-%Y").date()
    finish_date = date.today() if f is None else datetime.strptime(f, "%d-%b-%Y").date()
    if (
        date.today() - start_date
    ).days > 5 and "accounts-audit" not in user.permissions:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Accounts Audit",
        )

    return {
        "startDate": start_date.strftime("%d-%b-%Y"),
        "finishDate": finish_date.strftime("%d-%b-%Y"),
        "amounts": discount_report(start_date, finish_date, db),
    }


def discount_report(s: date, f: date, db: Session):
    start_date = s + timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES)
    finish_date = f + timedelta(days=1, minutes=settings.NEW_DAY_OFFSET_MINUTES)

    amount = func.sum(
        Inventory.quantity * Inventory.effective_price * Inventory.discount
    ).label("Amount")
    list_ = (
        db.query(SaleCategory.name, amount)
        .join(Voucher.kots)
        .join(Kot.inventories)
        .join(Inventory.product)
        .join(Product.sale_category)
        .filter(
            Inventory.discount != 0,
            Voucher.date >= start_date,
            Voucher.date <= finish_date,
            Voucher.voucher_type.in_(
                [VoucherType.REGULAR_BILL.value, VoucherType.KOT.value]
            ),
        )
        .group_by(SaleCategory.name)
        .order_by(SaleCategory.name)
        .all()
    )

    return [{"name": pg, "amount": amt} for pg, amt in list_]