140 lines
5.7 KiB
Python
140 lines
5.7 KiB
Python
from io import BytesIO
|
||
from typing import Optional
|
||
|
||
from fastapi import HTTPException
|
||
from reportlab.lib.units import mm
|
||
from reportlab.pdfgen.canvas import Canvas
|
||
from reportlab.platypus import Paragraph, SimpleDocTemplate, Frame, PageBreak
|
||
from reportlab_qrcode import QRCodeImage
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.orm import joinedload, selectinload
|
||
|
||
from constants import DOMAIN_NAME
|
||
from generators.base_pdf_card_generator.base_pdf_card_generator import BasePdfCardGenerator
|
||
from models import Deal, ShippingWarehouse, Pallet
|
||
from models.shipping import Box
|
||
|
||
|
||
class ShippingQRCodeGenerator(BasePdfCardGenerator):
|
||
async def _get_deal_by_id(self, deal_id: int) -> Optional[Deal]:
|
||
stmt = (
|
||
select(Deal)
|
||
.where(Deal.id == deal_id)
|
||
.options(
|
||
joinedload(Deal.shipping_warehouse),
|
||
selectinload(Deal.pallets),
|
||
)
|
||
)
|
||
deal = (await self._session.execute(stmt)).one_or_none()
|
||
return deal[0] if deal else None
|
||
|
||
async def generate_deal(self, deal_id: int) -> BytesIO:
|
||
deal = await self._get_deal_by_id(deal_id)
|
||
if not deal:
|
||
raise HTTPException(status_code=404, detail=f"Сделка с ID {deal_id}a не найдена")
|
||
|
||
buffer = BytesIO()
|
||
doc: SimpleDocTemplate = self._create_doc(buffer)
|
||
|
||
deal_link = f"{DOMAIN_NAME}/deals/{deal_id}"
|
||
shipping_warehouse = await self._session.get(ShippingWarehouse, deal.shipping_warehouse_id)
|
||
warehouse_name = shipping_warehouse.name if shipping_warehouse else ""
|
||
|
||
def on_first_page(canvas: Canvas, doc):
|
||
qr = QRCodeImage(deal_link, size=30 * mm)
|
||
qr.drawOn(canvas, 0, 30)
|
||
|
||
deal_id_paragraph = Paragraph(f"ID: {deal_id}", self.small_centered_style)
|
||
deal_name_paragraph = Paragraph(str(deal.name), self.small_centered_style)
|
||
|
||
frame = Frame(x1=28 * mm, y1=5 * mm, width=30 * mm, height=30 * mm)
|
||
frame.addFromList([deal_id_paragraph, deal_name_paragraph], canvas)
|
||
|
||
warehouse_paragraph = Paragraph(warehouse_name, self.small_centered_style)
|
||
frame = Frame(x1=0 * mm, y1=-7 * mm, width=58 * mm, height=20 * mm)
|
||
frame.addFromList([warehouse_paragraph], canvas)
|
||
|
||
empty_paragraph = Paragraph("", self.small_centered_style)
|
||
elements = [empty_paragraph]
|
||
doc.build(elements, on_first_page)
|
||
|
||
buffer.seek(0)
|
||
return buffer
|
||
|
||
async def generate_pallets(self, deal_id: int):
|
||
deal = await self._get_deal_by_id(deal_id)
|
||
if not deal:
|
||
raise HTTPException(status_code=404, detail=f"Сделка с ID {deal_id}a не найдена")
|
||
|
||
buffer = BytesIO()
|
||
doc: SimpleDocTemplate = self._create_doc(buffer)
|
||
|
||
shipping_warehouse = await self._session.get(ShippingWarehouse, deal.shipping_warehouse_id)
|
||
warehouse_name = shipping_warehouse.name if shipping_warehouse else ""
|
||
|
||
total_pallets = len(deal.pallets)
|
||
elements = []
|
||
|
||
for pallet_counter in range(total_pallets):
|
||
elements.append(Paragraph(f"ID: {deal_id}", self.medium_style))
|
||
elements.append(Paragraph(str(deal.name), self.medium_style))
|
||
elements.append(Paragraph(f"Паллет {pallet_counter + 1}/{total_pallets}", self.medium_style))
|
||
elements.append(Paragraph(warehouse_name, self.medium_style))
|
||
elements.append(PageBreak())
|
||
|
||
doc.build(elements)
|
||
|
||
buffer.seek(0)
|
||
return buffer
|
||
|
||
async def _get_boxes_on_pallets_count(self, deal_id):
|
||
stmt_boxes_on_pallets = (
|
||
select(
|
||
Pallet.id,
|
||
func.count(Box.id).label("box_count"),
|
||
)
|
||
.join(Box, isouter=True)
|
||
.where(Pallet.deal_id == deal_id)
|
||
.group_by(Pallet.id)
|
||
)
|
||
pallets = (await self._session.execute(stmt_boxes_on_pallets)).all()
|
||
return pallets
|
||
|
||
async def generate_boxes(self, deal_id: int) -> BytesIO:
|
||
deal = await self._get_deal_by_id(deal_id)
|
||
if not deal:
|
||
raise HTTPException(status_code=404, detail=f"Сделка с ID {deal_id}a не найдена")
|
||
|
||
shipping_warehouse = await self._session.get(ShippingWarehouse, deal.shipping_warehouse_id)
|
||
warehouse_name = shipping_warehouse.name if shipping_warehouse else ""
|
||
|
||
buffer = BytesIO()
|
||
doc: SimpleDocTemplate = self._create_doc(buffer)
|
||
|
||
elements = []
|
||
|
||
total_pallets = len(deal.pallets)
|
||
boxes_on_pallets = await self._get_boxes_on_pallets_count(deal_id)
|
||
boxes_without_pallets = len(deal.boxes)
|
||
|
||
for box_on_pallet in range(boxes_without_pallets):
|
||
elements.append(Paragraph(f"ID: {deal_id}", self.medium_style))
|
||
elements.append(Paragraph(str(deal.name), self.medium_style))
|
||
elements.append(Paragraph(f"Короб {box_on_pallet + 1}/{boxes_without_pallets}", self.medium_style))
|
||
elements.append(Paragraph(warehouse_name, self.medium_style))
|
||
elements.append(PageBreak())
|
||
|
||
for pallet_idx, [_, box_count] in enumerate(boxes_on_pallets):
|
||
for box_on_pallet in range(box_count):
|
||
elements.append(Paragraph(f"ID: {deal_id}", self.medium_style))
|
||
elements.append(Paragraph(str(deal.name), self.medium_style))
|
||
box_label = f"Паллет {pallet_idx + 1}/{total_pallets}, Короб {box_on_pallet + 1}/{box_count}"
|
||
elements.append(Paragraph(box_label, self.medium_style))
|
||
elements.append(Paragraph(warehouse_name, self.medium_style))
|
||
elements.append(PageBreak())
|
||
|
||
doc.build(elements)
|
||
|
||
buffer.seek(0)
|
||
return buffer
|