"""Background service that e-mails provider waybill PDFs to customers.""" import asyncio from collections.abc import Callable, Mapping, Sequence from contextlib import AbstractAsyncContextManager from dataclasses import dataclass from datetime import datetime, timezone from typing import Protocol import structlog logger = structlog.get_logger(__name__) _EMAIL_SUBJECT_TEMPLATE = "Накладная по заказу {order_uuid}" _EMAIL_BODY_TEMPLATE = ( "Здравствуйте!\n\n" "По вашему заказу {order_uuid} сформирована " "транспортная накладная.\n" "PDF-файл накладной приложен к этому письму.\n" ) _EMAIL_BODY_URL_LINE_TEMPLATE = ( "Также накладная доступна по ссылке: {waybill_url}\n" ) _ATTACHMENT_FILENAME_TEMPLATE = "waybill_{order_uuid}.pdf" class WaybillPDFDownloaderProtocol(Protocol): async def download_waybill_pdf(self, order: "OrderRecord") -> bytes: ... class EmailSenderProtocol(Protocol): async def send_email( self, *, to: str, subject: str, body: str, attachment_bytes: bytes, attachment_filename: str, ) -> None: ... class OrderRecord(Protocol): order_uuid: str provider: str account_email: str provider_waybill_id: str | None provider_waybill_url: str | None class WaybillEmailSenderRepositoryProtocol(Protocol): def session(self) -> AbstractAsyncContextManager[object]: ... async def list_orders_pending_waybill_email( self, session: object, *, limit: int ) -> Sequence[OrderRecord]: ... async def record_waybill_email_sent( self, session: object, *, order_uuid: str, sent_at: datetime, ) -> object | None: ... @dataclass(frozen=True) class SendBatchSummary: processed: int succeeded: int failed: int class WaybillEmailSenderService: def __init__( self, *, order_repository: WaybillEmailSenderRepositoryProtocol, waybill_downloaders: Mapping[str, WaybillPDFDownloaderProtocol], email_sender: EmailSenderProtocol, batch_size: int, datetime_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: self._repository = order_repository self._waybill_downloaders = dict(waybill_downloaders) self._email_sender = email_sender self._batch_size = batch_size self._datetime_now = datetime_now async def poll_once(self) -> SendBatchSummary: async with self._repository.session() as session: orders = await self._repository.list_orders_pending_waybill_email( session, limit=self._batch_size ) succeeded = 0 failed = 0 for order in orders: try: await self._handle_order(session, order) succeeded += 1 except Exception: failed += 1 logger.exception( "waybill_email_order_failed", order_uuid=order.order_uuid, provider=order.provider, account_email=order.account_email, ) return SendBatchSummary( processed=len(orders), succeeded=succeeded, failed=failed, ) async def run_forever( self, *, interval_seconds: float, stop_event: asyncio.Event, ) -> None: while not stop_event.is_set(): try: summary = await self.poll_once() logger.debug( "waybill_email_tick", processed=summary.processed, succeeded=summary.succeeded, failed=summary.failed, ) except Exception: logger.exception("waybill_email_tick_failed") try: await asyncio.wait_for( stop_event.wait(), timeout=interval_seconds ) except asyncio.TimeoutError: continue async def _handle_order(self, session: object, order: OrderRecord) -> None: if order.provider_waybill_url is None and order.provider_waybill_id is None: return downloader = self._resolve_downloader(order.provider) pdf_bytes = await downloader.download_waybill_pdf(order) subject = _EMAIL_SUBJECT_TEMPLATE.format(order_uuid=order.order_uuid) body = _EMAIL_BODY_TEMPLATE.format(order_uuid=order.order_uuid) if order.provider_waybill_url is not None: body += _EMAIL_BODY_URL_LINE_TEMPLATE.format( waybill_url=order.provider_waybill_url, ) filename = _ATTACHMENT_FILENAME_TEMPLATE.format(order_uuid=order.order_uuid) await self._email_sender.send_email( to=order.account_email, subject=subject, body=body, attachment_bytes=pdf_bytes, attachment_filename=filename, ) sent_at = self._datetime_now() await self._repository.record_waybill_email_sent( session, order_uuid=order.order_uuid, sent_at=sent_at, ) logger.info( "waybill_email_sent", order_uuid=order.order_uuid, provider=order.provider, account_email=order.account_email, sent_at=sent_at, ) def _resolve_downloader(self, provider: str) -> WaybillPDFDownloaderProtocol: downloader = self._waybill_downloaders.get(provider) if downloader is None: raise RuntimeError(f"Waybill downloader is not configured for {provider}.") return downloader