feat: pallets and boxes for deals

generators/shipping_qr_code_generator/generator.py

@@ -0,0 +1,139 @@
+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