barker/barker/barker/routers/reports/cashier_report.py

import uuid

from datetime import date, datetime, time, timedelta
from typing import Dict, List

from fastapi import APIRouter, Cookie, Depends, Security
from sqlalchemy import distinct
from sqlalchemy.orm import Session, joinedload

from ...core.config import settings
from ...core.security import get_current_active_user as get_user
from ...db.session import SessionLocal
from ...models.reporting_level import ReportingLevel
from ...models.settlement import Settlement
from ...models.user import User
from ...models.voucher import Voucher
from ...printing.cashier_report import print_cashier_report
from ...schemas.cashier_report import (
    CashierReport,
    InfoItem,
    NameAmount,
    SettleOptionSchema,
)
from ...schemas.user import UserLink
from ...schemas.user_token import UserToken
from . import check_audit_permission, report_finish_date, report_start_date


router = APIRouter()


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


@router.get("/active", response_model=List[UserLink])
def active_cashiers(
    start_date: date = Depends(report_start_date),
    finish_date: date = Depends(report_finish_date),
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["cashier-report"]),
) -> List[UserLink]:
    check_audit_permission(start_date, user.permissions)
    return get_active_cashiers(start_date, finish_date, db)


def get_active_cashiers(start_date: date, finish_date: date, db: Session) -> List[UserLink]:
    start_datetime = datetime.combine(start_date, time()) + timedelta(
        minutes=settings.TIMEZONE_OFFSET_MINUTES - settings.NEW_DAY_OFFSET_MINUTES
    )
    finish_datetime = datetime.combine(finish_date, time()) + timedelta(
        days=1, minutes=settings.TIMEZONE_OFFSET_MINUTES - settings.NEW_DAY_OFFSET_MINUTES
    )
    users = (
        db.query(User)
        .filter(
            User.id.in_(
                db.query(distinct(Voucher.user_id)).filter(
                    Voucher.date >= start_datetime,
                    Voucher.date <= finish_datetime,
                )
            )
        )
        .order_by(User.name)
        .all()
    )
    return [UserLink(id=u.id, name=u.name) for u in users]


@router.get("/{id_}", response_model=CashierReport)
def show_id(
    id_: uuid.UUID,
    start_date: date = Depends(report_start_date),
    finish_date: date = Depends(report_finish_date),
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["cashier-report"]),
) -> CashierReport:
    check_audit_permission(start_date, user.permissions)
    user = UserLink(id=user.id_, name=user.name)
    return get_id(id_, start_date, finish_date, user, db)


def get_id(id_: uuid.UUID, start_date: date, finish_date: date, user: UserLink, db: Session) -> CashierReport:
    cashier: str = db.query(User.name).filter(User.id == id_).scalar()
    start_datetime = datetime.combine(start_date, time()) + timedelta(
        minutes=settings.TIMEZONE_OFFSET_MINUTES - settings.NEW_DAY_OFFSET_MINUTES
    )
    finish_datetime = datetime.combine(finish_date, time()) + timedelta(
        days=1, minutes=settings.TIMEZONE_OFFSET_MINUTES - settings.NEW_DAY_OFFSET_MINUTES
    )
    vouchers = (
        db.query(Voucher)
        .options(joinedload(Voucher.settlements, innerjoin=True).joinedload(Settlement.settle_option, innerjoin=True))
        .filter(
            Voucher.date >= start_datetime,
            Voucher.date <= finish_datetime,
            Voucher.user_id == id_,
        )
        .order_by(Voucher.voucher_type)
        .order_by(Voucher.bill_id)
        .all()
    )

    info: Dict[SettleOptionSchema, List[InfoItem]] = {}
    amounts = {}
    for item in vouchers:
        for so in (
            so for so in item.settlements if so.settle_option.reporting_level >= ReportingLevel.Aggregate
        ):  # Only Roundoff and Amount is skipped
            if so.settle_option.name not in info:
                info[so.settle_option.name] = []
                amounts[so.settle_option.name] = 0
            amounts[so.settle_option.name] += so.amount
            if so.settle_option.reporting_level == ReportingLevel.Detailed:  # Only additionally Cash is skipped
                info[so.settle_option.name].append(
                    InfoItem(
                        date=item.date,
                        billId=item.full_bill_id,
                        customer=item.customer.name if item.customer is not None else "",
                        amount=so.amount,
                    )
                )
    return CashierReport(
        startDate=start_date,
        finishDate=finish_date,
        cashier=UserLink(id=id_, name=cashier),
        cashiers=", ".join(x.name for x in get_active_cashiers(start_date, finish_date, db)),
        amounts=[NameAmount(name=key, amount=value) for key, value in amounts.items()],
        info=info,
        user=user,
    )


@router.get("/print/{id_}", response_model=bool)
def print_report(
    id_: uuid.UUID,
    start_date: date = Depends(report_start_date),
    finish_date: date = Depends(report_finish_date),
    device_id: uuid.UUID = Cookie(None),
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["cashier-report"]),
) -> bool:
    check_audit_permission(start_date, user.permissions)
    report = get_id(id_, start_date, finish_date, UserLink(id=user.id_, name=user.name), db)
    print_cashier_report(report, device_id, db)
    return True


@router.get("", response_model=CashierReport)
def show_blank(
    start_date: date = Depends(report_start_date),
    finish_date: date = Depends(report_finish_date),
    db: Session = Depends(get_db),
    user: UserToken = Security(get_user, scopes=["cashier-report"]),
) -> CashierReport:
    check_audit_permission(start_date, user.permissions)
    return CashierReport(
        startDate=start_date,
        finishDate=finish_date,
        cashier=UserLink(id=None),
        cashiers="",
        amounts=[],
        info=[],
        user=UserLink(id=user.id_),
    )