feat: executors and grouping by article in deal document

generators/deal_pdf_generator/deal_data.py

@@ -0,0 +1,20 @@
+from typing import TypedDict, List, Dict, Tuple, Optional
+
+from models import DealProduct, Deal, DealStatusHistory
+
+
+class DocumentDealProductData(TypedDict):
+    deal_products: List[DealProduct]
+    total_one_product: int
+    quantity: int
+    additional_info: Optional[str]
+
+
+class DocumentDealData(TypedDict):
+    deal: Deal
+    general_services_total: int
+    products: Dict[str, DocumentDealProductData]
+    current_status_str: str
+    last_status: DealStatusHistory
+    product_images: Tuple[str]
+

generators/deal_pdf_generator/generator.py

@@ -0,0 +1,102 @@
+from io import BytesIO
+from typing import List, Dict, Optional
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload, joinedload
+from weasyprint import HTML, CSS
+
+from constants import DEAL_STATUS_STR, ENV, APP_PATH
+from generators.deal_pdf_generator.deal_data import DocumentDealProductData
+from models import Deal, DealProduct, DealService as DealServiceModel, Product
+from utils.images_fetcher import fetch_images
+
+
+class DealPdfGenerator:
+    def __init__(self, session: AsyncSession):
+        self._session = session
+
+    @staticmethod
+    async def _get_product_services_totals(deal: Deal) -> List[Dict[str, int]]:
+        totals: List[Dict[str, int]] = []
+
+        for product in deal.products:
+            total_one_product = sum((service.price for service in product.services))
+            total = total_one_product * product.quantity
+            totals.append({"total_one_product": total_one_product, "total": total})
+
+        return totals
+
+    @staticmethod
+    async def _group_deal_products_by_products(deal_products: List[DealProduct]) -> Dict[str, DocumentDealProductData]:
+        products: Dict[str, DocumentDealProductData] = {}
+        additional_info: Optional[str]
+
+        for deal_product in deal_products:
+            # Для группировки по артикулу и услугам
+            key = f"{deal_product.product.article} - " + ",".join(
+                str(service.service_id) for service in deal_product.services
+            )
+
+            if key not in products:
+                total_one_product = sum(service.price for service in deal_product.services)
+                products[key] = {
+                    "deal_products": [deal_product],
+                    "total_one_product": total_one_product,
+                    "quantity": deal_product.quantity,
+                    "additional_info": deal_product.product.additional_info,
+                }
+            else:
+                products[key]["deal_products"].append(deal_product)
+                products[key]["quantity"] += deal_product.quantity
+                if not products[key]["additional_info"]:
+                    products[key]["additional_info"] = deal_product.product.additional_info
+
+        return products
+
+    async def _create_detailed_deal_document_html(self, deal_id: int):
+        deal: Deal | None = await self._session.scalar(
+            select(Deal)
+            .where(Deal.id == deal_id)
+            .options(
+                selectinload(Deal.products).selectinload(DealProduct.services),
+                selectinload(Deal.products).selectinload(DealProduct.product).selectinload(Product.barcodes),
+                selectinload(Deal.services).selectinload(DealServiceModel.service),
+                selectinload(Deal.status_history),
+                joinedload(Deal.client),
+                joinedload(Deal.shipping_warehouse),
+            )
+        )
+
+        if not deal:
+            return ""
+
+        products = await self._group_deal_products_by_products(deal.products)
+
+        product_urls: List[Optional[str]] = []
+        for product in products.values():
+            if len(product["deal_products"][0].product.images) > 0:
+                product_urls.append(product["deal_products"][0].product.images[0].image_url)
+            else:
+                product_urls.append(None)
+        product_images = await fetch_images(product_urls)
+
+        document_deal_data = {
+            "deal": deal,
+            "general_services_total": sum((service.price * service.quantity for service in deal.services)),
+            "products": products,
+            "current_status_str": DEAL_STATUS_STR[deal.current_status],
+            "last_status": max(deal.status_history, key=lambda status: status.changed_at),
+            "product_images": product_images,
+        }
+
+        template = ENV.get_template("deal/deal.html")
+
+        result = template.render({"data": document_deal_data, "sign_place_text": "_" * 22})
+        return result
+
+    async def create_detailed_deal_document_pdf(self, deal_id) -> BytesIO:
+        doc = await self._create_detailed_deal_document_html(deal_id)
+        pdf_file = BytesIO()
+        HTML(string=doc).write_pdf(pdf_file, stylesheets=[CSS(APP_PATH + '/static/css/deal.css')])
+        return pdf_file