import uuid from datetime import date, datetime, time, timedelta from decimal import Decimal from fastapi import APIRouter, Cookie, Depends, Security from sqlalchemy import distinct, select 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 SessionFuture 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 from ...schemas.user import UserLink from ...schemas.user_token import UserToken from . import check_audit_permission, report_finish_date, report_start_date router = APIRouter() @router.get("/active", response_model=list[UserLink]) def active_cashiers( start_date: date = Depends(report_start_date), finish_date: date = Depends(report_finish_date), user: UserToken = Security(get_user, scopes=["cashier-report"]), ) -> list[UserLink]: check_audit_permission(start_date, user.permissions) with SessionFuture() as db: 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.execute( select(User) .where( User.id.in_( select(distinct(Voucher.user_id)).where( Voucher.date >= start_datetime, Voucher.date <= finish_datetime, ) ) ) .order_by(User.name) ) .scalars() .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), user: UserToken = Security(get_user, scopes=["cashier-report"]), ) -> CashierReport: check_audit_permission(start_date, user.permissions) user_link = UserLink(id=user.id_, name=user.name) with SessionFuture() as db: return get_id(id_, start_date, finish_date, user_link, db) def get_id(id_: uuid.UUID, start_date: date, finish_date: date, user: UserLink, db: Session) -> CashierReport: cashier: str = db.execute(select(User.name).where(User.id == id_)).scalar_one() 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.execute( select(Voucher) .options( joinedload(Voucher.settlements, innerjoin=True).joinedload(Settlement.settle_option, innerjoin=True) ) .where( Voucher.date >= start_datetime, Voucher.date <= finish_datetime, Voucher.user_id == id_, ) .order_by(Voucher.voucher_type) ) .unique() .scalars() .all() ) info: dict[str, list[InfoItem]] = {} amounts = {} for item in vouchers: for so in (so for so in item.settlements if so.settle_option.reporting_level >= ReportingLevel.Aggregate): if so.settle_option.name not in amounts: if so.settle_option.reporting_level == ReportingLevel.Detailed: info[so.settle_option.name] = [] amounts[so.settle_option.name] = Decimal(0) amounts[so.settle_option.name] += so.amount if so.settle_option.reporting_level == ReportingLevel.Detailed: info[so.settle_option.name].append( InfoItem( date=item.date, billId=", ".join(f"{x.regime.prefix}-{x.bill_number}" for x in item.bills), 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 or "" 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), user: UserToken = Security(get_user, scopes=["cashier-report"]), ) -> bool: check_audit_permission(start_date, user.permissions) with SessionFuture() as db: 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), 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_), )