89 lines
3.6 KiB
Python
89 lines
3.6 KiB
Python
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 _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:
|
|
products[key] = {
|
|
"deal_products": [deal_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,
|
|
"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
|