"""Background service that e-mails CDEK waybill PDFs to customers.""" import asyncio from collections.abc import Callable, 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} сформирована транспортная накладная CDEK.\n" "PDF-файл накладной приложен к этому письму.\n" "Также накладная доступна по ссылке: {waybill_url}\n" ) _ATTACHMENT_FILENAME_TEMPLATE = "waybill_{order_uuid}.pdf" class WaybillPDFDownloaderProtocol(Protocol): async def download_waybill_pdf(self, url: str) -> 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 account_email: str cdek_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_downloader: WaybillPDFDownloaderProtocol, email_sender: EmailSenderProtocol, batch_size: int, datetime_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: self._repository = order_repository self._waybill_downloader = waybill_downloader 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, 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.info( "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: waybill_url = order.cdek_waybill_url if waybill_url is None: return pdf_bytes = await self._waybill_downloader.download_waybill_pdf(waybill_url) subject = _EMAIL_SUBJECT_TEMPLATE.format(order_uuid=order.order_uuid) body = _EMAIL_BODY_TEMPLATE.format( order_uuid=order.order_uuid, waybill_url=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, account_email=order.account_email, sent_at=sent_at, )