g2s-aggregator/app/services/waybill_email_sender.py

"""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 "
                    f"processed={summary.processed} "
                    f"succeeded={summary.succeeded} "
                    f"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,
        )