Merge remote-tracking branch 'origin/master'

generators/price_list_pdf_generator/generator.py

@@ -0,0 +1,88 @@
+from collections import defaultdict
+from io import BytesIO
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import joinedload
+from weasyprint import HTML, CSS
+
+from constants import ENV, APP_PATH
+from enums.service import ServiceType
+from models import Service, ServiceCategory
+
+
+class PriceListPdfGenerator:
+    def __init__(self, session: AsyncSession):
+        self._session = session
+
+    async def _get_services_data(self):
+        # Получаем услуги из базы данных, отсортированные по рангу
+        services = (await (
+            self._session.scalars(
+                select(Service)
+                .options(joinedload(Service.category))
+                .filter(Service.is_deleted == False)
+                .order_by(Service.rank)
+            )
+        )).all()
+
+        # Группируем услуги по типу сервиса и категории
+        intermediate_result = defaultdict(lambda: defaultdict(list))
+
+        for service in services:
+            intermediate_result[service.service_type][service.category_id].append(service)
+
+        # Формируем окончательный результат с сортировкой категорий
+        final_result = defaultdict(dict)
+
+        for service_type, categories_dict in intermediate_result.items():
+            # Извлекаем уникальные категории
+            categories = {service.category for services in categories_dict.values() for service in services}
+
+            # Определяем функцию сортировки категорий по рангу
+            def category_sort_key(category):
+                if service_type == ServiceType.DEAL_SERVICE:
+                    return category.deal_service_rank
+                else:
+                    return category.product_service_rank
+
+            # Сортируем категории по определенному рангу
+            sorted_categories = sorted(categories, key=category_sort_key)
+
+            # Строим словарь категорий в отсортированном порядке
+            sorted_categories_dict = {}
+            for category in sorted_categories:
+                sorted_categories_dict[category.id] = categories_dict[category.id]
+
+            final_result[service_type] = sorted_categories_dict
+
+        final_final_result = {}
+        for service_type in [ServiceType.DEAL_SERVICE, ServiceType.PRODUCT_SERVICE]:
+            final_final_result[service_type] = final_result[service_type]
+        return dict(final_final_result)
+
+    async def _create_price_list_html(self):
+        categories = await self._session.scalars(select(ServiceCategory))
+        categories_dict = {category.id: category.name for category in categories}
+
+        services_data = await self._get_services_data()
+
+        service_type_dict = {
+            ServiceType.DEAL_SERVICE: "Общие услуги",
+            ServiceType.PRODUCT_SERVICE: "Услуги фулфилмента",
+        }
+
+        template = ENV.get_template("price-list.html")
+
+        result = template.render({
+            "services_data": services_data,
+            "categories_dict": categories_dict,
+            "service_type_dict": service_type_dict,
+        })
+        return result
+
+    async def create_price_list_pdf(self) -> BytesIO:
+        doc = await self._create_price_list_html()
+        pdf_file = BytesIO()
+        HTML(string=doc).write_pdf(pdf_file, stylesheets=[CSS(APP_PATH + '/static/css/price-list.css')])
+        return pdf_file