import asyncio import locale import re import uuid from datetime import timedelta from decimal import Decimal from typing import Dict, List, Tuple from arq import ArqRedis, create_pool from barker.core.config import settings from sqlalchemy import and_, or_, select from sqlalchemy.orm import Session from ..core.arq import settings as redis_settings from ..models.bill import Bill from ..models.db_setting import DbSetting from ..models.inventory import Inventory from ..models.printer import Printer from ..models.product_version import ProductVersion from ..models.section_printer import SectionPrinter from ..models.voucher import Voucher from ..models.voucher_type import VoucherType from . import currency_format, format_no_decimals def print_bill(voucher_id: uuid.UUID, db: Session): locale.setlocale(locale.LC_MONETARY, "en_IN") voucher: Voucher = db.execute(select(Voucher).where(Voucher.id == voucher_id)).scalar_one() printer = db.execute( select(Printer) .join(SectionPrinter.printer) .where(SectionPrinter.section_id == voucher.food_table.section_id) .where(SectionPrinter.sale_category_id == None) # noqa: E711 ).scalar_one() regimes = get_regimes(get_inventories(voucher), voucher.voucher_type, db) bills: List[str] = [] for regime_id, inventories in regimes: taxes_dict: dict[str, Decimal] = {} for i in inventories: get_tax_item(i.tax.id, i.tax.name, i.tax_amount, taxes_dict) taxes_list = [(t, a) for t, a in taxes_dict.items() if a != 0] bill_text = design_bill(voucher, regime_id, inventories, taxes_list, db) bills.append(bill_text) loop = asyncio.new_event_loop() redis: ArqRedis = loop.run_until_complete(create_pool(redis_settings)) if len(bills) > 1: total = design_total(voucher, regimes, db) loop.run_until_complete( redis.enqueue_job( "sent_to_printer", total, printer.address, printer.cut_code, _queue_name=f"barker:print:{printer.name}" ) ) for bill in bills: loop.run_until_complete( redis.enqueue_job( "sent_to_printer", bill, printer.address, printer.cut_code, _queue_name=f"barker:print:{printer.name}" ) ) loop.close() def get_inventories(voucher: Voucher) -> List[Inventory]: items_dict: dict[tuple[uuid.UUID, bool, Decimal, frozenset[uuid.UUID]], Inventory] = {} for i in [i for k in voucher.kots for i in k.inventories]: key = ( i.product_id, i.is_happy_hour, i.discount, frozenset(m.modifier_id for m in i.modifiers if m.modifier.show_in_bill), ) if key in items_dict: items_dict[key].quantity += i.quantity else: items_dict[key] = Inventory( product_id=i.product_id, quantity=i.quantity, price=i.price, discount=i.discount, is_hh=i.is_happy_hour, tax_rate=i.tax_rate, product=i.product, tax=i.tax, ) return list([item for item in items_dict.values() if item.quantity != 0]) def get_regimes( inventories: List[Inventory], voucher_type: VoucherType, db: Session ) -> List[Tuple[int, List[Inventory]]]: if voucher_type != VoucherType.REGULAR_BILL: return [ (int(voucher_type), inventories), ] items_dict: Dict[int, List[Inventory]] = {} for inv in inventories: key = inv.tax.regime_id if key not in items_dict: items_dict[key] = [] items_dict[key].append(inv) return list(items_dict.items()) def get_tax_item(id_: uuid.UUID, tax_name: str, amount: Decimal, taxes: Dict[str, Decimal]) -> None: items = tax_name.split(";") if len(items) == 1: if tax_name not in taxes: taxes[tax_name] = Decimal(0) taxes[tax_name] += round(amount, 2) else: for i, item in enumerate(it.strip() for it in items): match = re.match(r"(^.*)\s+\((.*?)/(.*?)\)[^(]*$", item) if not match or len(match.groups()) != 3: raise Exception("Error in tax as it has multiple items, but the format is wrong.") sub_amount: Decimal = round(amount * Decimal(match.group(2)) / Decimal(match.group(3)), 2) if not match.group(1) in taxes: taxes[match.group(1)] = Decimal(0) taxes[match.group(1)] += round(sub_amount, 2) def design_bill( voucher: Voucher, regime_id: int, inventories: List[Inventory], taxes: List[Tuple[str, Decimal]], db: Session, ) -> str: # Header s = "\n\r" + db.execute(select(DbSetting).where(DbSetting.name == "Header")).scalar_one().data["Text"] # Bill Regime bill_object: Bill = db.execute( select(Bill) .join(Bill.regime) .where(Bill.voucher_id == voucher.id, Bill.regime_id == regime_id, Bill.is_valid == True) # noqa: E712 ).scalar_one() s += "\n\r" + bill_object.regime.header # Products voucher_date = voucher.date + timedelta(minutes=settings.TIMEZONE_OFFSET_MINUTES) product_date = (voucher_date - timedelta(minutes=settings.NEW_DAY_OFFSET_MINUTES)).date() bill_number = f"{bill_object.regime.prefix}-{bill_object.bill_number}" s += "\n\r" + f"Bill No: {bill_number:>12} {voucher_date:%d-%b-%Y %H:%M}" s += "\n\r" + "Table No.: " + voucher.food_table.name s += "\n\r" + "-" * 42 s += "\n\r" + "Qty. Particulars Price Amount" s += "\n\r" + "-" * 42 for inv in inventories: product: ProductVersion = db.execute( select(ProductVersion).where( and_( ProductVersion.product_id == inv.product_id, or_( ProductVersion.valid_from == None, # noqa: E711 ProductVersion.valid_from <= product_date, ), or_( ProductVersion.valid_till == None, # noqa: E711 ProductVersion.valid_till >= product_date, ), ) ) ).scalar_one() name = "H H " + product.full_name if inv.is_happy_hour else product.full_name s += ( f"\n\r" f"{inv.quantity: >5.2f} {name:<22.22} {format_no_decimals(inv.price): >6}" f" {format_no_decimals(inv.price * inv.quantity): >6}" ) for m in [m for m in inv.modifiers if m.modifier.show_in_bill]: s += f"\n\r -- {m.modifier.name: <38.38}" s += "\n\r" + "------------------------------------------" # Totals amount = sum(round(i.quantity * i.price, 2) for i in inventories) s += f"\n\r{'Subtotal :': >32} {currency_format(amount): >9}" amount = sum(round(i.quantity * i.price, 2) for i in inventories if i.is_happy_hour) if amount != 0: s += f"\n\r{'Happy Hour Discount :': >32} {currency_format(amount): >9}" amount = sum(round(i.quantity * i.effective_price * i.discount, 2) for i in inventories) if amount != 0: s += f"\n\r{'Discount :': >32} {currency_format(amount): >9}" for t in taxes: s += f"\n\r{t[0]: >30} : {currency_format(t[1]): >9}" s += f"\n\r{'Total Amount :': >32} {currency_format(round(sum([i.amount for i in inventories]))): >9}" s += "\n\r" + "-" * 42 if voucher.narration: s += f"\n\r{voucher.narration: ^42}" s += "\n\r" + "-" * 42 if voucher.customer and voucher.customer.print_in_bill: s += f"\n\r{voucher.customer.name or ''}\n\r{voucher.customer.phone or ''}\n\r{voucher.customer.address or ''}" s += "\n\r" + "-" * 42 s += "\n\r" + "Cashier : " + voucher.user.name s += "\n\r" + db.execute(select(DbSetting).where(DbSetting.name == "Footer")).scalar_one().data["Text"] return s def design_total( voucher: Voucher, regimes: List[Tuple[int, List[Inventory]]], db: Session, ) -> str: # Header s = "\n\r" + db.execute(select(DbSetting).where(DbSetting.name == "Header")).scalar_one().data["Text"] voucher_date = voucher.date + timedelta(minutes=settings.TIMEZONE_OFFSET_MINUTES) s += "\n\r" + f"{voucher_date:%d-%b-%Y %H:%M} Table No.: {voucher.food_table.name} " s += "\n\r" + "-" * 42 s += "\n\r" + " Total Amount " s += "\n\r" + "------------------------------------------" # Bill Regime for regime_id, inventories in regimes: bill_object: Bill = db.execute( select(Bill) .join(Bill.regime) .where(Bill.voucher_id == voucher.id, Bill.regime_id == regime_id, Bill.is_valid == True) # noqa: E712 ).scalar_one() amount = currency_format(round(sum([i.amount for i in inventories]))) bill_number = f"{bill_object.regime.prefix}-{bill_object.bill_number}" s += "\n\r" + f"Bill No: {bill_number:>12} Amount: {amount: >9}" s += "\n\r" + "------------------------------------------" s += f"\n\r{'Total Amount :': >32} {currency_format(round(voucher.amount)): >9}" s += "\n\r" + "-" * 42 s += "\n\r" + db.execute(select(DbSetting).where(DbSetting.name == "Footer")).scalar_one().data["Text"] return s