112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
import uuid
|
|
|
|
from datetime import date, datetime, time, timedelta
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, Cookie, Depends, HTTPException, Security, status
|
|
from sqlalchemy import func, or_
|
|
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.inventory import Inventory
|
|
from ...models.kot import Kot
|
|
from ...models.product_version import ProductVersion
|
|
from ...models.sale_category import SaleCategory
|
|
from ...models.voucher import Voucher
|
|
from ...models.voucher_type import VoucherType
|
|
from ...printing.discount_report import print_discount_report
|
|
from ...schemas.discount_report import DiscountReport, DiscountReportItem
|
|
from ...schemas.user_token import UserToken
|
|
from . import report_finish_date, report_start_date
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# Dependency
|
|
def get_db() -> Session:
|
|
try:
|
|
db = SessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.get("", response_model=DiscountReport)
|
|
def discount_report(
|
|
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=["discount-report"]),
|
|
) -> DiscountReport:
|
|
today = (datetime.now() + timedelta(minutes=settings.TIMEZONE_OFFSET_MINUTES)).date()
|
|
if (today - start_date).days > 5 and "audit" not in user.permissions:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Accounts Audit",
|
|
)
|
|
return DiscountReport(
|
|
startDate=start_date,
|
|
finishDate=finish_date,
|
|
amounts=get_discount_report(start_date, finish_date, db),
|
|
)
|
|
|
|
|
|
def get_discount_report(s: date, f: date, db: Session) -> List[DiscountReportItem]:
|
|
start_date = datetime.combine(s, time()) + timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES)
|
|
finish_date = datetime.combine(f, time()) + timedelta(days=1, minutes=settings.NEW_DAY_OFFSET_MINUTES)
|
|
|
|
day = func.date_trunc("day", Voucher.date - timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES)).label("day")
|
|
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(ProductVersion.sale_category)
|
|
.filter(
|
|
Inventory.discount != 0,
|
|
Voucher.date >= start_date,
|
|
Voucher.date <= finish_date,
|
|
or_(
|
|
ProductVersion.valid_from == None, # noqa: E711
|
|
ProductVersion.valid_from <= day,
|
|
),
|
|
or_(
|
|
ProductVersion.valid_till == None, # noqa: E711
|
|
ProductVersion.valid_till >= day,
|
|
),
|
|
Voucher.voucher_type.in_([VoucherType.REGULAR_BILL, VoucherType.KOT]),
|
|
)
|
|
.group_by(SaleCategory.name)
|
|
.order_by(SaleCategory.name)
|
|
.all()
|
|
)
|
|
|
|
return [DiscountReportItem(name=pg, amount=amt) for pg, amt in list_]
|
|
|
|
|
|
@router.get("/print", response_model=bool)
|
|
def print_report(
|
|
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=["discount-report"]),
|
|
) -> bool:
|
|
today = (datetime.now() + timedelta(minutes=settings.TIMEZONE_OFFSET_MINUTES)).date()
|
|
if (today - start_date).days > 5 and "audit" not in user.permissions:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Accounts Audit",
|
|
)
|
|
report = DiscountReport(
|
|
startDate=start_date,
|
|
finishDate=finish_date,
|
|
amounts=get_discount_report(start_date, finish_date, db),
|
|
)
|
|
print_discount_report(report, device_id, db)
|
|
return True
|